root/mpich2/trunk/src/mpid/ch3/channels/nemesis/nemesis/netmod/tcp/tcp_send.c @ 4865

Revision 4865, 18.4 KB (checked in by buntinas, 5 months ago)

stomped some warnings
Line 
1/* -*- Mode: C; c-basic-offset:4 ; -*- */
2/*
3 *  (C) 2006 by Argonne National Laboratory.
4 *      See COPYRIGHT in top-level directory.
5 */
6
7#include "tcp_impl.h"
8
9#define NUM_PREALLOC_SENDQ 10
10#define MAX_SEND_IOV 10
11
12#define SENDQ_EMPTY(q) GENERIC_Q_EMPTY (q)
13#define SENDQ_HEAD(q) GENERIC_Q_HEAD (q)
14#define SENDQ_ENQUEUE(qp, ep) GENERIC_Q_ENQUEUE (qp, ep, dev.next)
15#define SENDQ_DEQUEUE(qp, ep) GENERIC_Q_DEQUEUE (qp, ep, dev.next)
16
17
18typedef struct MPID_nem_tcp_send_q_element
19{
20    struct MPID_nem_tcp_send_q_element *next;
21    size_t len;                        /* number of bytes left to send */
22    char *start;                       /* pointer to next byte to send */
23    MPID_nem_cell_ptr_t cell;
24    /*     char buf[MPID_NEM_MAX_PACKET_LEN];*/ /* data to be sent */
25} MPID_nem_tcp_send_q_element_t;
26
27static struct {MPID_nem_tcp_send_q_element_t *top;} free_buffers = {0};
28
29#define ALLOC_Q_ELEMENT(e) do {                                                                                                         \
30        if (S_EMPTY (free_buffers))                                                                                                     \
31        {                                                                                                                               \
32            MPIU_CHKPMEM_MALLOC (*(e), MPID_nem_tcp_send_q_element_t *, sizeof(MPID_nem_tcp_send_q_element_t),      \
33                                 mpi_errno, "send queue element");                                                                      \
34        }                                                                                                                               \
35        else                                                                                                                            \
36        {                                                                                                                               \
37            S_POP (&free_buffers, e);                                                                                                   \
38        }                                                                                                                               \
39    } while (0)
40
41/* FREE_Q_ELEMENTS() frees a list if elements starting at e0 through e1 */
42#define FREE_Q_ELEMENTS(e0, e1) S_PUSH_MULTIPLE (&free_buffers, e0, e1)
43#define FREE_Q_ELEMENT(e) S_PUSH (&free_buffers, e)
44
45int send_queued (MPIDI_VC_t *vc);
46
47#undef FUNCNAME
48#define FUNCNAME MPID_nem_tcp_send_init
49#undef FCNAME
50#define FCNAME MPIDI_QUOTE(FUNCNAME)
51int MPID_nem_tcp_send_init(void)
52{
53    int mpi_errno = MPI_SUCCESS;
54    int i;
55    MPIU_CHKPMEM_DECL (NUM_PREALLOC_SENDQ);
56   
57    /* preallocate sendq elements */
58    for (i = 0; i < NUM_PREALLOC_SENDQ; ++i)
59    {
60        MPID_nem_tcp_send_q_element_t *e;
61       
62        MPIU_CHKPMEM_MALLOC (e, MPID_nem_tcp_send_q_element_t *,
63                             sizeof(MPID_nem_tcp_send_q_element_t), mpi_errno, "send queue element");
64        S_PUSH (&free_buffers, e);
65    }
66
67    MPIU_CHKPMEM_COMMIT();
68    return mpi_errno;
69 fn_fail:
70    MPIU_CHKPMEM_REAP();
71    return mpi_errno;
72}
73
74#undef FUNCNAME
75#define FUNCNAME MPID_nem_tcp_send
76#undef FCNAME
77#define FCNAME MPIDI_QUOTE(FUNCNAME)
78int MPID_nem_tcp_send (MPIDI_VC_t *vc, MPID_nem_cell_ptr_t cell, int datalen)
79{
80    int mpi_errno = MPI_SUCCESS;
81    MPIU_Assert(0);
82    return mpi_errno;
83}
84
85
86#undef FUNCNAME
87#define FUNCNAME MPID_nem_tcp_send_queued
88#undef FCNAME
89#define FCNAME MPIDI_QUOTE(FUNCNAME)
90int MPID_nem_tcp_send_queued (MPIDI_VC_t *vc)
91{
92    int mpi_errno = MPI_SUCCESS;
93    MPID_Request *sreq;
94    MPIDI_msg_sz_t offset;
95    MPID_IOV *iov;
96    int complete;
97    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_SEND_QUEUED);
98
99    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_SEND_QUEUED);
100
101    MPIU_Assert(vc != NULL);
102
103    if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
104        goto fn_exit;
105
106    while (!SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
107    {
108        sreq = SENDQ_HEAD(VC_FIELD(vc, send_queue));
109       
110        iov = &sreq->dev.iov[sreq->dev.iov_offset];
111       
112        CHECK_EINTR(offset, writev(VC_FIELD(vc, sc)->fd, iov, sreq->dev.iov_count));
113        MPIU_ERR_CHKANDJUMP(offset == 0, mpi_errno, MPI_ERR_OTHER, "**sock_closed");
114        if (offset == -1)
115        {
116            if (errno == EAGAIN)
117            {
118                offset = 0;
119                MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, "EAGAIN");
120            }
121            else
122                MPIU_ERR_SETANDJUMP1(mpi_errno, MPI_ERR_OTHER, "**writev", "**writev %s", strerror (errno));
123        }
124        MPIU_DBG_MSG_D(CH3_CHANNEL, VERBOSE, "write " MPIDI_MSG_SZ_FMT, offset);
125
126        complete = 1;
127        for (iov = &sreq->dev.iov[sreq->dev.iov_offset]; iov < &sreq->dev.iov[sreq->dev.iov_offset + sreq->dev.iov_count]; ++iov)
128        {
129            if (offset < iov->MPID_IOV_LEN)
130            {
131                iov->MPID_IOV_BUF = (char *)iov->MPID_IOV_BUF + offset;
132                iov->MPID_IOV_LEN -= offset;
133                /* iov_count should be equal to the number of iov's remaining */
134                sreq->dev.iov_count -= ((iov - sreq->dev.iov) - sreq->dev.iov_offset);
135                sreq->dev.iov_offset = iov - sreq->dev.iov;
136                complete = 0;
137                break;
138            }
139            offset -= iov->MPID_IOV_LEN;
140        }
141        if (!complete)
142        {
143            /* writev couldn't write the entire iov, give up for now */
144            break;
145        }
146        else
147        {
148            /* sent whole message */
149            int (*reqFn)(MPIDI_VC_t *, MPID_Request *, int *);
150
151            reqFn = sreq->dev.OnDataAvail;
152            if (!reqFn)
153            {
154                MPIU_Assert(MPIDI_Request_get_type(sreq) != MPIDI_REQUEST_TYPE_GET_RESP);
155                MPIDI_CH3U_Request_complete(sreq);
156                MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
157                SENDQ_DEQUEUE(&VC_FIELD(vc, send_queue), &sreq);
158                continue;
159            }
160
161            complete = 0;
162            mpi_errno = reqFn(vc, sreq, &complete);
163            if (mpi_errno) MPIU_ERR_POP(mpi_errno);
164           
165            if (complete)
166            {
167                MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
168                SENDQ_DEQUEUE(&VC_FIELD(vc, send_queue), &sreq);
169                continue;
170            }
171            sreq->dev.iov_offset = 0;
172        }
173    }
174
175    if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
176        UNSET_PLFD(vc);
177   
178 fn_exit:
179    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_SEND_QUEUED);
180    return mpi_errno;
181 fn_fail:
182    goto fn_exit;
183}
184
185#undef FUNCNAME
186#define FUNCNAME MPID_nem_tcp_send_finalize
187#undef FCNAME
188#define FCNAME MPIDI_QUOTE(FUNCNAME)
189int MPID_nem_tcp_send_finalize(void)
190{
191    int mpi_errno = MPI_SUCCESS;
192
193    while (!S_EMPTY (free_buffers))
194    {
195        MPID_nem_tcp_send_q_element_t *e;
196        S_POP (&free_buffers, &e);
197        MPIU_Free (e);
198    }
199    return mpi_errno;
200}
201
202/* MPID_nem_tcp_conn_est -- this function is called when the
203   connection is finally established to send any pending sends */
204#undef FUNCNAME
205#define FUNCNAME MPID_nem_tcp_conn_est
206#undef FCNAME
207#define FCNAME MPIDI_QUOTE(FUNCNAME)
208int MPID_nem_tcp_conn_est (MPIDI_VC_t *vc)
209{
210    int mpi_errno = MPI_SUCCESS;
211    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_CONN_EST);
212
213    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_CONN_EST);
214
215    MPIDI_CHANGE_VC_STATE(vc, ACTIVE);
216
217    if (!SENDQ_EMPTY (VC_FIELD(vc, send_queue)))
218    {
219        SET_PLFD(vc);
220        mpi_errno = MPID_nem_tcp_send_queued (vc);
221        if (mpi_errno) MPIU_ERR_POP (mpi_errno);
222    }
223
224 fn_fail:   
225    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_CONN_EST);
226    return mpi_errno;
227}
228
229
230#undef FUNCNAME
231#define FUNCNAME MPID_nem_tcp_iStartContigMsg
232#undef FCNAME
233#define FCNAME MPIDI_QUOTE(FUNCNAME)
234int MPID_nem_tcp_iStartContigMsg(MPIDI_VC_t *vc, void *hdr, MPIDI_msg_sz_t hdr_sz, void *data, MPIDI_msg_sz_t data_sz,
235                                    MPID_Request **sreq_ptr)
236{
237    int mpi_errno = MPI_SUCCESS;
238    MPID_Request * sreq = NULL;
239    MPIDI_msg_sz_t offset = 0;
240    sockconn_t *sc = VC_FIELD(vc, sc);
241    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_ISTARTCONTIGMSG);
242
243    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_ISTARTCONTIGMSG);
244   
245    MPIU_Assert(hdr_sz <= sizeof(MPIDI_CH3_Pkt_t));
246   
247    MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, "tcp_iStartContigMsg");
248    MPIDI_DBG_Print_packet((MPIDI_CH3_Pkt_t *)hdr);
249    if (MPID_nem_tcp_vc_is_connected(vc))
250    {
251        if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
252        {
253            MPID_IOV iov[2];
254
255            iov[0].MPID_IOV_BUF = hdr;
256            iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t);
257            iov[1].MPID_IOV_BUF = data;
258            iov[1].MPID_IOV_LEN = data_sz;
259       
260            CHECK_EINTR(offset, writev(sc->fd, iov, 2));
261            MPIU_ERR_CHKANDJUMP(offset == 0, mpi_errno, MPI_ERR_OTHER, "**sock_closed");
262            if (offset == -1)
263            {
264                if (errno == EAGAIN)
265                    offset = 0;
266                else
267                    MPIU_ERR_SETANDJUMP1(mpi_errno, MPI_ERR_OTHER, "**writev", "**writev %s", strerror (errno));
268            }
269            MPIU_DBG_MSG_D(CH3_CHANNEL, VERBOSE, "write " MPIDI_MSG_SZ_FMT, offset);
270
271            if (offset == sizeof(MPIDI_CH3_PktGeneric_t) + data_sz)
272            {
273                /* sent whole message */
274                *sreq_ptr = NULL;
275                goto fn_exit;
276            }
277        }
278    }
279    else
280    {
281        mpi_errno = MPID_nem_tcp_connect(vc);
282        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
283    }
284
285    /* create and enqueue request */
286    MPIU_DBG_MSG (CH3_CHANNEL, VERBOSE, "enqueuing");
287
288    /* create a request */
289    sreq = MPID_Request_create();
290    MPIU_Assert (sreq != NULL);
291    MPIU_Object_set_ref (sreq, 2);
292    sreq->kind = MPID_REQUEST_SEND;
293
294    sreq->dev.OnDataAvail = 0;
295    sreq->ch.vc = vc;
296    sreq->dev.iov_offset = 0;
297
298    if (offset < sizeof(MPIDI_CH3_PktGeneric_t))
299    {
300        sreq->dev.pending_pkt = *(MPIDI_CH3_PktGeneric_t *)hdr;
301        sreq->dev.iov[0].MPID_IOV_BUF = (char *)&sreq->dev.pending_pkt + offset;
302        sreq->dev.iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t) - offset ;
303        if (data_sz)
304        {
305            sreq->dev.iov[1].MPID_IOV_BUF = data;
306            sreq->dev.iov[1].MPID_IOV_LEN = data_sz;
307            sreq->dev.iov_count = 2;
308        }
309        else
310            sreq->dev.iov_count = 1;
311    }
312    else
313    {
314        sreq->dev.iov[0].MPID_IOV_BUF = (char *)data + (offset - sizeof(MPIDI_CH3_PktGeneric_t));
315        sreq->dev.iov[0].MPID_IOV_LEN = data_sz - (offset - sizeof(MPIDI_CH3_PktGeneric_t));
316        sreq->dev.iov_count = 1;
317    }
318   
319    MPIU_Assert(sreq->dev.iov_count >= 1 && sreq->dev.iov[0].MPID_IOV_LEN > 0);
320
321    if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)) && MPID_nem_tcp_vc_is_connected(vc))
322        SET_PLFD(vc);
323    SENDQ_ENQUEUE(&VC_FIELD(vc, send_queue), sreq);
324
325    *sreq_ptr = sreq;
326   
327 fn_exit:
328    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_ISTARTCONTIGMSG);
329    return mpi_errno;
330 fn_fail:
331    goto fn_exit;
332}
333
334#undef FUNCNAME
335#define FUNCNAME MPID_nem_tcp_iSendContig
336#undef FCNAME
337#define FCNAME MPIDI_QUOTE(FUNCNAME)
338int MPID_nem_tcp_iSendContig(MPIDI_VC_t *vc, MPID_Request *sreq, void *hdr, MPIDI_msg_sz_t hdr_sz,
339                                void *data, MPIDI_msg_sz_t data_sz)
340{
341    int mpi_errno = MPI_SUCCESS;
342    MPIDI_msg_sz_t offset = 0;
343    sockconn_t *sc = VC_FIELD(vc, sc);
344    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_ISENDCONTIGMSG);
345
346    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_ISENDCONTIGMSG);
347   
348    MPIU_Assert(hdr_sz <= sizeof(MPIDI_CH3_Pkt_t));
349   
350    MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, "tcp_iSendContig");
351
352    MPIDI_DBG_Print_packet((MPIDI_CH3_Pkt_t *)hdr);
353    if (MPID_nem_tcp_vc_is_connected(vc))
354    {
355        if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
356        {
357            MPID_IOV iov[2];
358
359            iov[0].MPID_IOV_BUF = hdr;
360            iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t);
361            iov[1].MPID_IOV_BUF = data;
362            iov[1].MPID_IOV_LEN = data_sz;
363       
364            CHECK_EINTR(offset, writev(sc->fd, iov, 2));
365            MPIU_ERR_CHKANDJUMP(offset == 0, mpi_errno, MPI_ERR_OTHER, "**sock_closed");
366            if (offset == -1)
367            {
368                if (errno == EAGAIN)
369                    offset = 0;
370                else
371                    MPIU_ERR_SETANDJUMP1(mpi_errno, MPI_ERR_OTHER, "**writev", "**writev %s", strerror (errno));
372            }
373            MPIU_DBG_MSG_D(CH3_CHANNEL, VERBOSE, "write " MPIDI_MSG_SZ_FMT, offset);
374
375            if (offset == sizeof(MPIDI_CH3_PktGeneric_t) + data_sz)
376            {
377                /* sent whole message */
378                int (*reqFn)(MPIDI_VC_t *, MPID_Request *, int *);
379
380                reqFn = sreq->dev.OnDataAvail;
381                if (!reqFn)
382                {
383                    MPIU_Assert(MPIDI_Request_get_type(sreq) != MPIDI_REQUEST_TYPE_GET_RESP);
384                    MPIDI_CH3U_Request_complete(sreq);
385                    MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
386                    goto fn_exit;
387                }
388                else
389                {
390                    int complete = 0;
391               
392                    mpi_errno = reqFn(vc, sreq, &complete);
393                    if (mpi_errno) MPIU_ERR_POP(mpi_errno);
394
395                    if (complete)
396                    {
397                        MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
398                        goto fn_exit;
399                    }
400
401                    /* not completed: more to send */
402                    goto enqueue_request;
403                }
404            }
405        }
406    }
407    else
408    {
409        mpi_errno = MPID_nem_tcp_connect(vc);
410        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
411    }
412
413
414    /* save iov */
415    if (offset < sizeof(MPIDI_CH3_PktGeneric_t))
416    {
417        sreq->dev.pending_pkt = *(MPIDI_CH3_PktGeneric_t *)hdr;
418        sreq->dev.iov[0].MPID_IOV_BUF = (char *)&sreq->dev.pending_pkt + offset;
419        sreq->dev.iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t) - offset;
420        if (data_sz)
421        {
422            sreq->dev.iov[1].MPID_IOV_BUF = data;
423            sreq->dev.iov[1].MPID_IOV_LEN = data_sz;
424            sreq->dev.iov_count = 2;
425        }
426        else
427            sreq->dev.iov_count = 1;
428    }
429    else
430    {
431        sreq->dev.iov[0].MPID_IOV_BUF = (char *)data + (offset - sizeof(MPIDI_CH3_PktGeneric_t));
432        sreq->dev.iov[0].MPID_IOV_LEN = data_sz - (offset - sizeof(MPIDI_CH3_PktGeneric_t));
433        sreq->dev.iov_count = 1;
434    }
435
436 enqueue_request:
437    /* enqueue request */
438    MPIU_DBG_MSG (CH3_CHANNEL, VERBOSE, "enqueuing");
439    MPIU_Assert(sreq->dev.iov_count >= 1 && sreq->dev.iov[0].MPID_IOV_LEN > 0);
440
441    sreq->ch.vc = vc;
442    sreq->dev.iov_offset = 0;
443
444    if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)) && MPID_nem_tcp_vc_is_connected(vc))
445        SET_PLFD(vc);
446    SENDQ_ENQUEUE(&VC_FIELD(vc, send_queue), sreq);
447
448 fn_exit:
449    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_ISENDCONTIGMSG);
450    return mpi_errno;
451 fn_fail:
452    goto fn_exit;
453}
454
455
456#undef FUNCNAME
457#define FUNCNAME MPID_nem_tcp_SendNoncontig
458#undef FCNAME
459#define FCNAME MPIDI_QUOTE(FUNCNAME)
460int MPID_nem_tcp_SendNoncontig(MPIDI_VC_t *vc, MPID_Request *sreq, void *header, MPIDI_msg_sz_t hdr_sz)
461{
462    int mpi_errno = MPI_SUCCESS;
463    int iov_n;
464    MPID_IOV iov[MPID_IOV_LIMIT];
465    MPID_IOV *iov_p;
466    MPIDI_msg_sz_t offset;
467    int complete;
468
469    MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, "tcp_SendNoncontig");
470    MPIU_Assert(hdr_sz <= sizeof(MPIDI_CH3_PktGeneric_t));
471   
472    iov[0].MPID_IOV_BUF = header;
473    iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t);
474
475    iov_n = MPID_IOV_LIMIT - 1;
476
477    mpi_errno = MPIDI_CH3U_Request_load_send_iov(sreq, &iov[1], &iov_n);
478    MPIU_ERR_CHKANDJUMP(mpi_errno, mpi_errno, MPI_ERR_OTHER, "**ch3|loadsendiov");
479
480    iov_n += 1;
481    offset = 0;
482
483    if (MPID_nem_tcp_vc_is_connected(vc))
484    {
485        if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)))
486        {
487            CHECK_EINTR(offset, writev(VC_FIELD(vc, sc)->fd, iov, iov_n));
488            MPIU_ERR_CHKANDJUMP(offset == 0, mpi_errno, MPI_ERR_OTHER, "**sock_closed");
489            if (offset == -1)
490            {
491                if (errno == EAGAIN)
492                    offset = 0;
493                else
494                    MPIU_ERR_SETANDJUMP1(mpi_errno, MPI_ERR_OTHER, "**writev", "**writev %s", strerror (errno));
495            }
496            MPIU_DBG_MSG_D(CH3_CHANNEL, VERBOSE, "write noncontig " MPIDI_MSG_SZ_FMT, offset);
497        }
498    }
499    else
500    {
501        mpi_errno = MPID_nem_tcp_connect(vc);
502        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
503    }
504
505    if (offset < iov[0].MPID_IOV_LEN)
506    {
507        /* header was not yet sent, save it in req */
508        sreq->dev.pending_pkt = *(MPIDI_CH3_PktGeneric_t *)header;
509        iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->dev.pending_pkt;
510        iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_PktGeneric_t);
511    }
512
513    /* check if whole iov was sent, and save any unsent portion of iov */
514    sreq->dev.iov_count = 0;
515    complete = 1;
516    for (iov_p = &iov[0]; iov_p < &iov[iov_n]; ++iov_p)
517    {
518        if (offset < iov_p->MPID_IOV_LEN)
519        {
520            sreq->dev.iov[sreq->dev.iov_count].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)((char *)iov_p->MPID_IOV_BUF + offset);
521            sreq->dev.iov[sreq->dev.iov_count].MPID_IOV_LEN = iov_p->MPID_IOV_LEN - offset;
522            offset = 0;
523            ++sreq->dev.iov_count;
524            complete = 0;
525        }
526        else
527            offset -= iov_p->MPID_IOV_LEN;
528    }
529       
530    if (complete)
531    {
532        /* sent whole iov */
533        int (*reqFn)(MPIDI_VC_t *, MPID_Request *, int *);
534
535        reqFn = sreq->dev.OnDataAvail;
536        if (!reqFn)
537        {
538            MPIDI_CH3U_Request_complete(sreq);
539            MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
540            goto fn_exit;
541        }
542
543        complete = 0;
544        mpi_errno = reqFn(vc, sreq, &complete);
545        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
546           
547        if (complete)
548        {
549            MPIU_DBG_MSG(CH3_CHANNEL, VERBOSE, ".... complete");
550            goto fn_exit;
551        }
552    }
553       
554    /* enqueue request */
555    MPIU_DBG_MSG (CH3_CHANNEL, VERBOSE, "enqueuing");
556    MPIU_Assert(sreq->dev.iov_count >= 1 && sreq->dev.iov[0].MPID_IOV_LEN > 0);
557       
558    sreq->ch.vc = vc;
559    sreq->dev.iov_offset = 0;
560       
561    if (SENDQ_EMPTY(VC_FIELD(vc, send_queue)) && MPID_nem_tcp_vc_is_connected(vc))
562        SET_PLFD(vc);
563    SENDQ_ENQUEUE(&VC_FIELD(vc, send_queue), sreq);
564
565 fn_exit:
566    return mpi_errno;
567 fn_fail:
568    MPIU_Object_set_ref(sreq, 0);
569    MPIDI_CH3_Request_destroy(sreq);
570    goto fn_exit;
571}
Note: See TracBrowser for help on using the browser.