lnet/lnet/lib-move.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * lib/lib-move.c
   5  * Data movement routines
   6  *
   7  *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
   8  *  Copyright (c) 2001-2002 Sandia National Laboratories
   9  *
  10  *   This file is part of Lustre, http://www.sf.net/projects/lustre/
  11  *
  12  *   Lustre is free software; you can redistribute it and/or
  13  *   modify it under the terms of version 2 of the GNU General Public
  14  *   License as published by the Free Software Foundation.
  15  *
  16  *   Lustre is distributed in the hope that it will be useful,
  17  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  *   GNU General Public License for more details.
  20  *
  21  *   You should have received a copy of the GNU General Public License
  22  *   along with Lustre; if not, write to the Free Software
  23  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  24  */
  25
  26 #ifndef __KERNEL__
  27 # include <stdio.h>
  28 #else
  29 # define DEBUG_SUBSYSTEM S_PORTALS
  30 # include <linux/kp30.h>
  31 #endif
  32 #include <portals/p30.h>
  33 #include <portals/lib-p30.h>
  34 #include <portals/arg-blocks.h>
  35
  36 /*
  37  * Right now it does not check access control lists.
  38  *
  39  * We only support one MD per ME, which is how the Portals 3.1 spec is written.
  40  * All previous complication is removed.
  41  */
  42
  43 static lib_me_t *
  44 lib_find_me(nal_cb_t *nal, int index, int op_mask, ptl_nid_t src_nid,
  45             ptl_pid_t src_pid, ptl_size_t rlength, ptl_size_t roffset,
  46             ptl_match_bits_t match_bits, ptl_size_t *mlength_out,
  47             ptl_size_t *offset_out, int *unlink_out)
  48 {
  49         lib_ni_t         *ni = &nal->ni;
  50         struct list_head *match_list = &ni->tbl.tbl[index];
  51         struct list_head *tmp;
  52         lib_me_t         *me;
  53         lib_md_t         *md;
  54         ptl_size_t        mlength;
  55         ptl_size_t        offset;
  56
  57         ENTRY;
  58
  59         CDEBUG (D_NET, "Request from "LPU64".%d of length %d into portal %d "
  60                 "MB="LPX64"\n", src_nid, src_pid, rlength, index, match_bits);
  61
  62         if (index < 0 || index >= ni->tbl.size) {
  63                 CERROR("Invalid portal %d not in [0-%d]\n",
  64                        index, ni->tbl.size);
  65                 goto failed;
  66         }
  67
  68         list_for_each (tmp, match_list) {
  69                 me = list_entry(tmp, lib_me_t, me_list);
  70                 md = me->md;
  71
  72                  /* ME attached but MD not attached yet */
  73                 if (md == NULL)
  74                         continue;
  75
  76                 LASSERT (me == md->me);
  77
  78                 /* MD deactivated */
  79                 if (md->threshold == 0)
  80                         continue;
  81
  82                 /* mismatched MD op */
  83                 if ((md->options & op_mask) == 0)
  84                         continue;
  85
  86                 /* mismatched ME nid/pid? */
  87                 if (me->match_id.nid != PTL_NID_ANY &&
  88                     me->match_id.nid != src_nid)
  89                         continue;
  90
  91                 if (me->match_id.pid != PTL_PID_ANY &&
  92                     me->match_id.pid != src_pid)
  93                         continue;
  94
  95                 /* mismatched ME matchbits? */
  96                 if (((me->match_bits ^ match_bits) & ~me->ignore_bits) != 0)
  97                         continue;
  98
  99                 /* Hurrah! This _is_ a match; check it out... */
 100
 101                 if ((md->options & PTL_MD_MANAGE_REMOTE) == 0)
 102                         offset = md->offset;
 103                 else
 104                         offset = roffset;
 105
 106                 mlength = md->length - offset;
 107                 if ((md->options & PTL_MD_MAX_SIZE) != 0 &&
 108                     mlength > md->max_size)
 109                         mlength = md->max_size;
 110
 111                 if (rlength <= mlength) {        /* fits in allowed space */
 112                         mlength = rlength;
 113                 } else if ((md->options & PTL_MD_TRUNCATE) == 0) {
 114                         /* this packet _really_ is too big */
 115                         CERROR("Matching packet %d too big: %d left, "
 116                                "%d allowed\n", rlength, md->length - offset,
 117                                mlength);
 118                         goto failed;
 119                 }
 120
 121                 md->offset = offset + mlength;
 122
 123                 *offset_out = offset;
 124                 *mlength_out = mlength;
 125                 *unlink_out = ((md->options & PTL_MD_AUTO_UNLINK) != 0 &&
 126                                md->offset >= (md->length - md->max_size));
 127                 RETURN (me);
 128         }
 129
 130  failed:
 131         CERROR (LPU64": Dropping %s from "LPU64".%d portal %d match "LPX64
 132                 " offset %d length %d: no match\n",
 133                 ni->nid, (op_mask == PTL_MD_OP_GET) ? "GET" : "PUT",
 134                 src_nid, src_pid, index, match_bits, roffset, rlength);
 135         RETURN(NULL);
 136 }
 137
 138 int do_PtlFailNid (nal_cb_t *nal, void *private, void *v_args, void *v_ret)
 139 {
 140         PtlFailNid_in     *args = v_args;
 141         PtlFailNid_out    *ret  = v_ret;
 142         lib_test_peer_t   *tp;
 143         unsigned long      flags;
 144         struct list_head  *el;
 145         struct list_head  *next;
 146         struct list_head   cull;
 147
 148         if (args->threshold != 0) {
 149                 /* Adding a new entry */
 150                 tp = (lib_test_peer_t *)nal->cb_malloc (nal, sizeof (*tp));
 151                 if (tp == NULL)
 152                         return (ret->rc = PTL_FAIL);
 153
 154                 tp->tp_nid = args->nid;
 155                 tp->tp_threshold = args->threshold;
 156
 157                 state_lock (nal, &flags);
 158                 list_add (&tp->tp_list, &nal->ni.ni_test_peers);
 159                 state_unlock (nal, &flags);
 160                 return (ret->rc = PTL_OK);
 161         }
 162
 163         /* removing entries */
 164         INIT_LIST_HEAD (&cull);
 165
 166         state_lock (nal, &flags);
 167
 168         list_for_each_safe (el, next, &nal->ni.ni_test_peers) {
 169                 tp = list_entry (el, lib_test_peer_t, tp_list);
 170
 171                 if (tp->tp_threshold == 0 ||    /* needs culling anyway */
 172                     args->nid == PTL_NID_ANY || /* removing all entries */
 173                     tp->tp_nid == args->nid)    /* matched this one */
 174                 {
 175                         list_del (&tp->tp_list);
 176                         list_add (&tp->tp_list, &cull);
 177                 }
 178         }
 179
 180         state_unlock (nal, &flags);
 181
 182         while (!list_empty (&cull)) {
 183                 tp = list_entry (cull.next, lib_test_peer_t, tp_list);
 184
 185                 list_del (&tp->tp_list);
 186                 nal->cb_free (nal, tp, sizeof (*tp));
 187         }
 188         return (ret->rc = PTL_OK);
 189 }
 190
 191 static int
 192 fail_peer (nal_cb_t *nal, ptl_nid_t nid, int outgoing)
 193 {
 194         lib_test_peer_t  *tp;
 195         struct list_head *el;
 196         struct list_head *next;
 197         unsigned long     flags;
 198         struct list_head  cull;
 199         int               fail = 0;
 200
 201         INIT_LIST_HEAD (&cull);
 202
 203         state_lock (nal, &flags);
 204
 205         list_for_each_safe (el, next, &nal->ni.ni_test_peers) {
 206                 tp = list_entry (el, lib_test_peer_t, tp_list);
 207
 208                 if (tp->tp_threshold == 0) {
 209                         /* zombie entry */
 210                         if (outgoing) {
 211                                 /* only cull zombies on outgoing tests,
 212                                  * since we may be at interrupt priority on
 213                                  * incoming messages. */
 214                                 list_del (&tp->tp_list);
 215                                 list_add (&tp->tp_list, &cull);
 216                         }
 217                         continue;
 218                 }
 219
 220                 if (tp->tp_nid == PTL_NID_ANY || /* fail every peer */
 221                     nid == tp->tp_nid) {        /* fail this peer */
 222                         fail = 1;
 223
 224                         if (tp->tp_threshold != PTL_MD_THRESH_INF) {
 225                                 tp->tp_threshold--;
 226                                 if (outgoing &&
 227                                     tp->tp_threshold == 0) {
 228                                         /* see above */
 229                                         list_del (&tp->tp_list);
 230                                         list_add (&tp->tp_list, &cull);
 231                                 }
 232                         }
 233                         break;
 234                 }
 235         }
 236
 237         state_unlock (nal, &flags);
 238
 239         while (!list_empty (&cull)) {
 240                 tp = list_entry (cull.next, lib_test_peer_t, tp_list);
 241                 list_del (&tp->tp_list);
 242
 243                 nal->cb_free (nal, tp, sizeof (*tp));
 244         }
 245
 246         return (fail);
 247 }
 248
 249 ptl_size_t
 250 lib_iov_nob (int niov, struct iovec *iov)
 251 {
 252         ptl_size_t nob = 0;
 253
 254         while (niov-- > 0)
 255                 nob += (iov++)->iov_len;
 256
 257         return (nob);
 258 }
 259
 260 void
 261 lib_copy_iov2buf (char *dest, int niov, struct iovec *iov,
 262                   ptl_size_t offset, ptl_size_t len)
 263 {
 264         ptl_size_t nob;
 265
 266         if (len == 0)
 267                 return;
 268
 269         /* skip complete frags before 'offset' */
 270         LASSERT (niov > 0);
 271         while (offset >= iov->iov_len) {
 272                 offset -= iov->iov_len;
 273                 iov++;
 274                 niov--;
 275                 LASSERT (niov > 0);
 276         }
 277
 278         do {
 279                 LASSERT (niov > 0);
 280                 nob = MIN (iov->iov_len - offset, len);
 281                 memcpy (dest, iov->iov_base + offset, nob);
 282
 283                 len -= nob;
 284                 dest += nob;
 285                 niov--;
 286                 iov++;
 287                 offset = 0;
 288         } while (len > 0);
 289 }
 290
 291 void
 292 lib_copy_buf2iov (int niov, struct iovec *iov, ptl_size_t offset,
 293                   char *src, ptl_size_t len)
 294 {
 295         ptl_size_t nob;
 296
 297         if (len == 0)
 298                 return;
 299
 300         /* skip complete frags before 'offset' */
 301         LASSERT (niov > 0);
 302         while (offset >= iov->iov_len) {
 303                 offset -= iov->iov_len;
 304                 iov++;
 305                 niov--;
 306                 LASSERT (niov > 0);
 307         }
 308
 309         do {
 310                 LASSERT (niov > 0);
 311                 nob = MIN (iov->iov_len - offset, len);
 312                 memcpy (iov->iov_base + offset, src, nob);
 313
 314                 len -= nob;
 315                 src += nob;
 316                 niov--;
 317                 iov++;
 318                 offset = 0;
 319         } while (len > 0);
 320 }
 321
 322 int
 323 lib_extract_iov (int dst_niov, struct iovec *dst,
 324                  int src_niov, struct iovec *src,
 325                  ptl_size_t offset, ptl_size_t len)
 326 {
 327         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 328          * for exactly 'len' bytes, and return the number of entries.
 329          * NB not destructive to 'src' */
 330         ptl_size_t      frag_len;
 331         int             niov;
 332
 333         if (len == 0)                           /* no data => */
 334                 return (0);                     /* no frags */
 335
 336         LASSERT (src_niov > 0);
 337         while (offset >= src->iov_len) {      /* skip initial frags */
 338                 offset -= src->iov_len;
 339                 src_niov--;
 340                 src++;
 341                 LASSERT (src_niov > 0);
 342         }
 343
 344         niov = 1;
 345         for (;;) {
 346                 LASSERT (src_niov > 0);
 347                 LASSERT (niov <= dst_niov);
 348
 349                 frag_len = src->iov_len - offset;
 350                 dst->iov_base = ((char *)src->iov_base) + offset;
 351
 352                 if (len <= frag_len) {
 353                         dst->iov_len = len;
 354                         return (niov);
 355                 }
 356
 357                 dst->iov_len = frag_len;
 358
 359                 len -= frag_len;
 360                 dst++;
 361                 src++;
 362                 niov++;
 363                 src_niov--;
 364                 offset = 0;
 365         }
 366 }
 367
 368 #ifndef __KERNEL__
 369 ptl_size_t
 370 lib_kiov_nob (int niov, ptl_kiov_t *kiov)
 371 {
 372         LASSERT (0);
 373         return (0);
 374 }
 375
 376 void
 377 lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov,
 378                    ptl_size_t offset, ptl_size_t len)
 379 {
 380         LASSERT (0);
 381 }
 382
 383 void
 384 lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, ptl_size_t offset,
 385                    char *src, ptl_size_t len)
 386 {
 387         LASSERT (0);
 388 }
 389
 390 int
 391 lib_extract_kiov (int dst_niov, ptl_kiov_t *dst,
 392                   int src_niov, ptl_kiov_t *src,
 393                   ptl_size_t offset, ptl_size_t len)
 394 {
 395         LASSERT (0);
 396 }
 397
 398 #else
 399
 400 ptl_size_t
 401 lib_kiov_nob (int niov, ptl_kiov_t *kiov)
 402 {
 403         ptl_size_t  nob = 0;
 404
 405         while (niov-- > 0)
 406                 nob += (kiov++)->kiov_len;
 407
 408         return (nob);
 409 }
 410
 411 void
 412 lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov,
 413                    ptl_size_t offset, ptl_size_t len)
 414 {
 415         ptl_size_t  nob;
 416         char       *addr;
 417
 418         if (len == 0)
 419                 return;
 420
 421         LASSERT (!in_interrupt ());
 422
 423         LASSERT (niov > 0);
 424         while (offset > kiov->kiov_len) {
 425                 offset -= kiov->kiov_len;
 426                 kiov++;
 427                 niov--;
 428                 LASSERT (niov > 0);
 429         }
 430
 431         do{
 432                 LASSERT (niov > 0);
 433                 nob = MIN (kiov->kiov_len - offset, len);
 434
 435                 addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset + offset;
 436                 memcpy (dest, addr, nob);
 437                 kunmap (kiov->kiov_page);
 438
 439                 len -= nob;
 440                 dest += nob;
 441                 niov--;
 442                 kiov++;
 443                 offset = 0;
 444         } while (len > 0);
 445 }
 446
 447 void
 448 lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, ptl_size_t offset,
 449                    char *src, ptl_size_t len)
 450 {
 451         ptl_size_t  nob;
 452         char       *addr;
 453
 454         if (len == 0)
 455                 return;
 456
 457         LASSERT (!in_interrupt ());
 458
 459         LASSERT (niov > 0);
 460         while (offset >= kiov->kiov_len) {
 461                 offset -= kiov->kiov_len;
 462                 kiov++;
 463                 niov--;
 464                 LASSERT (niov > 0);
 465         }
 466
 467         do {
 468                 LASSERT (niov > 0);
 469                 nob = MIN (kiov->kiov_len - offset, len);
 470
 471                 addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset + offset;
 472                 memcpy (addr, src, nob);
 473                 kunmap (kiov->kiov_page);
 474
 475                 len -= nob;
 476                 src += nob;
 477                 niov--;
 478                 kiov++;
 479                 offset = 0;
 480         } while (len > 0);
 481 }
 482
 483 int
 484 lib_extract_kiov (int dst_niov, ptl_kiov_t *dst,
 485                   int src_niov, ptl_kiov_t *src,
 486                   ptl_size_t offset, ptl_size_t len)
 487 {
 488         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 489          * for exactly 'len' bytes, and return the number of entries.
 490          * NB not destructive to 'src' */
 491         ptl_size_t      frag_len;
 492         int             niov;
 493
 494         if (len == 0)                           /* no data => */
 495                 return (0);                     /* no frags */
 496
 497         LASSERT (src_niov > 0);
 498         while (offset >= src->kiov_len) {      /* skip initial frags */
 499                 offset -= src->kiov_len;
 500                 src_niov--;
 501                 src++;
 502                 LASSERT (src_niov > 0);
 503         }
 504
 505         niov = 1;
 506         for (;;) {
 507                 LASSERT (src_niov > 0);
 508                 LASSERT (niov <= dst_niov);
 509
 510                 frag_len = src->kiov_len - offset;
 511                 dst->kiov_page = src->kiov_page;
 512                 dst->kiov_offset = src->kiov_offset + offset;
 513
 514                 if (len <= frag_len) {
 515                         dst->kiov_len = len;
 516                         LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_SIZE);
 517                         return (niov);
 518                 }
 519
 520                 dst->kiov_len = frag_len;
 521                 LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_SIZE);
 522
 523                 len -= frag_len;
 524                 dst++;
 525                 src++;
 526                 niov++;
 527                 src_niov--;
 528                 offset = 0;
 529         }
 530 }
 531 #endif
 532
 533 ptl_err_t
 534 lib_recv (nal_cb_t *nal, void *private, lib_msg_t *msg, lib_md_t *md,
 535           ptl_size_t offset, ptl_size_t mlen, ptl_size_t rlen)
 536 {
 537         if (mlen == 0)
 538                 return (nal->cb_recv(nal, private, msg,
 539                                      0, NULL,
 540                                      offset, mlen, rlen));
 541
 542         if ((md->options & PTL_MD_KIOV) == 0)
 543                 return (nal->cb_recv(nal, private, msg,
 544                                      md->md_niov, md->md_iov.iov,
 545                                      offset, mlen, rlen));
 546
 547         return (nal->cb_recv_pages(nal, private, msg,
 548                                    md->md_niov, md->md_iov.kiov,
 549                                    offset, mlen, rlen));
 550 }
 551
 552 ptl_err_t
 553 lib_send (nal_cb_t *nal, void *private, lib_msg_t *msg,
 554           ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
 555           lib_md_t *md, ptl_size_t offset, ptl_size_t len)
 556 {
 557         if (len == 0)
 558                 return (nal->cb_send(nal, private, msg,
 559                                      hdr, type, nid, pid,
 560                                      0, NULL,
 561                                      offset, len));
 562
 563         if ((md->options & PTL_MD_KIOV) == 0)
 564                 return (nal->cb_send(nal, private, msg,
 565                                      hdr, type, nid, pid,
 566                                      md->md_niov, md->md_iov.iov,
 567                                      offset, len));
 568
 569         return (nal->cb_send_pages(nal, private, msg,
 570                                    hdr, type, nid, pid,
 571                                    md->md_niov, md->md_iov.kiov,
 572                                    offset, len));
 573 }
 574
 575 static void
 576 lib_commit_md (nal_cb_t *nal, lib_md_t *md, lib_msg_t *msg)
 577 {
 578         /* ALWAYS called holding the state_lock */
 579         lib_counters_t *counters = &nal->ni.counters;
 580
 581         /* Here, we commit the MD to a network OP by marking it busy and
 582          * decrementing its threshold.  Come what may, the network "owns"
 583          * the MD until a call to lib_finalize() signals completion. */
 584         msg->md = md;
 585
 586         md->pending++;
 587         if (md->threshold != PTL_MD_THRESH_INF) {
 588                 LASSERT (md->threshold > 0);
 589                 md->threshold--;
 590         }
 591
 592         counters->msgs_alloc++;
 593         if (counters->msgs_alloc > counters->msgs_max)
 594                 counters->msgs_max = counters->msgs_alloc;
 595
 596         list_add (&msg->msg_list, &nal->ni.ni_active_msgs);
 597 }
 598
 599 static void
 600 lib_drop_message (nal_cb_t *nal, void *private, ptl_hdr_t *hdr)
 601 {
 602         unsigned long flags;
 603
 604         /* CAVEAT EMPTOR: this only drops messages that we've not committed
 605          * to receive (init_msg() not called) and therefore can't cause an
 606          * event. */
 607
 608         state_lock(nal, &flags);
 609         nal->ni.counters.drop_count++;
 610         nal->ni.counters.drop_length += hdr->payload_length;
 611         state_unlock(nal, &flags);
 612
 613         /* NULL msg => if NAL calls lib_finalize it will be a noop */
 614         (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 615 }
 616
 617 /*
 618  * Incoming messages have a ptl_msg_t object associated with them
 619  * by the library.  This object encapsulates the state of the
 620  * message and allows the NAL to do non-blocking receives or sends
 621  * of long messages.
 622  *
 623  */
 624 static ptl_err_t
 625 parse_put(nal_cb_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg)
 626 {
 627         lib_ni_t        *ni = &nal->ni;
 628         ptl_size_t       mlength = 0;
 629         ptl_size_t       offset = 0;
 630         int              unlink = 0;
 631         ptl_err_t        rc;
 632         lib_me_t        *me;
 633         lib_md_t        *md;
 634         unsigned long    flags;
 635
 636         /* Convert put fields to host byte order */
 637         hdr->msg.put.match_bits = NTOH__u64 (hdr->msg.put.match_bits);
 638         hdr->msg.put.ptl_index = NTOH__u32 (hdr->msg.put.ptl_index);
 639         hdr->msg.put.offset = NTOH__u32 (hdr->msg.put.offset);
 640
 641         state_lock(nal, &flags);
 642
 643         me = lib_find_me(nal, hdr->msg.put.ptl_index, PTL_MD_OP_PUT,
 644                          hdr->src_nid, hdr->src_pid,
 645                          hdr->payload_length, hdr->msg.put.offset,
 646                          hdr->msg.put.match_bits,
 647                          &mlength, &offset, &unlink);
 648         if (me == NULL) {
 649                 state_unlock(nal, &flags);
 650                 return (PTL_FAIL);
 651         }
 652
 653         md = me->md;
 654         CDEBUG(D_NET, "Incoming put index %x from "LPU64"/%u of length %d/%d "
 655                "into md "LPX64" [%d] + %d\n", hdr->msg.put.ptl_index,
 656                hdr->src_nid, hdr->src_pid, mlength, hdr->payload_length,
 657                md->md_lh.lh_cookie, md->md_niov, offset);
 658
 659         lib_commit_md(nal, md, msg);
 660
 661         msg->ev.type = PTL_EVENT_PUT;
 662         msg->ev.initiator.nid = hdr->src_nid;
 663         msg->ev.initiator.pid = hdr->src_pid;
 664         msg->ev.portal = hdr->msg.put.ptl_index;
 665         msg->ev.match_bits = hdr->msg.put.match_bits;
 666         msg->ev.rlength = hdr->payload_length;
 667         msg->ev.mlength = mlength;
 668         msg->ev.offset = offset;
 669         msg->ev.hdr_data = hdr->msg.put.hdr_data;
 670
 671         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 672
 673         if (!ptl_is_wire_handle_none(&hdr->msg.put.ack_wmd) &&
 674             !(md->options & PTL_MD_ACK_DISABLE)) {
 675                 msg->ack_wmd = hdr->msg.put.ack_wmd;
 676         }
 677
 678         ni->counters.recv_count++;
 679         ni->counters.recv_length += mlength;
 680
 681         /* only unlink after MD's pending count has been bumped in
 682          * lib_commit_md() otherwise lib_me_unlink() will nuke it */
 683         if (unlink)
 684                 lib_me_unlink (nal, me);
 685
 686         state_unlock(nal, &flags);
 687
 688         rc = lib_recv(nal, private, msg, md, offset, mlength,
 689                       hdr->payload_length);
 690         if (rc != PTL_OK)
 691                 CERROR(LPU64": error on receiving PUT from "LPU64": %d\n",
 692                        ni->nid, hdr->src_nid, rc);
 693
 694         return (rc);
 695 }
 696
 697 static ptl_err_t
 698 parse_get(nal_cb_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg)
 699 {
 700         lib_ni_t        *ni = &nal->ni;
 701         ptl_size_t       mlength = 0;
 702         ptl_size_t       offset = 0;
 703         int              unlink = 0;
 704         lib_me_t        *me;
 705         lib_md_t        *md;
 706         ptl_hdr_t        reply;
 707         unsigned long    flags;
 708         int              rc;
 709
 710         /* Convert get fields to host byte order */
 711         hdr->msg.get.match_bits = NTOH__u64 (hdr->msg.get.match_bits);
 712         hdr->msg.get.ptl_index = NTOH__u32 (hdr->msg.get.ptl_index);
 713         hdr->msg.get.sink_length = NTOH__u32 (hdr->msg.get.sink_length);
 714         hdr->msg.get.src_offset = NTOH__u32 (hdr->msg.get.src_offset);
 715
 716         state_lock(nal, &flags);
 717
 718         me = lib_find_me(nal, hdr->msg.get.ptl_index, PTL_MD_OP_GET,
 719                          hdr->src_nid, hdr->src_pid,
 720                          hdr->msg.get.sink_length, hdr->msg.get.src_offset,
 721                          hdr->msg.get.match_bits,
 722                          &mlength, &offset, &unlink);
 723         if (me == NULL) {
 724                 state_unlock(nal, &flags);
 725                 return (PTL_FAIL);
 726         }
 727
 728         md = me->md;
 729         CDEBUG(D_NET, "Incoming get index %d from "LPU64".%u of length %d/%d "
 730                "from md "LPX64" [%d] + %d\n", hdr->msg.get.ptl_index,
 731                hdr->src_nid, hdr->src_pid, mlength, hdr->payload_length,
 732                md->md_lh.lh_cookie, md->md_niov, offset);
 733
 734         lib_commit_md(nal, md, msg);
 735
 736         msg->ev.type = PTL_EVENT_GET;
 737         msg->ev.initiator.nid = hdr->src_nid;
 738         msg->ev.initiator.pid = hdr->src_pid;
 739         msg->ev.portal = hdr->msg.get.ptl_index;
 740         msg->ev.match_bits = hdr->msg.get.match_bits;
 741         msg->ev.rlength = hdr->payload_length;
 742         msg->ev.mlength = mlength;
 743         msg->ev.offset = offset;
 744         msg->ev.hdr_data = 0;
 745
 746         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 747
 748         ni->counters.send_count++;
 749         ni->counters.send_length += mlength;
 750
 751         /* only unlink after MD's refcount has been bumped in
 752          * lib_commit_md() otherwise lib_me_unlink() will nuke it */
 753         if (unlink)
 754                 lib_me_unlink (nal, me);
 755
 756         state_unlock(nal, &flags);
 757
 758         memset (&reply, 0, sizeof (reply));
 759         reply.type     = HTON__u32 (PTL_MSG_REPLY);
 760         reply.dest_nid = HTON__u64 (hdr->src_nid);
 761         reply.src_nid  = HTON__u64 (ni->nid);
 762         reply.dest_pid = HTON__u32 (hdr->src_pid);
 763         reply.src_pid  = HTON__u32 (ni->pid);
 764         reply.payload_length = HTON__u32 (mlength);
 765
 766         reply.msg.reply.dst_wmd = hdr->msg.get.return_wmd;
 767
 768         /* NB call lib_send() _BEFORE_ lib_recv() completes the incoming
 769          * message.  Some NALs _require_ this to implement optimized GET */
 770
 771         rc = lib_send (nal, private, msg, &reply, PTL_MSG_REPLY,
 772                        hdr->src_nid, hdr->src_pid, md, offset, mlength);
 773         if (rc != PTL_OK)
 774                 CERROR(LPU64": Unable to send REPLY for GET from "LPU64": %d\n",
 775                        ni->nid, hdr->src_nid, rc);
 776
 777         /* Discard any junk after the hdr */
 778         (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 779
 780         return (rc);
 781 }
 782
 783 static ptl_err_t
 784 parse_reply(nal_cb_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg)
 785 {
 786         lib_ni_t        *ni = &nal->ni;
 787         lib_md_t        *md;
 788         int              rlength;
 789         int              length;
 790         unsigned long    flags;
 791         ptl_err_t        rc;
 792
 793         state_lock(nal, &flags);
 794
 795         /* NB handles only looked up by creator (no flips) */
 796         md = ptl_wire_handle2md(&hdr->msg.reply.dst_wmd, nal);
 797         if (md == NULL || md->threshold == 0) {
 798                 CERROR (LPU64": Dropping REPLY from "LPU64" for %s MD "LPX64"."LPX64"\n",
 799                         ni->nid, hdr->src_nid,
 800                         md == NULL ? "invalid" : "inactive",
 801                         hdr->msg.reply.dst_wmd.wh_interface_cookie,
 802                         hdr->msg.reply.dst_wmd.wh_object_cookie);
 803
 804                 state_unlock(nal, &flags);
 805                 return (PTL_FAIL);
 806         }
 807
 808         LASSERT (md->offset == 0);
 809
 810         length = rlength = hdr->payload_length;
 811
 812         if (length > md->length) {
 813                 if ((md->options & PTL_MD_TRUNCATE) == 0) {
 814                         CERROR (LPU64": Dropping REPLY from "LPU64
 815                                 " length %d for MD "LPX64" would overflow (%d)\n",
 816                                 ni->nid, hdr->src_nid, length,
 817                                 hdr->msg.reply.dst_wmd.wh_object_cookie,
 818                                 md->length);
 819                         state_unlock(nal, &flags);
 820                         return (PTL_FAIL);
 821                 }
 822                 length = md->length;
 823         }
 824
 825         CDEBUG(D_NET, "Reply from "LPU64" of length %d/%d into md "LPX64"\n",
 826                hdr->src_nid, length, rlength,
 827                hdr->msg.reply.dst_wmd.wh_object_cookie);
 828
 829         lib_commit_md(nal, md, msg);
 830
 831         msg->ev.type = PTL_EVENT_REPLY;
 832         msg->ev.initiator.nid = hdr->src_nid;
 833         msg->ev.initiator.pid = hdr->src_pid;
 834         msg->ev.rlength = rlength;
 835         msg->ev.mlength = length;
 836         msg->ev.offset = 0;
 837
 838         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 839
 840         ni->counters.recv_count++;
 841         ni->counters.recv_length += length;
 842
 843         state_unlock(nal, &flags);
 844
 845         rc = lib_recv(nal, private, msg, md, 0, length, rlength);
 846         if (rc != PTL_OK)
 847                 CERROR(LPU64": error on receiving REPLY from "LPU64": %d\n",
 848                        ni->nid, hdr->src_nid, rc);
 849
 850         return (rc);
 851 }
 852
 853 static ptl_err_t
 854 parse_ack(nal_cb_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg)
 855 {
 856         lib_ni_t      *ni = &nal->ni;
 857         lib_md_t      *md;
 858         unsigned long  flags;
 859
 860         /* Convert ack fields to host byte order */
 861         hdr->msg.ack.match_bits = NTOH__u64 (hdr->msg.ack.match_bits);
 862         hdr->msg.ack.mlength = NTOH__u32 (hdr->msg.ack.mlength);
 863
 864         state_lock(nal, &flags);
 865
 866         /* NB handles only looked up by creator (no flips) */
 867         md = ptl_wire_handle2md(&hdr->msg.ack.dst_wmd, nal);
 868         if (md == NULL || md->threshold == 0) {
 869                 CDEBUG(D_INFO, LPU64": Dropping ACK from "LPU64" to %s MD "
 870                        LPX64"."LPX64"\n", ni->nid, hdr->src_nid,
 871                        (md == NULL) ? "invalid" : "inactive",
 872                        hdr->msg.ack.dst_wmd.wh_interface_cookie,
 873                        hdr->msg.ack.dst_wmd.wh_object_cookie);
 874
 875                 state_unlock(nal, &flags);
 876                 return (PTL_FAIL);
 877         }
 878
 879         CDEBUG(D_NET, LPU64": ACK from "LPU64" into md "LPX64"\n",
 880                ni->nid, hdr->src_nid,
 881                hdr->msg.ack.dst_wmd.wh_object_cookie);
 882
 883         lib_commit_md(nal, md, msg);
 884
 885         msg->ev.type = PTL_EVENT_ACK;
 886         msg->ev.initiator.nid = hdr->src_nid;
 887         msg->ev.initiator.pid = hdr->src_pid;
 888         msg->ev.mlength = hdr->msg.ack.mlength;
 889         msg->ev.match_bits = hdr->msg.ack.match_bits;
 890
 891         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 892
 893         ni->counters.recv_count++;
 894
 895         state_unlock(nal, &flags);
 896
 897         /* We have received and matched up the ack OK, create the
 898          * completion event now... */
 899         lib_finalize(nal, private, msg, PTL_OK);
 900
 901         /* ...and now discard any junk after the hdr */
 902         (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 903
 904        return (PTL_OK);
 905 }
 906
 907 static char *
 908 hdr_type_string (ptl_hdr_t *hdr)
 909 {
 910         switch (hdr->type) {
 911         case PTL_MSG_ACK:
 912                 return ("ACK");
 913         case PTL_MSG_PUT:
 914                 return ("PUT");
 915         case PTL_MSG_GET:
 916                 return ("GET");
 917         case PTL_MSG_REPLY:
 918                 return ("REPLY");
 919         case PTL_MSG_HELLO:
 920                 return ("HELLO");
 921         default:
 922                 return ("<UNKNOWN>");
 923         }
 924 }
 925
 926 void print_hdr(nal_cb_t * nal, ptl_hdr_t * hdr)
 927 {
 928         char *type_str = hdr_type_string (hdr);
 929
 930         nal->cb_printf(nal, "P3 Header at %p of type %s\n", hdr, type_str);
 931         nal->cb_printf(nal, "    From nid/pid %Lu/%Lu", hdr->src_nid,
 932                        hdr->src_pid);
 933         nal->cb_printf(nal, "    To nid/pid %Lu/%Lu\n", hdr->dest_nid,
 934                        hdr->dest_pid);
 935
 936         switch (hdr->type) {
 937         default:
 938                 break;
 939
 940         case PTL_MSG_PUT:
 941                 nal->cb_printf(nal,
 942                                "    Ptl index %d, ack md "LPX64"."LPX64", "
 943                                "match bits "LPX64"\n",
 944                                hdr->msg.put.ptl_index,
 945                                hdr->msg.put.ack_wmd.wh_interface_cookie,
 946                                hdr->msg.put.ack_wmd.wh_object_cookie,
 947                                hdr->msg.put.match_bits);
 948                 nal->cb_printf(nal,
 949                                "    Length %d, offset %d, hdr data "LPX64"\n",
 950                                hdr->payload_length, hdr->msg.put.offset,
 951                                hdr->msg.put.hdr_data);
 952                 break;
 953
 954         case PTL_MSG_GET:
 955                 nal->cb_printf(nal,
 956                                "    Ptl index %d, return md "LPX64"."LPX64", "
 957                                "match bits "LPX64"\n", hdr->msg.get.ptl_index,
 958                                hdr->msg.get.return_wmd.wh_interface_cookie,
 959                                hdr->msg.get.return_wmd.wh_object_cookie,
 960                                hdr->msg.get.match_bits);
 961                 nal->cb_printf(nal,
 962                                "    Length %d, src offset %d\n",
 963                                hdr->msg.get.sink_length,
 964                                hdr->msg.get.src_offset);
 965                 break;
 966
 967         case PTL_MSG_ACK:
 968                 nal->cb_printf(nal, "    dst md "LPX64"."LPX64", "
 969                                "manipulated length %d\n",
 970                                hdr->msg.ack.dst_wmd.wh_interface_cookie,
 971                                hdr->msg.ack.dst_wmd.wh_object_cookie,
 972                                hdr->msg.ack.mlength);
 973                 break;
 974
 975         case PTL_MSG_REPLY:
 976                 nal->cb_printf(nal, "    dst md "LPX64"."LPX64", "
 977                                "length %d\n",
 978                                hdr->msg.reply.dst_wmd.wh_interface_cookie,
 979                                hdr->msg.reply.dst_wmd.wh_object_cookie,
 980                                hdr->payload_length);
 981         }
 982
 983 }                               /* end of print_hdr() */
 984
 985
 986 void
 987 lib_parse(nal_cb_t *nal, ptl_hdr_t *hdr, void *private)
 988 {
 989         unsigned long  flags;
 990         ptl_err_t      rc;
 991         lib_msg_t     *msg;
 992
 993         /* convert common fields to host byte order */
 994         hdr->dest_nid = NTOH__u64 (hdr->dest_nid);
 995         hdr->src_nid = NTOH__u64 (hdr->src_nid);
 996         hdr->dest_pid = NTOH__u32 (hdr->dest_pid);
 997         hdr->src_pid = NTOH__u32 (hdr->src_pid);
 998         hdr->type = NTOH__u32 (hdr->type);
 999         hdr->payload_length = NTOH__u32(hdr->payload_length);
1000 #if 0
1001         nal->cb_printf(nal, "%d: lib_parse: nal=%p hdr=%p type=%d\n",
1002                        nal->ni.nid, nal, hdr, hdr->type);
1003         print_hdr(nal, hdr);
1004 #endif
1005         if (hdr->type == PTL_MSG_HELLO) {
1006                 /* dest_nid is really ptl_magicversion_t */
1007                 ptl_magicversion_t *mv = (ptl_magicversion_t *)&hdr->dest_nid;
1008
1009                 CERROR (LPU64": Dropping unexpected HELLO message: "
1010                         "magic %d, version %d.%d from "LPD64"\n",
1011                         nal->ni.nid, mv->magic,
1012                         mv->version_major, mv->version_minor,
1013                         hdr->src_nid);
1014                 lib_drop_message(nal, private, hdr);
1015                 return;
1016         }
1017
1018         if (hdr->dest_nid != nal->ni.nid) {
1019                 CERROR(LPU64": Dropping %s message from "LPU64" to "LPU64
1020                        " (not me)\n", nal->ni.nid, hdr_type_string (hdr),
1021                        hdr->src_nid, hdr->dest_nid);
1022                 lib_drop_message(nal, private, hdr);
1023                 return;
1024         }
1025
1026         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1027             fail_peer (nal, hdr->src_nid, 0))      /* shall we now? */
1028         {
1029                 CERROR(LPU64": Dropping incoming %s from "LPU64
1030                        ": simulated failure\n",
1031                        nal->ni.nid, hdr_type_string (hdr),
1032                        hdr->src_nid);
1033                 lib_drop_message(nal, private, hdr);
1034                 return;
1035         }
1036
1037         msg = lib_msg_alloc(nal);
1038         if (msg == NULL) {
1039                 CERROR(LPU64": Dropping incoming %s from "LPU64
1040                        ": can't allocate a lib_msg_t\n",
1041                        nal->ni.nid, hdr_type_string (hdr),
1042                        hdr->src_nid);
1043                 lib_drop_message(nal, private, hdr);
1044                 return;
1045         }
1046
1047         do_gettimeofday(&msg->ev.arrival_time);
1048
1049         switch (hdr->type) {
1050         case PTL_MSG_ACK:
1051                 rc = parse_ack(nal, hdr, private, msg);
1052                 break;
1053         case PTL_MSG_PUT:
1054                 rc = parse_put(nal, hdr, private, msg);
1055                 break;
1056         case PTL_MSG_GET:
1057                 rc = parse_get(nal, hdr, private, msg);
1058                 break;
1059         case PTL_MSG_REPLY:
1060                 rc = parse_reply(nal, hdr, private, msg);
1061                 break;
1062         default:
1063                 CERROR(LPU64": Dropping <unknown> message from "LPU64
1064                        ": Bad type=0x%x\n",  nal->ni.nid, hdr->src_nid,
1065                        hdr->type);
1066                 rc = PTL_FAIL;
1067                 break;
1068         }
1069
1070         if (rc != PTL_OK) {
1071                 if (msg->md != NULL) {
1072                         /* committed... */
1073                         lib_finalize(nal, private, msg, rc);
1074                 } else {
1075                         state_lock(nal, &flags);
1076                         lib_msg_free(nal, msg); /* expects state_lock held */
1077                         state_unlock(nal, &flags);
1078
1079                         lib_drop_message(nal, private, hdr);
1080                 }
1081         }
1082 }
1083
1084 int
1085 do_PtlPut(nal_cb_t *nal, void *private, void *v_args, void *v_ret)
1086 {
1087         /*
1088          * Incoming:
1089          *      ptl_handle_md_t md_in
1090          *      ptl_ack_req_t ack_req_in
1091          *      ptl_process_id_t target_in
1092          *      ptl_pt_index_t portal_in
1093          *      ptl_ac_index_t cookie_in
1094          *      ptl_match_bits_t match_bits_in
1095          *      ptl_size_t offset_in
1096          *
1097          * Outgoing:
1098          */
1099
1100         PtlPut_in        *args = v_args;
1101         ptl_process_id_t *id = &args->target_in;
1102         PtlPut_out       *ret = v_ret;
1103         lib_ni_t         *ni = &nal->ni;
1104         lib_msg_t        *msg;
1105         ptl_hdr_t         hdr;
1106         lib_md_t         *md;
1107         unsigned long     flags;
1108         int               rc;
1109
1110         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1111             fail_peer (nal, id->nid, 1))           /* shall we now? */
1112         {
1113                 CERROR(LPU64": Dropping PUT to "LPU64": simulated failure\n",
1114                        nal->ni.nid, id->nid);
1115                 return (ret->rc = PTL_INV_PROC);
1116         }
1117
1118         msg = lib_msg_alloc(nal);
1119         if (msg == NULL) {
1120                 CERROR(LPU64": Dropping PUT to "LPU64": ENOMEM on lib_msg_t\n",
1121                        ni->nid, id->nid);
1122                 return (ret->rc = PTL_NOSPACE);
1123         }
1124
1125         state_lock(nal, &flags);
1126
1127         md = ptl_handle2md(&args->md_in, nal);
1128         if (md == NULL || md->threshold == 0) {
1129                 lib_msg_free(nal, msg);
1130                 state_unlock(nal, &flags);
1131
1132                 return (ret->rc = PTL_INV_MD);
1133         }
1134
1135         CDEBUG(D_NET, "PtlPut -> %Lu: %lu\n", (unsigned long long)id->nid,
1136                (unsigned long)id->pid);
1137
1138         memset (&hdr, 0, sizeof (hdr));
1139         hdr.type     = HTON__u32 (PTL_MSG_PUT);
1140         hdr.dest_nid = HTON__u64 (id->nid);
1141         hdr.src_nid  = HTON__u64 (ni->nid);
1142         hdr.dest_pid = HTON__u32 (id->pid);
1143         hdr.src_pid  = HTON__u32 (ni->pid);
1144         hdr.payload_length = HTON__u32 (md->length);
1145
1146         /* NB handles only looked up by creator (no flips) */
1147         if (args->ack_req_in == PTL_ACK_REQ) {
1148                 hdr.msg.put.ack_wmd.wh_interface_cookie = ni->ni_interface_cookie;
1149                 hdr.msg.put.ack_wmd.wh_object_cookie = md->md_lh.lh_cookie;
1150         } else {
1151                 hdr.msg.put.ack_wmd = PTL_WIRE_HANDLE_NONE;
1152         }
1153
1154         hdr.msg.put.match_bits = HTON__u64 (args->match_bits_in);
1155         hdr.msg.put.ptl_index = HTON__u32 (args->portal_in);
1156         hdr.msg.put.offset = HTON__u32 (args->offset_in);
1157         hdr.msg.put.hdr_data = args->hdr_data_in;
1158
1159         lib_commit_md(nal, md, msg);
1160
1161         msg->ev.type = PTL_EVENT_SENT;
1162         msg->ev.initiator.nid = ni->nid;
1163         msg->ev.initiator.pid = ni->pid;
1164         msg->ev.portal = args->portal_in;
1165         msg->ev.match_bits = args->match_bits_in;
1166         msg->ev.rlength = md->length;
1167         msg->ev.mlength = md->length;
1168         msg->ev.offset = args->offset_in;
1169         msg->ev.hdr_data = args->hdr_data_in;
1170
1171         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
1172
1173         ni->counters.send_count++;
1174         ni->counters.send_length += md->length;
1175
1176         state_unlock(nal, &flags);
1177
1178         rc = lib_send (nal, private, msg, &hdr, PTL_MSG_PUT,
1179                        id->nid, id->pid, md, 0, md->length);
1180         if (rc != PTL_OK) {
1181                 CERROR(LPU64": error sending PUT to "LPU64": %d\n",
1182                        ni->nid, id->nid, rc);
1183                 lib_finalize (nal, private, msg, rc);
1184         }
1185
1186         /* completion will be signalled by an event */
1187         return ret->rc = PTL_OK;
1188 }
1189
1190 lib_msg_t *
1191 lib_fake_reply_msg (nal_cb_t *nal, ptl_nid_t peer_nid, lib_md_t *getmd)
1192 {
1193         /* The NAL can DMA direct to the GET md (i.e. no REPLY msg).  This
1194          * returns a msg the NAL can pass to lib_finalize() so that a REPLY
1195          * event still occurs.
1196          *
1197          * CAVEAT EMPTOR: 'getmd' is passed by pointer so it MUST be valid.
1198          * This can only be guaranteed while a lib_msg_t holds a reference
1199          * on it (ie. pending > 0), so best call this before the
1200          * lib_finalize() of the original GET. */
1201
1202         lib_ni_t        *ni = &nal->ni;
1203         lib_msg_t       *msg = lib_msg_alloc(nal);
1204         unsigned long    flags;
1205
1206         state_lock(nal, &flags);
1207
1208         LASSERT (getmd->pending > 0);
1209
1210         if (msg == NULL) {
1211                 CERROR ("Dropping REPLY from "LPU64": can't allocate msg\n",
1212                         peer_nid);
1213                 goto drop;
1214         }
1215
1216         if (getmd->threshold == 0) {
1217                 CERROR ("Dropping REPLY from "LPU64" for inactive MD %p\n",
1218                         peer_nid, getmd);
1219                 goto drop_msg;
1220         }
1221
1222         LASSERT (getmd->offset == 0);
1223
1224         CDEBUG(D_NET, "Reply from "LPU64" md %p\n", peer_nid, getmd);
1225
1226         lib_commit_md (nal, getmd, msg);
1227
1228         msg->ev.type = PTL_EVENT_REPLY;
1229         msg->ev.initiator.nid = peer_nid;
1230         msg->ev.initiator.pid = 0;      /* XXX FIXME!!! */
1231         msg->ev.rlength = msg->ev.mlength = getmd->length;
1232         msg->ev.offset = 0;
1233
1234         lib_md_deconstruct(nal, getmd, &msg->ev.mem_desc);
1235
1236         ni->counters.recv_count++;
1237         ni->counters.recv_length += getmd->length;
1238
1239         state_unlock(nal, &flags);
1240
1241         return msg;
1242
1243  drop_msg:
1244         lib_msg_free(nal, msg);
1245  drop:
1246         nal->ni.counters.drop_count++;
1247         nal->ni.counters.drop_length += getmd->length;
1248
1249         state_unlock (nal, &flags);
1250
1251         return NULL;
1252 }
1253
1254 int
1255 do_PtlGet(nal_cb_t *nal, void *private, void *v_args, void *v_ret)
1256 {
1257         /*
1258          * Incoming:
1259          *      ptl_handle_md_t md_in
1260          *      ptl_process_id_t target_in
1261          *      ptl_pt_index_t portal_in
1262          *      ptl_ac_index_t cookie_in
1263          *      ptl_match_bits_t match_bits_in
1264          *      ptl_size_t offset_in
1265          *
1266          * Outgoing:
1267          */
1268
1269         PtlGet_in        *args = v_args;
1270         ptl_process_id_t *id = &args->target_in;
1271         PtlGet_out       *ret = v_ret;
1272         lib_ni_t         *ni = &nal->ni;
1273         lib_msg_t        *msg;
1274         ptl_hdr_t         hdr;
1275         lib_md_t         *md;
1276         unsigned long     flags;
1277         int               rc;
1278
1279         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1280             fail_peer (nal, id->nid, 1))           /* shall we now? */
1281         {
1282                 CERROR(LPU64": Dropping PUT to "LPU64": simulated failure\n",
1283                        nal->ni.nid, id->nid);
1284                 return (ret->rc = PTL_INV_PROC);
1285         }
1286
1287         msg = lib_msg_alloc(nal);
1288         if (msg == NULL) {
1289                 CERROR(LPU64": Dropping GET to "LPU64": ENOMEM on lib_msg_t\n",
1290                        ni->nid, id->nid);
1291                 return (ret->rc = PTL_NOSPACE);
1292         }
1293
1294         state_lock(nal, &flags);
1295
1296         md = ptl_handle2md(&args->md_in, nal);
1297         if (md == NULL || !md->threshold) {
1298                 lib_msg_free(nal, msg);
1299                 state_unlock(nal, &flags);
1300
1301                 return ret->rc = PTL_INV_MD;
1302         }
1303
1304         CDEBUG(D_NET, "PtlGet -> %Lu: %lu\n", (unsigned long long)id->nid,
1305                (unsigned long)id->pid);
1306
1307         memset (&hdr, 0, sizeof (hdr));
1308         hdr.type     = HTON__u32 (PTL_MSG_GET);
1309         hdr.dest_nid = HTON__u64 (id->nid);
1310         hdr.src_nid  = HTON__u64 (ni->nid);
1311         hdr.dest_pid = HTON__u32 (id->pid);
1312         hdr.src_pid  = HTON__u32 (ni->pid);
1313         hdr.payload_length = 0;
1314
1315         /* NB handles only looked up by creator (no flips) */
1316         hdr.msg.get.return_wmd.wh_interface_cookie = ni->ni_interface_cookie;
1317         hdr.msg.get.return_wmd.wh_object_cookie = md->md_lh.lh_cookie;
1318
1319         hdr.msg.get.match_bits = HTON__u64 (args->match_bits_in);
1320         hdr.msg.get.ptl_index = HTON__u32 (args->portal_in);
1321         hdr.msg.get.src_offset = HTON__u32 (args->offset_in);
1322         hdr.msg.get.sink_length = HTON__u32 (md->length);
1323
1324         lib_commit_md(nal, md, msg);
1325
1326         msg->ev.type = PTL_EVENT_SENT;
1327         msg->ev.initiator.nid = ni->nid;
1328         msg->ev.initiator.pid = ni->pid;
1329         msg->ev.portal = args->portal_in;
1330         msg->ev.match_bits = args->match_bits_in;
1331         msg->ev.rlength = md->length;
1332         msg->ev.mlength = md->length;
1333         msg->ev.offset = args->offset_in;
1334         msg->ev.hdr_data = 0;
1335
1336         lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
1337
1338         ni->counters.send_count++;
1339
1340         state_unlock(nal, &flags);
1341
1342         rc = lib_send (nal, private, msg, &hdr, PTL_MSG_GET,
1343                        id->nid, id->pid, NULL, 0, 0);
1344         if (rc != PTL_OK) {
1345                 CERROR(LPU64": error sending GET to "LPU64": %d\n",
1346                        ni->nid, id->nid, rc);
1347                 lib_finalize (nal, private, msg, rc);
1348         }
1349
1350         /* completion will be signalled by an event */
1351         return ret->rc = PTL_OK;
1352 }
1353
1354 void lib_assert_wire_constants (void)
1355 {
1356         /* Wire protocol assertions generated by 'wirecheck'
1357          * running on Linux robert.bartonsoftware.com 2.4.20-18.9 #1 Thu May 29 06:54:41 EDT 2003 i68
1358          * with gcc version 3.2.2 20030222 (Red Hat Linux 3.2.2-5) */
1359
1360
1361         /* Constants... */
1362         LASSERT (PORTALS_PROTO_MAGIC == 0xeebc0ded);
1363         LASSERT (PORTALS_PROTO_VERSION_MAJOR == 0);
1364         LASSERT (PORTALS_PROTO_VERSION_MINOR == 3);
1365         LASSERT (PTL_MSG_ACK == 0);
1366         LASSERT (PTL_MSG_PUT == 1);
1367         LASSERT (PTL_MSG_GET == 2);
1368         LASSERT (PTL_MSG_REPLY == 3);
1369         LASSERT (PTL_MSG_HELLO == 4);
1370
1371         /* Checks for struct ptl_handle_wire_t */
1372         LASSERT ((int)sizeof(ptl_handle_wire_t) == 16);
1373         LASSERT (offsetof(ptl_handle_wire_t, wh_interface_cookie) == 0);
1374         LASSERT ((int)sizeof(((ptl_handle_wire_t *)0)->wh_interface_cookie) == 8);
1375         LASSERT (offsetof(ptl_handle_wire_t, wh_object_cookie) == 8);
1376         LASSERT ((int)sizeof(((ptl_handle_wire_t *)0)->wh_object_cookie) == 8);
1377
1378         /* Checks for struct ptl_magicversion_t */
1379         LASSERT ((int)sizeof(ptl_magicversion_t) == 8);
1380         LASSERT (offsetof(ptl_magicversion_t, magic) == 0);
1381         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->magic) == 4);
1382         LASSERT (offsetof(ptl_magicversion_t, version_major) == 4);
1383         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->version_major) == 2);
1384         LASSERT (offsetof(ptl_magicversion_t, version_minor) == 6);
1385         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->version_minor) == 2);
1386
1387         /* Checks for struct ptl_hdr_t */
1388         LASSERT ((int)sizeof(ptl_hdr_t) == 72);
1389         LASSERT (offsetof(ptl_hdr_t, dest_nid) == 0);
1390         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->dest_nid) == 8);
1391         LASSERT (offsetof(ptl_hdr_t, src_nid) == 8);
1392         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->src_nid) == 8);
1393         LASSERT (offsetof(ptl_hdr_t, dest_pid) == 16);
1394         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->dest_pid) == 4);
1395         LASSERT (offsetof(ptl_hdr_t, src_pid) == 20);
1396         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->src_pid) == 4);
1397         LASSERT (offsetof(ptl_hdr_t, type) == 24);
1398         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->type) == 4);
1399         LASSERT (offsetof(ptl_hdr_t, payload_length) == 28);
1400         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->payload_length) == 4);
1401         LASSERT (offsetof(ptl_hdr_t, msg) == 32);
1402         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg) == 40);
1403
1404         /* Ack */
1405         LASSERT (offsetof(ptl_hdr_t, msg.ack.dst_wmd) == 32);
1406         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.dst_wmd) == 16);
1407         LASSERT (offsetof(ptl_hdr_t, msg.ack.match_bits) == 48);
1408         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.match_bits) == 8);
1409         LASSERT (offsetof(ptl_hdr_t, msg.ack.mlength) == 56);
1410         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.mlength) == 4);
1411
1412         /* Put */
1413         LASSERT (offsetof(ptl_hdr_t, msg.put.ack_wmd) == 32);
1414         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.ack_wmd) == 16);
1415         LASSERT (offsetof(ptl_hdr_t, msg.put.match_bits) == 48);
1416         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.match_bits) == 8);
1417         LASSERT (offsetof(ptl_hdr_t, msg.put.hdr_data) == 56);
1418         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.hdr_data) == 8);
1419         LASSERT (offsetof(ptl_hdr_t, msg.put.ptl_index) == 64);
1420         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.ptl_index) == 4);
1421         LASSERT (offsetof(ptl_hdr_t, msg.put.offset) == 68);
1422         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.offset) == 4);
1423
1424         /* Get */
1425         LASSERT (offsetof(ptl_hdr_t, msg.get.return_wmd) == 32);
1426         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.return_wmd) == 16);
1427         LASSERT (offsetof(ptl_hdr_t, msg.get.match_bits) == 48);
1428         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.match_bits) == 8);
1429         LASSERT (offsetof(ptl_hdr_t, msg.get.ptl_index) == 56);
1430         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.ptl_index) == 4);
1431         LASSERT (offsetof(ptl_hdr_t, msg.get.src_offset) == 60);
1432         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.src_offset) == 4);
1433         LASSERT (offsetof(ptl_hdr_t, msg.get.sink_length) == 64);
1434         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.sink_length) == 4);
1435
1436         /* Reply */
1437         LASSERT (offsetof(ptl_hdr_t, msg.reply.dst_wmd) == 32);
1438         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.reply.dst_wmd) == 16);
1439
1440         /* Hello */
1441         LASSERT (offsetof(ptl_hdr_t, msg.hello.incarnation) == 32);
1442         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.hello.incarnation) == 8);
1443         LASSERT (offsetof(ptl_hdr_t, msg.hello.type) == 40);
1444         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.hello.type) == 4);
1445 }