1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (c) 2003 Los Alamos National Laboratory (LANL)
6 * This file is part of Lustre, http://www.lustre.org/
8 * Lustre is free software; you can redistribute it and/or
9 * modify it under the terms of version 2 of the GNU General Public
10 * License as published by the Free Software Foundation.
12 * Lustre is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
17 * You should have received a copy of the GNU General Public License
18 * along with Lustre; if not, write to the Free Software
19 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
23 * This file contains all gmnal send and receive functions
29 * The caretaker thread
30 * This is main thread of execution for the NAL side
31 * This guy waits in gm_blocking_recvive and gets
32 * woken up when the myrinet adaptor gets an interrupt.
33 * Hands off receive operations to the receive thread
34 * This thread Looks after gm_callbacks etc inline.
37 gmnal_ct_thread(void *arg)
39 gmnal_data_t *nal_data;
40 gm_recv_event_t *rxevent = NULL;
41 gm_recv_t *recv = NULL;
44 CDEBUG(D_TRACE, "NO nal_data. Exiting\n");
48 nal_data = (gmnal_data_t*)arg;
49 CDEBUG(D_TRACE, "nal_data is [%p]\n", arg);
53 nal_data->ctthread_flag = GMNAL_CTTHREAD_STARTED;
55 GMNAL_GM_LOCK(nal_data);
56 while(nal_data->ctthread_flag == GMNAL_CTTHREAD_STARTED) {
57 CDEBUG(D_NET, "waiting\n");
58 rxevent = gm_blocking_receive_no_spin(nal_data->gm_port);
59 if (nal_data->ctthread_flag == GMNAL_THREAD_STOP) {
60 CDEBUG(D_INFO, "time to exit\n");
63 CDEBUG(D_INFO, "got [%s]\n", gmnal_rxevent(rxevent));
64 switch (GM_RECV_EVENT_TYPE(rxevent)) {
67 CDEBUG(D_NET, "CTTHREAD:: GM_RECV_EVENT\n");
68 recv = (gm_recv_t*)&rxevent->recv;
69 GMNAL_GM_UNLOCK(nal_data);
70 gmnal_add_rxtwe(nal_data, recv);
71 GMNAL_GM_LOCK(nal_data);
72 CDEBUG(D_NET, "CTTHREAD:: Added event to Q\n");
74 case(_GM_SLEEP_EVENT):
76 * Blocking receive above just returns
77 * immediatly with _GM_SLEEP_EVENT
78 * Don't know what this is
80 CDEBUG(D_NET, "Sleeping in gm_unknown\n");
81 GMNAL_GM_UNLOCK(nal_data);
82 gm_unknown(nal_data->gm_port, rxevent);
83 GMNAL_GM_LOCK(nal_data);
84 CDEBUG(D_INFO, "Awake from gm_unknown\n");
89 * Don't know what this is
90 * gm_unknown will make sense of it
91 * Should be able to do something with
92 * FAST_RECV_EVENTS here.
94 CDEBUG(D_NET, "Passing event to gm_unknown\n");
95 GMNAL_GM_UNLOCK(nal_data);
96 gm_unknown(nal_data->gm_port, rxevent);
97 GMNAL_GM_LOCK(nal_data);
98 CDEBUG(D_INFO, "Processed unknown event\n");
101 GMNAL_GM_UNLOCK(nal_data);
102 nal_data->ctthread_flag = GMNAL_THREAD_RESET;
103 CDEBUG(D_INFO, "thread nal_data [%p] is exiting\n", nal_data);
104 return(GMNAL_STATUS_OK);
109 * process a receive event
111 int gmnal_rx_thread(void *arg)
113 gmnal_data_t *nal_data;
115 gmnal_rxtwe_t *we = NULL;
118 CDEBUG(D_TRACE, "NO nal_data. Exiting\n");
122 nal_data = (gmnal_data_t*)arg;
123 CDEBUG(D_TRACE, "nal_data is [%p]\n", arg);
127 * set 1 bit for each thread started
128 * doesn't matter which bit
130 spin_lock(&nal_data->rxthread_flag_lock);
131 if (nal_data->rxthread_flag)
132 nal_data->rxthread_flag=nal_data->rxthread_flag*2 + 1;
134 nal_data->rxthread_flag = 1;
135 CDEBUG(D_INFO, "rxthread flag is [%ld]\n", nal_data->rxthread_flag);
136 spin_unlock(&nal_data->rxthread_flag_lock);
138 while(nal_data->rxthread_stop_flag != GMNAL_THREAD_STOP) {
139 CDEBUG(D_NET, "RXTHREAD:: Receive thread waiting\n");
140 we = gmnal_get_rxtwe(nal_data);
142 CDEBUG(D_INFO, "Receive thread time to exit\n");
147 switch(((gmnal_msghdr_t*)buffer)->type) {
148 case(GMNAL_SMALL_MESSAGE):
149 gmnal_pre_receive(nal_data, we,
150 GMNAL_SMALL_MESSAGE);
152 case(GMNAL_LARGE_MESSAGE_INIT):
153 gmnal_pre_receive(nal_data, we,
154 GMNAL_LARGE_MESSAGE_INIT);
156 case(GMNAL_LARGE_MESSAGE_ACK):
157 gmnal_pre_receive(nal_data, we,
158 GMNAL_LARGE_MESSAGE_ACK);
161 CDEBUG(D_ERROR, "Unsupported message type\n");
162 gmnal_rx_bad(nal_data, we, NULL);
164 PORTAL_FREE(we, sizeof(gmnal_rxtwe_t));
167 spin_lock(&nal_data->rxthread_flag_lock);
168 nal_data->rxthread_flag/=2;
169 CDEBUG(D_INFO, "rxthread flag is [%ld]\n", nal_data->rxthread_flag);
170 spin_unlock(&nal_data->rxthread_flag_lock);
171 CDEBUG(D_INFO, "thread nal_data [%p] is exiting\n", nal_data);
172 return(GMNAL_STATUS_OK);
178 * Start processing a small message receive
179 * Get here from gmnal_receive_thread
180 * Hand off to lib_parse, which calls cb_recv
181 * which hands back to gmnal_small_receive
182 * Deal with all endian stuff here.
185 gmnal_pre_receive(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, int gmnal_type)
187 gmnal_srxd_t *srxd = NULL;
189 unsigned int snode, sport, type, length;
190 gmnal_msghdr_t *gmnal_msghdr;
191 ptl_hdr_t *portals_hdr;
193 CDEBUG(D_INFO, "nal_data [%p], we[%p] type [%d]\n",
194 nal_data, we, gmnal_type);
203 gmnal_msghdr = (gmnal_msghdr_t*)buffer;
204 portals_hdr = (ptl_hdr_t*)(buffer+GMNAL_MSGHDR_SIZE);
206 CDEBUG(D_INFO, "rx_event:: Sender node [%d], Sender Port [%d],
207 type [%d], length [%d], buffer [%p]\n",
208 snode, sport, type, length, buffer);
209 CDEBUG(D_INFO, "gmnal_msghdr:: Sender node [%u], magic [%d],
210 gmnal_type [%d]\n", gmnal_msghdr->sender_node_id,
211 gmnal_msghdr->magic, gmnal_msghdr->type);
212 CDEBUG(D_INFO, "portals_hdr:: Sender node ["LPD64"],
213 dest_node ["LPD64"]\n", portals_hdr->src_nid,
214 portals_hdr->dest_nid);
218 * Get a receive descriptor for this message
220 srxd = gmnal_rxbuffer_to_srxd(nal_data, buffer);
221 CDEBUG(D_INFO, "Back from gmnal_rxbuffer_to_srxd\n");
222 srxd->nal_data = nal_data;
224 CDEBUG(D_ERROR, "Failed to get receive descriptor\n");
225 lib_parse(nal_data->nal_cb, portals_hdr, srxd);
226 return(GMNAL_STATUS_FAIL);
230 * no need to bother portals library with this
232 if (gmnal_type == GMNAL_LARGE_MESSAGE_ACK) {
233 gmnal_large_tx_ack_received(nal_data, srxd);
234 return(GMNAL_STATUS_OK);
237 srxd->type = gmnal_type;
238 srxd->nsiov = gmnal_msghdr->niov;
239 srxd->gm_source_node = gmnal_msghdr->sender_node_id;
241 CDEBUG(D_PORTALS, "Calling lib_parse buffer is [%p]\n",
242 buffer+GMNAL_MSGHDR_SIZE);
244 * control passes to lib, which calls cb_recv
245 * cb_recv is responsible for returning the buffer
248 lib_parse(nal_data->nal_cb, portals_hdr, srxd);
250 return(GMNAL_STATUS_OK);
256 * After a receive has been processed,
257 * hang out the receive buffer again.
258 * This implicitly returns a receive token.
261 gmnal_rx_requeue_buffer(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
263 CDEBUG(D_TRACE, "gmnal_rx_requeue_buffer\n");
265 CDEBUG(D_NET, "requeueing srxd[%p] nal_data[%p]\n", srxd, nal_data);
267 GMNAL_GM_LOCK(nal_data);
268 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
269 srxd->gmsize, GM_LOW_PRIORITY, 0 );
270 GMNAL_GM_UNLOCK(nal_data);
272 return(GMNAL_STATUS_OK);
277 * Handle a bad message
278 * A bad message is one we don't expect or can't interpret
281 gmnal_rx_bad(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, gmnal_srxd_t *srxd)
283 CDEBUG(D_TRACE, "Can't handle message\n");
286 srxd = gmnal_rxbuffer_to_srxd(nal_data,
289 gmnal_rx_requeue_buffer(nal_data, srxd);
291 CDEBUG(D_ERROR, "Can't find a descriptor for this buffer\n");
295 return(GMNAL_STATUS_FAIL);
298 return(GMNAL_STATUS_OK);
304 * Process a small message receive.
305 * Get here from gmnal_receive_thread, gmnal_pre_receive
307 * Put data from prewired receive buffer into users buffer(s)
308 * Hang out the receive buffer again for another receive
312 gmnal_small_rx(nal_cb_t *nal_cb, void *private, lib_msg_t *cookie,
313 unsigned int niov, struct iovec *iov, size_t mlen, size_t rlen)
315 gmnal_srxd_t *srxd = NULL;
317 gmnal_data_t *nal_data = (gmnal_data_t*)nal_cb->nal_data;
320 CDEBUG(D_TRACE, "niov [%d] mlen["LPSZ"]\n", niov, mlen);
323 CDEBUG(D_ERROR, "gmnal_small_rx no context\n");
324 lib_finalize(nal_cb, private, cookie);
328 srxd = (gmnal_srxd_t*)private;
329 buffer = srxd->buffer;
330 buffer += sizeof(gmnal_msghdr_t);
331 buffer += sizeof(ptl_hdr_t);
334 CDEBUG(D_INFO, "processing [%p] len ["LPSZ"]\n", iov,
336 gm_bcopy(buffer, iov->iov_base, iov->iov_len);
337 buffer += iov->iov_len;
343 * let portals library know receive is complete
345 CDEBUG(D_PORTALS, "calling lib_finalize\n");
346 if (lib_finalize(nal_cb, private, cookie) != PTL_OK) {
347 /* TO DO what to do with failed lib_finalise? */
348 CDEBUG(D_INFO, "lib_finalize failed\n");
351 * return buffer so it can be used again
353 CDEBUG(D_NET, "calling gm_provide_receive_buffer\n");
354 GMNAL_GM_LOCK(nal_data);
355 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
356 srxd->gmsize, GM_LOW_PRIORITY, 0);
357 GMNAL_GM_UNLOCK(nal_data);
364 * Start a small transmit.
365 * Get a send token (and wired transmit buffer).
366 * Copy data from senders buffer to wired buffer and
367 * initiate gm_send from the wired buffer.
368 * The callback function informs when the send is complete.
371 gmnal_small_tx(nal_cb_t *nal_cb, void *private, lib_msg_t *cookie,
372 ptl_hdr_t *hdr, int type, ptl_nid_t global_nid, ptl_pid_t pid,
373 unsigned int niov, struct iovec *iov, int size)
375 gmnal_data_t *nal_data = (gmnal_data_t*)nal_cb->nal_data;
376 gmnal_stxd_t *stxd = NULL;
378 gmnal_msghdr_t *msghdr = NULL;
380 unsigned int local_nid;
381 gm_status_t gm_status = GM_SUCCESS;
383 CDEBUG(D_TRACE, "gmnal_small_tx nal_cb [%p] private [%p] cookie [%p]
384 hdr [%p] type [%d] global_nid ["LPU64"] pid [%d] niov [%d]
385 iov [%p] size [%d]\n", nal_cb, private, cookie, hdr, type,
386 global_nid, pid, niov, iov, size);
388 CDEBUG(D_INFO, "portals_hdr:: dest_nid ["LPU64"], src_nid ["LPU64"]\n",
389 hdr->dest_nid, hdr->src_nid);
392 CDEBUG(D_ERROR, "no nal_data\n");
393 return(GMNAL_STATUS_FAIL);
395 CDEBUG(D_INFO, "nal_data [%p]\n", nal_data);
398 GMNAL_GM_LOCK(nal_data);
399 gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
401 GMNAL_GM_UNLOCK(nal_data);
402 if (gm_status != GM_SUCCESS) {
403 CDEBUG(D_ERROR, "Failed to obtain local id\n");
404 return(GMNAL_STATUS_FAIL);
406 CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
408 stxd = gmnal_get_stxd(nal_data, 1);
409 CDEBUG(D_INFO, "stxd [%p]\n", stxd);
411 stxd->type = GMNAL_SMALL_MESSAGE;
412 stxd->cookie = cookie;
415 * Copy gmnal_msg_hdr and portals header to the transmit buffer
416 * Then copy the data in
418 buffer = stxd->buffer;
419 msghdr = (gmnal_msghdr_t*)buffer;
421 msghdr->magic = GMNAL_MAGIC;
422 msghdr->type = GMNAL_SMALL_MESSAGE;
423 msghdr->sender_node_id = nal_data->gm_global_nid;
424 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
426 buffer += sizeof(gmnal_msghdr_t);
428 CDEBUG(D_INFO, "processing portals hdr at [%p]\n", buffer);
429 gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
431 buffer += sizeof(ptl_hdr_t);
434 CDEBUG(D_INFO, "processing iov [%p] len ["LPSZ"] to [%p]\n",
435 iov, iov->iov_len, buffer);
436 gm_bcopy(iov->iov_base, buffer, iov->iov_len);
437 buffer+= iov->iov_len;
441 CDEBUG(D_INFO, "sending\n");
442 tot_size = size+sizeof(ptl_hdr_t)+sizeof(gmnal_msghdr_t);
443 stxd->msg_size = tot_size;
446 CDEBUG(D_NET, "Calling gm_send_to_peer port [%p] buffer [%p]
447 gmsize [%lu] msize [%d] global_nid ["LPU64"] local_nid[%d]
448 stxd [%p]\n", nal_data->gm_port, stxd->buffer, stxd->gm_size,
449 stxd->msg_size, global_nid, local_nid, stxd);
451 GMNAL_GM_LOCK(nal_data);
452 stxd->gm_priority = GM_LOW_PRIORITY;
453 stxd->gm_target_node = local_nid;
454 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
455 stxd->gm_size, stxd->msg_size,
456 GM_LOW_PRIORITY, local_nid,
457 gmnal_small_tx_callback, (void*)stxd);
458 GMNAL_GM_UNLOCK(nal_data);
459 CDEBUG(D_INFO, "done\n");
466 * A callback to indicate the small transmit operation is compete
467 * Check for erros and try to deal with them.
468 * Call lib_finalise to inform the client application that the send
469 * is complete and the memory can be reused.
470 * Return the stxd when finished with it (returns a send token)
473 gmnal_small_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
475 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
476 lib_msg_t *cookie = stxd->cookie;
477 gmnal_data_t *nal_data = (gmnal_data_t*)stxd->nal_data;
478 nal_cb_t *nal_cb = nal_data->nal_cb;
481 CDEBUG(D_TRACE, "send completion event for unknown stxd\n");
484 if (status != GM_SUCCESS) {
485 CDEBUG(D_ERROR, "Result of send stxd [%p] is [%s]\n",
486 stxd, gmnal_gm_error(status));
495 case(GM_SEND_DROPPED):
497 * do a resend on the dropped ones
499 CDEBUG(D_ERROR, "send stxd [%p] was dropped
500 resending\n", context);
501 GMNAL_GM_LOCK(nal_data);
502 gm_send_to_peer_with_callback(nal_data->gm_port,
507 stxd->gm_target_node,
508 gmnal_small_tx_callback,
510 GMNAL_GM_UNLOCK(nal_data);
514 case(GM_SEND_TIMED_OUT):
518 CDEBUG(D_INFO, "calling gm_drop_sends\n");
519 GMNAL_GM_LOCK(nal_data);
520 gm_drop_sends(nal_data->gm_port, stxd->gm_priority,
521 stxd->gm_target_node, GMNAL_GM_PORT,
522 gmnal_drop_sends_callback, context);
523 GMNAL_GM_UNLOCK(nal_data);
532 case(GM_INTERRUPTED):
534 case(GM_INPUT_BUFFER_TOO_SMALL):
535 case(GM_OUTPUT_BUFFER_TOO_SMALL):
537 case(GM_MEMORY_FAULT):
538 case(GM_INVALID_PARAMETER):
539 case(GM_OUT_OF_MEMORY):
540 case(GM_INVALID_COMMAND):
541 case(GM_PERMISSION_DENIED):
542 case(GM_INTERNAL_ERROR):
544 case(GM_UNSUPPORTED_DEVICE):
545 case(GM_SEND_REJECTED):
546 case(GM_SEND_TARGET_PORT_CLOSED):
547 case(GM_SEND_TARGET_NODE_UNREACHABLE):
548 case(GM_SEND_PORT_CLOSED):
549 case(GM_NODE_ID_NOT_YET_SET):
550 case(GM_STILL_SHUTTING_DOWN):
552 case(GM_NO_SUCH_DEVICE):
554 case(GM_INCOMPATIBLE_LIB_AND_DRIVER):
555 case(GM_UNTRANSLATED_SYSTEM_ERROR):
556 case(GM_ACCESS_DENIED):
557 case(GM_NO_DRIVER_SUPPORT):
558 case(GM_PTE_REF_CNT_OVERFLOW):
559 case(GM_NOT_SUPPORTED_IN_KERNEL):
560 case(GM_NOT_SUPPORTED_ON_ARCH):
563 case(GM_DATA_CORRUPTED):
564 case(GM_HARDWARE_FAULT):
565 case(GM_SEND_ORPHANED):
566 case(GM_MINOR_OVERFLOW):
567 case(GM_PAGE_TABLE_FULL):
569 case(GM_INVALID_PORT_NUMBER):
570 case(GM_DEV_NOT_FOUND):
571 case(GM_FIRMWARE_NOT_RUNNING):
572 case(GM_YP_NO_MATCH):
574 CDEBUG(D_ERROR, "Unknown send error\n");
579 * If this is a large message init,
580 * we're not finished with the data yet,
581 * so can't call lib_finalise.
582 * However, we're also holding on to a
583 * stxd here (to keep track of the source
584 * iovec only). Should use another structure
585 * to keep track of iovec and return stxd to
588 if (stxd->type == GMNAL_LARGE_MESSAGE_INIT) {
589 CDEBUG(D_INFO, "large transmit done\n");
592 gmnal_return_stxd(nal_data, stxd);
593 if (lib_finalize(nal_cb, stxd, cookie) != PTL_OK) {
594 CDEBUG(D_INFO, "Call to lib_finalize failed for stxd [%p]\n",
602 void gmnal_drop_sends_callback(struct gm_port *gm_port, void *context,
605 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
606 gmnal_data_t *nal_data = stxd->nal_data;
608 CDEBUG(D_TRACE, "status is [%d] context is [%p]\n", status, context);
609 if (status == GM_SUCCESS) {
610 GMNAL_GM_LOCK(nal_data);
611 gm_send_to_peer_with_callback(gm_port, stxd->buffer,
612 stxd->gm_size, stxd->msg_size,
614 stxd->gm_target_node,
615 gmnal_small_tx_callback,
617 GMNAL_GM_LOCK(nal_data);
619 CDEBUG(D_ERROR, "send_to_peer status for stxd [%p] is
620 [%d][%s]\n", stxd, status, gmnal_gm_error(status));
629 * Begine a large transmit.
630 * Do a gm_register of the memory pointed to by the iovec
631 * and send details to the receiver. The receiver does a gm_get
632 * to pull the data and sends and ack when finished. Upon receipt of
633 * this ack, deregister the memory. Only 1 send token is required here.
636 gmnal_large_tx(nal_cb_t *nal_cb, void *private, lib_msg_t *cookie,
637 ptl_hdr_t *hdr, int type, ptl_nid_t global_nid, ptl_pid_t pid,
638 unsigned int niov, struct iovec *iov, int size)
641 gmnal_data_t *nal_data;
642 gmnal_stxd_t *stxd = NULL;
644 gmnal_msghdr_t *msghdr = NULL;
645 unsigned int local_nid;
646 int mlen = 0; /* the size of the init message data */
647 struct iovec *iov_dup = NULL;
648 gm_status_t gm_status;
652 CDEBUG(D_TRACE, "gmnal_large_tx nal_cb [%p] private [%p], cookie [%p]
653 hdr [%p], type [%d] global_nid ["LPU64"], pid [%d], niov [%d],
654 iov [%p], size [%d]\n", nal_cb, private, cookie, hdr, type,
655 global_nid, pid, niov, iov, size);
658 nal_data = (gmnal_data_t*)nal_cb->nal_data;
660 CDEBUG(D_ERROR, "no nal_cb.\n");
661 return(GMNAL_STATUS_FAIL);
666 * Get stxd and buffer. Put local address of data in buffer,
667 * send local addresses to target,
668 * wait for the target node to suck the data over.
669 * The stxd is used to ren
671 stxd = gmnal_get_stxd(nal_data, 1);
672 CDEBUG(D_INFO, "stxd [%p]\n", stxd);
674 stxd->type = GMNAL_LARGE_MESSAGE_INIT;
675 stxd->cookie = cookie;
678 * Copy gmnal_msg_hdr and portals header to the transmit buffer
679 * Then copy the iov in
681 buffer = stxd->buffer;
682 msghdr = (gmnal_msghdr_t*)buffer;
684 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
686 msghdr->magic = GMNAL_MAGIC;
687 msghdr->type = GMNAL_LARGE_MESSAGE_INIT;
688 msghdr->sender_node_id = nal_data->gm_global_nid;
690 msghdr->niov = niov ;
691 buffer += sizeof(gmnal_msghdr_t);
692 mlen = sizeof(gmnal_msghdr_t);
693 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
696 CDEBUG(D_INFO, "processing portals hdr at [%p]\n", buffer);
698 gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
699 buffer += sizeof(ptl_hdr_t);
700 mlen += sizeof(ptl_hdr_t);
701 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
704 * copy the iov to the buffer so target knows
705 * where to get the data from
707 CDEBUG(D_INFO, "processing iov to [%p]\n", buffer);
708 gm_bcopy(iov, buffer, niov*sizeof(struct iovec));
709 mlen += niov*(sizeof(struct iovec));
710 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
714 * Store the iovs in the stxd for we can get
715 * them later if we need them
717 CDEBUG(D_NET, "Copying iov [%p] to [%p]\n", iov, stxd->iov);
718 gm_bcopy(iov, stxd->iov, niov*sizeof(struct iovec));
723 * register the memory so the NIC can get hold of the data
724 * This is a slow process. it'd be good to overlap it
725 * with something else.
730 CDEBUG(D_INFO, "Registering memory [%p] len ["LPSZ"] \n",
731 iov->iov_base, iov->iov_len);
732 GMNAL_GM_LOCK(nal_data);
733 gm_status = gm_register_memory(nal_data->gm_port,
734 iov->iov_base, iov->iov_len);
735 if (gm_status != GM_SUCCESS) {
736 GMNAL_GM_UNLOCK(nal_data);
737 CDEBUG(D_ERROR, "gm_register_memory returns [%d][%s]
738 for memory [%p] len ["LPSZ"]\n",
739 gm_status, gmnal_gm_error(gm_status),
740 iov->iov_base, iov->iov_len);
741 GMNAL_GM_LOCK(nal_data);
742 while (iov_dup != iov) {
743 gm_deregister_memory(nal_data->gm_port,
748 GMNAL_GM_UNLOCK(nal_data);
749 gmnal_return_stxd(nal_data, stxd);
753 GMNAL_GM_UNLOCK(nal_data);
758 * Send the init message to the target
760 CDEBUG(D_INFO, "sending mlen [%d]\n", mlen);
761 GMNAL_GM_LOCK(nal_data);
762 gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
764 if (gm_status != GM_SUCCESS) {
765 GMNAL_GM_UNLOCK(nal_data);
766 CDEBUG(D_ERROR, "Failed to obtain local id\n");
767 gmnal_return_stxd(nal_data, stxd);
768 /* TO DO deregister memory on failure */
769 return(GMNAL_STATUS_FAIL);
771 CDEBUG(D_INFO, "Local Node_id is [%d]\n", local_nid);
772 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
773 stxd->gm_size, mlen, GM_LOW_PRIORITY,
774 local_nid, gmnal_large_tx_callback,
776 GMNAL_GM_UNLOCK(nal_data);
778 CDEBUG(D_INFO, "done\n");
784 * Callback function indicates that send of buffer with
785 * large message iovec has completed (or failed).
788 gmnal_large_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
790 gmnal_small_tx_callback(gm_port, context, status);
797 * Have received a buffer that contains an iovec of the sender.
798 * Do a gm_register_memory of the receivers buffer and then do a get
799 * data from the sender.
802 gmnal_large_rx(nal_cb_t *nal_cb, void *private, lib_msg_t *cookie,
803 unsigned int nriov, struct iovec *riov, size_t mlen,
806 gmnal_data_t *nal_data = nal_cb->nal_data;
807 gmnal_srxd_t *srxd = (gmnal_srxd_t*)private;
809 struct iovec *riov_dup;
811 gmnal_msghdr_t *msghdr = NULL;
812 gm_status_t gm_status;
814 CDEBUG(D_TRACE, "gmnal_large_rx :: nal_cb[%p], private[%p],
815 cookie[%p], niov[%d], iov[%p], mlen["LPSZ"], rlen["LPSZ"]\n",
816 nal_cb, private, cookie, nriov, riov, mlen, rlen);
819 CDEBUG(D_ERROR, "gmnal_large_rx no context\n");
820 lib_finalize(nal_cb, private, cookie);
824 buffer = srxd->buffer;
825 msghdr = (gmnal_msghdr_t*)buffer;
826 buffer += sizeof(gmnal_msghdr_t);
827 buffer += sizeof(ptl_hdr_t);
830 * Store the senders stxd address in the srxd for this message
831 * The gmnal_large_message_ack needs it to notify the sender
832 * the pull of data is complete
834 srxd->source_stxd = msghdr->stxd;
837 * Register the receivers memory
839 * tell the sender that we got the data
840 * then tell the receiver we got the data
842 * If the iovecs match, could interleave
843 * gm_registers and gm_gets for each element
848 CDEBUG(D_INFO, "Registering memory [%p] len ["LPSZ"] \n",
849 riov->iov_base, riov->iov_len);
850 GMNAL_GM_LOCK(nal_data);
851 gm_status = gm_register_memory(nal_data->gm_port,
852 riov->iov_base, riov->iov_len);
853 if (gm_status != GM_SUCCESS) {
854 GMNAL_GM_UNLOCK(nal_data);
855 CDEBUG(D_ERROR, "gm_register_memory returns [%d][%s]
856 for memory [%p] len ["LPSZ"]\n",
857 gm_status, gmnal_gm_error(gm_status),
858 riov->iov_base, riov->iov_len);
859 GMNAL_GM_LOCK(nal_data);
860 while (riov_dup != riov) {
861 gm_deregister_memory(nal_data->gm_port,
866 GMNAL_GM_LOCK(nal_data);
868 * give back srxd and buffer. Send NACK to sender
872 GMNAL_GM_UNLOCK(nal_data);
876 * do this so the final gm_get callback can deregister the memory
878 PORTAL_ALLOC(srxd->riov, nriov_dup*(sizeof(struct iovec)));
879 gm_bcopy(riov_dup, srxd->riov, nriov_dup*(sizeof(struct iovec)));
880 srxd->nriov = nriov_dup;
883 * now do gm_get to get the data
885 srxd->cookie = cookie;
886 if (gmnal_remote_get(srxd, srxd->nsiov, (struct iovec*)buffer,
887 nriov_dup, riov_dup) != GMNAL_STATUS_OK) {
888 CDEBUG(D_ERROR, "can't get the data");
891 CDEBUG(D_INFO, "lgmanl_large_rx done\n");
898 * Perform a number of remote gets as part of receiving
900 * The final one to complete (i.e. the last callback to get called)
902 * gm_get requires a send token.
905 gmnal_remote_get(gmnal_srxd_t *srxd, int nsiov, struct iovec *siov,
906 int nriov, struct iovec *riov)
911 CDEBUG(D_TRACE, "gmnal_remote_get srxd[%p], nriov[%d], riov[%p],
912 nsiov[%d], siov[%p]\n", srxd, nriov, riov, nsiov, siov);
915 ncalls = gmnal_copyiov(0, srxd, nsiov, siov, nriov, riov);
917 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
918 return(GMNAL_STATUS_FAIL);
920 CDEBUG(D_INFO, "gmnal_remote_get ncalls [%d]\n", ncalls);
921 spin_lock_init(&srxd->callback_lock);
922 srxd->ncallbacks = ncalls;
923 srxd->callback_status = 0;
925 ncalls = gmnal_copyiov(1, srxd, nsiov, siov, nriov, riov);
927 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
928 return(GMNAL_STATUS_FAIL);
931 return(GMNAL_STATUS_OK);
937 * pull data from source node (source iovec) to a local iovec.
938 * The iovecs may not match which adds the complications below.
939 * Count the number of gm_gets that will be required to the callbacks
940 * can determine who is the last one.
943 gmnal_copyiov(int do_copy, gmnal_srxd_t *srxd, int nsiov,
944 struct iovec *siov, int nriov, struct iovec *riov)
948 int slen = siov->iov_len, rlen = riov->iov_len;
949 char *sbuf = siov->iov_base, *rbuf = riov->iov_base;
950 unsigned long sbuf_long;
951 gm_remote_ptr_t remote_ptr = 0;
952 unsigned int source_node;
953 gmnal_ltxd_t *ltxd = NULL;
954 gmnal_data_t *nal_data = srxd->nal_data;
956 CDEBUG(D_TRACE, "copy[%d] nal_data[%p]\n", do_copy, nal_data);
959 CDEBUG(D_ERROR, "Bad args No nal_data\n");
960 return(GMNAL_STATUS_FAIL);
962 GMNAL_GM_LOCK(nal_data);
963 if (gm_global_id_to_node_id(nal_data->gm_port,
964 srxd->gm_source_node,
965 &source_node) != GM_SUCCESS) {
967 CDEBUG(D_ERROR, "cannot resolve global_id [%u]
968 to local node_id\n", srxd->gm_source_node);
969 GMNAL_GM_UNLOCK(nal_data);
970 return(GMNAL_STATUS_FAIL);
972 GMNAL_GM_UNLOCK(nal_data);
974 * We need a send token to use gm_get
975 * getting an stxd gets us a send token.
976 * the stxd is used as the context to the
977 * callback function (so stxd can be returned).
978 * Set pointer in stxd to srxd so callback count in srxd
979 * can be decremented to find last callback to complete
981 CDEBUG(D_INFO, "gmnal_copyiov source node is G[%u]L[%d]\n",
982 srxd->gm_source_node, source_node);
986 CDEBUG(D_INFO, "sbuf[%p] slen[%d] rbuf[%p], rlen[%d]\n",
987 sbuf, slen, rbuf, rlen);
991 CDEBUG(D_INFO, "slen>rlen\n");
992 ltxd = gmnal_get_ltxd(nal_data);
994 GMNAL_GM_LOCK(nal_data);
996 * funny business to get rid
997 * of compiler warning
999 sbuf_long = (unsigned long) sbuf;
1000 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1001 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1002 rlen, GM_LOW_PRIORITY, source_node,
1004 gmnal_remote_get_callback, ltxd);
1005 GMNAL_GM_UNLOCK(nal_data);
1008 * at the end of 1 iov element
1014 rbuf = riov->iov_base;
1015 rlen = riov->iov_len;
1016 } else if (rlen > slen) {
1019 CDEBUG(D_INFO, "slen<rlen\n");
1020 ltxd = gmnal_get_ltxd(nal_data);
1022 GMNAL_GM_LOCK(nal_data);
1023 sbuf_long = (unsigned long) sbuf;
1024 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1025 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1026 slen, GM_LOW_PRIORITY, source_node,
1028 gmnal_remote_get_callback, ltxd);
1029 GMNAL_GM_UNLOCK(nal_data);
1032 * at end of siov element
1037 sbuf = siov->iov_base;
1038 slen = siov->iov_len;
1042 CDEBUG(D_INFO, "rlen=slen\n");
1043 ltxd = gmnal_get_ltxd(nal_data);
1045 GMNAL_GM_LOCK(nal_data);
1046 sbuf_long = (unsigned long) sbuf;
1047 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1048 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1049 rlen, GM_LOW_PRIORITY, source_node,
1051 gmnal_remote_get_callback, ltxd);
1052 GMNAL_GM_UNLOCK(nal_data);
1055 * at end of siov and riov element
1058 sbuf = siov->iov_base;
1059 slen = siov->iov_len;
1062 rbuf = riov->iov_base;
1063 rlen = riov->iov_len;
1072 * The callback function that is invoked after each gm_get call completes.
1073 * Multiple callbacks may be invoked for 1 transaction, only the final
1074 * callback has work to do.
1077 gmnal_remote_get_callback(gm_port_t *gm_port, void *context,
1081 gmnal_ltxd_t *ltxd = (gmnal_ltxd_t*)context;
1082 gmnal_srxd_t *srxd = ltxd->srxd;
1083 nal_cb_t *nal_cb = srxd->nal_data->nal_cb;
1087 gmnal_data_t *nal_data;
1089 CDEBUG(D_TRACE, "called for context [%p]\n", context);
1091 if (status != GM_SUCCESS) {
1092 CDEBUG(D_ERROR, "reports error [%d][%s]\n", status,
1093 gmnal_gm_error(status));
1096 spin_lock(&srxd->callback_lock);
1098 srxd->callback_status |= status;
1099 lastone = srxd->ncallbacks?0:1;
1100 spin_unlock(&srxd->callback_lock);
1101 nal_data = srxd->nal_data;
1104 * everyone returns a send token
1106 gmnal_return_ltxd(nal_data, ltxd);
1109 CDEBUG(D_ERROR, "NOT final callback context[%p]\n", srxd);
1114 * Let our client application proceed
1116 CDEBUG(D_ERROR, "final callback context[%p]\n", srxd);
1117 if (lib_finalize(nal_cb, srxd, srxd->cookie) != PTL_OK) {
1118 CDEBUG(D_INFO, "Call to lib_finalize failed for srxd [%p]\n",
1123 * send an ack to the sender to let him know we got the data
1125 gmnal_large_tx_ack(nal_data, srxd);
1128 * Unregister the memory that was used
1129 * This is a very slow business (slower then register)
1131 nriov = srxd->nriov;
1133 GMNAL_GM_LOCK(nal_data);
1135 CDEBUG(D_ERROR, "deregister memory [%p]\n", riov->iov_base);
1136 if (gm_deregister_memory(srxd->nal_data->gm_port,
1137 riov->iov_base, riov->iov_len)) {
1138 CDEBUG(D_ERROR, "failed to deregister memory [%p]\n",
1143 GMNAL_GM_UNLOCK(nal_data);
1144 PORTAL_FREE(srxd->riov, sizeof(struct iovec)*nriov);
1147 * repost the receive buffer (return receive token)
1149 GMNAL_GM_LOCK(nal_data);
1150 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
1151 srxd->gmsize, GM_LOW_PRIORITY, 0);
1152 GMNAL_GM_UNLOCK(nal_data);
1159 * Called on target node.
1160 * After pulling data from a source node
1161 * send an ack message to indicate the large transmit is complete.
1164 gmnal_large_tx_ack(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1168 gmnal_msghdr_t *msghdr;
1169 void *buffer = NULL;
1170 unsigned int local_nid;
1171 gm_status_t gm_status = GM_SUCCESS;
1173 CDEBUG(D_TRACE, "srxd[%p] target_node [%u]\n", srxd,
1174 srxd->gm_source_node);
1176 GMNAL_GM_LOCK(nal_data);
1177 gm_status = gm_global_id_to_node_id(nal_data->gm_port,
1178 srxd->gm_source_node, &local_nid);
1179 GMNAL_GM_UNLOCK(nal_data);
1180 if (gm_status != GM_SUCCESS) {
1181 CDEBUG(D_ERROR, "Failed to obtain local id\n");
1184 CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
1186 stxd = gmnal_get_stxd(nal_data, 1);
1187 CDEBUG(D_TRACE, "gmnal_large_tx_ack got stxd[%p]\n", stxd);
1189 stxd->nal_data = nal_data;
1190 stxd->type = GMNAL_LARGE_MESSAGE_ACK;
1193 * Copy gmnal_msg_hdr and portals header to the transmit buffer
1194 * Then copy the data in
1196 buffer = stxd->buffer;
1197 msghdr = (gmnal_msghdr_t*)buffer;
1200 * Add in the address of the original stxd from the sender node
1201 * so it knows which thread to notify.
1203 msghdr->magic = GMNAL_MAGIC;
1204 msghdr->type = GMNAL_LARGE_MESSAGE_ACK;
1205 msghdr->sender_node_id = nal_data->gm_global_nid;
1206 msghdr->stxd = srxd->source_stxd;
1207 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
1209 CDEBUG(D_INFO, "sending\n");
1210 stxd->msg_size= sizeof(gmnal_msghdr_t);
1213 CDEBUG(D_NET, "Calling gm_send_to_peer port [%p] buffer [%p]
1214 gmsize [%lu] msize [%d] global_nid [%u] local_nid[%d]
1215 stxd [%p]\n", nal_data->gm_port, stxd->buffer, stxd->gm_size,
1216 stxd->msg_size, srxd->gm_source_node, local_nid, stxd);
1217 GMNAL_GM_LOCK(nal_data);
1218 stxd->gm_priority = GM_LOW_PRIORITY;
1219 stxd->gm_target_node = local_nid;
1220 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
1221 stxd->gm_size, stxd->msg_size,
1222 GM_LOW_PRIORITY, local_nid,
1223 gmnal_large_tx_ack_callback,
1226 GMNAL_GM_UNLOCK(nal_data);
1227 CDEBUG(D_INFO, "gmnal_large_tx_ack :: done\n");
1234 * A callback to indicate the small transmit operation is compete
1235 * Check for errors and try to deal with them.
1236 * Call lib_finalise to inform the client application that the
1237 * send is complete and the memory can be reused.
1238 * Return the stxd when finished with it (returns a send token)
1241 gmnal_large_tx_ack_callback(gm_port_t *gm_port, void *context,
1244 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
1245 gmnal_data_t *nal_data = (gmnal_data_t*)stxd->nal_data;
1248 CDEBUG(D_ERROR, "send completion event for unknown stxd\n");
1251 CDEBUG(D_TRACE, "send completion event for stxd [%p] status is [%d]\n",
1253 gmnal_return_stxd(stxd->nal_data, stxd);
1255 GMNAL_GM_UNLOCK(nal_data);
1260 * Indicates the large transmit operation is compete.
1261 * Called on transmit side (means data has been pulled by receiver
1263 * Call lib_finalise to inform the client application that the send
1264 * is complete, deregister the memory and return the stxd.
1265 * Finally, report the rx buffer that the ack message was delivered in.
1268 gmnal_large_tx_ack_received(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1270 nal_cb_t *nal_cb = nal_data->nal_cb;
1271 gmnal_stxd_t *stxd = NULL;
1272 gmnal_msghdr_t *msghdr = NULL;
1273 void *buffer = NULL;
1277 CDEBUG(D_TRACE, "gmnal_large_tx_ack_received buffer [%p]\n", buffer);
1279 buffer = srxd->buffer;
1280 msghdr = (gmnal_msghdr_t*)buffer;
1281 stxd = msghdr->stxd;
1283 CDEBUG(D_INFO, "gmnal_large_tx_ack_received stxd [%p]\n", stxd);
1285 if (lib_finalize(nal_cb, stxd, stxd->cookie) != PTL_OK) {
1286 CDEBUG(D_INFO, "Call to lib_finalize failed for stxd [%p]\n",
1291 * extract the iovec from the stxd, deregister the memory.
1292 * free the space used to store the iovec
1295 while(stxd->niov--) {
1296 CDEBUG(D_INFO, "deregister memory [%p] size ["LPSZ"]\n",
1297 iov->iov_base, iov->iov_len);
1298 GMNAL_GM_LOCK(nal_data);
1299 gm_deregister_memory(nal_data->gm_port, iov->iov_base,
1301 GMNAL_GM_UNLOCK(nal_data);
1306 * return the send token
1307 * TO DO It is bad to hold onto the send token so long?
1309 gmnal_return_stxd(nal_data, stxd);
1313 * requeue the receive buffer
1315 gmnal_rx_requeue_buffer(nal_data, srxd);