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