lnet/lnet/lib-msg.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  */
  30 /*
  31  * This file is part of Lustre, http://www.lustre.org/
  32  * Lustre is a trademark of Sun Microsystems, Inc.
  33  *
  34  * lnet/lnet/lib-msg.c
  35  *
  36  * Message decoding, parsing and finalizing routines
  37  */
  38
  39 #define DEBUG_SUBSYSTEM S_LNET
  40
  41 #include <lnet/lib-lnet.h>
  42
  43 void
  44 lnet_build_unlink_event (lnet_libmd_t *md, lnet_event_t *ev)
  45 {
  46         ENTRY;
  47
  48         memset(ev, 0, sizeof(*ev));
  49
  50         ev->status   = 0;
  51         ev->unlinked = 1;
  52         ev->type     = LNET_EVENT_UNLINK;
  53         lnet_md_deconstruct(md, &ev->md);
  54         lnet_md2handle(&ev->md_handle, md);
  55         EXIT;
  56 }
  57
  58 /*
  59  * Don't need any lock, must be called after lnet_commit_md
  60  */
  61 void
  62 lnet_build_msg_event(lnet_msg_t *msg, lnet_event_kind_t ev_type)
  63 {
  64         lnet_hdr_t      *hdr = &msg->msg_hdr;
  65         lnet_event_t    *ev  = &msg->msg_ev;
  66
  67         LASSERT(!msg->msg_routing);
  68
  69         ev->type = ev_type;
  70
  71         if (ev_type == LNET_EVENT_SEND) {
  72                 /* event for active message */
  73                 ev->target.nid    = le64_to_cpu(hdr->dest_nid);
  74                 ev->target.pid    = le32_to_cpu(hdr->dest_pid);
  75                 ev->initiator.nid = LNET_NID_ANY;
  76                 ev->initiator.pid = the_lnet.ln_pid;
  77                 ev->sender        = LNET_NID_ANY;
  78
  79         } else {
  80                 /* event for passive message */
  81                 ev->target.pid    = hdr->dest_pid;
  82                 ev->target.nid    = hdr->dest_nid;
  83                 ev->initiator.pid = hdr->src_pid;
  84                 ev->initiator.nid = hdr->src_nid;
  85                 ev->rlength       = hdr->payload_length;
  86                 ev->sender        = msg->msg_from;
  87                 ev->mlength       = msg->msg_wanted;
  88                 ev->offset        = msg->msg_offset;
  89         }
  90
  91         switch (ev_type) {
  92         default:
  93                 LBUG();
  94
  95         case LNET_EVENT_PUT: /* passive PUT */
  96                 ev->pt_index   = hdr->msg.put.ptl_index;
  97                 ev->match_bits = hdr->msg.put.match_bits;
  98                 ev->hdr_data   = hdr->msg.put.hdr_data;
  99                 return;
 100
 101         case LNET_EVENT_GET: /* passive GET */
 102                 ev->pt_index   = hdr->msg.get.ptl_index;
 103                 ev->match_bits = hdr->msg.get.match_bits;
 104                 ev->hdr_data   = 0;
 105                 return;
 106
 107         case LNET_EVENT_ACK: /* ACK */
 108                 ev->match_bits = hdr->msg.ack.match_bits;
 109                 ev->mlength    = hdr->msg.ack.mlength;
 110                 return;
 111
 112         case LNET_EVENT_REPLY: /* REPLY */
 113                 return;
 114
 115         case LNET_EVENT_SEND: /* active message */
 116                 if (msg->msg_type == LNET_MSG_PUT) {
 117                         ev->pt_index   = le32_to_cpu(hdr->msg.put.ptl_index);
 118                         ev->match_bits = le64_to_cpu(hdr->msg.put.match_bits);
 119                         ev->offset     = le32_to_cpu(hdr->msg.put.offset);
 120                         ev->mlength    =
 121                         ev->rlength    = le32_to_cpu(hdr->payload_length);
 122                         ev->hdr_data   = le64_to_cpu(hdr->msg.put.hdr_data);
 123
 124                 } else {
 125                         LASSERT(msg->msg_type == LNET_MSG_GET);
 126                         ev->pt_index   = le32_to_cpu(hdr->msg.get.ptl_index);
 127                         ev->match_bits = le64_to_cpu(hdr->msg.get.match_bits);
 128                         ev->mlength    =
 129                         ev->rlength    = le32_to_cpu(hdr->msg.get.sink_length);
 130                         ev->offset     = le32_to_cpu(hdr->msg.get.src_offset);
 131                         ev->hdr_data   = 0;
 132                 }
 133                 return;
 134         }
 135 }
 136
 137 void
 138 lnet_msg_commit(lnet_msg_t *msg, int cpt)
 139 {
 140         struct lnet_msg_container *container = the_lnet.ln_msg_containers[cpt];
 141         lnet_counters_t           *counters  = the_lnet.ln_counters[cpt];
 142
 143         /* routed message can be committed for both receiving and sending */
 144         LASSERT(!msg->msg_tx_committed);
 145
 146         if (msg->msg_sending) {
 147                 LASSERT(!msg->msg_receiving);
 148
 149                 msg->msg_tx_cpt = cpt;
 150                 msg->msg_tx_committed = 1;
 151                 if (msg->msg_rx_committed) { /* routed message REPLY */
 152                         LASSERT(msg->msg_onactivelist);
 153                         return;
 154                 }
 155         } else {
 156                 LASSERT(!msg->msg_sending);
 157                 msg->msg_rx_cpt = cpt;
 158                 msg->msg_rx_committed = 1;
 159         }
 160
 161         LASSERT(!msg->msg_onactivelist);
 162         msg->msg_onactivelist = 1;
 163         cfs_list_add(&msg->msg_activelist, &container->msc_active);
 164
 165         counters->msgs_alloc++;
 166         if (counters->msgs_alloc > counters->msgs_max)
 167                 counters->msgs_max = counters->msgs_alloc;
 168 }
 169
 170 static void
 171 lnet_msg_decommit_tx(lnet_msg_t *msg, int status)
 172 {
 173         lnet_counters_t *counters;
 174         lnet_event_t    *ev = &msg->msg_ev;
 175
 176         LASSERT(msg->msg_tx_committed);
 177         if (status != 0)
 178                 goto out;
 179
 180         counters = the_lnet.ln_counters[msg->msg_tx_cpt];
 181         switch (ev->type) {
 182         default: /* routed message */
 183                 LASSERT(msg->msg_routing);
 184                 LASSERT(msg->msg_rx_committed);
 185                 LASSERT(ev->type == 0);
 186
 187                 counters->route_length += msg->msg_len;
 188                 counters->route_count++;
 189                 goto out;
 190
 191         case LNET_EVENT_PUT:
 192                 /* should have been decommitted */
 193                 LASSERT(!msg->msg_rx_committed);
 194                 /* overwritten while sending ACK */
 195                 LASSERT(msg->msg_type == LNET_MSG_ACK);
 196                 msg->msg_type = LNET_MSG_PUT; /* fix type */
 197                 break;
 198
 199         case LNET_EVENT_SEND:
 200                 LASSERT(!msg->msg_rx_committed);
 201                 if (msg->msg_type == LNET_MSG_PUT)
 202                         counters->send_length += msg->msg_len;
 203                 break;
 204
 205         case LNET_EVENT_GET:
 206                 LASSERT(msg->msg_rx_committed);
 207                 /* overwritten while sending reply */
 208                 LASSERT(msg->msg_type == LNET_MSG_REPLY);
 209
 210                 msg->msg_type = LNET_MSG_GET; /* fix type */
 211                 counters->send_length += msg->msg_len;
 212                 break;
 213         }
 214
 215         counters->send_count++;
 216  out:
 217         lnet_return_tx_credits_locked(msg);
 218         msg->msg_tx_committed = 0;
 219 }
 220
 221 static void
 222 lnet_msg_decommit_rx(lnet_msg_t *msg, int status)
 223 {
 224         lnet_counters_t *counters;
 225         lnet_event_t    *ev = &msg->msg_ev;
 226
 227         LASSERT(!msg->msg_tx_committed); /* decommitted or never committed */
 228         LASSERT(msg->msg_rx_committed);
 229
 230         if (status != 0)
 231                 goto out;
 232
 233         switch (ev->type) {
 234         default:
 235                 LASSERT(ev->type == 0);
 236                 LASSERT(msg->msg_routing);
 237                 goto out;
 238
 239         case LNET_EVENT_ACK:
 240                 LASSERT(msg->msg_type == LNET_MSG_ACK);
 241                 break;
 242
 243         case LNET_EVENT_GET:
 244                 LASSERT(msg->msg_type == LNET_MSG_GET);
 245                 break;
 246
 247         case LNET_EVENT_PUT:
 248                 LASSERT(msg->msg_type == LNET_MSG_PUT);
 249                 break;
 250
 251         case LNET_EVENT_REPLY:
 252                 LASSERT(msg->msg_type == LNET_MSG_REPLY ||
 253                         msg->msg_type == LNET_MSG_GET); /* optimized GET */
 254                 break;
 255         }
 256
 257         counters = the_lnet.ln_counters[msg->msg_rx_cpt];
 258         counters->recv_count++;
 259         if (ev->type == LNET_EVENT_PUT || ev->type == LNET_EVENT_REPLY)
 260                 counters->recv_length += msg->msg_wanted;
 261
 262  out:
 263         lnet_return_rx_credits_locked(msg);
 264         msg->msg_rx_committed = 0;
 265 }
 266
 267 void
 268 lnet_msg_decommit(lnet_msg_t *msg, int cpt, int status)
 269 {
 270         int     cpt2 = cpt;
 271
 272         LASSERT(msg->msg_tx_committed || msg->msg_rx_committed);
 273         LASSERT(msg->msg_onactivelist);
 274
 275         if (msg->msg_tx_committed) { /* always decommit for sending first */
 276                 LASSERT(cpt == msg->msg_tx_cpt);
 277                 lnet_msg_decommit_tx(msg, status);
 278         }
 279
 280         if (msg->msg_rx_committed) {
 281                 /* forwarding msg committed for both receiving and sending */
 282                 if (cpt != msg->msg_rx_cpt) {
 283                         lnet_net_unlock(cpt);
 284                         cpt2 = msg->msg_rx_cpt;
 285                         lnet_net_lock(cpt2);
 286                 }
 287                 lnet_msg_decommit_rx(msg, status);
 288         }
 289
 290         cfs_list_del(&msg->msg_activelist);
 291         msg->msg_onactivelist = 0;
 292
 293         the_lnet.ln_counters[cpt2]->msgs_alloc--;
 294
 295         if (cpt2 != cpt) {
 296                 lnet_net_unlock(cpt2);
 297                 lnet_net_lock(cpt);
 298         }
 299 }
 300
 301 void
 302 lnet_msg_attach_md(lnet_msg_t *msg, lnet_libmd_t *md,
 303                    unsigned int offset, unsigned int mlen)
 304 {
 305         /* NB: @offset and @len are only useful for receiving */
 306         /* Here, we attach the MD on lnet_msg and mark it busy and
 307          * decrementing its threshold. Come what may, the lnet_msg "owns"
 308          * the MD until a call to lnet_msg_detach_md or lnet_finalize()
 309          * signals completion. */
 310         LASSERT(!msg->msg_routing);
 311
 312         msg->msg_md = md;
 313         if (msg->msg_receiving) { /* commited for receiving */
 314                 msg->msg_offset = offset;
 315                 msg->msg_wanted = mlen;
 316         }
 317
 318         md->md_refcount++;
 319         if (md->md_threshold != LNET_MD_THRESH_INF) {
 320                 LASSERT(md->md_threshold > 0);
 321                 md->md_threshold--;
 322         }
 323
 324         /* build umd in event */
 325         lnet_md2handle(&msg->msg_ev.md_handle, md);
 326         lnet_md_deconstruct(md, &msg->msg_ev.md);
 327 }
 328
 329 void
 330 lnet_msg_detach_md(lnet_msg_t *msg, int status)
 331 {
 332         lnet_libmd_t    *md = msg->msg_md;
 333         int             unlink;
 334
 335         /* Now it's safe to drop my caller's ref */
 336         md->md_refcount--;
 337         LASSERT(md->md_refcount >= 0);
 338
 339         unlink = lnet_md_unlinkable(md);
 340         if (md->md_eq != NULL) {
 341                 msg->msg_ev.status   = status;
 342                 msg->msg_ev.unlinked = unlink;
 343                 lnet_eq_enqueue_event(md->md_eq, &msg->msg_ev);
 344         }
 345
 346         if (unlink)
 347                 lnet_md_unlink(md);
 348
 349         msg->msg_md = NULL;
 350 }
 351
 352 void
 353 lnet_complete_msg_locked(lnet_msg_t *msg, int cpt)
 354 {
 355         lnet_handle_wire_t ack_wmd;
 356         int                rc;
 357         int                status = msg->msg_ev.status;
 358
 359         LASSERT (msg->msg_onactivelist);
 360
 361         if (status == 0 && msg->msg_ack) {
 362                 /* Only send an ACK if the PUT completed successfully */
 363
 364                 lnet_msg_decommit(msg, cpt, 0);
 365
 366                 msg->msg_ack = 0;
 367                 lnet_net_unlock(cpt);
 368
 369                 LASSERT(msg->msg_ev.type == LNET_EVENT_PUT);
 370                 LASSERT(!msg->msg_routing);
 371
 372                 ack_wmd = msg->msg_hdr.msg.put.ack_wmd;
 373
 374                 lnet_prep_send(msg, LNET_MSG_ACK, msg->msg_ev.initiator, 0, 0);
 375
 376                 msg->msg_hdr.msg.ack.dst_wmd = ack_wmd;
 377                 msg->msg_hdr.msg.ack.match_bits = msg->msg_ev.match_bits;
 378                 msg->msg_hdr.msg.ack.mlength = cpu_to_le32(msg->msg_ev.mlength);
 379
 380                 /* NB: we probably want to use NID of msg::msg_from as 3rd
 381                  * parameter (router NID) if it's routed message */
 382                 rc = lnet_send(msg->msg_ev.target.nid, msg, LNET_NID_ANY);
 383
 384                 lnet_net_lock(cpt);
 385
 386                 if (rc == 0)
 387                         return;
 388         } else if (status == 0 &&       /* OK so far */
 389                    (msg->msg_routing && !msg->msg_sending)) {
 390                 /* not forwarded */
 391                 LASSERT(!msg->msg_receiving);   /* called back recv already */
 392                 lnet_net_unlock(cpt);
 393
 394                 rc = lnet_send(LNET_NID_ANY, msg, LNET_NID_ANY);
 395
 396                 lnet_net_lock(cpt);
 397
 398                 if (rc == 0)
 399                         return;
 400         }
 401
 402         lnet_msg_decommit(msg, cpt, status);
 403         lnet_msg_free_locked(msg);
 404 }
 405
 406 void
 407 lnet_finalize (lnet_ni_t *ni, lnet_msg_t *msg, int status)
 408 {
 409         struct lnet_msg_container       *container;
 410         int                             my_slot;
 411         int                             cpt;
 412         int                             i;
 413
 414         LASSERT (!cfs_in_interrupt ());
 415
 416         if (msg == NULL)
 417                 return;
 418 #if 0
 419         CDEBUG(D_WARNING, "%s msg->%s Flags:%s%s%s%s%s%s%s%s%s%s%s txp %s rxp %s\n",
 420                lnet_msgtyp2str(msg->msg_type), libcfs_id2str(msg->msg_target),
 421                msg->msg_target_is_router ? "t" : "",
 422                msg->msg_routing ? "X" : "",
 423                msg->msg_ack ? "A" : "",
 424                msg->msg_sending ? "S" : "",
 425                msg->msg_receiving ? "R" : "",
 426                msg->msg_delayed ? "d" : "",
 427                msg->msg_txcredit ? "C" : "",
 428                msg->msg_peertxcredit ? "c" : "",
 429                msg->msg_rtrcredit ? "F" : "",
 430                msg->msg_peerrtrcredit ? "f" : "",
 431                msg->msg_onactivelist ? "!" : "",
 432                msg->msg_txpeer == NULL ? "<none>" : libcfs_nid2str(msg->msg_txpeer->lp_nid),
 433                msg->msg_rxpeer == NULL ? "<none>" : libcfs_nid2str(msg->msg_rxpeer->lp_nid));
 434 #endif
 435
 436         LASSERT (msg->msg_onactivelist);
 437
 438         msg->msg_ev.status = status;
 439
 440         if (msg->msg_md != NULL) {
 441                 cpt = lnet_cpt_of_cookie(msg->msg_md->md_lh.lh_cookie);
 442
 443                 lnet_res_lock(cpt);
 444                 lnet_msg_detach_md(msg, status);
 445                 lnet_res_unlock(cpt);
 446         }
 447
 448         if (!msg->msg_tx_committed && !msg->msg_rx_committed) {
 449                 /* not commited to network yet */
 450                 LASSERT(!msg->msg_onactivelist);
 451                 lnet_msg_free(msg);
 452                 return;
 453         }
 454
 455         /*
 456          * NB: routed message can be commited for both receiving and sending,
 457          * we should finalize in LIFO order and keep counters correct.
 458          * (finalize sending first then finalize receiving)
 459          */
 460         cpt = msg->msg_tx_committed ? msg->msg_tx_cpt : msg->msg_rx_cpt;
 461         lnet_net_lock(cpt);
 462
 463         container = the_lnet.ln_msg_containers[cpt];
 464         cfs_list_add_tail(&msg->msg_list, &container->msc_finalizing);
 465
 466         /* Recursion breaker.  Don't complete the message here if I am (or
 467          * enough other threads are) already completing messages */
 468
 469 #ifdef __KERNEL__
 470         my_slot = -1;
 471         for (i = 0; i < container->msc_nfinalizers; i++) {
 472                 if (container->msc_finalizers[i] == cfs_current())
 473                         goto out;
 474
 475                 if (my_slot < 0 && container->msc_finalizers[i] == NULL)
 476                         my_slot = i;
 477         }
 478
 479         if (my_slot < 0)
 480                 goto out;
 481
 482         container->msc_finalizers[my_slot] = cfs_current();
 483 #else
 484         LASSERT(container->msc_nfinalizers == 1);
 485         if (container->msc_finalizers[0] != NULL)
 486                 goto out;
 487
 488         my_slot = i = 0;
 489         container->msc_finalizers[0] = (struct lnet_msg_container *)1;
 490 #endif
 491
 492         while (!cfs_list_empty(&container->msc_finalizing)) {
 493                 msg = cfs_list_entry(container->msc_finalizing.next,
 494                                      lnet_msg_t, msg_list);
 495
 496                 cfs_list_del(&msg->msg_list);
 497
 498                 /* NB drops and regains the lnet lock if it actually does
 499                  * anything, so my finalizing friends can chomp along too */
 500                 lnet_complete_msg_locked(msg, cpt);
 501         }
 502
 503         container->msc_finalizers[my_slot] = NULL;
 504  out:
 505         lnet_net_unlock(cpt);
 506 }
 507
 508 void
 509 lnet_msg_container_cleanup(struct lnet_msg_container *container)
 510 {
 511         int     count = 0;
 512
 513         if (container->msc_init == 0)
 514                 return;
 515
 516         while (!cfs_list_empty(&container->msc_active)) {
 517                 lnet_msg_t *msg = cfs_list_entry(container->msc_active.next,
 518                                                  lnet_msg_t, msg_activelist);
 519
 520                 LASSERT(msg->msg_onactivelist);
 521                 msg->msg_onactivelist = 0;
 522                 cfs_list_del(&msg->msg_activelist);
 523                 lnet_msg_free(msg);
 524                 count++;
 525         }
 526
 527         if (count > 0)
 528                 CERROR("%d active msg on exit\n", count);
 529
 530         if (container->msc_finalizers != NULL) {
 531                 LIBCFS_FREE(container->msc_finalizers,
 532                             container->msc_nfinalizers *
 533                             sizeof(*container->msc_finalizers));
 534                 container->msc_finalizers = NULL;
 535         }
 536 #ifdef LNET_USE_LIB_FREELIST
 537         lnet_freelist_fini(&container->msc_freelist);
 538 #endif
 539         container->msc_init = 0;
 540 }
 541
 542 int
 543 lnet_msg_container_setup(struct lnet_msg_container *container, int cpt)
 544 {
 545         int     rc;
 546
 547         container->msc_init = 1;
 548
 549         CFS_INIT_LIST_HEAD(&container->msc_active);
 550         CFS_INIT_LIST_HEAD(&container->msc_finalizing);
 551
 552 #ifdef LNET_USE_LIB_FREELIST
 553         memset(&container->msc_freelist, 0, sizeof(lnet_freelist_t));
 554
 555         rc = lnet_freelist_init(&container->msc_freelist,
 556                                 LNET_FL_MAX_MSGS, sizeof(lnet_msg_t));
 557         if (rc != 0) {
 558                 CERROR("Failed to init freelist for message container\n");
 559                 lnet_msg_container_cleanup(container);
 560                 return rc;
 561         }
 562 #else
 563         rc = 0;
 564 #endif
 565         /* number of CPUs */
 566         container->msc_nfinalizers = cfs_cpt_weight(lnet_cpt_table(), cpt);
 567
 568         LIBCFS_CPT_ALLOC(container->msc_finalizers, lnet_cpt_table(), cpt,
 569                          container->msc_nfinalizers *
 570                          sizeof(*container->msc_finalizers));
 571
 572         if (container->msc_finalizers == NULL) {
 573                 CERROR("Failed to allocate message finalizers\n");
 574                 lnet_msg_container_cleanup(container);
 575                 return -ENOMEM;
 576         }
 577
 578         return rc;
 579 }
 580
 581 void
 582 lnet_msg_containers_destroy(void)
 583 {
 584         struct lnet_msg_container *container;
 585         int     i;
 586
 587         if (the_lnet.ln_msg_containers == NULL)
 588                 return;
 589
 590         cfs_percpt_for_each(container, i, the_lnet.ln_msg_containers)
 591                 lnet_msg_container_cleanup(container);
 592
 593         cfs_percpt_free(the_lnet.ln_msg_containers);
 594         the_lnet.ln_msg_containers = NULL;
 595 }
 596
 597 int
 598 lnet_msg_containers_create(void)
 599 {
 600         struct lnet_msg_container *container;
 601         int     rc;
 602         int     i;
 603
 604         the_lnet.ln_msg_containers = cfs_percpt_alloc(lnet_cpt_table(),
 605                                                       sizeof(*container));
 606
 607         if (the_lnet.ln_msg_containers == NULL) {
 608                 CERROR("Failed to allocate cpu-partition data for network\n");
 609                 return -ENOMEM;
 610         }
 611
 612         cfs_percpt_for_each(container, i, the_lnet.ln_msg_containers) {
 613                 rc = lnet_msg_container_setup(container, i);
 614                 if (rc != 0) {
 615                         lnet_msg_containers_destroy();
 616                         return rc;
 617                 }
 618         }
 619
 620         return 0;
 621 }