Whamcloud - gitweb
LU-3302 llog: Do not use ostid swab for llogid
[fs/lustre-release.git] / lnet / klnds / ralnd / ralnd.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lnet/klnds/ralnd/ralnd.c
37  *
38  * Author: Eric Barton <eric@bartonsoftware.com>
39  */
40 #include "ralnd.h"
41
42 static int        kranal_devids[RANAL_MAXDEVS] = {RAPK_MAIN_DEVICE_ID,
43                                                   RAPK_EXPANSION_DEVICE_ID};
44
45 lnd_t the_kralnd = {
46         .lnd_type       = RALND,
47         .lnd_startup    = kranal_startup,
48         .lnd_shutdown   = kranal_shutdown,
49         .lnd_ctl        = kranal_ctl,
50         .lnd_send       = kranal_send,
51         .lnd_recv       = kranal_recv,
52         .lnd_eager_recv = kranal_eager_recv,
53         .lnd_accept     = kranal_accept,
54 };
55
56 kra_data_t              kranal_data;
57
58 void
59 kranal_pack_connreq(kra_connreq_t *connreq, kra_conn_t *conn, lnet_nid_t dstnid)
60 {
61         RAP_RETURN   rrc;
62
63         memset(connreq, 0, sizeof(*connreq));
64
65         connreq->racr_magic     = RANAL_MSG_MAGIC;
66         connreq->racr_version   = RANAL_MSG_VERSION;
67
68         if (conn == NULL)                       /* prepping a "stub" reply */
69                 return;
70
71         connreq->racr_devid     = conn->rac_device->rad_id;
72         connreq->racr_srcnid    = kranal_data.kra_ni->ni_nid;
73         connreq->racr_dstnid    = dstnid;
74         connreq->racr_peerstamp = kranal_data.kra_peerstamp;
75         connreq->racr_connstamp = conn->rac_my_connstamp;
76         connreq->racr_timeout   = conn->rac_timeout;
77
78         rrc = RapkGetRiParams(conn->rac_rihandle, &connreq->racr_riparams);
79         LASSERT(rrc == RAP_SUCCESS);
80 }
81
82 int
83 kranal_recv_connreq(struct socket *sock, kra_connreq_t *connreq, int active)
84 {
85         int         timeout = active ? *kranal_tunables.kra_timeout :
86                                         lnet_acceptor_timeout();
87         int         swab;
88         int         rc;
89
90         /* return 0 on success, -ve on error, +ve to tell the peer I'm "old" */
91
92         rc = libcfs_sock_read(sock, &connreq->racr_magic, 
93                               sizeof(connreq->racr_magic), timeout);
94         if (rc != 0) {
95                 CERROR("Read(magic) failed(1): %d\n", rc);
96                 return -EIO;
97         }
98
99         if (connreq->racr_magic != RANAL_MSG_MAGIC &&
100             connreq->racr_magic != __swab32(RANAL_MSG_MAGIC)) {
101                 /* Unexpected magic! */
102                 if (!active &&
103                     (connreq->racr_magic == LNET_PROTO_MAGIC ||
104                      connreq->racr_magic == __swab32(LNET_PROTO_MAGIC))) {
105                         /* future protocol version compatibility!
106                          * When LNET unifies protocols over all LNDs, the first
107                          * thing sent will be a version query.  +ve rc means I
108                          * reply with my current magic/version */
109                         return EPROTO;
110                 }
111
112                 CERROR("Unexpected magic %08x (%s)\n",
113                        connreq->racr_magic, active ? "active" : "passive");
114                 return -EPROTO;
115         }
116
117         swab = (connreq->racr_magic == __swab32(RANAL_MSG_MAGIC));
118
119         rc = libcfs_sock_read(sock, &connreq->racr_version,
120                               sizeof(connreq->racr_version), timeout);
121         if (rc != 0) {
122                 CERROR("Read(version) failed: %d\n", rc);
123                 return -EIO;
124         }
125
126         if (swab)
127                 __swab16s(&connreq->racr_version);
128         
129         if (connreq->racr_version != RANAL_MSG_VERSION) {
130                 if (active) {
131                         CERROR("Unexpected version %d\n", connreq->racr_version);
132                         return -EPROTO;
133                 }
134                 /* If this is a future version of the ralnd protocol, and I'm
135                  * passive (accepted the connection), tell my peer I'm "old"
136                  * (+ve rc) */
137                 return EPROTO;
138         }
139
140         rc = libcfs_sock_read(sock, &connreq->racr_devid,
141                               sizeof(connreq->racr_version) -
142                               offsetof(kra_connreq_t, racr_devid),
143                               timeout);
144         if (rc != 0) {
145                 CERROR("Read(body) failed: %d\n", rc);
146                 return -EIO;
147         }
148
149         if (swab) {
150                 __swab32s(&connreq->racr_magic);
151                 __swab16s(&connreq->racr_version);
152                 __swab16s(&connreq->racr_devid);
153                 __swab64s(&connreq->racr_srcnid);
154                 __swab64s(&connreq->racr_dstnid);
155                 __swab64s(&connreq->racr_peerstamp);
156                 __swab64s(&connreq->racr_connstamp);
157                 __swab32s(&connreq->racr_timeout);
158
159                 __swab32s(&connreq->racr_riparams.HostId);
160                 __swab32s(&connreq->racr_riparams.FmaDomainHndl);
161                 __swab32s(&connreq->racr_riparams.PTag);
162                 __swab32s(&connreq->racr_riparams.CompletionCookie);
163         }
164
165         if (connreq->racr_srcnid == LNET_NID_ANY ||
166             connreq->racr_dstnid == LNET_NID_ANY) {
167                 CERROR("Received LNET_NID_ANY\n");
168                 return -EPROTO;
169         }
170
171         if (connreq->racr_timeout < RANAL_MIN_TIMEOUT) {
172                 CERROR("Received timeout %d < MIN %d\n",
173                        connreq->racr_timeout, RANAL_MIN_TIMEOUT);
174                 return -EPROTO;
175         }
176
177         return 0;
178 }
179
180 int
181 kranal_close_stale_conns_locked (kra_peer_t *peer, kra_conn_t *newconn)
182 {
183         kra_conn_t         *conn;
184         cfs_list_t         *ctmp;
185         cfs_list_t         *cnxt;
186         int                 loopback;
187         int                 count = 0;
188
189         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
190
191         cfs_list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
192                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_list);
193
194                 if (conn == newconn)
195                         continue;
196
197                 if (conn->rac_peerstamp != newconn->rac_peerstamp) {
198                         CDEBUG(D_NET, "Closing stale conn nid: %s "
199                                " peerstamp:"LPX64"("LPX64")\n", 
200                                libcfs_nid2str(peer->rap_nid),
201                                conn->rac_peerstamp, newconn->rac_peerstamp);
202                         LASSERT (conn->rac_peerstamp < newconn->rac_peerstamp);
203                         count++;
204                         kranal_close_conn_locked(conn, -ESTALE);
205                         continue;
206                 }
207
208                 if (conn->rac_device != newconn->rac_device)
209                         continue;
210
211                 if (loopback &&
212                     newconn->rac_my_connstamp == conn->rac_peer_connstamp &&
213                     newconn->rac_peer_connstamp == conn->rac_my_connstamp)
214                         continue;
215
216                 LASSERT (conn->rac_peer_connstamp < newconn->rac_peer_connstamp);
217
218                 CDEBUG(D_NET, "Closing stale conn nid: %s"
219                        " connstamp:"LPX64"("LPX64")\n", 
220                        libcfs_nid2str(peer->rap_nid),
221                        conn->rac_peer_connstamp, newconn->rac_peer_connstamp);
222
223                 count++;
224                 kranal_close_conn_locked(conn, -ESTALE);
225         }
226
227         return count;
228 }
229
230 int
231 kranal_conn_isdup_locked(kra_peer_t *peer, kra_conn_t *newconn)
232 {
233         kra_conn_t       *conn;
234         cfs_list_t       *tmp;
235         int               loopback;
236
237         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
238
239         cfs_list_for_each(tmp, &peer->rap_conns) {
240                 conn = cfs_list_entry(tmp, kra_conn_t, rac_list);
241
242                 /* 'newconn' is from an earlier version of 'peer'!!! */
243                 if (newconn->rac_peerstamp < conn->rac_peerstamp)
244                         return 1;
245
246                 /* 'conn' is from an earlier version of 'peer': it will be
247                  * removed when we cull stale conns later on... */
248                 if (newconn->rac_peerstamp > conn->rac_peerstamp)
249                         continue;
250
251                 /* Different devices are OK */
252                 if (conn->rac_device != newconn->rac_device)
253                         continue;
254
255                 /* It's me connecting to myself */
256                 if (loopback &&
257                     newconn->rac_my_connstamp == conn->rac_peer_connstamp &&
258                     newconn->rac_peer_connstamp == conn->rac_my_connstamp)
259                         continue;
260
261                 /* 'newconn' is an earlier connection from 'peer'!!! */
262                 if (newconn->rac_peer_connstamp < conn->rac_peer_connstamp)
263                         return 2;
264
265                 /* 'conn' is an earlier connection from 'peer': it will be
266                  * removed when we cull stale conns later on... */
267                 if (newconn->rac_peer_connstamp > conn->rac_peer_connstamp)
268                         continue;
269
270                 /* 'newconn' has the SAME connection stamp; 'peer' isn't
271                  * playing the game... */
272                 return 3;
273         }
274
275         return 0;
276 }
277
278 void
279 kranal_set_conn_uniqueness (kra_conn_t *conn)
280 {
281         unsigned long  flags;
282
283         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
284
285         conn->rac_my_connstamp = kranal_data.kra_connstamp++;
286
287         do {    /* allocate a unique cqid */
288                 conn->rac_cqid = kranal_data.kra_next_cqid++;
289         } while (kranal_cqid2conn_locked(conn->rac_cqid) != NULL);
290
291         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
292 }
293
294 int
295 kranal_create_conn(kra_conn_t **connp, kra_device_t *dev)
296 {
297         kra_conn_t    *conn;
298         RAP_RETURN     rrc;
299
300         LASSERT (!cfs_in_interrupt());
301         LIBCFS_ALLOC(conn, sizeof(*conn));
302
303         if (conn == NULL)
304                 return -ENOMEM;
305
306         memset(conn, 0, sizeof(*conn));
307         cfs_atomic_set(&conn->rac_refcount, 1);
308         CFS_INIT_LIST_HEAD(&conn->rac_list);
309         CFS_INIT_LIST_HEAD(&conn->rac_hashlist);
310         CFS_INIT_LIST_HEAD(&conn->rac_schedlist);
311         CFS_INIT_LIST_HEAD(&conn->rac_fmaq);
312         CFS_INIT_LIST_HEAD(&conn->rac_rdmaq);
313         CFS_INIT_LIST_HEAD(&conn->rac_replyq);
314         spin_lock_init(&conn->rac_lock);
315
316         kranal_set_conn_uniqueness(conn);
317
318         conn->rac_device = dev;
319         conn->rac_timeout = MAX(*kranal_tunables.kra_timeout, RANAL_MIN_TIMEOUT);
320         kranal_update_reaper_timeout(conn->rac_timeout);
321
322         rrc = RapkCreateRi(dev->rad_handle, conn->rac_cqid,
323                            &conn->rac_rihandle);
324         if (rrc != RAP_SUCCESS) {
325                 CERROR("RapkCreateRi failed: %d\n", rrc);
326                 LIBCFS_FREE(conn, sizeof(*conn));
327                 return -ENETDOWN;
328         }
329
330         cfs_atomic_inc(&kranal_data.kra_nconns);
331         *connp = conn;
332         return 0;
333 }
334
335 void
336 kranal_destroy_conn(kra_conn_t *conn)
337 {
338         RAP_RETURN         rrc;
339
340         LASSERT (!cfs_in_interrupt());
341         LASSERT (!conn->rac_scheduled);
342         LASSERT (cfs_list_empty(&conn->rac_list));
343         LASSERT (cfs_list_empty(&conn->rac_hashlist));
344         LASSERT (cfs_list_empty(&conn->rac_schedlist));
345         LASSERT (cfs_atomic_read(&conn->rac_refcount) == 0);
346         LASSERT (cfs_list_empty(&conn->rac_fmaq));
347         LASSERT (cfs_list_empty(&conn->rac_rdmaq));
348         LASSERT (cfs_list_empty(&conn->rac_replyq));
349
350         rrc = RapkDestroyRi(conn->rac_device->rad_handle,
351                             conn->rac_rihandle);
352         LASSERT (rrc == RAP_SUCCESS);
353
354         if (conn->rac_peer != NULL)
355                 kranal_peer_decref(conn->rac_peer);
356
357         LIBCFS_FREE(conn, sizeof(*conn));
358         cfs_atomic_dec(&kranal_data.kra_nconns);
359 }
360
361 void
362 kranal_terminate_conn_locked (kra_conn_t *conn)
363 {
364         LASSERT (!cfs_in_interrupt());
365         LASSERT (conn->rac_state == RANAL_CONN_CLOSING);
366         LASSERT (!cfs_list_empty(&conn->rac_hashlist));
367         LASSERT (cfs_list_empty(&conn->rac_list));
368
369         /* Remove from conn hash table: no new callbacks */
370         cfs_list_del_init(&conn->rac_hashlist);
371         kranal_conn_decref(conn);
372
373         conn->rac_state = RANAL_CONN_CLOSED;
374
375         /* schedule to clear out all uncompleted comms in context of dev's
376          * scheduler */
377         kranal_schedule_conn(conn);
378 }
379
380 void
381 kranal_close_conn_locked (kra_conn_t *conn, int error)
382 {
383         kra_peer_t        *peer = conn->rac_peer;
384
385         CDEBUG_LIMIT(error == 0 ? D_NET : D_NETERROR,
386                      "closing conn to %s: error %d\n",
387                      libcfs_nid2str(peer->rap_nid), error);
388
389         LASSERT (!cfs_in_interrupt());
390         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED);
391         LASSERT (!cfs_list_empty(&conn->rac_hashlist));
392         LASSERT (!cfs_list_empty(&conn->rac_list));
393
394         cfs_list_del_init(&conn->rac_list);
395
396         if (cfs_list_empty(&peer->rap_conns) &&
397             peer->rap_persistence == 0) {
398                 /* Non-persistent peer with no more conns... */
399                 kranal_unlink_peer_locked(peer);
400         }
401
402         /* Reset RX timeout to ensure we wait for an incoming CLOSE for the
403          * full timeout.  If we get a CLOSE we know the peer has stopped all
404          * RDMA.  Otherwise if we wait for the full timeout we can also be sure
405          * all RDMA has stopped. */
406         conn->rac_last_rx = jiffies;
407         cfs_mb();
408
409         conn->rac_state = RANAL_CONN_CLOSING;
410         kranal_schedule_conn(conn);             /* schedule sending CLOSE */
411
412         kranal_conn_decref(conn);               /* lose peer's ref */
413 }
414
415 void
416 kranal_close_conn (kra_conn_t *conn, int error)
417 {
418         unsigned long    flags;
419
420
421         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
422
423         if (conn->rac_state == RANAL_CONN_ESTABLISHED)
424                 kranal_close_conn_locked(conn, error);
425
426         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
427 }
428
429 int
430 kranal_set_conn_params(kra_conn_t *conn, kra_connreq_t *connreq,
431                        __u32 peer_ip, int peer_port)
432 {
433         kra_device_t  *dev = conn->rac_device;
434         unsigned long  flags;
435         RAP_RETURN     rrc;
436
437         /* CAVEAT EMPTOR: we're really overloading rac_last_tx + rac_keepalive
438          * to do RapkCompleteSync() timekeeping (see kibnal_scheduler). */
439         conn->rac_last_tx = jiffies;
440         conn->rac_keepalive = 0;
441
442         rrc = RapkSetRiParams(conn->rac_rihandle, &connreq->racr_riparams);
443         if (rrc != RAP_SUCCESS) {
444                 CERROR("Error setting riparams from %u.%u.%u.%u/%d: %d\n",
445                        HIPQUAD(peer_ip), peer_port, rrc);
446                 return -ECONNABORTED;
447         }
448
449         /* Schedule conn on rad_new_conns */
450         kranal_conn_addref(conn);
451         spin_lock_irqsave(&dev->rad_lock, flags);
452         cfs_list_add_tail(&conn->rac_schedlist, &dev->rad_new_conns);
453         cfs_waitq_signal(&dev->rad_waitq);
454         spin_unlock_irqrestore(&dev->rad_lock, flags);
455
456         rrc = RapkWaitToConnect(conn->rac_rihandle);
457         if (rrc != RAP_SUCCESS) {
458                 CERROR("Error waiting to connect to %u.%u.%u.%u/%d: %d\n",
459                        HIPQUAD(peer_ip), peer_port, rrc);
460                 return -ECONNABORTED;
461         }
462
463         /* Scheduler doesn't touch conn apart from to deschedule and decref it
464          * after RapkCompleteSync() return success, so conn is all mine */
465
466         conn->rac_peerstamp = connreq->racr_peerstamp;
467         conn->rac_peer_connstamp = connreq->racr_connstamp;
468         conn->rac_keepalive = RANAL_TIMEOUT2KEEPALIVE(connreq->racr_timeout);
469         kranal_update_reaper_timeout(conn->rac_keepalive);
470         return 0;
471 }
472
473 int
474 kranal_passive_conn_handshake (struct socket *sock, lnet_nid_t *src_nidp,
475                                lnet_nid_t *dst_nidp, kra_conn_t **connp)
476 {
477         __u32                peer_ip;
478         unsigned int         peer_port;
479         kra_connreq_t        rx_connreq;
480         kra_connreq_t        tx_connreq;
481         kra_conn_t          *conn;
482         kra_device_t        *dev;
483         int                  rc;
484         int                  i;
485
486         rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
487         if (rc != 0) {
488                 CERROR("Can't get peer's IP: %d\n", rc);
489                 return rc;
490         }
491
492         rc = kranal_recv_connreq(sock, &rx_connreq, 0);
493
494         if (rc < 0) {
495                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
496                        HIPQUAD(peer_ip), peer_port, rc);
497                 return rc;
498         }
499
500         if (rc > 0) {
501                 /* Request from "new" peer: send reply with my MAGIC/VERSION to
502                  * tell her I'm old... */
503                 kranal_pack_connreq(&tx_connreq, NULL, LNET_NID_ANY);
504
505                 rc = libcfs_sock_write(sock, &tx_connreq, sizeof(tx_connreq),
506                                        lnet_acceptor_timeout());
507                 if (rc != 0)
508                         CERROR("Can't tx stub connreq to %u.%u.%u.%u/%d: %d\n",
509                                HIPQUAD(peer_ip), peer_port, rc);
510
511                 return -EPROTO;
512         }
513
514         for (i = 0;;i++) {
515                 if (i == kranal_data.kra_ndevs) {
516                         CERROR("Can't match dev %d from %u.%u.%u.%u/%d\n",
517                                rx_connreq.racr_devid, HIPQUAD(peer_ip), peer_port);
518                         return -ENODEV;
519                 }
520                 dev = &kranal_data.kra_devices[i];
521                 if (dev->rad_id == rx_connreq.racr_devid)
522                         break;
523         }
524
525         rc = kranal_create_conn(&conn, dev);
526         if (rc != 0)
527                 return rc;
528
529         kranal_pack_connreq(&tx_connreq, conn, rx_connreq.racr_srcnid);
530
531         rc = libcfs_sock_write(sock, &tx_connreq, sizeof(tx_connreq),
532                                lnet_acceptor_timeout());
533         if (rc != 0) {
534                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
535                        HIPQUAD(peer_ip), peer_port, rc);
536                 kranal_conn_decref(conn);
537                 return rc;
538         }
539
540         rc = kranal_set_conn_params(conn, &rx_connreq, peer_ip, peer_port);
541         if (rc != 0) {
542                 kranal_conn_decref(conn);
543                 return rc;
544         }
545
546         *connp = conn;
547         *src_nidp = rx_connreq.racr_srcnid;
548         *dst_nidp = rx_connreq.racr_dstnid;
549         return 0;
550 }
551
552 int
553 kranal_active_conn_handshake(kra_peer_t *peer,
554                              lnet_nid_t *dst_nidp, kra_conn_t **connp)
555 {
556         kra_connreq_t       connreq;
557         kra_conn_t         *conn;
558         kra_device_t       *dev;
559         struct socket      *sock;
560         int                 rc;
561         unsigned int        idx;
562
563         /* spread connections over all devices using both peer NIDs to ensure
564          * all nids use all devices */
565         idx = peer->rap_nid + kranal_data.kra_ni->ni_nid;
566         dev = &kranal_data.kra_devices[idx % kranal_data.kra_ndevs];
567
568         rc = kranal_create_conn(&conn, dev);
569         if (rc != 0)
570                 return rc;
571
572         kranal_pack_connreq(&connreq, conn, peer->rap_nid);
573
574         if (the_lnet.ln_testprotocompat != 0) {
575                 /* single-shot proto test */
576                 LNET_LOCK();
577                 if ((the_lnet.ln_testprotocompat & 1) != 0) {
578                         connreq.racr_version++;
579                         the_lnet.ln_testprotocompat &= ~1;
580                 }
581                 if ((the_lnet.ln_testprotocompat & 2) != 0) {
582                         connreq.racr_magic = LNET_PROTO_MAGIC;
583                         the_lnet.ln_testprotocompat &= ~2;
584                 }
585                 LNET_UNLOCK();
586         }
587
588         rc = lnet_connect(&sock, peer->rap_nid,
589                          0, peer->rap_ip, peer->rap_port);
590         if (rc != 0)
591                 goto failed_0;
592
593         /* CAVEAT EMPTOR: the passive side receives with a SHORT rx timeout
594          * immediately after accepting a connection, so we connect and then
595          * send immediately. */
596
597         rc = libcfs_sock_write(sock, &connreq, sizeof(connreq),
598                                lnet_acceptor_timeout());
599         if (rc != 0) {
600                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
601                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
602                 goto failed_2;
603         }
604
605         rc = kranal_recv_connreq(sock, &connreq, 1);
606         if (rc != 0) {
607                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
608                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
609                 goto failed_2;
610         }
611
612         libcfs_sock_release(sock);
613         rc = -EPROTO;
614
615         if (connreq.racr_srcnid != peer->rap_nid) {
616                 CERROR("Unexpected srcnid from %u.%u.%u.%u/%d: "
617                        "received %s expected %s\n",
618                        HIPQUAD(peer->rap_ip), peer->rap_port,
619                        libcfs_nid2str(connreq.racr_srcnid), 
620                        libcfs_nid2str(peer->rap_nid));
621                 goto failed_1;
622         }
623
624         if (connreq.racr_devid != dev->rad_id) {
625                 CERROR("Unexpected device id from %u.%u.%u.%u/%d: "
626                        "received %d expected %d\n",
627                        HIPQUAD(peer->rap_ip), peer->rap_port,
628                        connreq.racr_devid, dev->rad_id);
629                 goto failed_1;
630         }
631
632         rc = kranal_set_conn_params(conn, &connreq,
633                                     peer->rap_ip, peer->rap_port);
634         if (rc != 0)
635                 goto failed_1;
636
637         *connp = conn;
638         *dst_nidp = connreq.racr_dstnid;
639         return 0;
640
641  failed_2:
642         libcfs_sock_release(sock);
643  failed_1:
644         lnet_connect_console_error(rc, peer->rap_nid,
645                                   peer->rap_ip, peer->rap_port);
646  failed_0:
647         kranal_conn_decref(conn);
648         return rc;
649 }
650
651 int
652 kranal_conn_handshake (struct socket *sock, kra_peer_t *peer)
653 {
654         kra_peer_t        *peer2;
655         kra_tx_t          *tx;
656         lnet_nid_t         peer_nid;
657         lnet_nid_t         dst_nid;
658         unsigned long      flags;
659         kra_conn_t        *conn;
660         int                rc;
661         int                nstale;
662         int                new_peer = 0;
663
664         if (sock == NULL) {
665                 /* active: connd wants to connect to 'peer' */
666                 LASSERT (peer != NULL);
667                 LASSERT (peer->rap_connecting);
668
669                 rc = kranal_active_conn_handshake(peer, &dst_nid, &conn);
670                 if (rc != 0)
671                         return rc;
672
673                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
674
675                 if (!kranal_peer_active(peer)) {
676                         /* raced with peer getting unlinked */
677                         write_unlock_irqrestore(&kranal_data. \
678                                                     kra_global_lock,
679                                                     flags);
680                         kranal_conn_decref(conn);
681                         return -ESTALE;
682                 }
683
684                 peer_nid = peer->rap_nid;
685         } else {
686                 /* passive: listener accepted 'sock' */
687                 LASSERT (peer == NULL);
688
689                 rc = kranal_passive_conn_handshake(sock, &peer_nid,
690                                                    &dst_nid, &conn);
691                 if (rc != 0)
692                         return rc;
693
694                 /* assume this is a new peer */
695                 rc = kranal_create_peer(&peer, peer_nid);
696                 if (rc != 0) {
697                         CERROR("Can't create conn for %s\n", 
698                                libcfs_nid2str(peer_nid));
699                         kranal_conn_decref(conn);
700                         return -ENOMEM;
701                 }
702
703                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
704
705                 peer2 = kranal_find_peer_locked(peer_nid);
706                 if (peer2 == NULL) {
707                         new_peer = 1;
708                 } else {
709                         /* peer_nid already in the peer table */
710                         kranal_peer_decref(peer);
711                         peer = peer2;
712                 }
713         }
714
715         LASSERT ((!new_peer) != (!kranal_peer_active(peer)));
716
717         /* Refuse connection if peer thinks we are a different NID.  We check
718          * this while holding the global lock, to synch with connection
719          * destruction on NID change. */
720         if (kranal_data.kra_ni->ni_nid != dst_nid) {
721                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
722                                             flags);
723
724                 CERROR("Stale/bad connection with %s: dst_nid %s, expected %s\n",
725                        libcfs_nid2str(peer_nid), libcfs_nid2str(dst_nid), 
726                        libcfs_nid2str(kranal_data.kra_ni->ni_nid));
727                 rc = -ESTALE;
728                 goto failed;
729         }
730
731         /* Refuse to duplicate an existing connection (both sides might try to
732          * connect at once).  NB we return success!  We _are_ connected so we
733          * _don't_ have any blocked txs to complete with failure. */
734         rc = kranal_conn_isdup_locked(peer, conn);
735         if (rc != 0) {
736                 LASSERT (!cfs_list_empty(&peer->rap_conns));
737                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
738                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
739                                             flags);
740                 CWARN("Not creating duplicate connection to %s: %d\n",
741                       libcfs_nid2str(peer_nid), rc);
742                 rc = 0;
743                 goto failed;
744         }
745
746         if (new_peer) {
747                 /* peer table takes my ref on the new peer */
748                 cfs_list_add_tail(&peer->rap_list,
749                                   kranal_nid2peerlist(peer_nid));
750         }
751
752         /* initialise timestamps before reaper looks at them */
753         conn->rac_last_tx = conn->rac_last_rx = jiffies;
754
755         kranal_peer_addref(peer);               /* +1 ref for conn */
756         conn->rac_peer = peer;
757         cfs_list_add_tail(&conn->rac_list, &peer->rap_conns);
758
759         kranal_conn_addref(conn);               /* +1 ref for conn table */
760         cfs_list_add_tail(&conn->rac_hashlist,
761                           kranal_cqid2connlist(conn->rac_cqid));
762
763         /* Schedule all packets blocking for a connection */
764         while (!cfs_list_empty(&peer->rap_tx_queue)) {
765                 tx = cfs_list_entry(peer->rap_tx_queue.next,
766                                     kra_tx_t, tx_list);
767
768                 cfs_list_del(&tx->tx_list);
769                 kranal_post_fma(conn, tx);
770         }
771
772         nstale = kranal_close_stale_conns_locked(peer, conn);
773
774         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
775
776         /* CAVEAT EMPTOR: passive peer can disappear NOW */
777
778         if (nstale != 0)
779                 CWARN("Closed %d stale conns to %s\n", nstale, 
780                       libcfs_nid2str(peer_nid));
781
782         CWARN("New connection to %s on devid[%d] = %d\n",
783                libcfs_nid2str(peer_nid), 
784                conn->rac_device->rad_idx, conn->rac_device->rad_id);
785
786         /* Ensure conn gets checked.  Transmits may have been queued and an
787          * FMA event may have happened before it got in the cq hash table */
788         kranal_schedule_conn(conn);
789         return 0;
790
791  failed:
792         if (new_peer)
793                 kranal_peer_decref(peer);
794         kranal_conn_decref(conn);
795         return rc;
796 }
797
798 void
799 kranal_connect (kra_peer_t *peer)
800 {
801         kra_tx_t          *tx;
802         unsigned long      flags;
803         cfs_list_t         zombies;
804         int                rc;
805
806         LASSERT (peer->rap_connecting);
807
808         CDEBUG(D_NET, "About to handshake %s\n", 
809                libcfs_nid2str(peer->rap_nid));
810
811         rc = kranal_conn_handshake(NULL, peer);
812
813         CDEBUG(D_NET, "Done handshake %s:%d \n", 
814                libcfs_nid2str(peer->rap_nid), rc);
815
816         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
817
818         LASSERT (peer->rap_connecting);
819         peer->rap_connecting = 0;
820
821         if (rc == 0) {
822                 /* kranal_conn_handshake() queues blocked txs immediately on
823                  * success to avoid messages jumping the queue */
824                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
825
826                 peer->rap_reconnect_interval = 0; /* OK to reconnect at any time */
827
828                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
829                                             flags);
830                 return;
831         }
832
833         peer->rap_reconnect_interval *= 2;
834         peer->rap_reconnect_interval =
835                 MAX(peer->rap_reconnect_interval,
836                     *kranal_tunables.kra_min_reconnect_interval);
837         peer->rap_reconnect_interval =
838                 MIN(peer->rap_reconnect_interval,
839                     *kranal_tunables.kra_max_reconnect_interval);
840
841         peer->rap_reconnect_time = jiffies + peer->rap_reconnect_interval *
842                 CFS_HZ;
843
844         /* Grab all blocked packets while we have the global lock */
845         cfs_list_add(&zombies, &peer->rap_tx_queue);
846         cfs_list_del_init(&peer->rap_tx_queue);
847
848         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
849
850         if (cfs_list_empty(&zombies))
851                 return;
852
853         CNETERR("Dropping packets for %s: connection failed\n",
854                 libcfs_nid2str(peer->rap_nid));
855
856         do {
857                 tx = cfs_list_entry(zombies.next, kra_tx_t, tx_list);
858
859                 cfs_list_del(&tx->tx_list);
860                 kranal_tx_done(tx, -EHOSTUNREACH);
861
862         } while (!cfs_list_empty(&zombies));
863 }
864
865 void
866 kranal_free_acceptsock (kra_acceptsock_t *ras)
867 {
868         libcfs_sock_release(ras->ras_sock);
869         LIBCFS_FREE(ras, sizeof(*ras));
870 }
871
872 int
873 kranal_accept (lnet_ni_t *ni, struct socket *sock)
874 {
875         kra_acceptsock_t  *ras;
876         int                rc;
877         __u32              peer_ip;
878         int                peer_port;
879         unsigned long      flags;
880
881         rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
882         LASSERT (rc == 0);                      /* we succeeded before */
883
884         LIBCFS_ALLOC(ras, sizeof(*ras));
885         if (ras == NULL) {
886                 CERROR("ENOMEM allocating connection request from "
887                        "%u.%u.%u.%u\n", HIPQUAD(peer_ip));
888                 return -ENOMEM;
889         }
890
891         ras->ras_sock = sock;
892
893         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
894
895         cfs_list_add_tail(&ras->ras_list, &kranal_data.kra_connd_acceptq);
896         cfs_waitq_signal(&kranal_data.kra_connd_waitq);
897
898         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
899         return 0;
900 }
901
902 int
903 kranal_create_peer (kra_peer_t **peerp, lnet_nid_t nid)
904 {
905         kra_peer_t    *peer;
906         unsigned long  flags;
907
908         LASSERT (nid != LNET_NID_ANY);
909
910         LIBCFS_ALLOC(peer, sizeof(*peer));
911         if (peer == NULL)
912                 return -ENOMEM;
913
914         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
915
916         peer->rap_nid = nid;
917         cfs_atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */
918
919         CFS_INIT_LIST_HEAD(&peer->rap_list);
920         CFS_INIT_LIST_HEAD(&peer->rap_connd_list);
921         CFS_INIT_LIST_HEAD(&peer->rap_conns);
922         CFS_INIT_LIST_HEAD(&peer->rap_tx_queue);
923
924         peer->rap_reconnect_interval = 0;       /* OK to connect at any time */
925
926         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
927
928         if (kranal_data.kra_nonewpeers) {
929                 /* shutdown has started already */
930                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
931                                             flags);
932
933                 LIBCFS_FREE(peer, sizeof(*peer));
934                 CERROR("Can't create peer: network shutdown\n");
935                 return -ESHUTDOWN;
936         }
937
938         cfs_atomic_inc(&kranal_data.kra_npeers);
939
940         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
941
942         *peerp = peer;
943         return 0;
944 }
945
946 void
947 kranal_destroy_peer (kra_peer_t *peer)
948 {
949         CDEBUG(D_NET, "peer %s %p deleted\n", 
950                libcfs_nid2str(peer->rap_nid), peer);
951
952         LASSERT (cfs_atomic_read(&peer->rap_refcount) == 0);
953         LASSERT (peer->rap_persistence == 0);
954         LASSERT (!kranal_peer_active(peer));
955         LASSERT (!peer->rap_connecting);
956         LASSERT (cfs_list_empty(&peer->rap_conns));
957         LASSERT (cfs_list_empty(&peer->rap_tx_queue));
958         LASSERT (cfs_list_empty(&peer->rap_connd_list));
959
960         LIBCFS_FREE(peer, sizeof(*peer));
961
962         /* NB a peer's connections keep a reference on their peer until
963          * they are destroyed, so we can be assured that _all_ state to do
964          * with this peer has been cleaned up when its refcount drops to
965          * zero. */
966         cfs_atomic_dec(&kranal_data.kra_npeers);
967 }
968
969 kra_peer_t *
970 kranal_find_peer_locked (lnet_nid_t nid)
971 {
972         cfs_list_t       *peer_list = kranal_nid2peerlist(nid);
973         cfs_list_t       *tmp;
974         kra_peer_t       *peer;
975
976         cfs_list_for_each (tmp, peer_list) {
977
978                 peer = cfs_list_entry(tmp, kra_peer_t, rap_list);
979
980                 LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
981                          !cfs_list_empty(&peer->rap_conns));  /* active conn */
982
983                 if (peer->rap_nid != nid)
984                         continue;
985
986                 CDEBUG(D_NET, "got peer [%p] -> %s (%d)\n",
987                        peer, libcfs_nid2str(nid), 
988                        cfs_atomic_read(&peer->rap_refcount));
989                 return peer;
990         }
991         return NULL;
992 }
993
994 kra_peer_t *
995 kranal_find_peer (lnet_nid_t nid)
996 {
997         kra_peer_t     *peer;
998
999         read_lock(&kranal_data.kra_global_lock);
1000         peer = kranal_find_peer_locked(nid);
1001         if (peer != NULL)                       /* +1 ref for caller? */
1002                 kranal_peer_addref(peer);
1003         read_unlock(&kranal_data.kra_global_lock);
1004
1005         return peer;
1006 }
1007
1008 void
1009 kranal_unlink_peer_locked (kra_peer_t *peer)
1010 {
1011         LASSERT (peer->rap_persistence == 0);
1012         LASSERT (cfs_list_empty(&peer->rap_conns));
1013
1014         LASSERT (kranal_peer_active(peer));
1015         cfs_list_del_init(&peer->rap_list);
1016
1017         /* lose peerlist's ref */
1018         kranal_peer_decref(peer);
1019 }
1020
1021 int
1022 kranal_get_peer_info (int index, lnet_nid_t *nidp, __u32 *ipp, int *portp,
1023                       int *persistencep)
1024 {
1025         kra_peer_t        *peer;
1026         cfs_list_t        *ptmp;
1027         int                i;
1028
1029         read_lock(&kranal_data.kra_global_lock);
1030
1031         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1032
1033                 cfs_list_for_each(ptmp, &kranal_data.kra_peers[i]) {
1034
1035                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1036                         LASSERT (peer->rap_persistence > 0 ||
1037                                  !cfs_list_empty(&peer->rap_conns));
1038
1039                         if (index-- > 0)
1040                                 continue;
1041
1042                         *nidp = peer->rap_nid;
1043                         *ipp = peer->rap_ip;
1044                         *portp = peer->rap_port;
1045                         *persistencep = peer->rap_persistence;
1046
1047                         read_unlock(&kranal_data.kra_global_lock);
1048                         return 0;
1049                 }
1050         }
1051
1052         read_unlock(&kranal_data.kra_global_lock);
1053         return -ENOENT;
1054 }
1055
1056 int
1057 kranal_add_persistent_peer (lnet_nid_t nid, __u32 ip, int port)
1058 {
1059         unsigned long      flags;
1060         kra_peer_t        *peer;
1061         kra_peer_t        *peer2;
1062         int                rc;
1063
1064         if (nid == LNET_NID_ANY)
1065                 return -EINVAL;
1066
1067         rc = kranal_create_peer(&peer, nid);
1068         if (rc != 0)
1069                 return rc;
1070
1071         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1072
1073         peer2 = kranal_find_peer_locked(nid);
1074         if (peer2 != NULL) {
1075                 kranal_peer_decref(peer);
1076                 peer = peer2;
1077         } else {
1078                 /* peer table takes existing ref on peer */
1079                 cfs_list_add_tail(&peer->rap_list,
1080                               kranal_nid2peerlist(nid));
1081         }
1082
1083         peer->rap_ip = ip;
1084         peer->rap_port = port;
1085         peer->rap_persistence++;
1086
1087         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1088         return 0;
1089 }
1090
1091 void
1092 kranal_del_peer_locked (kra_peer_t *peer)
1093 {
1094         cfs_list_t       *ctmp;
1095         cfs_list_t       *cnxt;
1096         kra_conn_t       *conn;
1097
1098         peer->rap_persistence = 0;
1099
1100         if (cfs_list_empty(&peer->rap_conns)) {
1101                 kranal_unlink_peer_locked(peer);
1102         } else {
1103                 cfs_list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
1104                         conn = cfs_list_entry(ctmp, kra_conn_t, rac_list);
1105
1106                         kranal_close_conn_locked(conn, 0);
1107                 }
1108                 /* peer unlinks itself when last conn is closed */
1109         }
1110 }
1111
1112 int
1113 kranal_del_peer (lnet_nid_t nid)
1114 {
1115         unsigned long      flags;
1116         cfs_list_t        *ptmp;
1117         cfs_list_t        *pnxt;
1118         kra_peer_t        *peer;
1119         int                lo;
1120         int                hi;
1121         int                i;
1122         int                rc = -ENOENT;
1123
1124         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1125
1126         if (nid != LNET_NID_ANY)
1127                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1128         else {
1129                 lo = 0;
1130                 hi = kranal_data.kra_peer_hash_size - 1;
1131         }
1132
1133         for (i = lo; i <= hi; i++) {
1134                 cfs_list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1135                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1136                         LASSERT (peer->rap_persistence > 0 ||
1137                                  !cfs_list_empty(&peer->rap_conns));
1138
1139                         if (!(nid == LNET_NID_ANY || peer->rap_nid == nid))
1140                                 continue;
1141
1142                         kranal_del_peer_locked(peer);
1143                         rc = 0;         /* matched something */
1144                 }
1145         }
1146
1147         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1148
1149         return rc;
1150 }
1151
1152 kra_conn_t *
1153 kranal_get_conn_by_idx (int index)
1154 {
1155         kra_peer_t        *peer;
1156         cfs_list_t        *ptmp;
1157         kra_conn_t        *conn;
1158         cfs_list_t        *ctmp;
1159         int                i;
1160
1161         read_lock(&kranal_data.kra_global_lock);
1162
1163         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1164                 cfs_list_for_each (ptmp, &kranal_data.kra_peers[i]) {
1165
1166                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1167                         LASSERT (peer->rap_persistence > 0 ||
1168                                  !cfs_list_empty(&peer->rap_conns));
1169
1170                         cfs_list_for_each (ctmp, &peer->rap_conns) {
1171                                 if (index-- > 0)
1172                                         continue;
1173
1174                                 conn = cfs_list_entry(ctmp, kra_conn_t,
1175                                                       rac_list);
1176                                 CDEBUG(D_NET, "++conn[%p] -> %s (%d)\n", conn,
1177                                        libcfs_nid2str(conn->rac_peer->rap_nid),
1178                                        cfs_atomic_read(&conn->rac_refcount));
1179                                 cfs_atomic_inc(&conn->rac_refcount);
1180                                 read_unlock(&kranal_data.kra_global_lock);
1181                                 return conn;
1182                         }
1183                 }
1184         }
1185
1186         read_unlock(&kranal_data.kra_global_lock);
1187         return NULL;
1188 }
1189
1190 int
1191 kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
1192 {
1193         kra_conn_t         *conn;
1194         cfs_list_t         *ctmp;
1195         cfs_list_t         *cnxt;
1196         int                 count = 0;
1197
1198         cfs_list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1199                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_list);
1200
1201                 count++;
1202                 kranal_close_conn_locked(conn, why);
1203         }
1204
1205         return count;
1206 }
1207
1208 int
1209 kranal_close_matching_conns (lnet_nid_t nid)
1210 {
1211         unsigned long       flags;
1212         kra_peer_t         *peer;
1213         cfs_list_t         *ptmp;
1214         cfs_list_t         *pnxt;
1215         int                 lo;
1216         int                 hi;
1217         int                 i;
1218         int                 count = 0;
1219
1220         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1221
1222         if (nid != LNET_NID_ANY)
1223                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1224         else {
1225                 lo = 0;
1226                 hi = kranal_data.kra_peer_hash_size - 1;
1227         }
1228
1229         for (i = lo; i <= hi; i++) {
1230                 cfs_list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1231
1232                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1233                         LASSERT (peer->rap_persistence > 0 ||
1234                                  !cfs_list_empty(&peer->rap_conns));
1235
1236                         if (!(nid == LNET_NID_ANY || nid == peer->rap_nid))
1237                                 continue;
1238
1239                         count += kranal_close_peer_conns_locked(peer, 0);
1240                 }
1241         }
1242
1243         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1244
1245         /* wildcards always succeed */
1246         if (nid == LNET_NID_ANY)
1247                 return 0;
1248
1249         return (count == 0) ? -ENOENT : 0;
1250 }
1251
1252 int
1253 kranal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1254 {
1255         struct libcfs_ioctl_data *data = arg;
1256         int                       rc = -EINVAL;
1257
1258         LASSERT (ni == kranal_data.kra_ni);
1259
1260         switch(cmd) {
1261         case IOC_LIBCFS_GET_PEER: {
1262                 lnet_nid_t   nid = 0;
1263                 __u32       ip = 0;
1264                 int         port = 0;
1265                 int         share_count = 0;
1266
1267                 rc = kranal_get_peer_info(data->ioc_count,
1268                                           &nid, &ip, &port, &share_count);
1269                 data->ioc_nid    = nid;
1270                 data->ioc_count  = share_count;
1271                 data->ioc_u32[0] = ip;
1272                 data->ioc_u32[1] = port;
1273                 break;
1274         }
1275         case IOC_LIBCFS_ADD_PEER: {
1276                 rc = kranal_add_persistent_peer(data->ioc_nid,
1277                                                 data->ioc_u32[0], /* IP */
1278                                                 data->ioc_u32[1]); /* port */
1279                 break;
1280         }
1281         case IOC_LIBCFS_DEL_PEER: {
1282                 rc = kranal_del_peer(data->ioc_nid);
1283                 break;
1284         }
1285         case IOC_LIBCFS_GET_CONN: {
1286                 kra_conn_t *conn = kranal_get_conn_by_idx(data->ioc_count);
1287
1288                 if (conn == NULL)
1289                         rc = -ENOENT;
1290                 else {
1291                         rc = 0;
1292                         data->ioc_nid    = conn->rac_peer->rap_nid;
1293                         data->ioc_u32[0] = conn->rac_device->rad_id;
1294                         kranal_conn_decref(conn);
1295                 }
1296                 break;
1297         }
1298         case IOC_LIBCFS_CLOSE_CONNECTION: {
1299                 rc = kranal_close_matching_conns(data->ioc_nid);
1300                 break;
1301         }
1302         case IOC_LIBCFS_REGISTER_MYNID: {
1303                 /* Ignore if this is a noop */
1304                 if (data->ioc_nid == ni->ni_nid) {
1305                         rc = 0;
1306                 } else {
1307                         CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID: %s(%s)\n",
1308                                libcfs_nid2str(data->ioc_nid),
1309                                libcfs_nid2str(ni->ni_nid));
1310                         rc = -EINVAL;
1311                 }
1312                 break;
1313         }
1314         }
1315
1316         return rc;
1317 }
1318
1319 void
1320 kranal_free_txdescs(cfs_list_t *freelist)
1321 {
1322         kra_tx_t    *tx;
1323
1324         while (!cfs_list_empty(freelist)) {
1325                 tx = cfs_list_entry(freelist->next, kra_tx_t, tx_list);
1326
1327                 cfs_list_del(&tx->tx_list);
1328                 LIBCFS_FREE(tx->tx_phys, LNET_MAX_IOV * sizeof(*tx->tx_phys));
1329                 LIBCFS_FREE(tx, sizeof(*tx));
1330         }
1331 }
1332
1333 int
1334 kranal_alloc_txdescs(cfs_list_t *freelist, int n)
1335 {
1336         int            i;
1337         kra_tx_t      *tx;
1338
1339         LASSERT (freelist == &kranal_data.kra_idle_txs);
1340         LASSERT (cfs_list_empty(freelist));
1341
1342         for (i = 0; i < n; i++) {
1343
1344                 LIBCFS_ALLOC(tx, sizeof(*tx));
1345                 if (tx == NULL) {
1346                         CERROR("Can't allocate tx[%d]\n", i);
1347                         kranal_free_txdescs(freelist);
1348                         return -ENOMEM;
1349                 }
1350
1351                 LIBCFS_ALLOC(tx->tx_phys,
1352                              LNET_MAX_IOV * sizeof(*tx->tx_phys));
1353                 if (tx->tx_phys == NULL) {
1354                         CERROR("Can't allocate tx[%d]->tx_phys\n", i);
1355
1356                         LIBCFS_FREE(tx, sizeof(*tx));
1357                         kranal_free_txdescs(freelist);
1358                         return -ENOMEM;
1359                 }
1360
1361                 tx->tx_buftype = RANAL_BUF_NONE;
1362                 tx->tx_msg.ram_type = RANAL_MSG_NONE;
1363
1364                 cfs_list_add(&tx->tx_list, freelist);
1365         }
1366
1367         return 0;
1368 }
1369
1370 int
1371 kranal_device_init(int id, kra_device_t *dev)
1372 {
1373         int               total_ntx = *kranal_tunables.kra_ntx;
1374         RAP_RETURN        rrc;
1375
1376         dev->rad_id = id;
1377         rrc = RapkGetDeviceByIndex(id, kranal_device_callback,
1378                                    &dev->rad_handle);
1379         if (rrc != RAP_SUCCESS) {
1380                 CERROR("Can't get Rapidarray Device %d: %d\n", id, rrc);
1381                 goto failed_0;
1382         }
1383
1384         rrc = RapkReserveRdma(dev->rad_handle, total_ntx);
1385         if (rrc != RAP_SUCCESS) {
1386                 CERROR("Can't reserve %d RDMA descriptors"
1387                        " for device %d: %d\n", total_ntx, id, rrc);
1388                 goto failed_1;
1389         }
1390
1391         rrc = RapkCreateCQ(dev->rad_handle, total_ntx, RAP_CQTYPE_SEND,
1392                            &dev->rad_rdma_cqh);
1393         if (rrc != RAP_SUCCESS) {
1394                 CERROR("Can't create rdma cq size %d for device %d: %d\n",
1395                        total_ntx, id, rrc);
1396                 goto failed_1;
1397         }
1398
1399         rrc = RapkCreateCQ(dev->rad_handle, 
1400                            *kranal_tunables.kra_fma_cq_size, 
1401                            RAP_CQTYPE_RECV, &dev->rad_fma_cqh);
1402         if (rrc != RAP_SUCCESS) {
1403                 CERROR("Can't create fma cq size %d for device %d: %d\n", 
1404                        *kranal_tunables.kra_fma_cq_size, id, rrc);
1405                 goto failed_2;
1406         }
1407
1408         return 0;
1409
1410  failed_2:
1411         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1412  failed_1:
1413         RapkReleaseDevice(dev->rad_handle);
1414  failed_0:
1415         return -ENODEV;
1416 }
1417
1418 void
1419 kranal_device_fini(kra_device_t *dev)
1420 {
1421         LASSERT (cfs_list_empty(&dev->rad_ready_conns));
1422         LASSERT (cfs_list_empty(&dev->rad_new_conns));
1423         LASSERT (dev->rad_nphysmap == 0);
1424         LASSERT (dev->rad_nppphysmap == 0);
1425         LASSERT (dev->rad_nvirtmap == 0);
1426         LASSERT (dev->rad_nobvirtmap == 0);
1427
1428         LASSERT(dev->rad_scheduler == NULL);
1429         RapkDestroyCQ(dev->rad_handle, dev->rad_fma_cqh);
1430         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1431         RapkReleaseDevice(dev->rad_handle);
1432 }
1433
1434 void
1435 kranal_shutdown (lnet_ni_t *ni)
1436 {
1437         int           i;
1438         unsigned long flags;
1439
1440         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1441                cfs_atomic_read(&libcfs_kmemory));
1442
1443         LASSERT (ni == kranal_data.kra_ni);
1444         LASSERT (ni->ni_data == &kranal_data);
1445
1446         switch (kranal_data.kra_init) {
1447         default:
1448                 CERROR("Unexpected state %d\n", kranal_data.kra_init);
1449                 LBUG();
1450
1451         case RANAL_INIT_ALL:
1452                 /* Prevent new peers from being created */
1453                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1454                 kranal_data.kra_nonewpeers = 1;
1455                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1456                                             flags);
1457
1458                 /* Remove all existing peers from the peer table */
1459                 kranal_del_peer(LNET_NID_ANY);
1460
1461                 /* Wait for pending conn reqs to be handled */
1462                 i = 2;
1463                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1464                 while (!cfs_list_empty(&kranal_data.kra_connd_acceptq)) {
1465                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1466                                                    flags);
1467                         i++;
1468                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1469                                "waiting for conn reqs to clean up\n");
1470                         cfs_pause(cfs_time_seconds(1));
1471
1472                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1473                                               flags);
1474                 }
1475                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1476
1477                 /* Wait for all peers to be freed */
1478                 i = 2;
1479                 while (cfs_atomic_read(&kranal_data.kra_npeers) != 0) {
1480                         i++;
1481                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1482                                "waiting for %d peers to close down\n",
1483                                cfs_atomic_read(&kranal_data.kra_npeers));
1484                         cfs_pause(cfs_time_seconds(1));
1485                 }
1486                 /* fall through */
1487
1488         case RANAL_INIT_DATA:
1489                 break;
1490         }
1491
1492         /* Peer state all cleaned up BEFORE setting shutdown, so threads don't
1493          * have to worry about shutdown races.  NB connections may be created
1494          * while there are still active connds, but these will be temporary
1495          * since peer creation always fails after the listener has started to
1496          * shut down. */
1497         LASSERT (cfs_atomic_read(&kranal_data.kra_npeers) == 0);
1498         
1499         /* Flag threads to terminate */
1500         kranal_data.kra_shutdown = 1;
1501
1502         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1503                 kra_device_t *dev = &kranal_data.kra_devices[i];
1504
1505                 spin_lock_irqsave(&dev->rad_lock, flags);
1506                 cfs_waitq_signal(&dev->rad_waitq);
1507                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1508         }
1509
1510         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1511         cfs_waitq_broadcast(&kranal_data.kra_reaper_waitq);
1512         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1513
1514         LASSERT (cfs_list_empty(&kranal_data.kra_connd_peers));
1515         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1516         cfs_waitq_broadcast(&kranal_data.kra_connd_waitq);
1517         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1518
1519         /* Wait for threads to exit */
1520         i = 2;
1521         while (cfs_atomic_read(&kranal_data.kra_nthreads) != 0) {
1522                 i++;
1523                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1524                        "Waiting for %d threads to terminate\n",
1525                        cfs_atomic_read(&kranal_data.kra_nthreads));
1526                 cfs_pause(cfs_time_seconds(1));
1527         }
1528
1529         LASSERT (cfs_atomic_read(&kranal_data.kra_npeers) == 0);
1530         if (kranal_data.kra_peers != NULL) {
1531                 for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1532                         LASSERT (cfs_list_empty(&kranal_data.kra_peers[i]));
1533
1534                 LIBCFS_FREE(kranal_data.kra_peers,
1535                             sizeof (cfs_list_t) *
1536                             kranal_data.kra_peer_hash_size);
1537         }
1538
1539         LASSERT (cfs_atomic_read(&kranal_data.kra_nconns) == 0);
1540         if (kranal_data.kra_conns != NULL) {
1541                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1542                         LASSERT (cfs_list_empty(&kranal_data.kra_conns[i]));
1543
1544                 LIBCFS_FREE(kranal_data.kra_conns,
1545                             sizeof (cfs_list_t) *
1546                             kranal_data.kra_conn_hash_size);
1547         }
1548
1549         for (i = 0; i < kranal_data.kra_ndevs; i++)
1550                 kranal_device_fini(&kranal_data.kra_devices[i]);
1551
1552         kranal_free_txdescs(&kranal_data.kra_idle_txs);
1553
1554         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1555                cfs_atomic_read(&libcfs_kmemory));
1556
1557         kranal_data.kra_init = RANAL_INIT_NOTHING;
1558         PORTAL_MODULE_UNUSE;
1559 }
1560
1561 int
1562 kranal_startup (lnet_ni_t *ni)
1563 {
1564         struct timeval    tv;
1565         int               pkmem = cfs_atomic_read(&libcfs_kmemory);
1566         int               rc;
1567         int               i;
1568         kra_device_t     *dev;
1569
1570         LASSERT (ni->ni_lnd == &the_kralnd);
1571
1572         /* Only 1 instance supported */
1573         if (kranal_data.kra_init != RANAL_INIT_NOTHING) {
1574                 CERROR ("Only 1 instance supported\n");
1575                 return -EPERM;
1576         }
1577
1578         if (lnet_set_ip_niaddr(ni) != 0) {
1579                 CERROR ("Can't determine my NID\n");
1580                 return -EPERM;
1581         }
1582
1583         if (*kranal_tunables.kra_credits > *kranal_tunables.kra_ntx) {
1584                 CERROR ("Can't set credits(%d) > ntx(%d)\n",
1585                         *kranal_tunables.kra_credits,
1586                         *kranal_tunables.kra_ntx);
1587                 return -EINVAL;
1588         }
1589         
1590         memset(&kranal_data, 0, sizeof(kranal_data)); /* zero pointers, flags etc */
1591
1592         ni->ni_maxtxcredits = *kranal_tunables.kra_credits;
1593         ni->ni_peertxcredits = *kranal_tunables.kra_peercredits;
1594
1595         ni->ni_data = &kranal_data;
1596         kranal_data.kra_ni = ni;
1597
1598         /* CAVEAT EMPTOR: Every 'Fma' message includes the sender's NID and
1599          * a unique (for all time) connstamp so we can uniquely identify
1600          * the sender.  The connstamp is an incrementing counter
1601          * initialised with seconds + microseconds at startup time.  So we
1602          * rely on NOT creating connections more frequently on average than
1603          * 1MHz to ensure we don't use old connstamps when we reboot. */
1604         cfs_gettimeofday(&tv);
1605         kranal_data.kra_connstamp =
1606         kranal_data.kra_peerstamp = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1607
1608         rwlock_init(&kranal_data.kra_global_lock);
1609
1610         for (i = 0; i < RANAL_MAXDEVS; i++ ) {
1611                 kra_device_t  *dev = &kranal_data.kra_devices[i];
1612
1613                 dev->rad_idx = i;
1614                 CFS_INIT_LIST_HEAD(&dev->rad_ready_conns);
1615                 CFS_INIT_LIST_HEAD(&dev->rad_new_conns);
1616                 cfs_waitq_init(&dev->rad_waitq);
1617                 spin_lock_init(&dev->rad_lock);
1618         }
1619
1620         kranal_data.kra_new_min_timeout = CFS_MAX_SCHEDULE_TIMEOUT;
1621         cfs_waitq_init(&kranal_data.kra_reaper_waitq);
1622         spin_lock_init(&kranal_data.kra_reaper_lock);
1623
1624         CFS_INIT_LIST_HEAD(&kranal_data.kra_connd_acceptq);
1625         CFS_INIT_LIST_HEAD(&kranal_data.kra_connd_peers);
1626         cfs_waitq_init(&kranal_data.kra_connd_waitq);
1627         spin_lock_init(&kranal_data.kra_connd_lock);
1628
1629         CFS_INIT_LIST_HEAD(&kranal_data.kra_idle_txs);
1630         spin_lock_init(&kranal_data.kra_tx_lock);
1631
1632         /* OK to call kranal_api_shutdown() to cleanup now */
1633         kranal_data.kra_init = RANAL_INIT_DATA;
1634         PORTAL_MODULE_USE;
1635
1636         kranal_data.kra_peer_hash_size = RANAL_PEER_HASH_SIZE;
1637         LIBCFS_ALLOC(kranal_data.kra_peers,
1638                      sizeof(cfs_list_t) *
1639                             kranal_data.kra_peer_hash_size);
1640         if (kranal_data.kra_peers == NULL)
1641                 goto failed;
1642
1643         for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1644                 CFS_INIT_LIST_HEAD(&kranal_data.kra_peers[i]);
1645
1646         kranal_data.kra_conn_hash_size = RANAL_PEER_HASH_SIZE;
1647         LIBCFS_ALLOC(kranal_data.kra_conns,
1648                      sizeof(cfs_list_t) *
1649                             kranal_data.kra_conn_hash_size);
1650         if (kranal_data.kra_conns == NULL)
1651                 goto failed;
1652
1653         for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1654                 CFS_INIT_LIST_HEAD(&kranal_data.kra_conns[i]);
1655
1656         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_txs, 
1657                                   *kranal_tunables.kra_ntx);
1658         if (rc != 0)
1659                 goto failed;
1660
1661         rc = kranal_thread_start(kranal_reaper, NULL);
1662         if (rc != 0) {
1663                 CERROR("Can't spawn ranal reaper: %d\n", rc);
1664                 goto failed;
1665         }
1666
1667         for (i = 0; i < *kranal_tunables.kra_n_connd; i++) {
1668                 rc = kranal_thread_start(kranal_connd, (void *)(unsigned long)i);
1669                 if (rc != 0) {
1670                         CERROR("Can't spawn ranal connd[%d]: %d\n",
1671                                i, rc);
1672                         goto failed;
1673                 }
1674         }
1675
1676         LASSERT (kranal_data.kra_ndevs == 0);
1677
1678         /* Use all available RapidArray devices */
1679         for (i = 0; i < RANAL_MAXDEVS; i++) {
1680                 dev = &kranal_data.kra_devices[kranal_data.kra_ndevs];
1681
1682                 rc = kranal_device_init(kranal_devids[i], dev);
1683                 if (rc == 0)
1684                         kranal_data.kra_ndevs++;
1685         }
1686
1687         if (kranal_data.kra_ndevs == 0) {
1688                 CERROR("Can't initialise any RapidArray devices\n");
1689                 goto failed;
1690         }
1691         
1692         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1693                 dev = &kranal_data.kra_devices[i];
1694                 rc = kranal_thread_start(kranal_scheduler, dev);
1695                 if (rc != 0) {
1696                         CERROR("Can't spawn ranal scheduler[%d]: %d\n",
1697                                i, rc);
1698                         goto failed;
1699                 }
1700         }
1701
1702         /* flag everything initialised */
1703         kranal_data.kra_init = RANAL_INIT_ALL;
1704         /*****************************************************/
1705
1706         CDEBUG(D_MALLOC, "initial kmem %d\n", pkmem);
1707         return 0;
1708
1709  failed:
1710         kranal_shutdown(ni);
1711         return -ENETDOWN;
1712 }
1713
1714 void __exit
1715 kranal_module_fini (void)
1716 {
1717         lnet_unregister_lnd(&the_kralnd);
1718         kranal_tunables_fini();
1719 }
1720
1721 int __init
1722 kranal_module_init (void)
1723 {
1724         int    rc;
1725
1726         rc = kranal_tunables_init();
1727         if (rc != 0)
1728                 return rc;
1729
1730         lnet_register_lnd(&the_kralnd);
1731
1732         return 0;
1733 }
1734
1735 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1736 MODULE_DESCRIPTION("Kernel RapidArray LND v0.01");
1737 MODULE_LICENSE("GPL");
1738
1739 module_init(kranal_module_init);
1740 module_exit(kranal_module_fini);