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