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;
194 CDEBUG(D_INFO, "nal_data [%p], we[%p] type [%d]\n",
195 nal_data, we, gmnal_type);
204 gmnal_msghdr = (gmnal_msghdr_t*)buffer;
205 portals_hdr = (ptl_hdr_t*)(buffer+GMNAL_MSGHDR_SIZE);
207 CDEBUG(D_INFO, "rx_event:: Sender node [%d], Sender Port [%d], "
208 "type [%d], length [%d], buffer [%p]\n",
209 snode, sport, type, length, buffer);
210 CDEBUG(D_INFO, "gmnal_msghdr:: Sender node [%u], magic [%d], "
211 "gmnal_type [%d]\n", gmnal_msghdr->sender_node_id,
212 gmnal_msghdr->magic, gmnal_msghdr->type);
213 CDEBUG(D_INFO, "portals_hdr:: Sender node ["LPD64"], "
214 "dest_node ["LPD64"]\n", portals_hdr->src_nid,
215 portals_hdr->dest_nid);
219 * Get a receive descriptor for this message
221 srxd = gmnal_rxbuffer_to_srxd(nal_data, buffer);
222 CDEBUG(D_INFO, "Back from gmnal_rxbuffer_to_srxd\n");
224 CDEBUG(D_ERROR, "Failed to get receive descriptor\n");
225 /* I think passing a NULL srxd to lib_parse will crash
228 lib_parse(nal_data->libnal, portals_hdr, srxd);
229 return(GMNAL_STATUS_FAIL);
233 * no need to bother portals library with this
235 if (gmnal_type == GMNAL_LARGE_MESSAGE_ACK) {
236 gmnal_large_tx_ack_received(nal_data, srxd);
237 return(GMNAL_STATUS_OK);
240 srxd->nal_data = nal_data;
241 srxd->type = gmnal_type;
242 srxd->nsiov = gmnal_msghdr->niov;
243 srxd->gm_source_node = gmnal_msghdr->sender_node_id;
245 CDEBUG(D_PORTALS, "Calling lib_parse buffer is [%p]\n",
246 buffer+GMNAL_MSGHDR_SIZE);
248 * control passes to lib, which calls cb_recv
249 * cb_recv is responsible for returning the buffer
252 rc = lib_parse(nal_data->libnal, portals_hdr, srxd);
255 /* I just received garbage; take appropriate action... */
259 return(GMNAL_STATUS_OK);
265 * After a receive has been processed,
266 * hang out the receive buffer again.
267 * This implicitly returns a receive token.
270 gmnal_rx_requeue_buffer(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
272 CDEBUG(D_TRACE, "gmnal_rx_requeue_buffer\n");
274 CDEBUG(D_NET, "requeueing srxd[%p] nal_data[%p]\n", srxd, nal_data);
276 GMNAL_GM_LOCK(nal_data);
277 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
278 srxd->gmsize, GM_LOW_PRIORITY, 0 );
279 GMNAL_GM_UNLOCK(nal_data);
281 return(GMNAL_STATUS_OK);
286 * Handle a bad message
287 * A bad message is one we don't expect or can't interpret
290 gmnal_rx_bad(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, gmnal_srxd_t *srxd)
292 CDEBUG(D_TRACE, "Can't handle message\n");
295 srxd = gmnal_rxbuffer_to_srxd(nal_data,
298 gmnal_rx_requeue_buffer(nal_data, srxd);
300 CDEBUG(D_ERROR, "Can't find a descriptor for this buffer\n");
304 return(GMNAL_STATUS_FAIL);
307 return(GMNAL_STATUS_OK);
313 * Process a small message receive.
314 * Get here from gmnal_receive_thread, gmnal_pre_receive
316 * Put data from prewired receive buffer into users buffer(s)
317 * Hang out the receive buffer again for another receive
321 gmnal_small_rx(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
322 unsigned int niov, struct iovec *iov, size_t mlen, size_t rlen)
324 gmnal_srxd_t *srxd = NULL;
326 gmnal_data_t *nal_data = (gmnal_data_t*)libnal->nal_data;
329 CDEBUG(D_TRACE, "niov [%d] mlen["LPSZ"]\n", niov, mlen);
332 CDEBUG(D_ERROR, "gmnal_small_rx no context\n");
333 lib_finalize(libnal, private, cookie, PTL_FAIL);
337 srxd = (gmnal_srxd_t*)private;
338 buffer = srxd->buffer;
339 buffer += sizeof(gmnal_msghdr_t);
340 buffer += sizeof(ptl_hdr_t);
343 CDEBUG(D_INFO, "processing [%p] len ["LPSZ"]\n", iov,
345 gm_bcopy(buffer, iov->iov_base, iov->iov_len);
346 buffer += iov->iov_len;
352 * let portals library know receive is complete
354 CDEBUG(D_PORTALS, "calling lib_finalize\n");
355 lib_finalize(libnal, private, cookie, PTL_OK);
357 * return buffer so it can be used again
359 CDEBUG(D_NET, "calling gm_provide_receive_buffer\n");
360 GMNAL_GM_LOCK(nal_data);
361 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
362 srxd->gmsize, GM_LOW_PRIORITY, 0);
363 GMNAL_GM_UNLOCK(nal_data);
370 * Start a small transmit.
371 * Get a send token (and wired transmit buffer).
372 * Copy data from senders buffer to wired buffer and
373 * initiate gm_send from the wired buffer.
374 * The callback function informs when the send is complete.
377 gmnal_small_tx(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
378 ptl_hdr_t *hdr, int type, ptl_nid_t global_nid, ptl_pid_t pid,
379 unsigned int niov, struct iovec *iov, int size)
381 gmnal_data_t *nal_data = (gmnal_data_t*)libnal->nal_data;
382 gmnal_stxd_t *stxd = NULL;
384 gmnal_msghdr_t *msghdr = NULL;
386 unsigned int local_nid;
387 gm_status_t gm_status = GM_SUCCESS;
389 CDEBUG(D_TRACE, "gmnal_small_tx libnal [%p] private [%p] cookie [%p] "
390 "hdr [%p] type [%d] global_nid ["LPU64"] pid [%d] niov [%d] "
391 "iov [%p] size [%d]\n", libnal, private, cookie, hdr, type,
392 global_nid, pid, niov, iov, size);
394 CDEBUG(D_INFO, "portals_hdr:: dest_nid ["LPU64"], src_nid ["LPU64"]\n",
395 hdr->dest_nid, hdr->src_nid);
398 CDEBUG(D_ERROR, "no nal_data\n");
399 return(GMNAL_STATUS_FAIL);
401 CDEBUG(D_INFO, "nal_data [%p]\n", nal_data);
404 GMNAL_GM_LOCK(nal_data);
405 gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
407 GMNAL_GM_UNLOCK(nal_data);
408 if (gm_status != GM_SUCCESS) {
409 CDEBUG(D_ERROR, "Failed to obtain local id\n");
410 return(GMNAL_STATUS_FAIL);
412 CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
414 stxd = gmnal_get_stxd(nal_data, 1);
415 CDEBUG(D_INFO, "stxd [%p]\n", stxd);
417 stxd->type = GMNAL_SMALL_MESSAGE;
418 stxd->cookie = cookie;
421 * Copy gmnal_msg_hdr and portals header to the transmit buffer
422 * Then copy the data in
424 buffer = stxd->buffer;
425 msghdr = (gmnal_msghdr_t*)buffer;
427 msghdr->magic = GMNAL_MAGIC;
428 msghdr->type = GMNAL_SMALL_MESSAGE;
429 msghdr->sender_node_id = nal_data->gm_global_nid;
430 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
432 buffer += sizeof(gmnal_msghdr_t);
434 CDEBUG(D_INFO, "processing portals hdr at [%p]\n", buffer);
435 gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
437 buffer += sizeof(ptl_hdr_t);
440 CDEBUG(D_INFO, "processing iov [%p] len ["LPSZ"] to [%p]\n",
441 iov, iov->iov_len, buffer);
442 gm_bcopy(iov->iov_base, buffer, iov->iov_len);
443 buffer+= iov->iov_len;
447 CDEBUG(D_INFO, "sending\n");
448 tot_size = size+sizeof(ptl_hdr_t)+sizeof(gmnal_msghdr_t);
449 stxd->msg_size = tot_size;
452 CDEBUG(D_NET, "Calling gm_send_to_peer port [%p] buffer [%p] "
453 "gmsize [%lu] msize [%d] global_nid ["LPU64"] local_nid[%d] "
454 "stxd [%p]\n", nal_data->gm_port, stxd->buffer, stxd->gm_size,
455 stxd->msg_size, global_nid, local_nid, stxd);
457 GMNAL_GM_LOCK(nal_data);
458 stxd->gm_priority = GM_LOW_PRIORITY;
459 stxd->gm_target_node = local_nid;
460 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
461 stxd->gm_size, stxd->msg_size,
462 GM_LOW_PRIORITY, local_nid,
463 gmnal_small_tx_callback, (void*)stxd);
464 GMNAL_GM_UNLOCK(nal_data);
465 CDEBUG(D_INFO, "done\n");
472 * A callback to indicate the small transmit operation is compete
473 * Check for erros and try to deal with them.
474 * Call lib_finalise to inform the client application that the send
475 * is complete and the memory can be reused.
476 * Return the stxd when finished with it (returns a send token)
479 gmnal_small_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
481 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
482 lib_msg_t *cookie = stxd->cookie;
483 gmnal_data_t *nal_data = (gmnal_data_t*)stxd->nal_data;
484 lib_nal_t *libnal = nal_data->libnal;
487 CDEBUG(D_TRACE, "send completion event for unknown stxd\n");
490 if (status != GM_SUCCESS) {
491 CDEBUG(D_ERROR, "Result of send stxd [%p] is [%s]\n",
492 stxd, gmnal_gm_error(status));
501 case(GM_SEND_DROPPED):
503 * do a resend on the dropped ones
505 CDEBUG(D_ERROR, "send stxd [%p] was dropped "
506 "resending\n", context);
507 GMNAL_GM_LOCK(nal_data);
508 gm_send_to_peer_with_callback(nal_data->gm_port,
513 stxd->gm_target_node,
514 gmnal_small_tx_callback,
516 GMNAL_GM_UNLOCK(nal_data);
520 case(GM_SEND_TIMED_OUT):
524 CDEBUG(D_INFO, "calling gm_drop_sends\n");
525 GMNAL_GM_LOCK(nal_data);
526 gm_drop_sends(nal_data->gm_port, stxd->gm_priority,
527 stxd->gm_target_node, GMNAL_GM_PORT,
528 gmnal_drop_sends_callback, context);
529 GMNAL_GM_UNLOCK(nal_data);
538 case(GM_INTERRUPTED):
540 case(GM_INPUT_BUFFER_TOO_SMALL):
541 case(GM_OUTPUT_BUFFER_TOO_SMALL):
543 case(GM_MEMORY_FAULT):
544 case(GM_INVALID_PARAMETER):
545 case(GM_OUT_OF_MEMORY):
546 case(GM_INVALID_COMMAND):
547 case(GM_PERMISSION_DENIED):
548 case(GM_INTERNAL_ERROR):
550 case(GM_UNSUPPORTED_DEVICE):
551 case(GM_SEND_REJECTED):
552 case(GM_SEND_TARGET_PORT_CLOSED):
553 case(GM_SEND_TARGET_NODE_UNREACHABLE):
554 case(GM_SEND_PORT_CLOSED):
555 case(GM_NODE_ID_NOT_YET_SET):
556 case(GM_STILL_SHUTTING_DOWN):
558 case(GM_NO_SUCH_DEVICE):
560 case(GM_INCOMPATIBLE_LIB_AND_DRIVER):
561 case(GM_UNTRANSLATED_SYSTEM_ERROR):
562 case(GM_ACCESS_DENIED):
563 case(GM_NO_DRIVER_SUPPORT):
564 case(GM_PTE_REF_CNT_OVERFLOW):
565 case(GM_NOT_SUPPORTED_IN_KERNEL):
566 case(GM_NOT_SUPPORTED_ON_ARCH):
569 case(GM_DATA_CORRUPTED):
570 case(GM_HARDWARE_FAULT):
571 case(GM_SEND_ORPHANED):
572 case(GM_MINOR_OVERFLOW):
573 case(GM_PAGE_TABLE_FULL):
575 case(GM_INVALID_PORT_NUMBER):
576 case(GM_DEV_NOT_FOUND):
577 case(GM_FIRMWARE_NOT_RUNNING):
578 case(GM_YP_NO_MATCH):
580 CDEBUG(D_ERROR, "Unknown send error\n");
581 gm_resume_sending(nal_data->gm_port, stxd->gm_priority,
582 stxd->gm_target_node, GMNAL_GM_PORT,
583 gmnal_resume_sending_callback, context);
590 * If this is a large message init,
591 * we're not finished with the data yet,
592 * so can't call lib_finalise.
593 * However, we're also holding on to a
594 * stxd here (to keep track of the source
595 * iovec only). Should use another structure
596 * to keep track of iovec and return stxd to
599 if (stxd->type == GMNAL_LARGE_MESSAGE_INIT) {
600 CDEBUG(D_INFO, "large transmit done\n");
603 gmnal_return_stxd(nal_data, stxd);
604 lib_finalize(libnal, stxd, cookie, PTL_OK);
609 * After an error on the port
610 * call this to allow future sends to complete
612 void gmnal_resume_sending_callback(struct gm_port *gm_port, void *context,
615 gmnal_data_t *nal_data;
616 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
617 CDEBUG(D_TRACE, "status is [%d] context is [%p]\n", status, context);
618 gmnal_return_stxd(stxd->nal_data, stxd);
623 void gmnal_drop_sends_callback(struct gm_port *gm_port, void *context,
626 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
627 gmnal_data_t *nal_data = stxd->nal_data;
629 CDEBUG(D_TRACE, "status is [%d] context is [%p]\n", status, context);
630 if (status == GM_SUCCESS) {
631 GMNAL_GM_LOCK(nal_data);
632 gm_send_to_peer_with_callback(gm_port, stxd->buffer,
633 stxd->gm_size, stxd->msg_size,
635 stxd->gm_target_node,
636 gmnal_small_tx_callback,
638 GMNAL_GM_LOCK(nal_data);
640 CDEBUG(D_ERROR, "send_to_peer status for stxd [%p] is "
641 "[%d][%s]\n", stxd, status, gmnal_gm_error(status));
650 * Begine a large transmit.
651 * Do a gm_register of the memory pointed to by the iovec
652 * and send details to the receiver. The receiver does a gm_get
653 * to pull the data and sends and ack when finished. Upon receipt of
654 * this ack, deregister the memory. Only 1 send token is required here.
657 gmnal_large_tx(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
658 ptl_hdr_t *hdr, int type, ptl_nid_t global_nid, ptl_pid_t pid,
659 unsigned int niov, struct iovec *iov, int size)
662 gmnal_data_t *nal_data;
663 gmnal_stxd_t *stxd = NULL;
665 gmnal_msghdr_t *msghdr = NULL;
666 unsigned int local_nid;
667 int mlen = 0; /* the size of the init message data */
668 struct iovec *iov_dup = NULL;
669 gm_status_t gm_status;
673 CDEBUG(D_TRACE, "gmnal_large_tx libnal [%p] private [%p], cookie [%p] "
674 "hdr [%p], type [%d] global_nid ["LPU64"], pid [%d], niov [%d], "
675 "iov [%p], size [%d]\n", libnal, private, cookie, hdr, type,
676 global_nid, pid, niov, iov, size);
679 nal_data = (gmnal_data_t*)libnal->nal_data;
681 CDEBUG(D_ERROR, "no libnal.\n");
682 return(GMNAL_STATUS_FAIL);
687 * Get stxd and buffer. Put local address of data in buffer,
688 * send local addresses to target,
689 * wait for the target node to suck the data over.
690 * The stxd is used to ren
692 stxd = gmnal_get_stxd(nal_data, 1);
693 CDEBUG(D_INFO, "stxd [%p]\n", stxd);
695 stxd->type = GMNAL_LARGE_MESSAGE_INIT;
696 stxd->cookie = cookie;
699 * Copy gmnal_msg_hdr and portals header to the transmit buffer
700 * Then copy the iov in
702 buffer = stxd->buffer;
703 msghdr = (gmnal_msghdr_t*)buffer;
705 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
707 msghdr->magic = GMNAL_MAGIC;
708 msghdr->type = GMNAL_LARGE_MESSAGE_INIT;
709 msghdr->sender_node_id = nal_data->gm_global_nid;
711 msghdr->niov = niov ;
712 buffer += sizeof(gmnal_msghdr_t);
713 mlen = sizeof(gmnal_msghdr_t);
714 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
717 CDEBUG(D_INFO, "processing portals hdr at [%p]\n", buffer);
719 gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
720 buffer += sizeof(ptl_hdr_t);
721 mlen += sizeof(ptl_hdr_t);
722 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
725 * copy the iov to the buffer so target knows
726 * where to get the data from
728 CDEBUG(D_INFO, "processing iov to [%p]\n", buffer);
729 gm_bcopy(iov, buffer, niov*sizeof(struct iovec));
730 mlen += niov*(sizeof(struct iovec));
731 CDEBUG(D_INFO, "mlen is [%d]\n", mlen);
735 * Store the iovs in the stxd for we can get
736 * them later if we need them
738 CDEBUG(D_NET, "Copying iov [%p] to [%p]\n", iov, stxd->iov);
739 gm_bcopy(iov, stxd->iov, niov*sizeof(struct iovec));
744 * register the memory so the NIC can get hold of the data
745 * This is a slow process. it'd be good to overlap it
746 * with something else.
751 CDEBUG(D_INFO, "Registering memory [%p] len ["LPSZ"] \n",
752 iov->iov_base, iov->iov_len);
753 GMNAL_GM_LOCK(nal_data);
754 gm_status = gm_register_memory(nal_data->gm_port,
755 iov->iov_base, iov->iov_len);
756 if (gm_status != GM_SUCCESS) {
757 GMNAL_GM_UNLOCK(nal_data);
758 CDEBUG(D_ERROR, "gm_register_memory returns [%d][%s] "
759 "for memory [%p] len ["LPSZ"]\n",
760 gm_status, gmnal_gm_error(gm_status),
761 iov->iov_base, iov->iov_len);
762 GMNAL_GM_LOCK(nal_data);
763 while (iov_dup != iov) {
764 gm_deregister_memory(nal_data->gm_port,
769 GMNAL_GM_UNLOCK(nal_data);
770 gmnal_return_stxd(nal_data, stxd);
774 GMNAL_GM_UNLOCK(nal_data);
779 * Send the init message to the target
781 CDEBUG(D_INFO, "sending mlen [%d]\n", mlen);
782 GMNAL_GM_LOCK(nal_data);
783 gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
785 if (gm_status != GM_SUCCESS) {
786 GMNAL_GM_UNLOCK(nal_data);
787 CDEBUG(D_ERROR, "Failed to obtain local id\n");
788 gmnal_return_stxd(nal_data, stxd);
789 /* TO DO deregister memory on failure */
790 return(GMNAL_STATUS_FAIL);
792 CDEBUG(D_INFO, "Local Node_id is [%d]\n", local_nid);
793 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
794 stxd->gm_size, mlen, GM_LOW_PRIORITY,
795 local_nid, gmnal_large_tx_callback,
797 GMNAL_GM_UNLOCK(nal_data);
799 CDEBUG(D_INFO, "done\n");
805 * Callback function indicates that send of buffer with
806 * large message iovec has completed (or failed).
809 gmnal_large_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
811 gmnal_small_tx_callback(gm_port, context, status);
818 * Have received a buffer that contains an iovec of the sender.
819 * Do a gm_register_memory of the receivers buffer and then do a get
820 * data from the sender.
823 gmnal_large_rx(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
824 unsigned int nriov, struct iovec *riov, size_t mlen,
827 gmnal_data_t *nal_data = libnal->nal_data;
828 gmnal_srxd_t *srxd = (gmnal_srxd_t*)private;
830 struct iovec *riov_dup;
832 gmnal_msghdr_t *msghdr = NULL;
833 gm_status_t gm_status;
835 CDEBUG(D_TRACE, "gmnal_large_rx :: libnal[%p], private[%p], "
836 "cookie[%p], niov[%d], iov[%p], mlen["LPSZ"], rlen["LPSZ"]\n",
837 libnal, private, cookie, nriov, riov, mlen, rlen);
840 CDEBUG(D_ERROR, "gmnal_large_rx no context\n");
841 lib_finalize(libnal, private, cookie, PTL_FAIL);
845 buffer = srxd->buffer;
846 msghdr = (gmnal_msghdr_t*)buffer;
847 buffer += sizeof(gmnal_msghdr_t);
848 buffer += sizeof(ptl_hdr_t);
851 * Store the senders stxd address in the srxd for this message
852 * The gmnal_large_message_ack needs it to notify the sender
853 * the pull of data is complete
855 srxd->source_stxd = msghdr->stxd;
858 * Register the receivers memory
860 * tell the sender that we got the data
861 * then tell the receiver we got the data
863 * If the iovecs match, could interleave
864 * gm_registers and gm_gets for each element
869 CDEBUG(D_INFO, "Registering memory [%p] len ["LPSZ"] \n",
870 riov->iov_base, riov->iov_len);
871 GMNAL_GM_LOCK(nal_data);
872 gm_status = gm_register_memory(nal_data->gm_port,
873 riov->iov_base, riov->iov_len);
874 if (gm_status != GM_SUCCESS) {
875 GMNAL_GM_UNLOCK(nal_data);
876 CDEBUG(D_ERROR, "gm_register_memory returns [%d][%s] "
877 "for memory [%p] len ["LPSZ"]\n",
878 gm_status, gmnal_gm_error(gm_status),
879 riov->iov_base, riov->iov_len);
880 GMNAL_GM_LOCK(nal_data);
881 while (riov_dup != riov) {
882 gm_deregister_memory(nal_data->gm_port,
887 GMNAL_GM_LOCK(nal_data);
889 * give back srxd and buffer. Send NACK to sender
893 GMNAL_GM_UNLOCK(nal_data);
897 * do this so the final gm_get callback can deregister the memory
899 PORTAL_ALLOC(srxd->riov, nriov_dup*(sizeof(struct iovec)));
900 gm_bcopy(riov_dup, srxd->riov, nriov_dup*(sizeof(struct iovec)));
901 srxd->nriov = nriov_dup;
904 * now do gm_get to get the data
906 srxd->cookie = cookie;
907 if (gmnal_remote_get(srxd, srxd->nsiov, (struct iovec*)buffer,
908 nriov_dup, riov_dup) != GMNAL_STATUS_OK) {
909 CDEBUG(D_ERROR, "can't get the data");
912 CDEBUG(D_INFO, "lgmanl_large_rx done\n");
919 * Perform a number of remote gets as part of receiving
921 * The final one to complete (i.e. the last callback to get called)
923 * gm_get requires a send token.
926 gmnal_remote_get(gmnal_srxd_t *srxd, int nsiov, struct iovec *siov,
927 int nriov, struct iovec *riov)
932 CDEBUG(D_TRACE, "gmnal_remote_get srxd[%p], nriov[%d], riov[%p], "
933 "nsiov[%d], siov[%p]\n", srxd, nriov, riov, nsiov, siov);
936 ncalls = gmnal_copyiov(0, srxd, nsiov, siov, nriov, riov);
938 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
939 return(GMNAL_STATUS_FAIL);
941 CDEBUG(D_INFO, "gmnal_remote_get ncalls [%d]\n", ncalls);
942 spin_lock_init(&srxd->callback_lock);
943 srxd->ncallbacks = ncalls;
944 srxd->callback_status = 0;
946 ncalls = gmnal_copyiov(1, srxd, nsiov, siov, nriov, riov);
948 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
949 return(GMNAL_STATUS_FAIL);
952 return(GMNAL_STATUS_OK);
958 * pull data from source node (source iovec) to a local iovec.
959 * The iovecs may not match which adds the complications below.
960 * Count the number of gm_gets that will be required to the callbacks
961 * can determine who is the last one.
964 gmnal_copyiov(int do_copy, gmnal_srxd_t *srxd, int nsiov,
965 struct iovec *siov, int nriov, struct iovec *riov)
969 int slen = siov->iov_len, rlen = riov->iov_len;
970 char *sbuf = siov->iov_base, *rbuf = riov->iov_base;
971 unsigned long sbuf_long;
972 gm_remote_ptr_t remote_ptr = 0;
973 unsigned int source_node;
974 gmnal_ltxd_t *ltxd = NULL;
975 gmnal_data_t *nal_data = srxd->nal_data;
977 CDEBUG(D_TRACE, "copy[%d] nal_data[%p]\n", do_copy, nal_data);
980 CDEBUG(D_ERROR, "Bad args No nal_data\n");
981 return(GMNAL_STATUS_FAIL);
983 GMNAL_GM_LOCK(nal_data);
984 if (gm_global_id_to_node_id(nal_data->gm_port,
985 srxd->gm_source_node,
986 &source_node) != GM_SUCCESS) {
988 CDEBUG(D_ERROR, "cannot resolve global_id [%u] "
989 "to local node_id\n", srxd->gm_source_node);
990 GMNAL_GM_UNLOCK(nal_data);
991 return(GMNAL_STATUS_FAIL);
993 GMNAL_GM_UNLOCK(nal_data);
995 * We need a send token to use gm_get
996 * getting an stxd gets us a send token.
997 * the stxd is used as the context to the
998 * callback function (so stxd can be returned).
999 * Set pointer in stxd to srxd so callback count in srxd
1000 * can be decremented to find last callback to complete
1002 CDEBUG(D_INFO, "gmnal_copyiov source node is G[%u]L[%d]\n",
1003 srxd->gm_source_node, source_node);
1007 CDEBUG(D_INFO, "sbuf[%p] slen[%d] rbuf[%p], rlen[%d]\n",
1008 sbuf, slen, rbuf, rlen);
1012 CDEBUG(D_INFO, "slen>rlen\n");
1013 ltxd = gmnal_get_ltxd(nal_data);
1015 GMNAL_GM_LOCK(nal_data);
1017 * funny business to get rid
1018 * of compiler warning
1020 sbuf_long = (unsigned long) sbuf;
1021 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1022 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1023 rlen, GM_LOW_PRIORITY, source_node,
1025 gmnal_remote_get_callback, ltxd);
1026 GMNAL_GM_UNLOCK(nal_data);
1029 * at the end of 1 iov element
1035 rbuf = riov->iov_base;
1036 rlen = riov->iov_len;
1037 } else if (rlen > slen) {
1040 CDEBUG(D_INFO, "slen<rlen\n");
1041 ltxd = gmnal_get_ltxd(nal_data);
1043 GMNAL_GM_LOCK(nal_data);
1044 sbuf_long = (unsigned long) sbuf;
1045 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1046 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1047 slen, GM_LOW_PRIORITY, source_node,
1049 gmnal_remote_get_callback, ltxd);
1050 GMNAL_GM_UNLOCK(nal_data);
1053 * at end of siov element
1058 sbuf = siov->iov_base;
1059 slen = siov->iov_len;
1063 CDEBUG(D_INFO, "rlen=slen\n");
1064 ltxd = gmnal_get_ltxd(nal_data);
1066 GMNAL_GM_LOCK(nal_data);
1067 sbuf_long = (unsigned long) sbuf;
1068 remote_ptr = (gm_remote_ptr_t)sbuf_long;
1069 gm_get(nal_data->gm_port, remote_ptr, rbuf,
1070 rlen, GM_LOW_PRIORITY, source_node,
1072 gmnal_remote_get_callback, ltxd);
1073 GMNAL_GM_UNLOCK(nal_data);
1076 * at end of siov and riov element
1079 sbuf = siov->iov_base;
1080 slen = siov->iov_len;
1083 rbuf = riov->iov_base;
1084 rlen = riov->iov_len;
1093 * The callback function that is invoked after each gm_get call completes.
1094 * Multiple callbacks may be invoked for 1 transaction, only the final
1095 * callback has work to do.
1098 gmnal_remote_get_callback(gm_port_t *gm_port, void *context,
1102 gmnal_ltxd_t *ltxd = (gmnal_ltxd_t*)context;
1103 gmnal_srxd_t *srxd = ltxd->srxd;
1104 lib_nal_t *libnal = srxd->nal_data->libnal;
1108 gmnal_data_t *nal_data;
1110 CDEBUG(D_TRACE, "called for context [%p]\n", context);
1112 if (status != GM_SUCCESS) {
1113 CDEBUG(D_ERROR, "reports error [%d][%s]\n", status,
1114 gmnal_gm_error(status));
1117 spin_lock(&srxd->callback_lock);
1119 srxd->callback_status |= status;
1120 lastone = srxd->ncallbacks?0:1;
1121 spin_unlock(&srxd->callback_lock);
1122 nal_data = srxd->nal_data;
1125 * everyone returns a send token
1127 gmnal_return_ltxd(nal_data, ltxd);
1130 CDEBUG(D_ERROR, "NOT final callback context[%p]\n", srxd);
1135 * Let our client application proceed
1137 CDEBUG(D_ERROR, "final callback context[%p]\n", srxd);
1138 lib_finalize(libnal, srxd, srxd->cookie, PTL_OK);
1141 * send an ack to the sender to let him know we got the data
1143 gmnal_large_tx_ack(nal_data, srxd);
1146 * Unregister the memory that was used
1147 * This is a very slow business (slower then register)
1149 nriov = srxd->nriov;
1151 GMNAL_GM_LOCK(nal_data);
1153 CDEBUG(D_ERROR, "deregister memory [%p]\n", riov->iov_base);
1154 if (gm_deregister_memory(srxd->nal_data->gm_port,
1155 riov->iov_base, riov->iov_len)) {
1156 CDEBUG(D_ERROR, "failed to deregister memory [%p]\n",
1161 GMNAL_GM_UNLOCK(nal_data);
1162 PORTAL_FREE(srxd->riov, sizeof(struct iovec)*nriov);
1165 * repost the receive buffer (return receive token)
1167 GMNAL_GM_LOCK(nal_data);
1168 gm_provide_receive_buffer_with_tag(nal_data->gm_port, srxd->buffer,
1169 srxd->gmsize, GM_LOW_PRIORITY, 0);
1170 GMNAL_GM_UNLOCK(nal_data);
1177 * Called on target node.
1178 * After pulling data from a source node
1179 * send an ack message to indicate the large transmit is complete.
1182 gmnal_large_tx_ack(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1186 gmnal_msghdr_t *msghdr;
1187 void *buffer = NULL;
1188 unsigned int local_nid;
1189 gm_status_t gm_status = GM_SUCCESS;
1191 CDEBUG(D_TRACE, "srxd[%p] target_node [%u]\n", srxd,
1192 srxd->gm_source_node);
1194 GMNAL_GM_LOCK(nal_data);
1195 gm_status = gm_global_id_to_node_id(nal_data->gm_port,
1196 srxd->gm_source_node, &local_nid);
1197 GMNAL_GM_UNLOCK(nal_data);
1198 if (gm_status != GM_SUCCESS) {
1199 CDEBUG(D_ERROR, "Failed to obtain local id\n");
1202 CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
1204 stxd = gmnal_get_stxd(nal_data, 1);
1205 CDEBUG(D_TRACE, "gmnal_large_tx_ack got stxd[%p]\n", stxd);
1207 stxd->nal_data = nal_data;
1208 stxd->type = GMNAL_LARGE_MESSAGE_ACK;
1211 * Copy gmnal_msg_hdr and portals header to the transmit buffer
1212 * Then copy the data in
1214 buffer = stxd->buffer;
1215 msghdr = (gmnal_msghdr_t*)buffer;
1218 * Add in the address of the original stxd from the sender node
1219 * so it knows which thread to notify.
1221 msghdr->magic = GMNAL_MAGIC;
1222 msghdr->type = GMNAL_LARGE_MESSAGE_ACK;
1223 msghdr->sender_node_id = nal_data->gm_global_nid;
1224 msghdr->stxd = srxd->source_stxd;
1225 CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
1227 CDEBUG(D_INFO, "sending\n");
1228 stxd->msg_size= sizeof(gmnal_msghdr_t);
1231 CDEBUG(D_NET, "Calling gm_send_to_peer port [%p] buffer [%p] "
1232 "gmsize [%lu] msize [%d] global_nid [%u] local_nid[%d] "
1233 "stxd [%p]\n", nal_data->gm_port, stxd->buffer, stxd->gm_size,
1234 stxd->msg_size, srxd->gm_source_node, local_nid, stxd);
1235 GMNAL_GM_LOCK(nal_data);
1236 stxd->gm_priority = GM_LOW_PRIORITY;
1237 stxd->gm_target_node = local_nid;
1238 gm_send_to_peer_with_callback(nal_data->gm_port, stxd->buffer,
1239 stxd->gm_size, stxd->msg_size,
1240 GM_LOW_PRIORITY, local_nid,
1241 gmnal_large_tx_ack_callback,
1244 GMNAL_GM_UNLOCK(nal_data);
1245 CDEBUG(D_INFO, "gmnal_large_tx_ack :: done\n");
1252 * A callback to indicate the small transmit operation is compete
1253 * Check for errors and try to deal with them.
1254 * Call lib_finalise to inform the client application that the
1255 * send is complete and the memory can be reused.
1256 * Return the stxd when finished with it (returns a send token)
1259 gmnal_large_tx_ack_callback(gm_port_t *gm_port, void *context,
1262 gmnal_stxd_t *stxd = (gmnal_stxd_t*)context;
1263 gmnal_data_t *nal_data = (gmnal_data_t*)stxd->nal_data;
1266 CDEBUG(D_ERROR, "send completion event for unknown stxd\n");
1269 CDEBUG(D_TRACE, "send completion event for stxd [%p] status is [%d]\n",
1271 gmnal_return_stxd(stxd->nal_data, stxd);
1273 GMNAL_GM_UNLOCK(nal_data);
1278 * Indicates the large transmit operation is compete.
1279 * Called on transmit side (means data has been pulled by receiver
1281 * Call lib_finalise to inform the client application that the send
1282 * is complete, deregister the memory and return the stxd.
1283 * Finally, report the rx buffer that the ack message was delivered in.
1286 gmnal_large_tx_ack_received(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1288 lib_nal_t *libnal = nal_data->libnal;
1289 gmnal_stxd_t *stxd = NULL;
1290 gmnal_msghdr_t *msghdr = NULL;
1291 void *buffer = NULL;
1295 CDEBUG(D_TRACE, "gmnal_large_tx_ack_received buffer [%p]\n", buffer);
1297 buffer = srxd->buffer;
1298 msghdr = (gmnal_msghdr_t*)buffer;
1299 stxd = msghdr->stxd;
1301 CDEBUG(D_INFO, "gmnal_large_tx_ack_received stxd [%p]\n", stxd);
1303 lib_finalize(libnal, stxd, stxd->cookie, PTL_OK);
1306 * extract the iovec from the stxd, deregister the memory.
1307 * free the space used to store the iovec
1310 while(stxd->niov--) {
1311 CDEBUG(D_INFO, "deregister memory [%p] size ["LPSZ"]\n",
1312 iov->iov_base, iov->iov_len);
1313 GMNAL_GM_LOCK(nal_data);
1314 gm_deregister_memory(nal_data->gm_port, iov->iov_base,
1316 GMNAL_GM_UNLOCK(nal_data);
1321 * return the send token
1322 * TO DO It is bad to hold onto the send token so long?
1324 gmnal_return_stxd(nal_data, stxd);
1328 * requeue the receive buffer
1330 gmnal_rx_requeue_buffer(nal_data, srxd);