lnet/lnet/router.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
   5  *
   6  *   This file is part of Portals
   7  *   http://sourceforge.net/projects/sandiaportals/
   8  *
   9  *   Portals is free software; you can redistribute it and/or
  10  *   modify it under the terms of version 2 of the GNU General Public
  11  *   License as published by the Free Software Foundation.
  12  *
  13  *   Portals is distributed in the hope that it will be useful,
  14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *   GNU General Public License for more details.
  17  *
  18  *   You should have received a copy of the GNU General Public License
  19  *   along with Portals; if not, write to the Free Software
  20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21  *
  22  */
  23
  24 #define DEBUG_SUBSYSTEM S_LNET
  25 #include <lnet/lib-lnet.h>
  26
  27 #if defined(__KERNEL__) && defined(LNET_ROUTER)
  28
  29 static char *forwarding = "";
  30 CFS_MODULE_PARM(forwarding, "s", charp, 0444,
  31                 "Explicitly enable/disable forwarding between networks");
  32
  33 static int tiny_router_buffers = 1024;
  34 CFS_MODULE_PARM(tiny_router_buffers, "i", int, 0444,
  35                 "# of 0 payload messages to buffer in the router");
  36 static int small_router_buffers = 8192;
  37 CFS_MODULE_PARM(small_router_buffers, "i", int, 0444,
  38                 "# of small (1 page) messages to buffer in the router");
  39 static int large_router_buffers = 512;
  40 CFS_MODULE_PARM(large_router_buffers, "i", int, 0444,
  41                 "# of large messages to buffer in the router");
  42
  43 static int auto_down = 1;
  44 CFS_MODULE_PARM(auto_down, "i", int, 0444,
  45                 "Automatically mark peers down on comms error");
  46
  47 static int check_routers_before_use = 0;
  48 CFS_MODULE_PARM(check_routers_before_use, "i", int, 0444,
  49                 "Assume routers are down and ping them before use");
  50
  51 static int dead_router_check_interval = 0;
  52 CFS_MODULE_PARM(dead_router_check_interval, "i", int, 0444,
  53                 "Seconds between dead router health checks (<= 0 to disable)");
  54
  55 static int live_router_check_interval = 0;
  56 CFS_MODULE_PARM(live_router_check_interval, "i", int, 0444,
  57                 "Seconds between live router health checks (<= 0 to disable)");
  58
  59 static int router_ping_timeout = 50;
  60 CFS_MODULE_PARM(router_ping_timeout, "i", int, 0444,
  61                 "Seconds to wait for the reply to a router health query");
  62
  63 int
  64 lnet_peers_start_down(void)
  65 {
  66         return check_routers_before_use;
  67 }
  68
  69 void
  70 lnet_notify_locked(lnet_peer_t *lp, int notifylnd, int alive, time_t when)
  71 {
  72         if (when < lp->lp_timestamp) {          /* out of date information */
  73                 CDEBUG(D_NET, "Out of date\n");
  74                 return;
  75         }
  76
  77         lp->lp_timestamp = when;                /* update timestamp */
  78         lp->lp_ping_deadline = 0;               /* disable ping timeout */
  79
  80         if (lp->lp_alive_count != 0 &&          /* got old news */
  81             (!lp->lp_alive) == (!alive)) {      /* new date for old news */
  82                 CDEBUG(D_NET, "Old news\n");
  83                 return;
  84         }
  85
  86         /* Flag that notification is outstanding */
  87
  88         lp->lp_alive_count++;
  89         lp->lp_alive = !(!alive);               /* 1 bit! */
  90         lp->lp_notify = 1;
  91         lp->lp_notifylnd |= notifylnd;
  92
  93         CDEBUG(D_NET, "set %s %d\n", libcfs_nid2str(lp->lp_nid), alive);
  94 }
  95
  96 void
  97 lnet_do_notify (lnet_peer_t *lp)
  98 {
  99         lnet_ni_t *ni = lp->lp_ni;
 100         int        alive;
 101         int        notifylnd;
 102
 103         LNET_LOCK();
 104
 105         /* Notify only in 1 thread at any time to ensure ordered notification.
 106          * NB individual events can be missed; the only guarantee is that you
 107          * always get the most recent news */
 108
 109         if (lp->lp_notifying) {
 110                 LNET_UNLOCK();
 111                 return;
 112         }
 113
 114         lp->lp_notifying = 1;
 115
 116         while (lp->lp_notify) {
 117                 alive     = lp->lp_alive;
 118                 notifylnd = lp->lp_notifylnd;
 119
 120                 lp->lp_notifylnd = 0;
 121                 lp->lp_notify    = 0;
 122
 123                 if (notifylnd && ni->ni_lnd->lnd_notify != NULL) {
 124                         LNET_UNLOCK();
 125
 126                         /* A new notification could happen now; I'll handle it
 127                          * when control returns to me */
 128
 129                         (ni->ni_lnd->lnd_notify)(ni, lp->lp_nid, alive);
 130
 131                         LNET_LOCK();
 132                 }
 133         }
 134
 135         lp->lp_notifying = 0;
 136
 137         LNET_UNLOCK();
 138 }
 139
 140 int
 141 lnet_notify (lnet_ni_t *ni, lnet_nid_t nid, int alive, time_t when)
 142 {
 143         lnet_peer_t         *lp = NULL;
 144         time_t               now = cfs_time_current_sec();
 145
 146         LASSERT (!in_interrupt ());
 147
 148         CDEBUG (D_NET, "%s notifying %s: %s\n",
 149                 (ni == NULL) ? "userspace" : libcfs_nid2str(ni->ni_nid),
 150                 libcfs_nid2str(nid),
 151                 alive ? "up" : "down");
 152
 153         if (ni != NULL &&
 154             LNET_NIDNET(ni->ni_nid) != LNET_NIDNET(nid)) {
 155                 CWARN ("Ignoring notification of %s %s by %s (different net)\n",
 156                         libcfs_nid2str(nid), alive ? "birth" : "death",
 157                         libcfs_nid2str(ni->ni_nid));
 158                 return -EINVAL;
 159         }
 160
 161         /* can't do predictions... */
 162         if (when > now) {
 163                 CWARN ("Ignoring prediction from %s of %s %s "
 164                        "%ld seconds in the future\n",
 165                        (ni == NULL) ? "userspace" : libcfs_nid2str(ni->ni_nid),
 166                        libcfs_nid2str(nid), alive ? "up" : "down",
 167                        when - now);
 168                 return -EINVAL;
 169         }
 170
 171         if (ni != NULL && !alive &&             /* LND telling me she's down */
 172             !auto_down) {                       /* auto-down disabled */
 173                 CDEBUG(D_NET, "Auto-down disabled\n");
 174                 return 0;
 175         }
 176
 177         LNET_LOCK();
 178
 179         lp = lnet_find_peer_locked(nid);
 180         if (lp == NULL) {
 181                 /* nid not found */
 182                 LNET_UNLOCK();
 183                 CDEBUG(D_NET, "%s not found\n", libcfs_nid2str(nid));
 184                 return 0;
 185         }
 186
 187         lnet_notify_locked(lp, ni == NULL, alive, when);
 188
 189         LNET_UNLOCK();
 190
 191         lnet_do_notify(lp);
 192
 193         LNET_LOCK();
 194
 195         lnet_peer_decref_locked(lp);
 196
 197         LNET_UNLOCK();
 198         return 0;
 199 }
 200 EXPORT_SYMBOL(lnet_notify);
 201
 202 #else
 203
 204 int
 205 lnet_notify (lnet_ni_t *ni, lnet_nid_t nid, int alive, time_t when)
 206 {
 207         return -EOPNOTSUPP;
 208 }
 209
 210 void
 211 lnet_notify_locked (lnet_peer_t *lp, int notifylnd, int alive, time_t when)
 212 {
 213         return;
 214 }
 215
 216 #endif
 217
 218 static void
 219 lnet_rtr_addref_locked(lnet_peer_t *lp)
 220 {
 221         LASSERT (lp->lp_refcount > 0);
 222         LASSERT (lp->lp_rtr_refcount >= 0);
 223
 224         lp->lp_rtr_refcount++;
 225         if (lp->lp_rtr_refcount == 1) {
 226                 struct list_head *pos;
 227
 228                 /* a simple insertion sort */
 229                 list_for_each_prev(pos, &the_lnet.ln_routers) {
 230                         lnet_peer_t *rtr = list_entry(pos, lnet_peer_t,
 231                                                       lp_rtr_list);
 232
 233                         if (rtr->lp_nid < lp->lp_nid)
 234                                 break;
 235                 }
 236
 237                 list_add(&lp->lp_rtr_list, pos);
 238                 /* addref for the_lnet.ln_routers */
 239                 lnet_peer_addref_locked(lp);
 240                 the_lnet.ln_routers_version++;
 241         }
 242 }
 243
 244 static void
 245 lnet_rtr_decref_locked(lnet_peer_t *lp)
 246 {
 247         LASSERT (lp->lp_refcount > 0);
 248         LASSERT (lp->lp_rtr_refcount > 0);
 249
 250         lp->lp_rtr_refcount--;
 251         if (lp->lp_rtr_refcount == 0) {
 252                 list_del(&lp->lp_rtr_list);
 253                 /* decref for the_lnet.ln_routers */
 254                 lnet_peer_decref_locked(lp);
 255                 the_lnet.ln_routers_version++;
 256         }
 257 }
 258
 259 lnet_remotenet_t *
 260 lnet_find_net_locked (__u32 net)
 261 {
 262         lnet_remotenet_t *rnet;
 263         struct list_head *tmp;
 264
 265         LASSERT (!the_lnet.ln_shutdown);
 266
 267         list_for_each (tmp, &the_lnet.ln_remote_nets) {
 268                 rnet = list_entry(tmp, lnet_remotenet_t, lrn_list);
 269
 270                 if (rnet->lrn_net == net)
 271                         return rnet;
 272         }
 273         return NULL;
 274 }
 275
 276 int
 277 lnet_add_route (__u32 net, unsigned int hops, lnet_nid_t gateway)
 278 {
 279         struct list_head     zombies;
 280         struct list_head    *e;
 281         lnet_remotenet_t    *rnet;
 282         lnet_remotenet_t    *rnet2;
 283         lnet_route_t        *route;
 284         lnet_route_t        *route2;
 285         lnet_ni_t           *ni;
 286         int                  add_route;
 287         int                  rc;
 288
 289         CDEBUG(D_NET, "Add route: net %s hops %u gw %s\n",
 290                libcfs_net2str(net), hops, libcfs_nid2str(gateway));
 291
 292         if (gateway == LNET_NID_ANY ||
 293             LNET_NETTYP(LNET_NIDNET(gateway)) == LOLND ||
 294             net == LNET_NIDNET(LNET_NID_ANY) ||
 295             LNET_NETTYP(net) == LOLND ||
 296             LNET_NIDNET(gateway) == net ||
 297             hops < 1 || hops > 255)
 298                 return (-EINVAL);
 299
 300         if (lnet_islocalnet(net))               /* it's a local network */
 301                 return 0;                       /* ignore the route entry */
 302
 303         /* Assume net, route, all new */
 304         LIBCFS_ALLOC(route, sizeof(*route));
 305         LIBCFS_ALLOC(rnet, sizeof(*rnet));
 306         if (route == NULL || rnet == NULL) {
 307                 CERROR("Out of memory creating route %s %d %s\n",
 308                        libcfs_net2str(net), hops, libcfs_nid2str(gateway));
 309                 if (route != NULL)
 310                         LIBCFS_FREE(route, sizeof(*route));
 311                 if (rnet != NULL)
 312                         LIBCFS_FREE(rnet, sizeof(*rnet));
 313                 return -ENOMEM;
 314         }
 315
 316         CFS_INIT_LIST_HEAD(&rnet->lrn_routes);
 317         rnet->lrn_net = net;
 318         rnet->lrn_hops = hops;
 319
 320         LNET_LOCK();
 321
 322         rc = lnet_nid2peer_locked(&route->lr_gateway, gateway);
 323         if (rc != 0) {
 324                 LNET_UNLOCK();
 325
 326                 LIBCFS_FREE(route, sizeof(*route));
 327                 LIBCFS_FREE(rnet, sizeof(*rnet));
 328
 329                 if (rc == -EHOSTUNREACH)        /* gateway is not on a local net */
 330                         return 0;               /* ignore the route entry */
 331
 332                 CERROR("Error %d creating route %s %d %s\n", rc,
 333                        libcfs_net2str(net), hops, libcfs_nid2str(gateway));
 334                 return rc;
 335         }
 336
 337         LASSERT (!the_lnet.ln_shutdown);
 338         CFS_INIT_LIST_HEAD(&zombies);
 339
 340         rnet2 = lnet_find_net_locked(net);
 341         if (rnet2 == NULL) {
 342                 /* new network */
 343                 list_add_tail(&rnet->lrn_list, &the_lnet.ln_remote_nets);
 344                 rnet2 = rnet;
 345         }
 346
 347         if (hops > rnet2->lrn_hops) {
 348                 /* New route is longer; ignore it */
 349                 add_route = 0;
 350         } else if (hops < rnet2->lrn_hops) {
 351                 /* new route supercedes all currently known routes to this
 352                  * net */
 353                 list_add(&zombies, &rnet2->lrn_routes);
 354                 list_del_init(&rnet2->lrn_routes);
 355                 add_route = 1;
 356         } else {
 357                 add_route = 1;
 358                 /* New route has the same hopcount as existing routes; search
 359                  * for a duplicate route (it's a NOOP if it is) */
 360                 list_for_each (e, &rnet2->lrn_routes) {
 361                         route2 = list_entry(e, lnet_route_t, lr_list);
 362
 363                         if (route2->lr_gateway == route->lr_gateway) {
 364                                 add_route = 0;
 365                                 break;
 366                         }
 367
 368                         /* our loopups must be true */
 369                         LASSERT (route2->lr_gateway->lp_nid != gateway);
 370                 }
 371         }
 372
 373         if (add_route) {
 374                 ni = route->lr_gateway->lp_ni;
 375                 lnet_ni_addref_locked(ni);
 376
 377                 LASSERT (rc == 0);
 378                 list_add_tail(&route->lr_list, &rnet2->lrn_routes);
 379                 the_lnet.ln_remote_nets_version++;
 380
 381                 lnet_rtr_addref_locked(route->lr_gateway);
 382
 383                 LNET_UNLOCK();
 384
 385                 /* XXX Assume alive */
 386                 if (ni->ni_lnd->lnd_notify != NULL)
 387                         (ni->ni_lnd->lnd_notify)(ni, gateway, 1);
 388
 389                 lnet_ni_decref(ni);
 390         } else {
 391                 lnet_peer_decref_locked(route->lr_gateway);
 392                 LNET_UNLOCK();
 393                 LIBCFS_FREE(route, sizeof(*route));
 394         }
 395
 396         if (rnet != rnet2)
 397                 LIBCFS_FREE(rnet, sizeof(*rnet));
 398
 399         while (!list_empty(&zombies)) {
 400                 route = list_entry(zombies.next, lnet_route_t, lr_list);
 401                 list_del(&route->lr_list);
 402
 403                 LNET_LOCK();
 404                 lnet_rtr_decref_locked(route->lr_gateway);
 405                 lnet_peer_decref_locked(route->lr_gateway);
 406                 LNET_UNLOCK();
 407                 LIBCFS_FREE(route, sizeof(*route));
 408         }
 409
 410         return rc;
 411 }
 412
 413 int
 414 lnet_check_routes (void)
 415 {
 416         lnet_remotenet_t    *rnet;
 417         lnet_route_t        *route;
 418         lnet_route_t        *route2;
 419         struct list_head    *e1;
 420         struct list_head    *e2;
 421
 422         LNET_LOCK();
 423
 424         list_for_each (e1, &the_lnet.ln_remote_nets) {
 425                 rnet = list_entry(e1, lnet_remotenet_t, lrn_list);
 426
 427                 route2 = NULL;
 428                 list_for_each (e2, &rnet->lrn_routes) {
 429                         route = list_entry(e2, lnet_route_t, lr_list);
 430
 431                         if (route2 == NULL)
 432                                 route2 = route;
 433                         else if (route->lr_gateway->lp_ni !=
 434                                  route2->lr_gateway->lp_ni) {
 435                                 LNET_UNLOCK();
 436
 437                                 CERROR("Routes to %s via %s and %s not supported\n",
 438                                        libcfs_net2str(rnet->lrn_net),
 439                                        libcfs_nid2str(route->lr_gateway->lp_nid),
 440                                        libcfs_nid2str(route2->lr_gateway->lp_nid));
 441                                 return -EINVAL;
 442                         }
 443                 }
 444         }
 445
 446         LNET_UNLOCK();
 447         return 0;
 448 }
 449
 450 int
 451 lnet_del_route (__u32 net, lnet_nid_t gw_nid)
 452 {
 453         lnet_remotenet_t    *rnet;
 454         lnet_route_t        *route;
 455         struct list_head    *e1;
 456         struct list_head    *e2;
 457         int                  rc = -ENOENT;
 458
 459         CDEBUG(D_NET, "Del route: net %s : gw %s\n",
 460                libcfs_net2str(net), libcfs_nid2str(gw_nid));
 461
 462         /* NB Caller may specify either all routes via the given gateway
 463          * or a specific route entry actual NIDs) */
 464
 465  again:
 466         LNET_LOCK();
 467
 468         list_for_each (e1, &the_lnet.ln_remote_nets) {
 469                 rnet = list_entry(e1, lnet_remotenet_t, lrn_list);
 470
 471                 if (!(net == LNET_NIDNET(LNET_NID_ANY) ||
 472                       net == rnet->lrn_net))
 473                         continue;
 474
 475                 list_for_each (e2, &rnet->lrn_routes) {
 476                         route = list_entry(e2, lnet_route_t, lr_list);
 477
 478                         if (!(gw_nid == LNET_NID_ANY ||
 479                               gw_nid == route->lr_gateway->lp_nid))
 480                                 continue;
 481
 482                         list_del(&route->lr_list);
 483                         the_lnet.ln_remote_nets_version++;
 484
 485                         if (list_empty(&rnet->lrn_routes))
 486                                 list_del(&rnet->lrn_list);
 487                         else
 488                                 rnet = NULL;
 489
 490                         lnet_rtr_decref_locked(route->lr_gateway);
 491                         lnet_peer_decref_locked(route->lr_gateway);
 492                         LNET_UNLOCK();
 493
 494                         LIBCFS_FREE(route, sizeof (*route));
 495
 496                         if (rnet != NULL)
 497                                 LIBCFS_FREE(rnet, sizeof(*rnet));
 498
 499                         rc = 0;
 500                         goto again;
 501                 }
 502         }
 503
 504         LNET_UNLOCK();
 505         return rc;
 506 }
 507
 508 void
 509 lnet_destroy_routes (void)
 510 {
 511         lnet_del_route(LNET_NIDNET(LNET_NID_ANY), LNET_NID_ANY);
 512 }
 513
 514 int
 515 lnet_get_route (int idx, __u32 *net, __u32 *hops,
 516                lnet_nid_t *gateway, __u32 *alive)
 517 {
 518         struct list_head    *e1;
 519         struct list_head    *e2;
 520         lnet_remotenet_t    *rnet;
 521         lnet_route_t        *route;
 522
 523         LNET_LOCK();
 524
 525         list_for_each (e1, &the_lnet.ln_remote_nets) {
 526                 rnet = list_entry(e1, lnet_remotenet_t, lrn_list);
 527
 528                 list_for_each (e2, &rnet->lrn_routes) {
 529                         route = list_entry(e2, lnet_route_t, lr_list);
 530
 531                         if (idx-- == 0) {
 532                                 *net     = rnet->lrn_net;
 533                                 *hops    = rnet->lrn_hops;
 534                                 *gateway = route->lr_gateway->lp_nid;
 535                                 *alive   = route->lr_gateway->lp_alive;
 536                                 LNET_UNLOCK();
 537                                 return 0;
 538                         }
 539                 }
 540         }
 541
 542         LNET_UNLOCK();
 543         return -ENOENT;
 544 }
 545
 546 #if defined(__KERNEL__) && defined(LNET_ROUTER)
 547 static void
 548 lnet_router_checker_event (lnet_event_t *event)
 549 {
 550         /* CAVEAT EMPTOR: I'm called with LNET_LOCKed and I'm not allowed to
 551          * drop it (that's how come I see _every_ event, even ones that would
 552          * overflow my EQ) */
 553         lnet_peer_t   *lp;
 554         lnet_nid_t     nid;
 555
 556         if (event->unlinked) {
 557                 /* The router checker thread has unlinked the rc_md
 558                  * and exited. */
 559                 LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_UNLINKING);
 560                 the_lnet.ln_rc_state = LNET_RC_STATE_UNLINKED;
 561                 mutex_up(&the_lnet.ln_rc_signal);
 562                 return;
 563         }
 564
 565         LASSERT (event->type == LNET_EVENT_SEND ||
 566                  event->type == LNET_EVENT_REPLY);
 567
 568         nid = (event->type == LNET_EVENT_SEND) ?
 569               event->target.nid : event->initiator.nid;
 570
 571         lp = lnet_find_peer_locked(nid);
 572         if (lp == NULL) {
 573                 /* router may have been removed */
 574                 CDEBUG(D_NET, "Router %s not found\n", libcfs_nid2str(nid));
 575                 return;
 576         }
 577
 578         if (event->type == LNET_EVENT_SEND)     /* re-enable another ping */
 579                 lp->lp_ping_notsent = 0;
 580
 581         if (lnet_isrouter(lp) &&                /* ignore if no longer a router */
 582             (event->status != 0 ||
 583              event->type == LNET_EVENT_REPLY)) {
 584
 585                 /* A successful REPLY means the router is up.  If _any_ comms
 586                  * to the router fail I assume it's down (this will happen if
 587                  * we ping alive routers to try to detect router death before
 588                  * apps get burned). */
 589
 590                 lnet_notify_locked(lp, 1, (event->status == 0),
 591                                    cfs_time_current_sec());
 592
 593                 /* The router checker will wake up very shortly and do the
 594                  * actual notification.
 595                  * XXX If 'lp' stops being a router before then, it will still
 596                  * have the notification pending!!! */
 597         }
 598
 599         /* This decref will NOT drop LNET_LOCK (it had to have 1 ref when it
 600          * was in the peer table and I've not dropped the lock, so no-one else
 601          * can have reduced the refcount) */
 602         LASSERT(lp->lp_refcount > 1);
 603
 604         lnet_peer_decref_locked(lp);
 605 }
 606
 607 static int
 608 lnet_router_checker(void *arg)
 609 {
 610         static lnet_ping_info_t   pinginfo;
 611
 612         int                  rc;
 613         lnet_handle_md_t     mdh;
 614         lnet_peer_t         *rtr;
 615         lnet_md_t            md = {0};
 616         struct list_head    *entry;
 617         time_t               now;
 618         lnet_process_id_t    rtr_id;
 619         int                  secs;
 620
 621         cfs_daemonize("router_checker");
 622         cfs_block_allsigs();
 623
 624         rtr_id.pid = LUSTRE_SRV_LNET_PID;
 625
 626         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);
 627
 628         /* initialize md content */
 629         md.start     = &pinginfo;
 630         md.length    = sizeof(pinginfo);
 631         md.threshold = LNET_MD_THRESH_INF;
 632         md.max_size  = 0;
 633         md.options   = LNET_MD_TRUNCATE,
 634         md.user_ptr  = NULL;
 635         md.eq_handle = the_lnet.ln_rc_eqh;
 636
 637         rc = LNetMDBind(md, LNET_UNLINK, &mdh);
 638
 639         if (rc < 0) {
 640                 CERROR("Can't bind MD: %d\n", rc);
 641                 the_lnet.ln_rc_state = rc;
 642                 mutex_up(&the_lnet.ln_rc_signal);
 643                 return rc;
 644         }
 645
 646         LASSERT (rc == 0);
 647
 648         the_lnet.ln_rc_state = LNET_RC_STATE_RUNNING;
 649         mutex_up(&the_lnet.ln_rc_signal);       /* let my parent go */
 650
 651         while (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING) {
 652                 __u64 version;
 653
 654                 LNET_LOCK();
 655 rescan:
 656                 version = the_lnet.ln_routers_version;
 657
 658                 list_for_each (entry, &the_lnet.ln_routers) {
 659                         rtr = list_entry(entry, lnet_peer_t, lp_rtr_list);
 660
 661                         lnet_peer_addref_locked(rtr);
 662
 663                         now = cfs_time_current_sec();
 664
 665                         if (rtr->lp_ping_deadline != 0 && /* ping timed out? */
 666                             now > rtr->lp_ping_deadline)
 667                                 lnet_notify_locked(rtr, 1, 0, now);
 668
 669                         LNET_UNLOCK();
 670
 671                         /* Run any outstanding notificiations */
 672                         lnet_do_notify(rtr);
 673
 674                         if (rtr->lp_alive) {
 675                                 secs = live_router_check_interval;
 676                         } else {
 677                                 secs = dead_router_check_interval;
 678                         }
 679                         if (secs <= 0)
 680                                 secs = 0;
 681
 682                         if (secs != 0 &&
 683                             !rtr->lp_ping_notsent &&
 684                             now > rtr->lp_ping_timestamp + secs) {
 685                                 CDEBUG(D_NET, "Check: %s\n",
 686                                        libcfs_nid2str(rtr->lp_nid));
 687
 688                                 LNET_LOCK();
 689                                 rtr_id.nid = rtr->lp_nid;
 690                                 rtr->lp_ping_notsent = 1;
 691                                 rtr->lp_ping_timestamp = now;
 692
 693                                 if (rtr->lp_ping_deadline == 0)
 694                                         rtr->lp_ping_deadline =
 695                                                 now + router_ping_timeout;
 696
 697                                 LNET_UNLOCK();
 698
 699                                 LNetGet(LNET_NID_ANY, mdh, rtr_id,
 700                                         LNET_RESERVED_PORTAL,
 701                                         LNET_PROTO_PING_MATCHBITS, 0);
 702                         }
 703
 704                         LNET_LOCK();
 705                         lnet_peer_decref_locked(rtr);
 706
 707                         if (version != the_lnet.ln_routers_version) {
 708                                 /* the routers list has changed */
 709                                 goto rescan;
 710                         }
 711                 }
 712
 713                 LNET_UNLOCK();
 714
 715                 /* Call cfs_pause() here always adds 1 to load average
 716                  * because kernel counts # active tasks as nr_running
 717                  * + nr_uninterruptible. */
 718                 cfs_schedule_timeout(CFS_TASK_INTERRUPTIBLE,
 719                                      cfs_time_seconds(1));
 720         }
 721
 722         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_STOPTHREAD);
 723         the_lnet.ln_rc_state = LNET_RC_STATE_UNLINKING;
 724
 725         rc = LNetMDUnlink(mdh);
 726         LASSERT (rc == 0);
 727
 728         /* The unlink event callback will signal final completion */
 729         return 0;
 730 }
 731
 732
 733 void
 734 lnet_wait_known_routerstate(void)
 735 {
 736         lnet_peer_t         *rtr;
 737         struct list_head    *entry;
 738         int                  all_known;
 739
 740         for (;;) {
 741                 LNET_LOCK();
 742
 743                 all_known = 1;
 744                 list_for_each (entry, &the_lnet.ln_routers) {
 745                         rtr = list_entry(entry, lnet_peer_t, lp_rtr_list);
 746
 747                         if (rtr->lp_alive_count == 0) {
 748                                 all_known = 0;
 749                                 break;
 750                         }
 751                 }
 752
 753                 LNET_UNLOCK();
 754
 755                 if (all_known)
 756                         return;
 757
 758                 cfs_pause(cfs_time_seconds(1));
 759         }
 760 }
 761
 762 void
 763 lnet_router_checker_stop(void)
 764 {
 765         int       rc;
 766
 767         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING ||
 768                  the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);
 769
 770         if (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN)
 771                 return;
 772
 773         the_lnet.ln_rc_state = LNET_RC_STATE_STOPTHREAD;
 774         /* block until event callback signals exit */
 775         mutex_down(&the_lnet.ln_rc_signal);
 776
 777         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_UNLINKED);
 778
 779         rc = LNetEQFree(the_lnet.ln_rc_eqh);
 780         LASSERT (rc == 0);
 781
 782         the_lnet.ln_rc_state = LNET_RC_STATE_SHUTDOWN;
 783 }
 784
 785 int
 786 lnet_router_checker_start(void)
 787 {
 788         int  rc;
 789
 790         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);
 791
 792         if (check_routers_before_use &&
 793             dead_router_check_interval <= 0) {
 794                 LCONSOLE_ERROR_MSG(0x10a, "'dead_router_check_interval' must be"
 795                                    " set if 'check_routers_before_use' is set"
 796                                    "\n");
 797                 return -EINVAL;
 798         }
 799
 800         if (live_router_check_interval <= 0 &&
 801             dead_router_check_interval <= 0)
 802                 return 0;
 803
 804         init_mutex_locked(&the_lnet.ln_rc_signal);
 805
 806         /* EQ size doesn't matter; the callback is guaranteed to get every
 807          * event */
 808         rc = LNetEQAlloc(1, lnet_router_checker_event,
 809                          &the_lnet.ln_rc_eqh);
 810         if (rc != 0) {
 811                 CERROR("Can't allocate EQ: %d\n", rc);
 812                 return -ENOMEM;
 813         }
 814
 815         rc = (int)cfs_kernel_thread(lnet_router_checker, NULL, 0);
 816         if (rc < 0) {
 817                 CERROR("Can't start router checker thread: %d\n", rc);
 818                 goto failed;
 819         }
 820
 821         mutex_down(&the_lnet.ln_rc_signal);     /* wait for checker to startup */
 822
 823         rc = the_lnet.ln_rc_state;
 824         if (rc < 0) {
 825                 the_lnet.ln_rc_state = LNET_RC_STATE_SHUTDOWN;
 826                 goto failed;
 827         }
 828
 829         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING);
 830
 831         if (check_routers_before_use) {
 832                 /* Note that a helpful side-effect of pinging all known routers
 833                  * at startup is that it makes them drop stale connections they
 834                  * may have to a previous instance of me. */
 835                 lnet_wait_known_routerstate();
 836         }
 837
 838         return 0;
 839
 840  failed:
 841         rc = LNetEQFree(the_lnet.ln_rc_eqh);
 842         LASSERT (rc == 0);
 843         return rc;
 844 }
 845
 846 void
 847 lnet_destroy_rtrbuf(lnet_rtrbuf_t *rb, int npages)
 848 {
 849         int sz = offsetof(lnet_rtrbuf_t, rb_kiov[npages]);
 850
 851         while (--npages >= 0)
 852                 cfs_free_page(rb->rb_kiov[npages].kiov_page);
 853
 854         LIBCFS_FREE(rb, sz);
 855 }
 856
 857 lnet_rtrbuf_t *
 858 lnet_new_rtrbuf(lnet_rtrbufpool_t *rbp)
 859 {
 860         int            npages = rbp->rbp_npages;
 861         int            sz = offsetof(lnet_rtrbuf_t, rb_kiov[npages]);
 862         struct page   *page;
 863         lnet_rtrbuf_t *rb;
 864         int            i;
 865
 866         LIBCFS_ALLOC(rb, sz);
 867         if (rb == NULL)
 868                 return NULL;
 869
 870         rb->rb_pool = rbp;
 871
 872         for (i = 0; i < npages; i++) {
 873                 page = cfs_alloc_page(CFS_ALLOC_ZERO | CFS_ALLOC_STD);
 874                 if (page == NULL) {
 875                         while (--i >= 0)
 876                                 cfs_free_page(rb->rb_kiov[i].kiov_page);
 877
 878                         LIBCFS_FREE(rb, sz);
 879                         return NULL;
 880                 }
 881
 882                 rb->rb_kiov[i].kiov_len = CFS_PAGE_SIZE;
 883                 rb->rb_kiov[i].kiov_offset = 0;
 884                 rb->rb_kiov[i].kiov_page = page;
 885         }
 886
 887         return rb;
 888 }
 889
 890 void
 891 lnet_rtrpool_free_bufs(lnet_rtrbufpool_t *rbp)
 892 {
 893         int            npages = rbp->rbp_npages;
 894         int            nbuffers = 0;
 895         lnet_rtrbuf_t *rb;
 896
 897         LASSERT (list_empty(&rbp->rbp_msgs));
 898         LASSERT (rbp->rbp_credits == rbp->rbp_nbuffers);
 899
 900         while (!list_empty(&rbp->rbp_bufs)) {
 901                 LASSERT (rbp->rbp_credits > 0);
 902
 903                 rb = list_entry(rbp->rbp_bufs.next,
 904                                 lnet_rtrbuf_t, rb_list);
 905                 list_del(&rb->rb_list);
 906                 lnet_destroy_rtrbuf(rb, npages);
 907                 nbuffers++;
 908         }
 909
 910         LASSERT (rbp->rbp_nbuffers == nbuffers);
 911         LASSERT (rbp->rbp_credits == nbuffers);
 912
 913         rbp->rbp_nbuffers = rbp->rbp_credits = 0;
 914 }
 915
 916 int
 917 lnet_rtrpool_alloc_bufs(lnet_rtrbufpool_t *rbp, int nbufs)
 918 {
 919         lnet_rtrbuf_t *rb;
 920         int            i;
 921
 922         if (rbp->rbp_nbuffers != 0) {
 923                 LASSERT (rbp->rbp_nbuffers == nbufs);
 924                 return 0;
 925         }
 926
 927         for (i = 0; i < nbufs; i++) {
 928                 rb = lnet_new_rtrbuf(rbp);
 929
 930                 if (rb == NULL) {
 931                         CERROR("Failed to allocate %d router bufs of %d pages\n",
 932                                nbufs, rbp->rbp_npages);
 933                         return -ENOMEM;
 934                 }
 935
 936                 rbp->rbp_nbuffers++;
 937                 rbp->rbp_credits++;
 938                 rbp->rbp_mincredits++;
 939                 list_add(&rb->rb_list, &rbp->rbp_bufs);
 940
 941                 /* No allocation "under fire" */
 942                 /* Otherwise we'd need code to schedule blocked msgs etc */
 943                 LASSERT (!the_lnet.ln_routing);
 944         }
 945
 946         LASSERT (rbp->rbp_credits == nbufs);
 947         return 0;
 948 }
 949
 950 void
 951 lnet_rtrpool_init(lnet_rtrbufpool_t *rbp, int npages)
 952 {
 953         CFS_INIT_LIST_HEAD(&rbp->rbp_msgs);
 954         CFS_INIT_LIST_HEAD(&rbp->rbp_bufs);
 955
 956         rbp->rbp_npages = npages;
 957         rbp->rbp_credits = 0;
 958         rbp->rbp_mincredits = 0;
 959 }
 960
 961 void
 962 lnet_free_rtrpools(void)
 963 {
 964         lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[0]);
 965         lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[1]);
 966         lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[2]);
 967 }
 968
 969 void
 970 lnet_init_rtrpools(void)
 971 {
 972         int small_pages = 1;
 973         int large_pages = (LNET_MTU + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT;
 974
 975         lnet_rtrpool_init(&the_lnet.ln_rtrpools[0], 0);
 976         lnet_rtrpool_init(&the_lnet.ln_rtrpools[1], small_pages);
 977         lnet_rtrpool_init(&the_lnet.ln_rtrpools[2], large_pages);
 978 }
 979
 980
 981 int
 982 lnet_alloc_rtrpools(int im_a_router)
 983 {
 984         int       rc;
 985
 986         if (!strcmp(forwarding, "")) {
 987                 /* not set either way */
 988                 if (!im_a_router)
 989                         return 0;
 990         } else if (!strcmp(forwarding, "disabled")) {
 991                 /* explicitly disabled */
 992                 return 0;
 993         } else if (!strcmp(forwarding, "enabled")) {
 994                 /* explicitly enabled */
 995         } else {
 996                 LCONSOLE_ERROR_MSG(0x10b, "'forwarding' not set to either "
 997                                    "'enabled' or 'disabled'\n");
 998                 return -EINVAL;
 999         }
1000
1001         if (tiny_router_buffers <= 0) {
1002                 LCONSOLE_ERROR_MSG(0x10c, "tiny_router_buffers=%d invalid when "
1003                                    "routing enabled\n", tiny_router_buffers);
1004                 rc = -EINVAL;
1005                 goto failed;
1006         }
1007
1008         rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[0],
1009                                      tiny_router_buffers);
1010         if (rc != 0)
1011                 goto failed;
1012
1013         if (small_router_buffers <= 0) {
1014                 LCONSOLE_ERROR_MSG(0x10d, "small_router_buffers=%d invalid when"
1015                                    " routing enabled\n", small_router_buffers);
1016                 rc = -EINVAL;
1017                 goto failed;
1018         }
1019
1020         rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[1],
1021                                      small_router_buffers);
1022         if (rc != 0)
1023                 goto failed;
1024
1025         if (large_router_buffers <= 0) {
1026                 LCONSOLE_ERROR_MSG(0x10e, "large_router_buffers=%d invalid when"
1027                                    " routing enabled\n", large_router_buffers);
1028                 rc = -EINVAL;
1029                 goto failed;
1030         }
1031
1032         rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[2],
1033                                      large_router_buffers);
1034         if (rc != 0)
1035                 goto failed;
1036
1037         LNET_LOCK();
1038         the_lnet.ln_routing = 1;
1039         LNET_UNLOCK();
1040
1041         return 0;
1042
1043  failed:
1044         lnet_free_rtrpools();
1045         return rc;
1046 }
1047
1048 #else
1049
1050 int
1051 lnet_peers_start_down(void)
1052 {
1053         return 0;
1054 }
1055
1056 void
1057 lnet_router_checker_stop(void)
1058 {
1059         return;
1060 }
1061
1062 int
1063 lnet_router_checker_start(void)
1064 {
1065         return 0;
1066 }
1067
1068 void
1069 lnet_free_rtrpools (void)
1070 {
1071 }
1072
1073 void
1074 lnet_init_rtrpools (void)
1075 {
1076 }
1077
1078 int
1079 lnet_alloc_rtrpools (int im_a_arouter)
1080 {
1081         return 0;
1082 }
1083
1084 #endif