lustre/portals/knals/gmnal/gmnal_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         int              rc;
 193
 194         CDEBUG(D_INFO, "nal_data [%p], we[%p] type [%d]\n",
 195                nal_data, we, gmnal_type);
 196
 197         buffer = we->buffer;
 198         snode = we->snode;
 199         sport = we->sport;
 200         type = we->type;
 201         buffer = we->buffer;
 202         length = we->length;
 203
 204         gmnal_msghdr = (gmnal_msghdr_t*)buffer;
 205         portals_hdr = (ptl_hdr_t*)(buffer+GMNAL_MSGHDR_SIZE);
 206
 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);
 216
 217
 218         /*
 219          *      Get a receive descriptor for this message
 220          */
 221         srxd = gmnal_rxbuffer_to_srxd(nal_data, buffer);
 222         CDEBUG(D_INFO, "Back from gmnal_rxbuffer_to_srxd\n");
 223         if (!srxd) {
 224                 CDEBUG(D_ERROR, "Failed to get receive descriptor\n");
 225                 /* I think passing a NULL srxd to lib_parse will crash
 226                  * gmnal_recv() */
 227                 LBUG();
 228                 lib_parse(nal_data->libnal, portals_hdr, srxd);
 229                 return(GMNAL_STATUS_FAIL);
 230         }
 231
 232         /*
 233          *      no need to bother portals library with this
 234          */
 235         if (gmnal_type == GMNAL_LARGE_MESSAGE_ACK) {
 236                 gmnal_large_tx_ack_received(nal_data, srxd);
 237                 return(GMNAL_STATUS_OK);
 238         }
 239
 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;
 244
 245         CDEBUG(D_PORTALS, "Calling lib_parse buffer is [%p]\n",
 246                buffer+GMNAL_MSGHDR_SIZE);
 247         /*
 248          *      control passes to lib, which calls cb_recv
 249          *      cb_recv is responsible for returning the buffer
 250          *      for future receive
 251          */
 252         rc = lib_parse(nal_data->libnal, portals_hdr, srxd);
 253
 254         if (rc != PTL_OK) {
 255                 /* I just received garbage; take appropriate action... */
 256                 LBUG();
 257         }
 258
 259         return(GMNAL_STATUS_OK);
 260 }
 261
 262
 263
 264 /*
 265  *      After a receive has been processed,
 266  *      hang out the receive buffer again.
 267  *      This implicitly returns a receive token.
 268  */
 269 int
 270 gmnal_rx_requeue_buffer(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
 271 {
 272         CDEBUG(D_TRACE, "gmnal_rx_requeue_buffer\n");
 273
 274         CDEBUG(D_NET, "requeueing srxd[%p] nal_data[%p]\n", srxd, nal_data);
 275
 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);
 280
 281         return(GMNAL_STATUS_OK);
 282 }
 283
 284
 285 /*
 286  *      Handle a bad message
 287  *      A bad message is one we don't expect or can't interpret
 288  */
 289 int
 290 gmnal_rx_bad(gmnal_data_t *nal_data, gmnal_rxtwe_t *we, gmnal_srxd_t *srxd)
 291 {
 292         CDEBUG(D_TRACE, "Can't handle message\n");
 293
 294         if (!srxd)
 295                 srxd = gmnal_rxbuffer_to_srxd(nal_data,
 296                                                we->buffer);
 297         if (srxd) {
 298                 gmnal_rx_requeue_buffer(nal_data, srxd);
 299         } else {
 300                 CDEBUG(D_ERROR, "Can't find a descriptor for this buffer\n");
 301                 /*
 302                  *      get rid of it ?
 303                  */
 304                 return(GMNAL_STATUS_FAIL);
 305         }
 306
 307         return(GMNAL_STATUS_OK);
 308 }
 309
 310
 311
 312 /*
 313  *      Process a small message receive.
 314  *      Get here from gmnal_receive_thread, gmnal_pre_receive
 315  *      lib_parse, cb_recv
 316  *      Put data from prewired receive buffer into users buffer(s)
 317  *      Hang out the receive buffer again for another receive
 318  *      Call lib_finalize
 319  */
 320 int
 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)
 323 {
 324         gmnal_srxd_t    *srxd = NULL;
 325         void    *buffer = NULL;
 326         gmnal_data_t    *nal_data = (gmnal_data_t*)libnal->nal_data;
 327
 328
 329         CDEBUG(D_TRACE, "niov [%d] mlen["LPSZ"]\n", niov, mlen);
 330
 331         if (!private) {
 332                 CDEBUG(D_ERROR, "gmnal_small_rx no context\n");
 333                 lib_finalize(libnal, private, cookie, PTL_FAIL);
 334                 return(PTL_FAIL);
 335         }
 336
 337         srxd = (gmnal_srxd_t*)private;
 338         buffer = srxd->buffer;
 339         buffer += sizeof(gmnal_msghdr_t);
 340         buffer += sizeof(ptl_hdr_t);
 341
 342         while(niov--) {
 343                 CDEBUG(D_INFO, "processing [%p] len ["LPSZ"]\n", iov,
 344                        iov->iov_len);
 345                 gm_bcopy(buffer, iov->iov_base, iov->iov_len);
 346                 buffer += iov->iov_len;
 347                 iov++;
 348         }
 349
 350
 351         /*
 352          *      let portals library know receive is complete
 353          */
 354         CDEBUG(D_PORTALS, "calling lib_finalize\n");
 355         lib_finalize(libnal, private, cookie, PTL_OK);
 356         /*
 357          *      return buffer so it can be used again
 358          */
 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);
 364
 365         return(PTL_OK);
 366 }
 367
 368
 369 /*
 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.
 375  */
 376 int
 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)
 380 {
 381         gmnal_data_t    *nal_data = (gmnal_data_t*)libnal->nal_data;
 382         gmnal_stxd_t    *stxd = NULL;
 383         void            *buffer = NULL;
 384         gmnal_msghdr_t  *msghdr = NULL;
 385         int             tot_size = 0;
 386         unsigned int    local_nid;
 387         gm_status_t     gm_status = GM_SUCCESS;
 388
 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);
 393
 394         CDEBUG(D_INFO, "portals_hdr:: dest_nid ["LPU64"], src_nid ["LPU64"]\n",
 395                hdr->dest_nid, hdr->src_nid);
 396
 397         if (!nal_data) {
 398                 CDEBUG(D_ERROR, "no nal_data\n");
 399                 return(GMNAL_STATUS_FAIL);
 400         } else {
 401                 CDEBUG(D_INFO, "nal_data [%p]\n", nal_data);
 402         }
 403
 404         GMNAL_GM_LOCK(nal_data);
 405         gm_status = gm_global_id_to_node_id(nal_data->gm_port, global_nid,
 406                                             &local_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);
 411         }
 412         CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
 413
 414         stxd = gmnal_get_stxd(nal_data, 1);
 415         CDEBUG(D_INFO, "stxd [%p]\n", stxd);
 416
 417         stxd->type = GMNAL_SMALL_MESSAGE;
 418         stxd->cookie = cookie;
 419
 420         /*
 421          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
 422          *      Then copy the data in
 423          */
 424         buffer = stxd->buffer;
 425         msghdr = (gmnal_msghdr_t*)buffer;
 426
 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);
 431
 432         buffer += sizeof(gmnal_msghdr_t);
 433
 434         CDEBUG(D_INFO, "processing  portals hdr at [%p]\n", buffer);
 435         gm_bcopy(hdr, buffer, sizeof(ptl_hdr_t));
 436
 437         buffer += sizeof(ptl_hdr_t);
 438
 439         while(niov--) {
 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;
 444                 iov++;
 445         }
 446
 447         CDEBUG(D_INFO, "sending\n");
 448         tot_size = size+sizeof(ptl_hdr_t)+sizeof(gmnal_msghdr_t);
 449         stxd->msg_size = tot_size;
 450
 451
 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);
 456
 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");
 466
 467         return(PTL_OK);
 468 }
 469
 470
 471 /*
 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)
 477  */
 478 void
 479 gmnal_small_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
 480 {
 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;
 485
 486         if (!stxd) {
 487                 CDEBUG(D_TRACE, "send completion event for unknown stxd\n");
 488                 return;
 489         }
 490         if (status != GM_SUCCESS) {
 491                 CDEBUG(D_ERROR, "Result of send stxd [%p] is [%s]\n",
 492                        stxd, gmnal_gm_error(status));
 493         }
 494
 495         switch(status) {
 496                 case(GM_SUCCESS):
 497                 break;
 498
 499
 500
 501                 case(GM_SEND_DROPPED):
 502                 /*
 503                  *      do a resend on the dropped ones
 504                  */
 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,
 509                                                       stxd->buffer,
 510                                                       stxd->gm_size,
 511                                                       stxd->msg_size,
 512                                                       stxd->gm_priority,
 513                                                       stxd->gm_target_node,
 514                                                       gmnal_small_tx_callback,
 515                                                       context);
 516                         GMNAL_GM_UNLOCK(nal_data);
 517
 518                 return;
 519                 case(GM_TIMED_OUT):
 520                 case(GM_SEND_TIMED_OUT):
 521                 /*
 522                  *      drop these ones
 523                  */
 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);
 530
 531                 return;
 532
 533
 534                 /*
 535                  *      abort on these ?
 536                  */
 537                 case(GM_TRY_AGAIN):
 538                 case(GM_INTERRUPTED):
 539                 case(GM_FAILURE):
 540                 case(GM_INPUT_BUFFER_TOO_SMALL):
 541                 case(GM_OUTPUT_BUFFER_TOO_SMALL):
 542                 case(GM_BUSY):
 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):
 549                 case(GM_UNATTACHED):
 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):
 557                 case(GM_CLONE_BUSY):
 558                 case(GM_NO_SUCH_DEVICE):
 559                 case(GM_ABORTED):
 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):
 567                 case(GM_NO_MATCH):
 568                 case(GM_USER_ERROR):
 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):
 574                 case(GM_UC_ERROR):
 575                 case(GM_INVALID_PORT_NUMBER):
 576                 case(GM_DEV_NOT_FOUND):
 577                 case(GM_FIRMWARE_NOT_RUNNING):
 578                 case(GM_YP_NO_MATCH):
 579                 default:
 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);
 584                 return;
 585
 586         }
 587
 588         /*
 589          *      TO DO
 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
 597          *      free list earlier.
 598          */
 599         if (stxd->type == GMNAL_LARGE_MESSAGE_INIT) {
 600                 CDEBUG(D_INFO, "large transmit done\n");
 601                 return;
 602         }
 603         gmnal_return_stxd(nal_data, stxd);
 604         lib_finalize(libnal, stxd, cookie, PTL_OK);
 605         return;
 606 }
 607
 608 /*
 609  *      After an error on the port
 610  *      call this to allow future sends to complete
 611  */
 612 void gmnal_resume_sending_callback(struct gm_port *gm_port, void *context,
 613                                  gm_status_t status)
 614 {
 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);
 619         return;
 620 }
 621
 622
 623 void gmnal_drop_sends_callback(struct gm_port *gm_port, void *context,
 624                                 gm_status_t status)
 625 {
 626         gmnal_stxd_t    *stxd = (gmnal_stxd_t*)context;
 627         gmnal_data_t    *nal_data = stxd->nal_data;
 628
 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,
 634                                               stxd->gm_priority,
 635                                               stxd->gm_target_node,
 636                                               gmnal_small_tx_callback,
 637                                               context);
 638                 GMNAL_GM_LOCK(nal_data);
 639         } else {
 640                 CDEBUG(D_ERROR, "send_to_peer status for stxd [%p] is "
 641                        "[%d][%s]\n", stxd, status, gmnal_gm_error(status));
 642         }
 643
 644
 645         return;
 646 }
 647
 648
 649 /*
 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.
 655  */
 656 int
 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)
 660 {
 661
 662         gmnal_data_t    *nal_data;
 663         gmnal_stxd_t    *stxd = NULL;
 664         void            *buffer = 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;
 670         int             niov_dup;
 671
 672
 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);
 677
 678         if (libnal)
 679                 nal_data = (gmnal_data_t*)libnal->nal_data;
 680         else  {
 681                 CDEBUG(D_ERROR, "no libnal.\n");
 682                 return(GMNAL_STATUS_FAIL);
 683         }
 684
 685
 686         /*
 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
 691          */
 692         stxd = gmnal_get_stxd(nal_data, 1);
 693         CDEBUG(D_INFO, "stxd [%p]\n", stxd);
 694
 695         stxd->type = GMNAL_LARGE_MESSAGE_INIT;
 696         stxd->cookie = cookie;
 697
 698         /*
 699          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
 700          *      Then copy the iov in
 701          */
 702         buffer = stxd->buffer;
 703         msghdr = (gmnal_msghdr_t*)buffer;
 704
 705         CDEBUG(D_INFO, "processing msghdr at [%p]\n", buffer);
 706
 707         msghdr->magic = GMNAL_MAGIC;
 708         msghdr->type = GMNAL_LARGE_MESSAGE_INIT;
 709         msghdr->sender_node_id = nal_data->gm_global_nid;
 710         msghdr->stxd = stxd;
 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);
 715
 716
 717         CDEBUG(D_INFO, "processing  portals hdr at [%p]\n", buffer);
 718
 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);
 723
 724         /*
 725          *      copy the iov to the buffer so target knows
 726          *      where to get the data from
 727          */
 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);
 732
 733
 734         /*
 735          *      Store the iovs in the stxd for we can get
 736          *      them later if we need them
 737          */
 738         CDEBUG(D_NET, "Copying iov [%p] to [%p]\n", iov, stxd->iov);
 739         gm_bcopy(iov, stxd->iov, niov*sizeof(struct iovec));
 740         stxd->niov = niov;
 741
 742
 743         /*
 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.
 747          */
 748         iov_dup = iov;
 749         niov_dup = niov;
 750         while(niov--) {
 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,
 765                                                      iov_dup->iov_base,
 766                                                      iov_dup->iov_len);
 767                                 iov_dup++;
 768                         }
 769                         GMNAL_GM_UNLOCK(nal_data);
 770                         gmnal_return_stxd(nal_data, stxd);
 771                         return(PTL_FAIL);
 772                 }
 773
 774                 GMNAL_GM_UNLOCK(nal_data);
 775                 iov++;
 776         }
 777
 778         /*
 779          *      Send the init message to the target
 780          */
 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,
 784                                             &local_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);
 791         }
 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,
 796                                       (void*)stxd);
 797         GMNAL_GM_UNLOCK(nal_data);
 798
 799         CDEBUG(D_INFO, "done\n");
 800
 801         return(PTL_OK);
 802 }
 803
 804 /*
 805  *      Callback function indicates that send of buffer with
 806  *      large message iovec has completed (or failed).
 807  */
 808 void
 809 gmnal_large_tx_callback(gm_port_t *gm_port, void *context, gm_status_t status)
 810 {
 811         gmnal_small_tx_callback(gm_port, context, status);
 812
 813 }
 814
 815
 816
 817 /*
 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.
 821  */
 822 int
 823 gmnal_large_rx(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
 824                 unsigned int nriov, struct iovec *riov, size_t mlen,
 825                 size_t rlen)
 826 {
 827         gmnal_data_t    *nal_data = libnal->nal_data;
 828         gmnal_srxd_t    *srxd = (gmnal_srxd_t*)private;
 829         void            *buffer = NULL;
 830         struct  iovec   *riov_dup;
 831         int             nriov_dup;
 832         gmnal_msghdr_t  *msghdr = NULL;
 833         gm_status_t     gm_status;
 834
 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);
 838
 839         if (!srxd) {
 840                 CDEBUG(D_ERROR, "gmnal_large_rx no context\n");
 841                 lib_finalize(libnal, private, cookie, PTL_FAIL);
 842                 return(PTL_FAIL);
 843         }
 844
 845         buffer = srxd->buffer;
 846         msghdr = (gmnal_msghdr_t*)buffer;
 847         buffer += sizeof(gmnal_msghdr_t);
 848         buffer += sizeof(ptl_hdr_t);
 849
 850         /*
 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
 854          */
 855         srxd->source_stxd = msghdr->stxd;
 856
 857         /*
 858          *      Register the receivers memory
 859          *      get the data,
 860          *      tell the sender that we got the data
 861          *      then tell the receiver we got the data
 862          *      TO DO
 863          *      If the iovecs match, could interleave
 864          *      gm_registers and gm_gets for each element
 865          */
 866         nriov_dup = nriov;
 867         riov_dup = riov;
 868         while(nriov--) {
 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,
 883                                                      riov_dup->iov_base,
 884                                                      riov_dup->iov_len);
 885                                 riov_dup++;
 886                         }
 887                         GMNAL_GM_LOCK(nal_data);
 888                         /*
 889                          *      give back srxd and buffer. Send NACK to sender
 890                          */
 891                         return(PTL_FAIL);
 892                 }
 893                 GMNAL_GM_UNLOCK(nal_data);
 894                 riov++;
 895         }
 896         /*
 897          *      do this so the final gm_get callback can deregister the memory
 898          */
 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;
 902
 903         /*
 904          *      now do gm_get to get the data
 905          */
 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");
 910         }
 911
 912         CDEBUG(D_INFO, "lgmanl_large_rx done\n");
 913
 914         return(PTL_OK);
 915 }
 916
 917
 918 /*
 919  *      Perform a number of remote gets as part of receiving
 920  *      a large message.
 921  *      The final one to complete (i.e. the last callback to get called)
 922  *      tidies up.
 923  *      gm_get requires a send token.
 924  */
 925 int
 926 gmnal_remote_get(gmnal_srxd_t *srxd, int nsiov, struct iovec *siov,
 927                   int nriov, struct iovec *riov)
 928 {
 929
 930         int     ncalls = 0;
 931
 932         CDEBUG(D_TRACE, "gmnal_remote_get srxd[%p], nriov[%d], riov[%p], "
 933                "nsiov[%d], siov[%p]\n", srxd, nriov, riov, nsiov, siov);
 934
 935
 936         ncalls = gmnal_copyiov(0, srxd, nsiov, siov, nriov, riov);
 937         if (ncalls < 0) {
 938                 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
 939                 return(GMNAL_STATUS_FAIL);
 940         }
 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;
 945
 946         ncalls = gmnal_copyiov(1, srxd, nsiov, siov, nriov, riov);
 947         if (ncalls < 0) {
 948                 CDEBUG(D_ERROR, "there's something wrong with the iovecs\n");
 949                 return(GMNAL_STATUS_FAIL);
 950         }
 951
 952         return(GMNAL_STATUS_OK);
 953
 954 }
 955
 956
 957 /*
 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.
 962  */
 963 int
 964 gmnal_copyiov(int do_copy, gmnal_srxd_t *srxd, int nsiov,
 965                struct iovec *siov, int nriov, struct iovec *riov)
 966 {
 967
 968         int     ncalls = 0;
 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;
 976
 977         CDEBUG(D_TRACE, "copy[%d] nal_data[%p]\n", do_copy, nal_data);
 978         if (do_copy) {
 979                 if (!nal_data) {
 980                         CDEBUG(D_ERROR, "Bad args No nal_data\n");
 981                         return(GMNAL_STATUS_FAIL);
 982                 }
 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) {
 987
 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);
 992                 }
 993                 GMNAL_GM_UNLOCK(nal_data);
 994                 /*
 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
1001                  */
1002                 CDEBUG(D_INFO, "gmnal_copyiov source node is G[%u]L[%d]\n",
1003                        srxd->gm_source_node, source_node);
1004         }
1005
1006         do {
1007                 CDEBUG(D_INFO, "sbuf[%p] slen[%d] rbuf[%p], rlen[%d]\n",
1008                                 sbuf, slen, rbuf, rlen);
1009                 if (slen > rlen) {
1010                         ncalls++;
1011                         if (do_copy) {
1012                                 CDEBUG(D_INFO, "slen>rlen\n");
1013                                 ltxd = gmnal_get_ltxd(nal_data);
1014                                 ltxd->srxd = srxd;
1015                                 GMNAL_GM_LOCK(nal_data);
1016                                 /*
1017                                  *      funny business to get rid
1018                                  *      of compiler warning
1019                                  */
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,
1024                                        GMNAL_GM_PORT,
1025                                        gmnal_remote_get_callback, ltxd);
1026                                 GMNAL_GM_UNLOCK(nal_data);
1027                         }
1028                         /*
1029                          *      at the end of 1 iov element
1030                          */
1031                         sbuf+=rlen;
1032                         slen-=rlen;
1033                         riov++;
1034                         nriov--;
1035                         rbuf = riov->iov_base;
1036                         rlen = riov->iov_len;
1037                 } else if (rlen > slen) {
1038                         ncalls++;
1039                         if (do_copy) {
1040                                 CDEBUG(D_INFO, "slen<rlen\n");
1041                                 ltxd = gmnal_get_ltxd(nal_data);
1042                                 ltxd->srxd = srxd;
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,
1048                                        GMNAL_GM_PORT,
1049                                        gmnal_remote_get_callback, ltxd);
1050                                 GMNAL_GM_UNLOCK(nal_data);
1051                         }
1052                         /*
1053                          *      at end of siov element
1054                          */
1055                         rbuf+=slen;
1056                         rlen-=slen;
1057                         siov++;
1058                         sbuf = siov->iov_base;
1059                         slen = siov->iov_len;
1060                 } else {
1061                         ncalls++;
1062                         if (do_copy) {
1063                                 CDEBUG(D_INFO, "rlen=slen\n");
1064                                 ltxd = gmnal_get_ltxd(nal_data);
1065                                 ltxd->srxd = srxd;
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,
1071                                        GMNAL_GM_PORT,
1072                                        gmnal_remote_get_callback, ltxd);
1073                                 GMNAL_GM_UNLOCK(nal_data);
1074                         }
1075                         /*
1076                          *      at end of siov and riov element
1077                          */
1078                         siov++;
1079                         sbuf = siov->iov_base;
1080                         slen = siov->iov_len;
1081                         riov++;
1082                         nriov--;
1083                         rbuf = riov->iov_base;
1084                         rlen = riov->iov_len;
1085                 }
1086
1087         } while (nriov);
1088         return(ncalls);
1089 }
1090
1091
1092 /*
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.
1096  */
1097 void
1098 gmnal_remote_get_callback(gm_port_t *gm_port, void *context,
1099                            gm_status_t status)
1100 {
1101
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;
1105         int             lastone;
1106         struct  iovec   *riov;
1107         int             nriov;
1108         gmnal_data_t    *nal_data;
1109
1110         CDEBUG(D_TRACE, "called for context [%p]\n", context);
1111
1112         if (status != GM_SUCCESS) {
1113                 CDEBUG(D_ERROR, "reports error [%d][%s]\n", status,
1114                        gmnal_gm_error(status));
1115         }
1116
1117         spin_lock(&srxd->callback_lock);
1118         srxd->ncallbacks--;
1119         srxd->callback_status |= status;
1120         lastone = srxd->ncallbacks?0:1;
1121         spin_unlock(&srxd->callback_lock);
1122         nal_data = srxd->nal_data;
1123
1124         /*
1125          *      everyone returns a send token
1126          */
1127         gmnal_return_ltxd(nal_data, ltxd);
1128
1129         if (!lastone) {
1130                 CDEBUG(D_ERROR, "NOT final callback context[%p]\n", srxd);
1131                 return;
1132         }
1133
1134         /*
1135          *      Let our client application proceed
1136          */
1137         CDEBUG(D_ERROR, "final callback context[%p]\n", srxd);
1138         lib_finalize(libnal, srxd, srxd->cookie, PTL_OK);
1139
1140         /*
1141          *      send an ack to the sender to let him know we got the data
1142          */
1143         gmnal_large_tx_ack(nal_data, srxd);
1144
1145         /*
1146          *      Unregister the memory that was used
1147          *      This is a very slow business (slower then register)
1148          */
1149         nriov = srxd->nriov;
1150         riov = srxd->riov;
1151         GMNAL_GM_LOCK(nal_data);
1152         while (nriov--) {
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",
1157                                riov->iov_base);
1158                 }
1159                 riov++;
1160         }
1161         GMNAL_GM_UNLOCK(nal_data);
1162         PORTAL_FREE(srxd->riov, sizeof(struct iovec)*nriov);
1163
1164         /*
1165          *      repost the receive buffer (return receive token)
1166          */
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);
1171
1172         return;
1173 }
1174
1175
1176 /*
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.
1180  */
1181 void
1182 gmnal_large_tx_ack(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1183 {
1184
1185         gmnal_stxd_t    *stxd;
1186         gmnal_msghdr_t *msghdr;
1187         void            *buffer = NULL;
1188         unsigned int    local_nid;
1189         gm_status_t     gm_status = GM_SUCCESS;
1190
1191         CDEBUG(D_TRACE, "srxd[%p] target_node [%u]\n", srxd,
1192                srxd->gm_source_node);
1193
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");
1200                 return;
1201         }
1202         CDEBUG(D_INFO, "Local Node_id is [%u][%x]\n", local_nid, local_nid);
1203
1204         stxd = gmnal_get_stxd(nal_data, 1);
1205         CDEBUG(D_TRACE, "gmnal_large_tx_ack got stxd[%p]\n", stxd);
1206
1207         stxd->nal_data = nal_data;
1208         stxd->type = GMNAL_LARGE_MESSAGE_ACK;
1209
1210         /*
1211          *      Copy gmnal_msg_hdr and portals header to the transmit buffer
1212          *      Then copy the data in
1213          */
1214         buffer = stxd->buffer;
1215         msghdr = (gmnal_msghdr_t*)buffer;
1216
1217         /*
1218          *      Add in the address of the original stxd from the sender node
1219          *      so it knows which thread to notify.
1220          */
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);
1226
1227         CDEBUG(D_INFO, "sending\n");
1228         stxd->msg_size= sizeof(gmnal_msghdr_t);
1229
1230
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,
1242                                       (void*)stxd);
1243
1244         GMNAL_GM_UNLOCK(nal_data);
1245         CDEBUG(D_INFO, "gmnal_large_tx_ack :: done\n");
1246
1247         return;
1248 }
1249
1250
1251 /*
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)
1257  */
1258 void
1259 gmnal_large_tx_ack_callback(gm_port_t *gm_port, void *context,
1260                              gm_status_t status)
1261 {
1262         gmnal_stxd_t    *stxd = (gmnal_stxd_t*)context;
1263         gmnal_data_t    *nal_data = (gmnal_data_t*)stxd->nal_data;
1264
1265         if (!stxd) {
1266                 CDEBUG(D_ERROR, "send completion event for unknown stxd\n");
1267                 return;
1268         }
1269         CDEBUG(D_TRACE, "send completion event for stxd [%p] status is [%d]\n",
1270                stxd, status);
1271         gmnal_return_stxd(stxd->nal_data, stxd);
1272
1273         GMNAL_GM_UNLOCK(nal_data);
1274         return;
1275 }
1276
1277 /*
1278  *      Indicates the large transmit operation is compete.
1279  *      Called on transmit side (means data has been pulled  by receiver
1280  *      or failed).
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.
1284  */
1285 void
1286 gmnal_large_tx_ack_received(gmnal_data_t *nal_data, gmnal_srxd_t *srxd)
1287 {
1288         lib_nal_t       *libnal = nal_data->libnal;
1289         gmnal_stxd_t    *stxd = NULL;
1290         gmnal_msghdr_t  *msghdr = NULL;
1291         void            *buffer = NULL;
1292         struct  iovec   *iov;
1293
1294
1295         CDEBUG(D_TRACE, "gmnal_large_tx_ack_received buffer [%p]\n", buffer);
1296
1297         buffer = srxd->buffer;
1298         msghdr = (gmnal_msghdr_t*)buffer;
1299         stxd = msghdr->stxd;
1300
1301         CDEBUG(D_INFO, "gmnal_large_tx_ack_received stxd [%p]\n", stxd);
1302
1303         lib_finalize(libnal, stxd, stxd->cookie, PTL_OK);
1304
1305         /*
1306          *      extract the iovec from the stxd, deregister the memory.
1307          *      free the space used to store the iovec
1308          */
1309         iov = stxd->iov;
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,
1315                                      iov->iov_len);
1316                 GMNAL_GM_UNLOCK(nal_data);
1317                 iov++;
1318         }
1319
1320         /*
1321          *      return the send token
1322          *      TO DO It is bad to hold onto the send token so long?
1323          */
1324         gmnal_return_stxd(nal_data, stxd);
1325
1326
1327         /*
1328          *      requeue the receive buffer
1329          */
1330         gmnal_rx_requeue_buffer(nal_data, srxd);
1331
1332
1333         return;
1334 }