lnet/klnds/gmlnd/gmlnd_comm.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2003 Los Alamos National Laboratory (LANL)
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org/
   7  *
   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.
  11  *
  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.
  16  *
  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.
  20  */
  21
  22 /*
  23  *      This file contains all gmnal send and receive functions
  24  */
  25
  26 #include "gmnal.h"
  27
  28 /*
  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.
  35  */
  36 int
  37 gmnal_ct_thread(void *arg)
  38 {
  39         gmnal_data_t            *nal_data;
  40         gm_recv_event_t         *rxevent = NULL;
  41         gm_recv_t               *recv = NULL;
  42
  43         if (!arg) {
  44                 CDEBUG(D_TRACE, "NO nal_data. Exiting\n");
  45                 return(-1);
  46         }
  47
  48         nal_data = (gmnal_data_t*)arg;
  49         CDEBUG(D_TRACE, "nal_data is [%p]\n", arg);
  50
  51         daemonize();
  52
  53         nal_data->ctthread_flag = GMNAL_CTTHREAD_STARTED;
  54
  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");
  61                         break;
  62                 }
  63                 CDEBUG(D_INFO, "got [%s]\n", gmnal_rxevent(rxevent));
  64                 switch (GM_RECV_EVENT_TYPE(rxevent)) {
  65
  66                         case(GM_RECV_EVENT):
  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");
  73                         break;
  74                         case(_GM_SLEEP_EVENT):
  75                                 /*
  76                                  *      Blocking receive above just returns
  77                                  *      immediatly with _GM_SLEEP_EVENT
  78                                  *      Don't know what this is
  79                                  */
  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");
  85                                 break;
  86
  87                         default:
  88                                 /*
  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.
  93                                  */
  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");
  99                 }
 100         }
 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);
 105 }
 106
 107
 108 /*
 109  *      process a receive event
 110  */
 111 int gmnal_rx_thread(void *arg)
 112 {
 113         gmnal_data_t            *nal_data;
 114         void                    *buffer;
 115         gmnal_rxtwe_t           *we = NULL;
 116
 117         if (!arg) {
 118                 CDEBUG(D_TRACE, "NO nal_data. Exiting\n");
 119                 return(-1);
 120         }
 121
 122         nal_data = (gmnal_data_t*)arg;
 123         CDEBUG(D_TRACE, "nal_data is [%p]\n", arg);
 124
 125         daemonize();
 126         /*
 127          *      set 1 bit for each thread started
 128          *      doesn't matter which bit
 129          */
 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;
 133         else
 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);
 137
 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);
 141                 if (!we) {
 142                         CDEBUG(D_INFO, "Receive thread time to exit\n");
 143                         break;
 144                 }
 145
 146                 buffer = we->buffer;
 147                 switch(((gmnal_msghdr_t*)buffer)->type) {
 148                 case(GMNAL_SMALL_MESSAGE):
 149                         gmnal_pre_receive(nal_data, we,
 150                                            GMNAL_SMALL_MESSAGE);
 151                 break;
 152                 case(GMNAL_LARGE_MESSAGE_INIT):
 153                         gmnal_pre_receive(nal_data, we,
 154                                            GMNAL_LARGE_MESSAGE_INIT);
 155                 break;
 156                 case(GMNAL_LARGE_MESSAGE_ACK):
 157                         gmnal_pre_receive(nal_data, we,
 158                                            GMNAL_LARGE_MESSAGE_ACK);
 159                 break;
 160                 default:
 161                         CDEBUG(D_ERROR, "Unsupported message type\n");
 162                         gmnal_rx_bad(nal_data, we, NULL);
 163                 }
 164                 PORTAL_FREE(we, sizeof(gmnal_rxtwe_t));
 165         }
 166
 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);
 173 }
 174
 175
 176
 177 /*
 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.
 183  */
 184 int
 185 gmnal_pre_receive(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, int gmnal_type)
 186 {
 187         gmnal_srxd_t    *srxd = NULL;
 188         void            *buffer = NULL;
 189         unsigned int snode, sport, type, length;
 190         gmnal_msghdr_t  *gmnal_msghdr;
 191         ptl_hdr_t       *portals_hdr;
 192
 193         CDEBUG(D_INFO, "nal_data [%p], we[%p] type [%d]\n",
 194                nal_data, we, gmnal_type);
 195
 196         buffer = we->buffer;
 197         snode = we->snode;
 198         sport = we->sport;
 199         type = we->type;
 200         buffer = we->buffer;
 201         length = we->length;
 202
 203         gmnal_msghdr = (gmnal_msghdr_t*)buffer;
 204         portals_hdr = (ptl_hdr_t*)(buffer+GMNAL_MSGHDR_SIZE);
 205
 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);
 215
 216
 217         /*
 218          *      Get a receive descriptor for this message
 219          */
 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;
 223         if (!srxd) {
 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);
 227         }
 228
 229         /*
 230          *      no need to bother portals library with this
 231          */
 232         if (gmnal_type == GMNAL_LARGE_MESSAGE_ACK) {
 233                 gmnal_large_tx_ack_received(nal_data, srxd);
 234                 return(GMNAL_STATUS_OK);
 235         }
 236
 237         srxd->type = gmnal_type;
 238         srxd->nsiov = gmnal_msghdr->niov;
 239         srxd->gm_source_node = gmnal_msghdr->sender_node_id;
 240
 241         CDEBUG(D_PORTALS, "Calling lib_parse buffer is [%p]\n",
 242                buffer+GMNAL_MSGHDR_SIZE);
 243         /*
 244          *      control passes to lib, which calls cb_recv
 245          *      cb_recv is responsible for returning the buffer
 246          *      for future receive
 247          */
 248         lib_parse(nal_data->nal_cb, portals_hdr, srxd);
 249
 250         return(GMNAL_STATUS_OK);
 251 }
 252
 253
 254
 255 /*
 256  *      After a receive has been processed,
 257  *      hang out the receive buffer again.
 258  *      This implicitly returns a receive token.
 259  */
 260 int
 261 gmnal_rx_requeue_buffer(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
 262 {
 263         CDEBUG(D_TRACE, "gmnal_rx_requeue_buffer\n");
 264
 265         CDEBUG(D_NET, "requeueing srxd[%p] nal_data[%p]\n", srxd, nal_data);
 266
 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);
 271
 272         return(GMNAL_STATUS_OK);
 273 }
 274
 275
 276 /*
 277  *      Handle a bad message
 278  *      A bad message is one we don't expect or can't interpret
 279  */
 280 int
 281 gmnal_rx_bad(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, gmnal_srxd_t *srxd)
 282 {
 283         CDEBUG(D_TRACE, "Can't handle message\n");
 284
 285         if (!srxd)
 286                 srxd = gmnal_rxbuffer_to_srxd(nal_data,
 287                                                we->buffer);
 288         if (srxd) {
 289                 gmnal_rx_requeue_buffer(nal_data, srxd);
 290         } else {
 291                 CDEBUG(D_ERROR, "Can't find a descriptor for this buffer\n");
 292                 /*
 293                  *      get rid of it ?
 294                  */
 295                 return(GMNAL_STATUS_FAIL);
 296         }
 297
 298         return(GMNAL_STATUS_OK);
 299 }
 300
 301
 302
 303 /*
 304  *      Process a small message receive.
 305  *      Get here from gmnal_receive_thread, gmnal_pre_receive
 306  *      lib_parse, cb_recv
 307  *      Put data from prewired receive buffer into users buffer(s)
 308  *      Hang out the receive buffer again for another receive
 309  *      Call lib_finalize
 310  */
 311 int
 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)
 314 {
 315         gmnal_srxd_t    *srxd = NULL;
 316         void    *buffer = NULL;
 317         gmnal_data_t    *nal_data = (gmnal_data_t*)nal_cb->nal_data;
 318
 319
 320         CDEBUG(D_TRACE, "niov [%d] mlen["LPSZ"]\n", niov, mlen);
 321
 322         if (!private) {
 323                 CDEBUG(D_ERROR, "gmnal_small_rx no context\n");
 324                 lib_finalize(nal_cb, private, cookie);
 325                 return(PTL_FAIL);
 326         }
 327
 328         srxd = (gmnal_srxd_t*)private;
 329         buffer = srxd->buffer;
 330         buffer += sizeof(gmnal_msghdr_t);
 331         buffer += sizeof(ptl_hdr_t);
 332
 333         while(niov--) {
 334                 CDEBUG(D_INFO, "processing [%p] len ["LPSZ"]\n", iov,
 335                        iov->iov_len);
 336                 gm_bcopy(buffer, iov->iov_base, iov->iov_len);
 337                 buffer += iov->iov_len;
 338                 iov++;
 339         }
 340
 341
 342         /*
 343          *      let portals library know receive is complete
 344          */
 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");
 349         }
 350         /*
 351          *      return buffer so it can be used again
 352          */
 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);
 358
 359         return(PTL_OK);
 360 }
 361
 362
 363 /*
 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.
 369  */
 370 int
 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)
 374 {
 375         gmnal_data_t    *nal_data = (gmnal_data_t*)nal_cb->nal_data;
 376         gmnal_stxd_t    *stxd = NULL;
 377         void            *buffer = NULL;
 378         gmnal_msghdr_t  *msghdr = NULL;
 379         int             tot_size = 0;
 380         unsigned int    local_nid;
 381         gm_status_t     gm_status = GM_SUCCESS;
 382
 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);
 387
 388         CDEBUG(D_INFO, "portals_hdr:: dest_nid ["LPU64"], src_nid ["LPU64"]\n",
 389                hdr->dest_nid, hdr->src_nid);
 390
 391         if (!nal_data) {
 392                 CDEBUG(D_ERROR, "no nal_data\n");
 393                 return(GMNAL_STATUS_FAIL);
 394         } else {
 395                 CDEBUG(D_INFO, "nal_data [%p]\n", nal_data);
 396         }
 397
 398         GMNAL_GM_LOCK(nal_data);
 399         gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
 400                                             &local_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);
 405         }
 406         CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
 407
 408         stxd = gmnal_get_stxd(nal_data, 1);
 409         CDEBUG(D_INFO, "stxd [%p]\n", stxd);
 410
 411         stxd->type = GMNAL_SMALL_MESSAGE;
 412         stxd->cookie = cookie;
 413
 414         /*
 415          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
 416          *      Then copy the data in
 417          */
 418         buffer = stxd->buffer;
 419         msghdr = (gmnal_msghdr_t*)buffer;
 420
 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);
 425
 426         buffer += sizeof(gmnal_msghdr_t);
 427
 428         CDEBUG(D_INFO, "processing  portals hdr at [%p]\n", buffer);
 429         gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
 430
 431         buffer += sizeof(ptl_hdr_t);
 432
 433         while(niov--) {
 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;
 438                 iov++;
 439         }
 440
 441         CDEBUG(D_INFO, "sending\n");
 442         tot_size = size+sizeof(ptl_hdr_t)+sizeof(gmnal_msghdr_t);
 443         stxd->msg_size = tot_size;
 444
 445
 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);
 450
 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");
 460
 461         return(PTL_OK);
 462 }
 463
 464
 465 /*
 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)
 471  */
 472 void
 473 gmnal_small_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
 474 {
 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;
 479
 480         if (!stxd) {
 481                 CDEBUG(D_TRACE, "send completion event for unknown stxd\n");
 482                 return;
 483         }
 484         if (status != GM_SUCCESS) {
 485                 CDEBUG(D_ERROR, "Result of send stxd [%p] is [%s]\n",
 486                        stxd, gmnal_gm_error(status));
 487         }
 488
 489         switch(status) {
 490                 case(GM_SUCCESS):
 491                 break;
 492
 493
 494
 495                 case(GM_SEND_DROPPED):
 496                 /*
 497                  *      do a resend on the dropped ones
 498                  */
 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,
 503                                                       stxd->buffer,
 504                                                       stxd->gm_size,
 505                                                       stxd->msg_size,
 506                                                       stxd->gm_priority,
 507                                                       stxd->gm_target_node,
 508                                                       gmnal_small_tx_callback,
 509                                                       context);
 510                         GMNAL_GM_UNLOCK(nal_data);
 511
 512                 return;
 513                 case(GM_TIMED_OUT):
 514                 case(GM_SEND_TIMED_OUT):
 515                 /*
 516                  *      drop these ones
 517                  */
 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);
 524
 525                 return;
 526
 527
 528                 /*
 529                  *      abort on these ?
 530                  */
 531                 case(GM_TRY_AGAIN):
 532                 case(GM_INTERRUPTED):
 533                 case(GM_FAILURE):
 534                 case(GM_INPUT_BUFFER_TOO_SMALL):
 535                 case(GM_OUTPUT_BUFFER_TOO_SMALL):
 536                 case(GM_BUSY):
 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):
 543                 case(GM_UNATTACHED):
 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):
 551                 case(GM_CLONE_BUSY):
 552                 case(GM_NO_SUCH_DEVICE):
 553                 case(GM_ABORTED):
 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):
 561                 case(GM_NO_MATCH):
 562                 case(GM_USER_ERROR):
 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):
 568                 case(GM_UC_ERROR):
 569                 case(GM_INVALID_PORT_NUMBER):
 570                 case(GM_DEV_NOT_FOUND):
 571                 case(GM_FIRMWARE_NOT_RUNNING):
 572                 case(GM_YP_NO_MATCH):
 573                 default:
 574                         CDEBUG(D_ERROR, "Unknown send error\n");
 575         }
 576
 577         /*
 578          *      TO DO
 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
 586          *      free list earlier.
 587          */
 588         if (stxd->type == GMNAL_LARGE_MESSAGE_INIT) {
 589                 CDEBUG(D_INFO, "large transmit done\n");
 590                 return;
 591         }
 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",
 595                        stxd);
 596         }
 597         return;
 598 }
 599
 600
 601
 602 void gmnal_drop_sends_callback(struct gm_port *gm_port, void *context,
 603                                 gm_status_t status)
 604 {
 605         gmnal_stxd_t    *stxd = (gmnal_stxd_t*)context;
 606         gmnal_data_t    *nal_data = stxd->nal_data;
 607
 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,
 613                                               stxd->gm_priority,
 614                                               stxd->gm_target_node,
 615                                               gmnal_small_tx_callback,
 616                                               context);
 617                 GMNAL_GM_LOCK(nal_data);
 618         } else {
 619                 CDEBUG(D_ERROR, "send_to_peer status for stxd [%p] is
 620                        [%d][%s]\n", stxd, status, gmnal_gm_error(status));
 621         }
 622
 623
 624         return;
 625 }
 626
 627
 628 /*
 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.
 634  */
 635 int
 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)
 639 {
 640
 641         gmnal_data_t    *nal_data;
 642         gmnal_stxd_t    *stxd = NULL;
 643         void            *buffer = 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;
 649         int             niov_dup;
 650
 651
 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);
 656
 657         if (nal_cb)
 658                 nal_data = (gmnal_data_t*)nal_cb->nal_data;
 659         else  {
 660                 CDEBUG(D_ERROR, "no nal_cb.\n");
 661                 return(GMNAL_STATUS_FAIL);
 662         }
 663
 664
 665         /*
 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
 670          */
 671         stxd = gmnal_get_stxd(nal_data, 1);
 672         CDEBUG(D_INFO, "stxd [%p]\n", stxd);
 673
 674         stxd->type = GMNAL_LARGE_MESSAGE_INIT;
 675         stxd->cookie = cookie;
 676
 677         /*
 678          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
 679          *      Then copy the iov in
 680          */
 681         buffer = stxd->buffer;
 682         msghdr = (gmnal_msghdr_t*)buffer;
 683
 684         CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
 685
 686         msghdr->magic = GMNAL_MAGIC;
 687         msghdr->type = GMNAL_LARGE_MESSAGE_INIT;
 688         msghdr->sender_node_id = nal_data->gm_global_nid;
 689         msghdr->stxd = stxd;
 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);
 694
 695
 696         CDEBUG(D_INFO, "processing  portals hdr at [%p]\n", buffer);
 697
 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);
 702
 703         /*
 704          *      copy the iov to the buffer so target knows
 705          *      where to get the data from
 706          */
 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);
 711
 712
 713         /*
 714          *      Store the iovs in the stxd for we can get
 715          *      them later if we need them
 716          */
 717         CDEBUG(D_NET, "Copying iov [%p] to [%p]\n", iov, stxd->iov);
 718         gm_bcopy(iov, stxd->iov, niov*sizeof(struct iovec));
 719         stxd->niov = niov;
 720
 721
 722         /*
 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.
 726          */
 727         iov_dup = iov;
 728         niov_dup = niov;
 729         while(niov--) {
 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,
 744                                                      iov_dup->iov_base,
 745                                                      iov_dup->iov_len);
 746                                 iov_dup++;
 747                         }
 748                         GMNAL_GM_UNLOCK(nal_data);
 749                         gmnal_return_stxd(nal_data, stxd);
 750                         return(PTL_FAIL);
 751                 }
 752
 753                 GMNAL_GM_UNLOCK(nal_data);
 754                 iov++;
 755         }
 756
 757         /*
 758          *      Send the init message to the target
 759          */
 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,
 763                                             &local_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);
 770         }
 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,
 775                                       (void*)stxd);
 776         GMNAL_GM_UNLOCK(nal_data);
 777
 778         CDEBUG(D_INFO, "done\n");
 779
 780         return(PTL_OK);
 781 }
 782
 783 /*
 784  *      Callback function indicates that send of buffer with
 785  *      large message iovec has completed (or failed).
 786  */
 787 void
 788 gmnal_large_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
 789 {
 790         gmnal_small_tx_callback(gm_port, context, status);
 791
 792 }
 793
 794
 795
 796 /*
 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.
 800  */
 801 int
 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,
 804                 size_t rlen)
 805 {
 806         gmnal_data_t    *nal_data = nal_cb->nal_data;
 807         gmnal_srxd_t    *srxd = (gmnal_srxd_t*)private;
 808         void            *buffer = NULL;
 809         struct  iovec   *riov_dup;
 810         int             nriov_dup;
 811         gmnal_msghdr_t  *msghdr = NULL;
 812         gm_status_t     gm_status;
 813
 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);
 817
 818         if (!srxd) {
 819                 CDEBUG(D_ERROR, "gmnal_large_rx no context\n");
 820                 lib_finalize(nal_cb, private, cookie);
 821                 return(PTL_FAIL);
 822         }
 823
 824         buffer = srxd->buffer;
 825         msghdr = (gmnal_msghdr_t*)buffer;
 826         buffer += sizeof(gmnal_msghdr_t);
 827         buffer += sizeof(ptl_hdr_t);
 828
 829         /*
 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
 833          */
 834         srxd->source_stxd = msghdr->stxd;
 835
 836         /*
 837          *      Register the receivers memory
 838          *      get the data,
 839          *      tell the sender that we got the data
 840          *      then tell the receiver we got the data
 841          *      TO DO
 842          *      If the iovecs match, could interleave
 843          *      gm_registers and gm_gets for each element
 844          */
 845         nriov_dup = nriov;
 846         riov_dup = riov;
 847         while(nriov--) {
 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,
 862                                                      riov_dup->iov_base,
 863                                                      riov_dup->iov_len);
 864                                 riov_dup++;
 865                         }
 866                         GMNAL_GM_LOCK(nal_data);
 867                         /*
 868                          *      give back srxd and buffer. Send NACK to sender
 869                          */
 870                         return(PTL_FAIL);
 871                 }
 872                 GMNAL_GM_UNLOCK(nal_data);
 873                 riov++;
 874         }
 875         /*
 876          *      do this so the final gm_get callback can deregister the memory
 877          */
 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;
 881
 882         /*
 883          *      now do gm_get to get the data
 884          */
 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");
 889         }
 890
 891         CDEBUG(D_INFO, "lgmanl_large_rx done\n");
 892
 893         return(PTL_OK);
 894 }
 895
 896
 897 /*
 898  *      Perform a number of remote gets as part of receiving
 899  *      a large message.
 900  *      The final one to complete (i.e. the last callback to get called)
 901  *      tidies up.
 902  *      gm_get requires a send token.
 903  */
 904 int
 905 gmnal_remote_get(gmnal_srxd_t *srxd, int nsiov, struct iovec *siov,
 906                   int nriov, struct iovec *riov)
 907 {
 908
 909         int     ncalls = 0;
 910
 911         CDEBUG(D_TRACE, "gmnal_remote_get srxd[%p], nriov[%d], riov[%p],
 912                nsiov[%d], siov[%p]\n", srxd, nriov, riov, nsiov, siov);
 913
 914
 915         ncalls = gmnal_copyiov(0, srxd, nsiov, siov, nriov, riov);
 916         if (ncalls < 0) {
 917                 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
 918                 return(GMNAL_STATUS_FAIL);
 919         }
 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;
 924
 925         ncalls = gmnal_copyiov(1, srxd, nsiov, siov, nriov, riov);
 926         if (ncalls < 0) {
 927                 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
 928                 return(GMNAL_STATUS_FAIL);
 929         }
 930
 931         return(GMNAL_STATUS_OK);
 932
 933 }
 934
 935
 936 /*
 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.
 941  */
 942 int
 943 gmnal_copyiov(int do_copy, gmnal_srxd_t *srxd, int nsiov,
 944                struct iovec *siov, int nriov, struct iovec *riov)
 945 {
 946
 947         int     ncalls = 0;
 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;
 955
 956         CDEBUG(D_TRACE, "copy[%d] nal_data[%p]\n", do_copy, nal_data);
 957         if (do_copy) {
 958                 if (!nal_data) {
 959                         CDEBUG(D_ERROR, "Bad args No nal_data\n");
 960                         return(GMNAL_STATUS_FAIL);
 961                 }
 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) {
 966
 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);
 971                 }
 972                 GMNAL_GM_UNLOCK(nal_data);
 973                 /*
 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
 980                  */
 981                 CDEBUG(D_INFO, "gmnal_copyiov source node is G[%u]L[%d]\n",
 982                        srxd->gm_source_node, source_node);
 983         }
 984
 985         do {
 986                 CDEBUG(D_INFO, "sbuf[%p] slen[%d] rbuf[%p], rlen[%d]\n",
 987                                 sbuf, slen, rbuf, rlen);
 988                 if (slen > rlen) {
 989                         ncalls++;
 990                         if (do_copy) {
 991                                 CDEBUG(D_INFO, "slen>rlen\n");
 992                                 ltxd = gmnal_get_ltxd(nal_data);
 993                                 ltxd->srxd = srxd;
 994                                 GMNAL_GM_LOCK(nal_data);
 995                                 /*
 996                                  *      funny business to get rid
 997                                  *      of compiler warning
 998                                  */
 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,
1003                                        GMNAL_GM_PORT,
1004                                        gmnal_remote_get_callback, ltxd);
1005                                 GMNAL_GM_UNLOCK(nal_data);
1006                         }
1007                         /*
1008                          *      at the end of 1 iov element
1009                          */
1010                         sbuf+=rlen;
1011                         slen-=rlen;
1012                         riov++;
1013                         nriov--;
1014                         rbuf = riov->iov_base;
1015                         rlen = riov->iov_len;
1016                 } else if (rlen > slen) {
1017                         ncalls++;
1018                         if (do_copy) {
1019                                 CDEBUG(D_INFO, "slen<rlen\n");
1020                                 ltxd = gmnal_get_ltxd(nal_data);
1021                                 ltxd->srxd = srxd;
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,
1027                                        GMNAL_GM_PORT,
1028                                        gmnal_remote_get_callback, ltxd);
1029                                 GMNAL_GM_UNLOCK(nal_data);
1030                         }
1031                         /*
1032                          *      at end of siov element
1033                          */
1034                         rbuf+=slen;
1035                         rlen-=slen;
1036                         siov++;
1037                         sbuf = siov->iov_base;
1038                         slen = siov->iov_len;
1039                 } else {
1040                         ncalls++;
1041                         if (do_copy) {
1042                                 CDEBUG(D_INFO, "rlen=slen\n");
1043                                 ltxd = gmnal_get_ltxd(nal_data);
1044                                 ltxd->srxd = srxd;
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,
1050                                        GMNAL_GM_PORT,
1051                                        gmnal_remote_get_callback, ltxd);
1052                                 GMNAL_GM_UNLOCK(nal_data);
1053                         }
1054                         /*
1055                          *      at end of siov and riov element
1056                          */
1057                         siov++;
1058                         sbuf = siov->iov_base;
1059                         slen = siov->iov_len;
1060                         riov++;
1061                         nriov--;
1062                         rbuf = riov->iov_base;
1063                         rlen = riov->iov_len;
1064                 }
1065
1066         } while (nriov);
1067         return(ncalls);
1068 }
1069
1070
1071 /*
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.
1075  */
1076 void
1077 gmnal_remote_get_callback(gm_port_t *gm_port, void *context,
1078                            gm_status_t status)
1079 {
1080
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;
1084         int             lastone;
1085         struct  iovec   *riov;
1086         int             nriov;
1087         gmnal_data_t    *nal_data;
1088
1089         CDEBUG(D_TRACE, "called for context [%p]\n", context);
1090
1091         if (status != GM_SUCCESS) {
1092                 CDEBUG(D_ERROR, "reports error [%d][%s]\n", status,
1093                        gmnal_gm_error(status));
1094         }
1095
1096         spin_lock(&srxd->callback_lock);
1097         srxd->ncallbacks--;
1098         srxd->callback_status |= status;
1099         lastone = srxd->ncallbacks?0:1;
1100         spin_unlock(&srxd->callback_lock);
1101         nal_data = srxd->nal_data;
1102
1103         /*
1104          *      everyone returns a send token
1105          */
1106         gmnal_return_ltxd(nal_data, ltxd);
1107
1108         if (!lastone) {
1109                 CDEBUG(D_ERROR, "NOT final callback context[%p]\n", srxd);
1110                 return;
1111         }
1112
1113         /*
1114          *      Let our client application proceed
1115          */
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",
1119                        srxd);
1120         }
1121
1122         /*
1123          *      send an ack to the sender to let him know we got the data
1124          */
1125         gmnal_large_tx_ack(nal_data, srxd);
1126
1127         /*
1128          *      Unregister the memory that was used
1129          *      This is a very slow business (slower then register)
1130          */
1131         nriov = srxd->nriov;
1132         riov = srxd->riov;
1133         GMNAL_GM_LOCK(nal_data);
1134         while (nriov--) {
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",
1139                                riov->iov_base);
1140                 }
1141                 riov++;
1142         }
1143         GMNAL_GM_UNLOCK(nal_data);
1144         PORTAL_FREE(srxd->riov, sizeof(struct iovec)*nriov);
1145
1146         /*
1147          *      repost the receive buffer (return receive token)
1148          */
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);
1153
1154         return;
1155 }
1156
1157
1158 /*
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.
1162  */
1163 void
1164 gmnal_large_tx_ack(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1165 {
1166
1167         gmnal_stxd_t    *stxd;
1168         gmnal_msghdr_t *msghdr;
1169         void            *buffer = NULL;
1170         unsigned int    local_nid;
1171         gm_status_t     gm_status = GM_SUCCESS;
1172
1173         CDEBUG(D_TRACE, "srxd[%p] target_node [%u]\n", srxd,
1174                srxd->gm_source_node);
1175
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");
1182                 return;
1183         }
1184         CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
1185
1186         stxd = gmnal_get_stxd(nal_data, 1);
1187         CDEBUG(D_TRACE, "gmnal_large_tx_ack got stxd[%p]\n", stxd);
1188
1189         stxd->nal_data = nal_data;
1190         stxd->type = GMNAL_LARGE_MESSAGE_ACK;
1191
1192         /*
1193          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
1194          *      Then copy the data in
1195          */
1196         buffer = stxd->buffer;
1197         msghdr = (gmnal_msghdr_t*)buffer;
1198
1199         /*
1200          *      Add in the address of the original stxd from the sender node
1201          *      so it knows which thread to notify.
1202          */
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);
1208
1209         CDEBUG(D_INFO, "sending\n");
1210         stxd->msg_size= sizeof(gmnal_msghdr_t);
1211
1212
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,
1224                                       (void*)stxd);
1225
1226         GMNAL_GM_UNLOCK(nal_data);
1227         CDEBUG(D_INFO, "gmnal_large_tx_ack :: done\n");
1228
1229         return;
1230 }
1231
1232
1233 /*
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)
1239  */
1240 void
1241 gmnal_large_tx_ack_callback(gm_port_t *gm_port, void *context,
1242                              gm_status_t status)
1243 {
1244         gmnal_stxd_t    *stxd = (gmnal_stxd_t*)context;
1245         gmnal_data_t    *nal_data = (gmnal_data_t*)stxd->nal_data;
1246
1247         if (!stxd) {
1248                 CDEBUG(D_ERROR, "send completion event for unknown stxd\n");
1249                 return;
1250         }
1251         CDEBUG(D_TRACE, "send completion event for stxd [%p] status is [%d]\n",
1252                stxd, status);
1253         gmnal_return_stxd(stxd->nal_data, stxd);
1254
1255         GMNAL_GM_UNLOCK(nal_data);
1256         return;
1257 }
1258
1259 /*
1260  *      Indicates the large transmit operation is compete.
1261  *      Called on transmit side (means data has been pulled  by receiver
1262  *      or failed).
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.
1266  */
1267 void
1268 gmnal_large_tx_ack_received(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1269 {
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;
1274         struct  iovec   *iov;
1275
1276
1277         CDEBUG(D_TRACE, "gmnal_large_tx_ack_received buffer [%p]\n", buffer);
1278
1279         buffer = srxd->buffer;
1280         msghdr = (gmnal_msghdr_t*)buffer;
1281         stxd = msghdr->stxd;
1282
1283         CDEBUG(D_INFO, "gmnal_large_tx_ack_received stxd [%p]\n", stxd);
1284
1285         if (lib_finalize(nal_cb, stxd, stxd->cookie) != PTL_OK) {
1286                 CDEBUG(D_INFO, "Call to lib_finalize failed for stxd [%p]\n",
1287                        stxd);
1288         }
1289
1290         /*
1291          *      extract the iovec from the stxd, deregister the memory.
1292          *      free the space used to store the iovec
1293          */
1294         iov = stxd->iov;
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,
1300                                      iov->iov_len);
1301                 GMNAL_GM_UNLOCK(nal_data);
1302                 iov++;
1303         }
1304
1305         /*
1306          *      return the send token
1307          *      TO DO It is bad to hold onto the send token so long?
1308          */
1309         gmnal_return_stxd(nal_data, stxd);
1310
1311
1312         /*
1313          *      requeue the receive buffer
1314          */
1315         gmnal_rx_requeue_buffer(nal_data, srxd);
1316
1317
1318         return;
1319 }