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