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, ptl_size_t len)
 262 {
 263         ptl_size_t nob;
 264
 265         while (len > 0)
 266         {
 267                 LASSERT (niov > 0);
 268                 nob = MIN (iov->iov_len, len);
 269                 memcpy (dest, iov->iov_base, nob);
 270
 271                 len -= nob;
 272                 dest += nob;
 273                 niov--;
 274                 iov++;
 275         }
 276 }
 277
 278 void
 279 lib_copy_buf2iov (int niov, struct iovec *iov, char *src, ptl_size_t len)
 280 {
 281         ptl_size_t nob;
 282
 283         while (len > 0)
 284         {
 285                 LASSERT (niov > 0);
 286                 nob = MIN (iov->iov_len, len);
 287                 memcpy (iov->iov_base, src, nob);
 288
 289                 len -= nob;
 290                 src += nob;
 291                 niov--;
 292                 iov++;
 293         }
 294 }
 295
 296 static int
 297 lib_extract_iov (struct iovec *dst, lib_md_t *md,
 298                  ptl_size_t offset, ptl_size_t len)
 299 {
 300         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 301          * for exactly 'len' bytes, and return the number of entries.
 302          * NB not destructive to 'src' */
 303         int             src_niov = md->md_niov;
 304         struct iovec   *src = md->md_iov.iov;
 305         ptl_size_t      frag_len;
 306         int             dst_niov;
 307
 308         LASSERT (offset + len <= md->length);
 309
 310         if (len == 0)                           /* no data => */
 311                 return (0);                     /* no frags */
 312
 313         LASSERT (src_niov > 0);
 314         while (offset >= src->iov_len) {      /* skip initial frags */
 315                 offset -= src->iov_len;
 316                 src_niov--;
 317                 src++;
 318                 LASSERT (src_niov > 0);
 319         }
 320
 321         dst_niov = 1;
 322         for (;;) {
 323                 LASSERT (src_niov > 0);
 324                 LASSERT (dst_niov <= PTL_MD_MAX_IOV);
 325
 326                 frag_len = src->iov_len - offset;
 327                 dst->iov_base = ((char *)src->iov_base) + offset;
 328
 329                 if (len <= frag_len) {
 330                         dst->iov_len = len;
 331                         return (dst_niov);
 332                 }
 333
 334                 dst->iov_len = frag_len;
 335
 336                 len -= frag_len;
 337                 dst++;
 338                 src++;
 339                 dst_niov++;
 340                 src_niov--;
 341                 offset = 0;
 342         }
 343 }
 344
 345 #ifndef __KERNEL__
 346 ptl_size_t
 347 lib_kiov_nob (int niov, ptl_kiov_t *kiov)
 348 {
 349         LASSERT (0);
 350         return (0);
 351 }
 352
 353 void
 354 lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, ptl_size_t len)
 355 {
 356         LASSERT (0);
 357 }
 358
 359 void
 360 lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, char *dest, ptl_size_t len)
 361 {
 362         LASSERT (0);
 363 }
 364
 365 static int
 366 lib_extract_kiov (ptl_kiov_t *dst, lib_md_t *md,
 367                   ptl_size_t offset, ptl_size_t len)
 368 {
 369         LASSERT (0);
 370 }
 371
 372 #else
 373
 374 ptl_size_t
 375 lib_kiov_nob (int niov, ptl_kiov_t *kiov)
 376 {
 377         ptl_size_t  nob = 0;
 378
 379         while (niov-- > 0)
 380                 nob += (kiov++)->kiov_len;
 381
 382         return (nob);
 383 }
 384
 385 void
 386 lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, ptl_size_t len)
 387 {
 388         ptl_size_t  nob;
 389         char       *addr;
 390
 391         LASSERT (!in_interrupt ());
 392         while (len > 0)
 393         {
 394                 LASSERT (niov > 0);
 395                 nob = MIN (kiov->kiov_len, len);
 396
 397                 addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset;
 398                 memcpy (dest, addr, nob);
 399                 kunmap (kiov->kiov_page);
 400
 401                 len -= nob;
 402                 dest += nob;
 403                 niov--;
 404                 kiov++;
 405         }
 406 }
 407
 408 void
 409 lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, char *src, ptl_size_t len)
 410 {
 411         ptl_size_t  nob;
 412         char       *addr;
 413
 414         LASSERT (!in_interrupt ());
 415         while (len > 0)
 416         {
 417                 LASSERT (niov > 0);
 418                 nob = MIN (kiov->kiov_len, len);
 419
 420                 addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset;
 421                 memcpy (addr, src, nob);
 422                 kunmap (kiov->kiov_page);
 423
 424                 len -= nob;
 425                 src += nob;
 426                 niov--;
 427                 kiov++;
 428         }
 429 }
 430
 431 static int
 432 lib_extract_kiov (ptl_kiov_t *dst, lib_md_t *md,
 433                   ptl_size_t offset, ptl_size_t len)
 434 {
 435         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 436          * for exactly 'len' bytes, and return the number of entries.
 437          * NB not destructive to 'src' */
 438         int             src_niov = md->md_niov;
 439         ptl_kiov_t     *src = md->md_iov.kiov;
 440         ptl_size_t      frag_len;
 441         int             dst_niov;
 442
 443         LASSERT (offset + len <= md->length);
 444
 445         if (len == 0)                           /* no data => */
 446                 return (0);                     /* no frags */
 447
 448         LASSERT (src_niov > 0);
 449         while (offset >= src->kiov_len) {      /* skip initial frags */
 450                 offset -= src->kiov_len;
 451                 src_niov--;
 452                 src++;
 453                 LASSERT (src_niov > 0);
 454         }
 455
 456         dst_niov = 1;
 457         for (;;) {
 458                 LASSERT (src_niov > 0);
 459                 LASSERT (dst_niov <= PTL_MD_MAX_IOV);
 460
 461                 frag_len = src->kiov_len - offset;
 462                 dst->kiov_page = src->kiov_page;
 463                 dst->kiov_offset = src->kiov_offset + offset;
 464
 465                 if (len <= frag_len) {
 466                         dst->kiov_len = len;
 467                         LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_SIZE);
 468                         return (dst_niov);
 469                 }
 470
 471                 dst->kiov_len = frag_len;
 472                 LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_SIZE);
 473
 474                 len -= frag_len;
 475                 dst++;
 476                 src++;
 477                 dst_niov++;
 478                 src_niov--;
 479                 offset = 0;
 480         }
 481 }
 482 #endif
 483
 484 void
 485 lib_recv (nal_cb_t *nal, void *private, lib_msg_t *msg, lib_md_t *md,
 486           ptl_size_t offset, ptl_size_t mlen, ptl_size_t rlen)
 487 {
 488         int   niov;
 489
 490         if (mlen == 0)
 491                 nal->cb_recv (nal, private, msg, 0, NULL, 0, rlen);
 492         else if ((md->options & PTL_MD_KIOV) == 0) {
 493                 niov = lib_extract_iov (msg->msg_iov.iov, md, offset, mlen);
 494                 nal->cb_recv (nal, private, msg,
 495                               niov, msg->msg_iov.iov, mlen, rlen);
 496         } else {
 497                 niov = lib_extract_kiov (msg->msg_iov.kiov, md, offset, mlen);
 498                 nal->cb_recv_pages (nal, private, msg,
 499                                     niov, msg->msg_iov.kiov, mlen, rlen);
 500         }
 501 }
 502
 503 int
 504 lib_send (nal_cb_t *nal, void *private, lib_msg_t *msg,
 505           ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
 506           lib_md_t *md, ptl_size_t offset, ptl_size_t len)
 507 {
 508         int   niov;
 509
 510         if (len == 0)
 511                 return (nal->cb_send (nal, private, msg,
 512                                       hdr, type, nid, pid,
 513                                       0, NULL, 0));
 514
 515         if ((md->options & PTL_MD_KIOV) == 0) {
 516                 niov = lib_extract_iov (msg->msg_iov.iov, md, offset, len);
 517                 return (nal->cb_send (nal, private, msg,
 518                                       hdr, type, nid, pid,
 519                                       niov, msg->msg_iov.iov, len));
 520         }
 521
 522         niov = lib_extract_kiov (msg->msg_iov.kiov, md, offset, len);
 523         return (nal->cb_send_pages (nal, private, msg,
 524                                     hdr, type, nid, pid,
 525                                     niov, msg->msg_iov.kiov, len));
 526 }
 527
 528 static lib_msg_t *
 529 get_new_msg (nal_cb_t *nal, lib_md_t *md)
 530 {
 531         /* ALWAYS called holding the state_lock */
 532         lib_counters_t *counters = &nal->ni.counters;
 533         lib_msg_t      *msg      = lib_msg_alloc (nal);
 534
 535         if (msg == NULL)
 536                 return (NULL);
 537
 538         memset (msg, 0, sizeof (*msg));
 539
 540         msg->send_ack = 0;
 541
 542         msg->md = md;
 543         do_gettimeofday(&msg->ev.arrival_time);
 544         md->pending++;
 545         if (md->threshold != PTL_MD_THRESH_INF) {
 546                 LASSERT (md->threshold > 0);
 547                 md->threshold--;
 548         }
 549
 550         counters->msgs_alloc++;
 551         if (counters->msgs_alloc > counters->msgs_max)
 552                 counters->msgs_max = counters->msgs_alloc;
 553
 554         list_add (&msg->msg_list, &nal->ni.ni_active_msgs);
 555
 556         return (msg);
 557 }
 558
 559 /*
 560  * Incoming messages have a ptl_msg_t object associated with them
 561  * by the library.  This object encapsulates the state of the
 562  * message and allows the NAL to do non-blocking receives or sends
 563  * of long messages.
 564  *
 565  */
 566 static int parse_put(nal_cb_t * nal, ptl_hdr_t * hdr, void *private)
 567 {
 568         lib_ni_t        *ni = &nal->ni;
 569         ptl_size_t       mlength = 0;
 570         ptl_size_t       offset = 0;
 571         int              unlink = 0;
 572         lib_me_t        *me;
 573         lib_md_t        *md;
 574         lib_msg_t       *msg;
 575         unsigned long    flags;
 576
 577         /* Convert put fields to host byte order */
 578         hdr->msg.put.match_bits = NTOH__u64 (hdr->msg.put.match_bits);
 579         hdr->msg.put.ptl_index = NTOH__u32 (hdr->msg.put.ptl_index);
 580         hdr->msg.put.offset = NTOH__u32 (hdr->msg.put.offset);
 581
 582         state_lock(nal, &flags);
 583
 584         me = lib_find_me(nal, hdr->msg.put.ptl_index, PTL_MD_OP_PUT,
 585                          hdr->src_nid, hdr->src_pid,
 586                          hdr->payload_length, hdr->msg.put.offset,
 587                          hdr->msg.put.match_bits,
 588                          &mlength, &offset, &unlink);
 589         if (me == NULL)
 590                 goto drop;
 591
 592         md = me->md;
 593         CDEBUG(D_NET, "Incoming put index %x from "LPU64"/%u of length %d/%d "
 594                "into md "LPX64" [%d] + %d\n", hdr->msg.put.ptl_index,
 595                hdr->src_nid, hdr->src_pid, mlength, hdr->payload_length,
 596                md->md_lh.lh_cookie, md->md_niov, offset);
 597
 598         msg = get_new_msg (nal, md);
 599         if (msg == NULL) {
 600                 CERROR(LPU64": Dropping PUT from "LPU64": can't allocate msg\n",
 601                        ni->nid, hdr->src_nid);
 602                 goto drop;
 603         }
 604
 605         if (!ptl_is_wire_handle_none(&hdr->msg.put.ack_wmd) &&
 606             !(md->options & PTL_MD_ACK_DISABLE)) {
 607                 msg->send_ack = 1;
 608                 msg->ack_wmd = hdr->msg.put.ack_wmd;
 609                 msg->nid = hdr->src_nid;
 610                 msg->pid = hdr->src_pid;
 611                 msg->ev.match_bits = hdr->msg.put.match_bits;
 612         }
 613
 614         if (md->eq) {
 615                 msg->ev.type = PTL_EVENT_PUT;
 616                 msg->ev.initiator.nid = hdr->src_nid;
 617                 msg->ev.initiator.pid = hdr->src_pid;
 618                 msg->ev.portal = hdr->msg.put.ptl_index;
 619                 msg->ev.match_bits = hdr->msg.put.match_bits;
 620                 msg->ev.rlength = hdr->payload_length;
 621                 msg->ev.mlength = mlength;
 622                 msg->ev.offset = offset;
 623                 msg->ev.hdr_data = hdr->msg.put.hdr_data;
 624
 625                 /* NB if this match has exhausted the MD, we can't be sure
 626                  * that this event will the the last one associated with
 627                  * this MD in the event queue (another message already
 628                  * matching this ME/MD could end up being last).  So we
 629                  * remember the ME handle anyway and check again when we're
 630                  * allocating our slot in the event queue.
 631                  */
 632                 ptl_me2handle (&msg->ev.unlinked_me, me);
 633
 634                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 635         }
 636
 637         ni->counters.recv_count++;
 638         ni->counters.recv_length += mlength;
 639
 640         /* only unlink after MD's pending count has been bumped
 641          * in get_new_msg() otherwise lib_me_unlink() will nuke it */
 642         if (unlink) {
 643                 md->md_flags |= PTL_MD_FLAG_AUTO_UNLINKED;
 644                 lib_me_unlink (nal, me);
 645         }
 646
 647         state_unlock(nal, &flags);
 648
 649         lib_recv (nal, private, msg, md, offset, mlength, hdr->payload_length);
 650         return 0;
 651
 652  drop:
 653         nal->ni.counters.drop_count++;
 654         nal->ni.counters.drop_length += hdr->payload_length;
 655         state_unlock (nal, &flags);
 656         lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 657         return -1;
 658 }
 659
 660 static int parse_get(nal_cb_t * nal, ptl_hdr_t * hdr, void *private)
 661 {
 662         lib_ni_t        *ni = &nal->ni;
 663         ptl_size_t       mlength = 0;
 664         ptl_size_t       offset = 0;
 665         int              unlink = 0;
 666         lib_me_t        *me;
 667         lib_md_t        *md;
 668         lib_msg_t       *msg;
 669         ptl_hdr_t        reply;
 670         unsigned long    flags;
 671         int              rc;
 672
 673         /* Convert get fields to host byte order */
 674         hdr->msg.get.match_bits = NTOH__u64 (hdr->msg.get.match_bits);
 675         hdr->msg.get.ptl_index = NTOH__u32 (hdr->msg.get.ptl_index);
 676         hdr->msg.get.sink_length = NTOH__u32 (hdr->msg.get.sink_length);
 677         hdr->msg.get.src_offset = NTOH__u32 (hdr->msg.get.src_offset);
 678
 679         state_lock(nal, &flags);
 680
 681         me = lib_find_me(nal, hdr->msg.get.ptl_index, PTL_MD_OP_GET,
 682                          hdr->src_nid, hdr->src_pid,
 683                          hdr->msg.get.sink_length, hdr->msg.get.src_offset,
 684                          hdr->msg.get.match_bits,
 685                          &mlength, &offset, &unlink);
 686         if (me == NULL)
 687                 goto drop;
 688
 689         md = me->md;
 690         CDEBUG(D_NET, "Incoming get index %d from "LPU64".%u of length %d/%d "
 691                "from md "LPX64" [%d] + %d\n", hdr->msg.get.ptl_index,
 692                hdr->src_nid, hdr->src_pid, mlength, hdr->payload_length,
 693                md->md_lh.lh_cookie, md->md_niov, offset);
 694
 695         msg = get_new_msg (nal, md);
 696         if (msg == NULL) {
 697                 CERROR(LPU64": Dropping GET from "LPU64": can't allocate msg\n",
 698                        ni->nid, hdr->src_nid);
 699                 goto drop;
 700         }
 701
 702         if (md->eq) {
 703                 msg->ev.type = PTL_EVENT_GET;
 704                 msg->ev.initiator.nid = hdr->src_nid;
 705                 msg->ev.initiator.pid = hdr->src_pid;
 706                 msg->ev.portal = hdr->msg.get.ptl_index;
 707                 msg->ev.match_bits = hdr->msg.get.match_bits;
 708                 msg->ev.rlength = hdr->payload_length;
 709                 msg->ev.mlength = mlength;
 710                 msg->ev.offset = offset;
 711                 msg->ev.hdr_data = 0;
 712
 713                 /* NB if this match has exhausted the MD, we can't be sure
 714                  * that this event will the the last one associated with
 715                  * this MD in the event queue (another message already
 716                  * matching this ME/MD could end up being last).  So we
 717                  * remember the ME handle anyway and check again when we're
 718                  * allocating our slot in the event queue.
 719                  */
 720                 ptl_me2handle (&msg->ev.unlinked_me, me);
 721
 722                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 723         }
 724
 725         ni->counters.send_count++;
 726         ni->counters.send_length += mlength;
 727
 728         /* only unlink after MD's refcount has been bumped
 729          * in get_new_msg() otherwise lib_me_unlink() will nuke it */
 730         if (unlink) {
 731                 md->md_flags |= PTL_MD_FLAG_AUTO_UNLINKED;
 732                 lib_me_unlink (nal, me);
 733         }
 734
 735         state_unlock(nal, &flags);
 736
 737         memset (&reply, 0, sizeof (reply));
 738         reply.type     = HTON__u32 (PTL_MSG_REPLY);
 739         reply.dest_nid = HTON__u64 (hdr->src_nid);
 740         reply.src_nid  = HTON__u64 (ni->nid);
 741         reply.dest_pid = HTON__u32 (hdr->src_pid);
 742         reply.src_pid  = HTON__u32 (ni->pid);
 743         reply.payload_length = HTON__u32 (mlength);
 744
 745         reply.msg.reply.dst_wmd = hdr->msg.get.return_wmd;
 746
 747         /* NB call lib_send() _BEFORE_ lib_recv() completes the incoming
 748          * message.  Some NALs _require_ this to implement optimized GET */
 749
 750         rc = lib_send (nal, private, msg, &reply, PTL_MSG_REPLY,
 751                        hdr->src_nid, hdr->src_pid, md, offset, mlength);
 752         if (rc != PTL_OK) {
 753                 CERROR(LPU64": Dropping GET from "LPU64": send REPLY failed\n",
 754                        ni->nid, hdr->src_nid);
 755                 /* Hmm, this will create a GET event and make believe
 756                  * the reply completed, which it kind of did, only the
 757                  * source won't get her reply */
 758                 lib_finalize (nal, private, msg);
 759                 state_lock (nal, &flags);
 760                 goto drop;
 761         }
 762
 763         /* Complete the incoming message */
 764         lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 765         return (rc);
 766  drop:
 767         ni->counters.drop_count++;
 768         ni->counters.drop_length += hdr->msg.get.sink_length;
 769         state_unlock(nal, &flags);
 770         lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 771         return -1;
 772 }
 773
 774 static int parse_reply(nal_cb_t * nal, ptl_hdr_t * hdr, void *private)
 775 {
 776         lib_ni_t        *ni = &nal->ni;
 777         lib_md_t        *md;
 778         int              rlength;
 779         int              length;
 780         lib_msg_t       *msg;
 781         unsigned long    flags;
 782
 783         state_lock(nal, &flags);
 784
 785         /* NB handles only looked up by creator (no flips) */
 786         md = ptl_wire_handle2md(&hdr->msg.reply.dst_wmd, nal);
 787         if (md == NULL || md->threshold == 0) {
 788                 CERROR (LPU64": Dropping REPLY from "LPU64" for %s MD "LPX64"."LPX64"\n",
 789                         ni->nid, hdr->src_nid,
 790                         md == NULL ? "invalid" : "inactive",
 791                         hdr->msg.reply.dst_wmd.wh_interface_cookie,
 792                         hdr->msg.reply.dst_wmd.wh_object_cookie);
 793                 goto drop;
 794         }
 795
 796         LASSERT (md->offset == 0);
 797
 798         length = rlength = hdr->payload_length;
 799
 800         if (length > md->length) {
 801                 if ((md->options & PTL_MD_TRUNCATE) == 0) {
 802                         CERROR (LPU64": Dropping REPLY from "LPU64
 803                                 " length %d for MD "LPX64" would overflow (%d)\n",
 804                                 ni->nid, hdr->src_nid, length,
 805                                 hdr->msg.reply.dst_wmd.wh_object_cookie,
 806                                 md->length);
 807                         goto drop;
 808                 }
 809                 length = md->length;
 810         }
 811
 812         CDEBUG(D_NET, "Reply from "LPU64" of length %d/%d into md "LPX64"\n",
 813                hdr->src_nid, length, rlength,
 814                hdr->msg.reply.dst_wmd.wh_object_cookie);
 815
 816         msg = get_new_msg (nal, md);
 817         if (msg == NULL) {
 818                 CERROR(LPU64": Dropping REPLY from "LPU64": can't "
 819                        "allocate msg\n", ni->nid, hdr->src_nid);
 820                 goto drop;
 821         }
 822
 823         if (md->eq) {
 824                 msg->ev.type = PTL_EVENT_REPLY;
 825                 msg->ev.initiator.nid = hdr->src_nid;
 826                 msg->ev.initiator.pid = hdr->src_pid;
 827                 msg->ev.rlength = rlength;
 828                 msg->ev.mlength = length;
 829                 msg->ev.offset = 0;
 830
 831                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 832         }
 833
 834         ni->counters.recv_count++;
 835         ni->counters.recv_length += length;
 836
 837         state_unlock(nal, &flags);
 838
 839         lib_recv (nal, private, msg, md, 0, length, rlength);
 840         return 0;
 841
 842  drop:
 843         nal->ni.counters.drop_count++;
 844         nal->ni.counters.drop_length += hdr->payload_length;
 845         state_unlock (nal, &flags);
 846         lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 847         return -1;
 848 }
 849
 850 static int parse_ack(nal_cb_t * nal, ptl_hdr_t * hdr, void *private)
 851 {
 852         lib_ni_t *ni = &nal->ni;
 853         lib_md_t *md;
 854         lib_msg_t *msg = NULL;
 855         unsigned long flags;
 856
 857         /* Convert ack fields to host byte order */
 858         hdr->msg.ack.match_bits = NTOH__u64 (hdr->msg.ack.match_bits);
 859         hdr->msg.ack.mlength = NTOH__u32 (hdr->msg.ack.mlength);
 860
 861         state_lock(nal, &flags);
 862
 863         /* NB handles only looked up by creator (no flips) */
 864         md = ptl_wire_handle2md(&hdr->msg.ack.dst_wmd, nal);
 865         if (md == NULL || md->threshold == 0) {
 866                 CDEBUG(D_INFO, LPU64": Dropping ACK from "LPU64" to %s MD "
 867                        LPX64"."LPX64"\n", ni->nid, hdr->src_nid,
 868                        (md == NULL) ? "invalid" : "inactive",
 869                        hdr->msg.ack.dst_wmd.wh_interface_cookie,
 870                        hdr->msg.ack.dst_wmd.wh_object_cookie);
 871                 goto drop;
 872         }
 873
 874         CDEBUG(D_NET, LPU64": ACK from "LPU64" into md "LPX64"\n",
 875                ni->nid, hdr->src_nid,
 876                hdr->msg.ack.dst_wmd.wh_object_cookie);
 877
 878         msg = get_new_msg (nal, md);
 879         if (msg == NULL) {
 880                 CERROR(LPU64": Dropping ACK from "LPU64": can't allocate msg\n",
 881                        ni->nid, hdr->src_nid);
 882                 goto drop;
 883         }
 884
 885         if (md->eq) {
 886                 msg->ev.type = PTL_EVENT_ACK;
 887                 msg->ev.initiator.nid = hdr->src_nid;
 888                 msg->ev.initiator.pid = hdr->src_pid;
 889                 msg->ev.mlength = hdr->msg.ack.mlength;
 890                 msg->ev.match_bits = hdr->msg.ack.match_bits;
 891
 892                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
 893         }
 894
 895         ni->counters.recv_count++;
 896         state_unlock(nal, &flags);
 897         lib_recv (nal, private, msg, NULL, 0, 0, hdr->payload_length);
 898         return 0;
 899
 900  drop:
 901         nal->ni.counters.drop_count++;
 902         state_unlock (nal, &flags);
 903         lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
 904         return -1;
 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 int lib_parse(nal_cb_t * nal, ptl_hdr_t * hdr, void *private)
 987 {
 988         unsigned long  flags;
 989
 990         /* convert common fields to host byte order */
 991         hdr->dest_nid = NTOH__u64 (hdr->dest_nid);
 992         hdr->src_nid = NTOH__u64 (hdr->src_nid);
 993         hdr->dest_pid = NTOH__u32 (hdr->dest_pid);
 994         hdr->src_pid = NTOH__u32 (hdr->src_pid);
 995         hdr->type = NTOH__u32 (hdr->type);
 996         hdr->payload_length = NTOH__u32(hdr->payload_length);
 997 #if 0
 998         nal->cb_printf(nal, "%d: lib_parse: nal=%p hdr=%p type=%d\n",
 999                        nal->ni.nid, nal, hdr, hdr->type);
1000         print_hdr(nal, hdr);
1001 #endif
1002         if (hdr->type == PTL_MSG_HELLO) {
1003                 /* dest_nid is really ptl_magicversion_t */
1004                 ptl_magicversion_t *mv = (ptl_magicversion_t *)&hdr->dest_nid;
1005
1006                 CERROR (LPU64": Dropping unexpected HELLO message: "
1007                         "magic %d, version %d.%d from "LPD64"\n",
1008                         nal->ni.nid, mv->magic,
1009                         mv->version_major, mv->version_minor,
1010                         hdr->src_nid);
1011                 lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
1012                 return (-1);
1013         }
1014
1015         if (hdr->dest_nid != nal->ni.nid) {
1016                 CERROR(LPU64": Dropping %s message from "LPU64" to "LPU64
1017                        " (not me)\n", nal->ni.nid, hdr_type_string (hdr),
1018                        hdr->src_nid, hdr->dest_nid);
1019
1020                 state_lock (nal, &flags);
1021                 nal->ni.counters.drop_count++;
1022                 nal->ni.counters.drop_length += hdr->payload_length;
1023                 state_unlock (nal, &flags);
1024
1025                 lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
1026                 return (-1);
1027         }
1028
1029         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1030             fail_peer (nal, hdr->src_nid, 0))      /* shall we now? */
1031         {
1032                 CERROR(LPU64": Dropping incoming %s from "LPU64
1033                        ": simulated failure\n",
1034                        nal->ni.nid, hdr_type_string (hdr),
1035                        hdr->src_nid);
1036                 lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
1037                 return (-1);
1038         }
1039
1040         switch (hdr->type) {
1041         case PTL_MSG_ACK:
1042                 return (parse_ack(nal, hdr, private));
1043         case PTL_MSG_PUT:
1044                 return (parse_put(nal, hdr, private));
1045                 break;
1046         case PTL_MSG_GET:
1047                 return (parse_get(nal, hdr, private));
1048                 break;
1049         case PTL_MSG_REPLY:
1050                 return (parse_reply(nal, hdr, private));
1051                 break;
1052         default:
1053                 CERROR(LPU64": Dropping <unknown> message from "LPU64
1054                        ": Bad type=0x%x\n",  nal->ni.nid, hdr->src_nid,
1055                        hdr->type);
1056
1057                 lib_recv (nal, private, NULL, NULL, 0, 0, hdr->payload_length);
1058                 return (-1);
1059         }
1060 }
1061
1062
1063 int do_PtlPut(nal_cb_t * nal, void *private, void *v_args, void *v_ret)
1064 {
1065         /*
1066          * Incoming:
1067          *      ptl_handle_md_t md_in
1068          *      ptl_ack_req_t ack_req_in
1069          *      ptl_process_id_t target_in
1070          *      ptl_pt_index_t portal_in
1071          *      ptl_ac_index_t cookie_in
1072          *      ptl_match_bits_t match_bits_in
1073          *      ptl_size_t offset_in
1074          *
1075          * Outgoing:
1076          */
1077
1078         PtlPut_in *args = v_args;
1079         PtlPut_out *ret = v_ret;
1080         ptl_hdr_t hdr;
1081
1082         lib_ni_t *ni = &nal->ni;
1083         lib_md_t *md;
1084         lib_msg_t *msg = NULL;
1085         ptl_process_id_t *id = &args->target_in;
1086         unsigned long flags;
1087         int           rc;
1088
1089         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1090             fail_peer (nal, id->nid, 1))           /* shall we now? */
1091         {
1092                 CERROR(LPU64": Dropping PUT to "LPU64": simulated failure\n",
1093                        nal->ni.nid, id->nid);
1094                 return (ret->rc = PTL_INV_PROC);
1095         }
1096
1097         ret->rc = PTL_OK;
1098         state_lock(nal, &flags);
1099         md = ptl_handle2md(&args->md_in, nal);
1100         if (md == NULL || !md->threshold) {
1101                 state_unlock(nal, &flags);
1102                 return ret->rc = PTL_INV_MD;
1103         }
1104
1105         CDEBUG(D_NET, "PtlPut -> %Lu: %lu\n", (unsigned long long)id->nid,
1106                (unsigned long)id->pid);
1107
1108         memset (&hdr, 0, sizeof (hdr));
1109         hdr.type     = HTON__u32 (PTL_MSG_PUT);
1110         hdr.dest_nid = HTON__u64 (id->nid);
1111         hdr.src_nid  = HTON__u64 (ni->nid);
1112         hdr.dest_pid = HTON__u32 (id->pid);
1113         hdr.src_pid  = HTON__u32 (ni->pid);
1114         hdr.payload_length = HTON__u32 (md->length);
1115
1116         /* NB handles only looked up by creator (no flips) */
1117         if (args->ack_req_in == PTL_ACK_REQ) {
1118                 hdr.msg.put.ack_wmd.wh_interface_cookie = ni->ni_interface_cookie;
1119                 hdr.msg.put.ack_wmd.wh_object_cookie = md->md_lh.lh_cookie;
1120         } else {
1121                 hdr.msg.put.ack_wmd = PTL_WIRE_HANDLE_NONE;
1122         }
1123
1124         hdr.msg.put.match_bits = HTON__u64 (args->match_bits_in);
1125         hdr.msg.put.ptl_index = HTON__u32 (args->portal_in);
1126         hdr.msg.put.offset = HTON__u32 (args->offset_in);
1127         hdr.msg.put.hdr_data = args->hdr_data_in;
1128
1129         ni->counters.send_count++;
1130         ni->counters.send_length += md->length;
1131
1132         msg = get_new_msg (nal, md);
1133         if (msg == NULL) {
1134                 CERROR("BAD: could not allocate msg!\n");
1135                 state_unlock(nal, &flags);
1136                 return ret->rc = PTL_NOSPACE;
1137         }
1138
1139         /*
1140          * If this memory descriptor has an event queue associated with
1141          * it we need to allocate a message state object and record the
1142          * information about this operation that will be recorded into
1143          * event queue once the message has been completed.
1144          *
1145          * NB. We're now committed to the GET, since we just marked the MD
1146          * busy.  Callers who observe this (by getting PTL_MD_INUSE from
1147          * PtlMDUnlink()) expect a completion event to tell them when the
1148          * MD becomes idle.
1149          */
1150         if (md->eq) {
1151                 msg->ev.type = PTL_EVENT_SENT;
1152                 msg->ev.initiator.nid = ni->nid;
1153                 msg->ev.initiator.pid = ni->pid;
1154                 msg->ev.portal = args->portal_in;
1155                 msg->ev.match_bits = args->match_bits_in;
1156                 msg->ev.rlength = md->length;
1157                 msg->ev.mlength = md->length;
1158                 msg->ev.offset = args->offset_in;
1159                 msg->ev.hdr_data = args->hdr_data_in;
1160
1161                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
1162         }
1163
1164         state_unlock(nal, &flags);
1165
1166         rc = lib_send (nal, private, msg, &hdr, PTL_MSG_PUT,
1167                        id->nid, id->pid, md, 0, md->length);
1168         if (rc != PTL_OK) {
1169                 /* get_new_msg() committed us to sending by decrementing
1170                  * md->threshold, so we have to act like we did send, but
1171                  * the network dropped it. */
1172                 lib_finalize (nal, private, msg);
1173         }
1174
1175         return ret->rc = PTL_OK;
1176 }
1177
1178 lib_msg_t * lib_fake_reply_msg (nal_cb_t *nal, ptl_nid_t peer_nid,
1179                                 lib_md_t *getmd)
1180 {
1181         /* The NAL can DMA direct to the GET md (i.e. no REPLY msg).  This
1182          * returns a msg the NAL can pass to lib_finalize() so that a REPLY
1183          * event still occurs.
1184          *
1185          * CAVEAT EMPTOR: 'getmd' is passed by pointer so it MUST be valid.
1186          * This can only be guaranteed while a lib_msg_t holds a reference
1187          * on it (ie. pending > 0), so best call this before the
1188          * lib_finalize() of the original GET. */
1189
1190         lib_ni_t        *ni = &nal->ni;
1191         lib_msg_t       *msg;
1192         unsigned long    flags;
1193
1194         state_lock(nal, &flags);
1195
1196         LASSERT (getmd->pending > 0);
1197
1198         if (getmd->threshold == 0) {
1199                 CERROR ("Dropping REPLY from "LPU64" for inactive MD %p\n",
1200                         peer_nid, getmd);
1201                 goto drop;
1202         }
1203
1204         LASSERT (getmd->offset == 0);
1205
1206         CDEBUG(D_NET, "Reply from "LPU64" md %p\n", peer_nid, getmd);
1207
1208         msg = get_new_msg (nal, getmd);
1209         if (msg == NULL) {
1210                 CERROR("Dropping REPLY from "LPU64" md %p: can't allocate msg\n",
1211                        peer_nid, getmd);
1212                 goto drop;
1213         }
1214
1215         if (getmd->eq) {
1216                 msg->ev.type = PTL_EVENT_REPLY;
1217                 msg->ev.initiator.nid = peer_nid;
1218                 msg->ev.initiator.pid = 0;      /* XXX FIXME!!! */
1219                 msg->ev.rlength = msg->ev.mlength = getmd->length;
1220                 msg->ev.offset = 0;
1221
1222                 lib_md_deconstruct(nal, getmd, &msg->ev.mem_desc);
1223         }
1224
1225         ni->counters.recv_count++;
1226         ni->counters.recv_length += getmd->length;
1227
1228         state_unlock(nal, &flags);
1229
1230         return msg;
1231
1232  drop:
1233         nal->ni.counters.drop_count++;
1234         nal->ni.counters.drop_length += getmd->length;
1235
1236         state_unlock (nal, &flags);
1237
1238         return NULL;
1239 }
1240
1241 int do_PtlGet(nal_cb_t * nal, void *private, void *v_args, void *v_ret)
1242 {
1243         /*
1244          * Incoming:
1245          *      ptl_handle_md_t md_in
1246          *      ptl_process_id_t target_in
1247          *      ptl_pt_index_t portal_in
1248          *      ptl_ac_index_t cookie_in
1249          *      ptl_match_bits_t match_bits_in
1250          *      ptl_size_t offset_in
1251          *
1252          * Outgoing:
1253          */
1254
1255         PtlGet_in *args = v_args;
1256         PtlGet_out *ret = v_ret;
1257         ptl_hdr_t hdr;
1258         lib_msg_t *msg = NULL;
1259         lib_ni_t *ni = &nal->ni;
1260         ptl_process_id_t *id = &args->target_in;
1261         lib_md_t *md;
1262         unsigned long flags;
1263         int           rc;
1264
1265         if (!list_empty (&nal->ni.ni_test_peers) && /* normally we don't */
1266             fail_peer (nal, id->nid, 1))           /* shall we now? */
1267         {
1268                 CERROR(LPU64": Dropping PUT to "LPU64": simulated failure\n",
1269                        nal->ni.nid, id->nid);
1270                 return (ret->rc = PTL_INV_PROC);
1271         }
1272
1273         state_lock(nal, &flags);
1274         md = ptl_handle2md(&args->md_in, nal);
1275         if (md == NULL || !md->threshold) {
1276                 state_unlock(nal, &flags);
1277                 return ret->rc = PTL_INV_MD;
1278         }
1279
1280         LASSERT (md->offset == 0);
1281
1282         CDEBUG(D_NET, "PtlGet -> %Lu: %lu\n", (unsigned long long)id->nid,
1283                (unsigned long)id->pid);
1284
1285         memset (&hdr, 0, sizeof (hdr));
1286         hdr.type     = HTON__u32 (PTL_MSG_GET);
1287         hdr.dest_nid = HTON__u64 (id->nid);
1288         hdr.src_nid  = HTON__u64 (ni->nid);
1289         hdr.dest_pid = HTON__u32 (id->pid);
1290         hdr.src_pid  = HTON__u32 (ni->pid);
1291         hdr.payload_length = 0;
1292
1293         /* NB handles only looked up by creator (no flips) */
1294         hdr.msg.get.return_wmd.wh_interface_cookie = ni->ni_interface_cookie;
1295         hdr.msg.get.return_wmd.wh_object_cookie = md->md_lh.lh_cookie;
1296
1297         hdr.msg.get.match_bits = HTON__u64 (args->match_bits_in);
1298         hdr.msg.get.ptl_index = HTON__u32 (args->portal_in);
1299         hdr.msg.get.src_offset = HTON__u32 (args->offset_in);
1300         hdr.msg.get.sink_length = HTON__u32 (md->length);
1301
1302         ni->counters.send_count++;
1303
1304         msg = get_new_msg (nal, md);
1305         if (msg == NULL) {
1306                 CERROR("do_PtlGet: BAD - could not allocate cookie!\n");
1307                 state_unlock(nal, &flags);
1308                 return ret->rc = PTL_NOSPACE;
1309         }
1310
1311         /*
1312          * If this memory descriptor has an event queue associated with
1313          * it we must allocate a message state object that will record
1314          * the information to be filled in once the message has been
1315          * completed.  More information is in the do_PtlPut() comments.
1316          *
1317          * NB. We're now committed to the GET, since we just marked the MD
1318          * busy.  Callers who observe this (by getting PTL_MD_INUSE from
1319          * PtlMDUnlink()) expect a completion event to tell them when the
1320          * MD becomes idle.
1321          */
1322         if (md->eq) {
1323                 msg->ev.type = PTL_EVENT_SENT;
1324                 msg->ev.initiator.nid = ni->nid;
1325                 msg->ev.initiator.pid = ni->pid;
1326                 msg->ev.portal = args->portal_in;
1327                 msg->ev.match_bits = args->match_bits_in;
1328                 msg->ev.rlength = md->length;
1329                 msg->ev.mlength = md->length;
1330                 msg->ev.offset = args->offset_in;
1331                 msg->ev.hdr_data = 0;
1332
1333                 lib_md_deconstruct(nal, md, &msg->ev.mem_desc);
1334         }
1335
1336         state_unlock(nal, &flags);
1337
1338         rc = lib_send (nal, private, msg, &hdr, PTL_MSG_GET,
1339                        id->nid, id->pid, NULL, 0, 0);
1340         if (rc != PTL_OK) {
1341                 /* get_new_msg() committed us to sending by decrementing
1342                  * md->threshold, so we have to act like we did send, but
1343                  * the network dropped it. */
1344                 lib_finalize (nal, private, msg);
1345         }
1346
1347         return ret->rc = PTL_OK;
1348 }
1349
1350 void lib_assert_wire_constants (void)
1351 {
1352         /* Wire protocol assertions generated by 'wirecheck'
1353          * running on Linux robert.bartonsoftware.com 2.4.20-18.9 #1 Thu May 29 06:54:41 EDT 2003 i68
1354          * with gcc version 3.2.2 20030222 (Red Hat Linux 3.2.2-5) */
1355
1356
1357         /* Constants... */
1358         LASSERT (PORTALS_PROTO_MAGIC == 0xeebc0ded);
1359         LASSERT (PORTALS_PROTO_VERSION_MAJOR == 0);
1360         LASSERT (PORTALS_PROTO_VERSION_MINOR == 3);
1361         LASSERT (PTL_MSG_ACK == 0);
1362         LASSERT (PTL_MSG_PUT == 1);
1363         LASSERT (PTL_MSG_GET == 2);
1364         LASSERT (PTL_MSG_REPLY == 3);
1365         LASSERT (PTL_MSG_HELLO == 4);
1366
1367         /* Checks for struct ptl_handle_wire_t */
1368         LASSERT ((int)sizeof(ptl_handle_wire_t) == 16);
1369         LASSERT (offsetof(ptl_handle_wire_t, wh_interface_cookie) == 0);
1370         LASSERT ((int)sizeof(((ptl_handle_wire_t *)0)->wh_interface_cookie) == 8);
1371         LASSERT (offsetof(ptl_handle_wire_t, wh_object_cookie) == 8);
1372         LASSERT ((int)sizeof(((ptl_handle_wire_t *)0)->wh_object_cookie) == 8);
1373
1374         /* Checks for struct ptl_magicversion_t */
1375         LASSERT ((int)sizeof(ptl_magicversion_t) == 8);
1376         LASSERT (offsetof(ptl_magicversion_t, magic) == 0);
1377         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->magic) == 4);
1378         LASSERT (offsetof(ptl_magicversion_t, version_major) == 4);
1379         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->version_major) == 2);
1380         LASSERT (offsetof(ptl_magicversion_t, version_minor) == 6);
1381         LASSERT ((int)sizeof(((ptl_magicversion_t *)0)->version_minor) == 2);
1382
1383         /* Checks for struct ptl_hdr_t */
1384         LASSERT ((int)sizeof(ptl_hdr_t) == 72);
1385         LASSERT (offsetof(ptl_hdr_t, dest_nid) == 0);
1386         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->dest_nid) == 8);
1387         LASSERT (offsetof(ptl_hdr_t, src_nid) == 8);
1388         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->src_nid) == 8);
1389         LASSERT (offsetof(ptl_hdr_t, dest_pid) == 16);
1390         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->dest_pid) == 4);
1391         LASSERT (offsetof(ptl_hdr_t, src_pid) == 20);
1392         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->src_pid) == 4);
1393         LASSERT (offsetof(ptl_hdr_t, type) == 24);
1394         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->type) == 4);
1395         LASSERT (offsetof(ptl_hdr_t, payload_length) == 28);
1396         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->payload_length) == 4);
1397         LASSERT (offsetof(ptl_hdr_t, msg) == 32);
1398         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg) == 40);
1399
1400         /* Ack */
1401         LASSERT (offsetof(ptl_hdr_t, msg.ack.dst_wmd) == 32);
1402         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.dst_wmd) == 16);
1403         LASSERT (offsetof(ptl_hdr_t, msg.ack.match_bits) == 48);
1404         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.match_bits) == 8);
1405         LASSERT (offsetof(ptl_hdr_t, msg.ack.mlength) == 56);
1406         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.ack.mlength) == 4);
1407
1408         /* Put */
1409         LASSERT (offsetof(ptl_hdr_t, msg.put.ack_wmd) == 32);
1410         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.ack_wmd) == 16);
1411         LASSERT (offsetof(ptl_hdr_t, msg.put.match_bits) == 48);
1412         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.match_bits) == 8);
1413         LASSERT (offsetof(ptl_hdr_t, msg.put.hdr_data) == 56);
1414         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.hdr_data) == 8);
1415         LASSERT (offsetof(ptl_hdr_t, msg.put.ptl_index) == 64);
1416         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.ptl_index) == 4);
1417         LASSERT (offsetof(ptl_hdr_t, msg.put.offset) == 68);
1418         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.put.offset) == 4);
1419
1420         /* Get */
1421         LASSERT (offsetof(ptl_hdr_t, msg.get.return_wmd) == 32);
1422         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.return_wmd) == 16);
1423         LASSERT (offsetof(ptl_hdr_t, msg.get.match_bits) == 48);
1424         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.match_bits) == 8);
1425         LASSERT (offsetof(ptl_hdr_t, msg.get.ptl_index) == 56);
1426         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.ptl_index) == 4);
1427         LASSERT (offsetof(ptl_hdr_t, msg.get.src_offset) == 60);
1428         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.src_offset) == 4);
1429         LASSERT (offsetof(ptl_hdr_t, msg.get.sink_length) == 64);
1430         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.get.sink_length) == 4);
1431
1432         /* Reply */
1433         LASSERT (offsetof(ptl_hdr_t, msg.reply.dst_wmd) == 32);
1434         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.reply.dst_wmd) == 16);
1435
1436         /* Hello */
1437         LASSERT (offsetof(ptl_hdr_t, msg.hello.incarnation) == 32);
1438         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.hello.incarnation) == 8);
1439         LASSERT (offsetof(ptl_hdr_t, msg.hello.type) == 40);
1440         LASSERT ((int)sizeof(((ptl_hdr_t *)0)->msg.hello.type) == 4);
1441 }