Whamcloud - gitweb
LU-6142 lnet: convert kiblnd/ksocknal_thread_start to vararg
[fs/lustre-release.git] / lnet / klnds / socklnd / socklnd.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  *
31  * lnet/klnds/socklnd/socklnd.c
32  *
33  * Author: Zach Brown <zab@zabbo.net>
34  * Author: Peter J. Braam <braam@clusterfs.com>
35  * Author: Phil Schwan <phil@clusterfs.com>
36  * Author: Eric Barton <eric@bartonsoftware.com>
37  */
38
39 #include <linux/inetdevice.h>
40 #include "socklnd.h"
41 #include <linux/sunrpc/addr.h>
42
43 static const struct lnet_lnd the_ksocklnd;
44 struct ksock_nal_data ksocknal_data;
45
46 static struct ksock_interface *
47 ksocknal_ip2iface(struct lnet_ni *ni, struct sockaddr *addr)
48 {
49         struct ksock_net *net = ni->ni_data;
50         struct ksock_interface *iface;
51
52         iface = &net->ksnn_interface;
53
54         if (rpc_cmp_addr((struct sockaddr *)&iface->ksni_addr, addr))
55                 return iface;
56
57         return NULL;
58 }
59
60 static struct ksock_interface *
61 ksocknal_index2iface(struct lnet_ni *ni, int index)
62 {
63         struct ksock_net *net = ni->ni_data;
64         struct ksock_interface *iface;
65
66         iface = &net->ksnn_interface;
67
68         if (iface->ksni_index == index)
69                 return iface;
70
71         return NULL;
72 }
73
74 static int ksocknal_ip2index(struct sockaddr *addr, struct lnet_ni *ni)
75 {
76         struct net_device *dev;
77         int ret = -1;
78         DECLARE_CONST_IN_IFADDR(ifa);
79
80         if (addr->sa_family != AF_INET)
81                 /* No IPv6 support yet */
82                 return ret;
83
84         rcu_read_lock();
85         for_each_netdev(ni->ni_net_ns, dev) {
86                 int flags = dev_get_flags(dev);
87                 struct in_device *in_dev;
88
89                 if (flags & IFF_LOOPBACK) /* skip the loopback IF */
90                         continue;
91
92                 if (!(flags & IFF_UP))
93                         continue;
94
95                 in_dev = __in_dev_get_rcu(dev);
96                 if (!in_dev)
97                         continue;
98
99                 in_dev_for_each_ifa_rcu(ifa, in_dev) {
100                         if (ifa->ifa_local ==
101                             ((struct sockaddr_in *)addr)->sin_addr.s_addr)
102                                 ret = dev->ifindex;
103                 }
104                 endfor_ifa(in_dev);
105                 if (ret >= 0)
106                         break;
107         }
108         rcu_read_unlock();
109
110         return ret;
111 }
112
113 static struct ksock_conn_cb *
114 ksocknal_create_conn_cb(struct sockaddr *addr)
115 {
116         struct ksock_conn_cb *conn_cb;
117
118         LIBCFS_ALLOC(conn_cb, sizeof(*conn_cb));
119         if (!conn_cb)
120                 return NULL;
121
122         refcount_set(&conn_cb->ksnr_refcount, 1);
123         conn_cb->ksnr_peer = NULL;
124         conn_cb->ksnr_retry_interval = 0;         /* OK to connect at any time */
125         rpc_copy_addr((struct sockaddr *)&conn_cb->ksnr_addr, addr);
126         rpc_set_port((struct sockaddr *)&conn_cb->ksnr_addr,
127                      rpc_get_port(addr));
128         conn_cb->ksnr_myiface = -1;
129         conn_cb->ksnr_scheduled = 0;
130         conn_cb->ksnr_connecting = 0;
131         conn_cb->ksnr_connected = 0;
132         conn_cb->ksnr_deleted = 0;
133         conn_cb->ksnr_conn_count = 0;
134         conn_cb->ksnr_ctrl_conn_count = 0;
135         conn_cb->ksnr_blki_conn_count = 0;
136         conn_cb->ksnr_blko_conn_count = 0;
137
138         return conn_cb;
139 }
140
141 void
142 ksocknal_destroy_conn_cb(struct ksock_conn_cb *conn_cb)
143 {
144         LASSERT(refcount_read(&conn_cb->ksnr_refcount) == 0);
145
146         if (conn_cb->ksnr_peer)
147                 ksocknal_peer_decref(conn_cb->ksnr_peer);
148
149         LIBCFS_FREE(conn_cb, sizeof(*conn_cb));
150 }
151
152 static struct ksock_peer_ni *
153 ksocknal_create_peer(struct lnet_ni *ni, struct lnet_process_id id)
154 {
155         int cpt = lnet_cpt_of_nid(id.nid, ni);
156         struct ksock_net *net = ni->ni_data;
157         struct ksock_peer_ni *peer_ni;
158
159         LASSERT(id.nid != LNET_NID_ANY);
160         LASSERT(id.pid != LNET_PID_ANY);
161         LASSERT(!in_interrupt());
162
163         if (!atomic_inc_unless_negative(&net->ksnn_npeers)) {
164                 CERROR("Can't create peer_ni: network shutdown\n");
165                 return ERR_PTR(-ESHUTDOWN);
166         }
167
168         LIBCFS_CPT_ALLOC(peer_ni, lnet_cpt_table(), cpt, sizeof(*peer_ni));
169         if (!peer_ni) {
170                 atomic_dec(&net->ksnn_npeers);
171                 return ERR_PTR(-ENOMEM);
172         }
173
174         peer_ni->ksnp_ni = ni;
175         peer_ni->ksnp_id = id;
176         refcount_set(&peer_ni->ksnp_refcount, 1); /* 1 ref for caller */
177         peer_ni->ksnp_closing = 0;
178         peer_ni->ksnp_accepting = 0;
179         peer_ni->ksnp_proto = NULL;
180         peer_ni->ksnp_last_alive = 0;
181         peer_ni->ksnp_zc_next_cookie = SOCKNAL_KEEPALIVE_PING + 1;
182         peer_ni->ksnp_conn_cb = NULL;
183
184         INIT_LIST_HEAD(&peer_ni->ksnp_conns);
185         INIT_LIST_HEAD(&peer_ni->ksnp_tx_queue);
186         INIT_LIST_HEAD(&peer_ni->ksnp_zc_req_list);
187         spin_lock_init(&peer_ni->ksnp_lock);
188
189         return peer_ni;
190 }
191
192 void
193 ksocknal_destroy_peer(struct ksock_peer_ni *peer_ni)
194 {
195         struct ksock_net *net = peer_ni->ksnp_ni->ni_data;
196
197         CDEBUG (D_NET, "peer_ni %s %p deleted\n",
198                 libcfs_id2str(peer_ni->ksnp_id), peer_ni);
199
200         LASSERT(refcount_read(&peer_ni->ksnp_refcount) == 0);
201         LASSERT(peer_ni->ksnp_accepting == 0);
202         LASSERT(list_empty(&peer_ni->ksnp_conns));
203         LASSERT(peer_ni->ksnp_conn_cb == NULL);
204         LASSERT(list_empty(&peer_ni->ksnp_tx_queue));
205         LASSERT(list_empty(&peer_ni->ksnp_zc_req_list));
206
207         LIBCFS_FREE(peer_ni, sizeof(*peer_ni));
208
209         /* NB a peer_ni's connections and conn_cb keep a reference on their
210          * peer_ni until they are destroyed, so we can be assured that _all_
211          * state to do with this peer_ni has been cleaned up when its refcount
212          * drops to zero.
213          */
214         if (atomic_dec_and_test(&net->ksnn_npeers))
215                 wake_up_var(&net->ksnn_npeers);
216 }
217
218 struct ksock_peer_ni *
219 ksocknal_find_peer_locked(struct lnet_ni *ni, struct lnet_process_id id)
220 {
221         struct ksock_peer_ni *peer_ni;
222
223         hash_for_each_possible(ksocknal_data.ksnd_peers, peer_ni,
224                                ksnp_list, id.nid) {
225                 LASSERT(!peer_ni->ksnp_closing);
226
227                 if (peer_ni->ksnp_ni != ni)
228                         continue;
229
230                 if (peer_ni->ksnp_id.nid != id.nid ||
231                     peer_ni->ksnp_id.pid != id.pid)
232                         continue;
233
234                 CDEBUG(D_NET, "got peer_ni [%p] -> %s (%d)\n",
235                        peer_ni, libcfs_id2str(id),
236                        refcount_read(&peer_ni->ksnp_refcount));
237                 return peer_ni;
238         }
239         return NULL;
240 }
241
242 struct ksock_peer_ni *
243 ksocknal_find_peer(struct lnet_ni *ni, struct lnet_process_id id)
244 {
245         struct ksock_peer_ni *peer_ni;
246
247         read_lock(&ksocknal_data.ksnd_global_lock);
248         peer_ni = ksocknal_find_peer_locked(ni, id);
249         if (peer_ni != NULL)                    /* +1 ref for caller? */
250                 ksocknal_peer_addref(peer_ni);
251         read_unlock(&ksocknal_data.ksnd_global_lock);
252
253         return (peer_ni);
254 }
255
256 static void
257 ksocknal_unlink_peer_locked(struct ksock_peer_ni *peer_ni)
258 {
259         int i;
260         struct ksock_interface *iface;
261
262         for (i = 0; i < peer_ni->ksnp_n_passive_ips; i++) {
263                 struct sockaddr_in sa = { .sin_family = AF_INET };
264                 LASSERT(i < LNET_INTERFACES_NUM);
265                 sa.sin_addr.s_addr = htonl(peer_ni->ksnp_passive_ips[i]);
266
267                 iface = ksocknal_ip2iface(peer_ni->ksnp_ni,
268                                           (struct sockaddr *)&sa);
269                 /*
270                  * All IPs in peer_ni->ksnp_passive_ips[] come from the
271                  * interface list, therefore the call must succeed.
272                  */
273                 LASSERT(iface != NULL);
274
275                 CDEBUG(D_NET, "peer_ni=%p iface=%p ksni_nroutes=%d\n",
276                        peer_ni, iface, iface->ksni_nroutes);
277                 iface->ksni_npeers--;
278         }
279
280         LASSERT(list_empty(&peer_ni->ksnp_conns));
281         LASSERT(peer_ni->ksnp_conn_cb == NULL);
282         LASSERT(!peer_ni->ksnp_closing);
283         peer_ni->ksnp_closing = 1;
284         hlist_del(&peer_ni->ksnp_list);
285         /* lose peerlist's ref */
286         ksocknal_peer_decref(peer_ni);
287 }
288
289 static int
290 ksocknal_get_peer_info(struct lnet_ni *ni, int index,
291                        struct lnet_process_id *id, __u32 *myip, __u32 *peer_ip,
292                        int *port, int *conn_count, int *share_count)
293 {
294         struct ksock_peer_ni *peer_ni;
295         struct ksock_conn_cb *conn_cb;
296         int i;
297         int j;
298         int rc = -ENOENT;
299
300         read_lock(&ksocknal_data.ksnd_global_lock);
301
302         hash_for_each(ksocknal_data.ksnd_peers, i, peer_ni, ksnp_list) {
303
304                 if (peer_ni->ksnp_ni != ni)
305                         continue;
306
307                 if (peer_ni->ksnp_n_passive_ips == 0 &&
308                     peer_ni->ksnp_conn_cb == NULL) {
309                         if (index-- > 0)
310                                 continue;
311
312                         *id = peer_ni->ksnp_id;
313                         *myip = 0;
314                         *peer_ip = 0;
315                         *port = 0;
316                         *conn_count = 0;
317                         *share_count = 0;
318                         rc = 0;
319                         goto out;
320                 }
321
322                 for (j = 0; j < peer_ni->ksnp_n_passive_ips; j++) {
323                         if (index-- > 0)
324                                 continue;
325
326                         *id = peer_ni->ksnp_id;
327                         *myip = peer_ni->ksnp_passive_ips[j];
328                         *peer_ip = 0;
329                         *port = 0;
330                         *conn_count = 0;
331                         *share_count = 0;
332                         rc = 0;
333                         goto out;
334                 }
335
336                 if (peer_ni->ksnp_conn_cb) {
337                         if (index-- > 0)
338                                 continue;
339
340                         conn_cb = peer_ni->ksnp_conn_cb;
341
342                         *id = peer_ni->ksnp_id;
343                         if (conn_cb->ksnr_addr.ss_family == AF_INET) {
344                                 struct sockaddr_in *sa =
345                                         (void *)&conn_cb->ksnr_addr;
346
347                                 rc = choose_ipv4_src(myip,
348                                                      conn_cb->ksnr_myiface,
349                                                      ntohl(sa->sin_addr.s_addr),
350                                                      ni->ni_net_ns);
351                                 *peer_ip = ntohl(sa->sin_addr.s_addr);
352                                 *port = ntohs(sa->sin_port);
353                         } else {
354                                 *myip = 0xFFFFFFFF;
355                                 *peer_ip = 0xFFFFFFFF;
356                                 *port = 0;
357                                 rc = -ENOTSUPP;
358                         }
359                         *conn_count = conn_cb->ksnr_conn_count;
360                         *share_count = 1;
361                         goto out;
362                 }
363         }
364 out:
365         read_unlock(&ksocknal_data.ksnd_global_lock);
366         return rc;
367 }
368
369 static unsigned int
370 ksocknal_get_conn_count_by_type(struct ksock_conn_cb *conn_cb,
371                                 int type)
372 {
373         unsigned int count = 0;
374
375         switch (type) {
376         case SOCKLND_CONN_CONTROL:
377                 count = conn_cb->ksnr_ctrl_conn_count;
378                 break;
379         case SOCKLND_CONN_BULK_IN:
380                 count = conn_cb->ksnr_blki_conn_count;
381                 break;
382         case SOCKLND_CONN_BULK_OUT:
383                 count = conn_cb->ksnr_blko_conn_count;
384                 break;
385         case SOCKLND_CONN_ANY:
386                 count = conn_cb->ksnr_conn_count;
387                 break;
388         default:
389                 LBUG();
390                 break;
391         }
392
393         return count;
394 }
395
396 static void
397 ksocknal_incr_conn_count(struct ksock_conn_cb *conn_cb,
398                          int type)
399 {
400         conn_cb->ksnr_conn_count++;
401
402         /* check if all connections of the given type got created */
403         switch (type) {
404         case SOCKLND_CONN_CONTROL:
405                 conn_cb->ksnr_ctrl_conn_count++;
406                 /* there's a single control connection per peer */
407                 conn_cb->ksnr_connected |= BIT(type);
408                 break;
409         case SOCKLND_CONN_BULK_IN:
410                 conn_cb->ksnr_blki_conn_count++;
411                 if (conn_cb->ksnr_blki_conn_count >=
412                     *ksocknal_tunables.ksnd_conns_per_peer)
413                         conn_cb->ksnr_connected |= BIT(type);
414                 break;
415         case SOCKLND_CONN_BULK_OUT:
416                 conn_cb->ksnr_blko_conn_count++;
417                 if (conn_cb->ksnr_blko_conn_count >=
418                     *ksocknal_tunables.ksnd_conns_per_peer)
419                         conn_cb->ksnr_connected |= BIT(type);
420                 break;
421         case SOCKLND_CONN_ANY:
422                 if (conn_cb->ksnr_conn_count >=
423                     *ksocknal_tunables.ksnd_conns_per_peer)
424                         conn_cb->ksnr_connected |= BIT(type);
425                 break;
426         default:
427                 LBUG();
428                 break;
429
430         }
431
432         CDEBUG(D_NET, "Add conn type %d, ksnr_connected %x conns_per_peer %d\n",
433                type, conn_cb->ksnr_connected, *ksocknal_tunables.ksnd_conns_per_peer);
434 }
435
436 static void
437 ksocknal_associate_cb_conn_locked(struct ksock_conn_cb *conn_cb,
438                                   struct ksock_conn *conn)
439 {
440         struct ksock_peer_ni *peer_ni = conn_cb->ksnr_peer;
441         int type = conn->ksnc_type;
442         struct ksock_interface *iface;
443         int conn_iface;
444
445         conn_iface = ksocknal_ip2index((struct sockaddr *)&conn->ksnc_myaddr,
446                                        peer_ni->ksnp_ni);
447         conn->ksnc_conn_cb = conn_cb;
448         ksocknal_conn_cb_addref(conn_cb);
449
450         if (conn_cb->ksnr_myiface != conn_iface) {
451                 if (conn_cb->ksnr_myiface < 0) {
452                         /* route wasn't bound locally yet (the initial route) */
453                         CDEBUG(D_NET, "Binding %s %pIS to interface %d\n",
454                                libcfs_id2str(peer_ni->ksnp_id),
455                                &conn_cb->ksnr_addr,
456                                conn_iface);
457                 } else {
458                         CDEBUG(D_NET,
459                                "Rebinding %s %pIS from interface %d to %d\n",
460                                libcfs_id2str(peer_ni->ksnp_id),
461                                &conn_cb->ksnr_addr,
462                                conn_cb->ksnr_myiface,
463                                conn_iface);
464
465                         iface = ksocknal_index2iface(peer_ni->ksnp_ni,
466                                                      conn_cb->ksnr_myiface);
467                         if (iface)
468                                 iface->ksni_nroutes--;
469                 }
470                 conn_cb->ksnr_myiface = conn_iface;
471                 iface = ksocknal_index2iface(peer_ni->ksnp_ni,
472                                              conn_cb->ksnr_myiface);
473                 if (iface)
474                         iface->ksni_nroutes++;
475         }
476
477         ksocknal_incr_conn_count(conn_cb, type);
478
479         /* Successful connection => further attempts can
480          * proceed immediately
481          */
482         conn_cb->ksnr_retry_interval = 0;
483 }
484
485 static void
486 ksocknal_add_conn_cb_locked(struct ksock_peer_ni *peer_ni,
487                             struct ksock_conn_cb *conn_cb)
488 {
489         struct ksock_conn *conn;
490         struct ksock_net *net = peer_ni->ksnp_ni->ni_data;
491
492         LASSERT(!peer_ni->ksnp_closing);
493         LASSERT(!conn_cb->ksnr_peer);
494         LASSERT(!conn_cb->ksnr_scheduled);
495         LASSERT(!conn_cb->ksnr_connecting);
496         LASSERT(conn_cb->ksnr_connected == 0);
497
498         conn_cb->ksnr_peer = peer_ni;
499         ksocknal_peer_addref(peer_ni);
500
501         /* set the conn_cb's interface to the current net's interface */
502         conn_cb->ksnr_myiface = net->ksnn_interface.ksni_index;
503         net->ksnn_interface.ksni_nroutes++;
504
505         /* peer_ni's route list takes over my ref on 'route' */
506         peer_ni->ksnp_conn_cb = conn_cb;
507
508         list_for_each_entry(conn, &peer_ni->ksnp_conns, ksnc_list) {
509                 if (!rpc_cmp_addr((struct sockaddr *)&conn->ksnc_peeraddr,
510                                   (struct sockaddr *)&conn_cb->ksnr_addr))
511                         continue;
512
513                 ksocknal_associate_cb_conn_locked(conn_cb, conn);
514                 /* keep going (typed conns) */
515         }
516 }
517
518 static void
519 ksocknal_del_conn_cb_locked(struct ksock_conn_cb *conn_cb)
520 {
521         struct ksock_peer_ni *peer_ni = conn_cb->ksnr_peer;
522         struct ksock_interface *iface;
523         struct ksock_conn *conn;
524         struct ksock_conn *cnxt;
525
526         LASSERT(!conn_cb->ksnr_deleted);
527
528         /* Close associated conns */
529         list_for_each_entry_safe(conn, cnxt, &peer_ni->ksnp_conns, ksnc_list) {
530                 if (conn->ksnc_conn_cb != conn_cb)
531                         continue;
532
533                 ksocknal_close_conn_locked(conn, 0);
534         }
535
536         if (conn_cb->ksnr_myiface >= 0) {
537                 iface = ksocknal_index2iface(peer_ni->ksnp_ni,
538                                              conn_cb->ksnr_myiface);
539                 if (iface)
540                         iface->ksni_nroutes--;
541         }
542
543         conn_cb->ksnr_deleted = 1;
544         ksocknal_conn_cb_decref(conn_cb);               /* drop peer_ni's ref */
545         peer_ni->ksnp_conn_cb = NULL;
546
547         if (list_empty(&peer_ni->ksnp_conns)) {
548                 /* I've just removed the last route to a peer_ni with no active
549                  * connections
550                  */
551                 ksocknal_unlink_peer_locked(peer_ni);
552         }
553 }
554
555 int
556 ksocknal_add_peer(struct lnet_ni *ni, struct lnet_process_id id,
557                   struct sockaddr *addr)
558 {
559         struct ksock_peer_ni *peer_ni;
560         struct ksock_peer_ni *peer2;
561         struct ksock_conn_cb *conn_cb;
562
563         if (id.nid == LNET_NID_ANY ||
564             id.pid == LNET_PID_ANY)
565                 return (-EINVAL);
566
567         /* Have a brand new peer_ni ready... */
568         peer_ni = ksocknal_create_peer(ni, id);
569         if (IS_ERR(peer_ni))
570                 return PTR_ERR(peer_ni);
571
572         conn_cb = ksocknal_create_conn_cb(addr);
573         if (!conn_cb) {
574                 ksocknal_peer_decref(peer_ni);
575                 return -ENOMEM;
576         }
577
578         write_lock_bh(&ksocknal_data.ksnd_global_lock);
579
580         /* always called with a ref on ni, so shutdown can't have started */
581         LASSERT(atomic_read(&((struct ksock_net *)ni->ni_data)->ksnn_npeers)
582                 >= 0);
583
584         peer2 = ksocknal_find_peer_locked(ni, id);
585         if (peer2 != NULL) {
586                 ksocknal_peer_decref(peer_ni);
587                 peer_ni = peer2;
588         } else {
589                 /* peer_ni table takes my ref on peer_ni */
590                 hash_add(ksocknal_data.ksnd_peers, &peer_ni->ksnp_list, id.nid);
591         }
592
593         ksocknal_add_conn_cb_locked(peer_ni, conn_cb);
594
595         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
596
597         return 0;
598 }
599
600 static void
601 ksocknal_del_peer_locked(struct ksock_peer_ni *peer_ni, __u32 ip)
602 {
603         struct ksock_conn *conn;
604         struct ksock_conn *cnxt;
605         struct ksock_conn_cb *conn_cb;
606
607         LASSERT(!peer_ni->ksnp_closing);
608
609         /* Extra ref prevents peer_ni disappearing until I'm done with it */
610         ksocknal_peer_addref(peer_ni);
611         conn_cb = peer_ni->ksnp_conn_cb;
612         if (conn_cb)
613                 ksocknal_del_conn_cb_locked(conn_cb);
614
615         list_for_each_entry_safe(conn, cnxt, &peer_ni->ksnp_conns,
616                                  ksnc_list)
617                 ksocknal_close_conn_locked(conn, 0);
618
619         ksocknal_peer_decref(peer_ni);
620         /* NB peer_ni unlinks itself when last conn/conn_cb is removed */
621 }
622
623 static int
624 ksocknal_del_peer(struct lnet_ni *ni, struct lnet_process_id id, __u32 ip)
625 {
626         LIST_HEAD(zombies);
627         struct hlist_node *pnxt;
628         struct ksock_peer_ni *peer_ni;
629         int lo;
630         int hi;
631         int i;
632         int rc = -ENOENT;
633
634         write_lock_bh(&ksocknal_data.ksnd_global_lock);
635
636         if (id.nid != LNET_NID_ANY) {
637                 lo = hash_min(id.nid, HASH_BITS(ksocknal_data.ksnd_peers));
638                 hi = lo;
639         } else {
640                 lo = 0;
641                 hi = HASH_SIZE(ksocknal_data.ksnd_peers) - 1;
642         }
643
644         for (i = lo; i <= hi; i++) {
645                 hlist_for_each_entry_safe(peer_ni, pnxt,
646                                           &ksocknal_data.ksnd_peers[i],
647                                           ksnp_list) {
648                         if (peer_ni->ksnp_ni != ni)
649                                 continue;
650
651                         if (!((id.nid == LNET_NID_ANY ||
652                                peer_ni->ksnp_id.nid == id.nid) &&
653                               (id.pid == LNET_PID_ANY ||
654                                peer_ni->ksnp_id.pid == id.pid)))
655                                 continue;
656
657                         ksocknal_peer_addref(peer_ni);  /* a ref for me... */
658
659                         ksocknal_del_peer_locked(peer_ni, ip);
660
661                         if (peer_ni->ksnp_closing &&
662                             !list_empty(&peer_ni->ksnp_tx_queue)) {
663                                 LASSERT(list_empty(&peer_ni->ksnp_conns));
664                                 LASSERT(peer_ni->ksnp_conn_cb == NULL);
665
666                                 list_splice_init(&peer_ni->ksnp_tx_queue,
667                                                  &zombies);
668                         }
669
670                         ksocknal_peer_decref(peer_ni);  /* ...till here */
671
672                         rc = 0;                         /* matched! */
673                 }
674         }
675
676         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
677
678         ksocknal_txlist_done(ni, &zombies, -ENETDOWN);
679
680         return rc;
681 }
682
683 static struct ksock_conn *
684 ksocknal_get_conn_by_idx(struct lnet_ni *ni, int index)
685 {
686         struct ksock_peer_ni *peer_ni;
687         struct ksock_conn *conn;
688         int i;
689
690         read_lock(&ksocknal_data.ksnd_global_lock);
691
692         hash_for_each(ksocknal_data.ksnd_peers, i, peer_ni, ksnp_list) {
693                 LASSERT(!peer_ni->ksnp_closing);
694
695                 if (peer_ni->ksnp_ni != ni)
696                         continue;
697
698                 list_for_each_entry(conn, &peer_ni->ksnp_conns,
699                                     ksnc_list) {
700                         if (index-- > 0)
701                                 continue;
702
703                         ksocknal_conn_addref(conn);
704                         read_unlock(&ksocknal_data.ksnd_global_lock);
705                         return conn;
706                 }
707         }
708
709         read_unlock(&ksocknal_data.ksnd_global_lock);
710         return NULL;
711 }
712
713 static struct ksock_sched *
714 ksocknal_choose_scheduler_locked(unsigned int cpt)
715 {
716         struct ksock_sched *sched = ksocknal_data.ksnd_schedulers[cpt];
717         int i;
718
719         if (sched->kss_nthreads == 0) {
720                 cfs_percpt_for_each(sched, i, ksocknal_data.ksnd_schedulers) {
721                         if (sched->kss_nthreads > 0) {
722                                 CDEBUG(D_NET, "scheduler[%d] has no threads. selected scheduler[%d]\n",
723                                        cpt, sched->kss_cpt);
724                                 return sched;
725                         }
726                 }
727                 return NULL;
728         }
729
730         return sched;
731 }
732
733 int
734 ksocknal_accept(struct lnet_ni *ni, struct socket *sock)
735 {
736         struct ksock_connreq *cr;
737         int rc;
738         struct sockaddr_storage peer;
739
740         rc = lnet_sock_getaddr(sock, true, &peer);
741         if (rc != 0) {
742                 CERROR("Can't determine new connection's address\n");
743                 return rc;
744         }
745
746         LIBCFS_ALLOC(cr, sizeof(*cr));
747         if (cr == NULL) {
748                 LCONSOLE_ERROR_MSG(0x12f,
749                                    "Dropping connection request from %pIS: memory exhausted\n",
750                                    &peer);
751                 return -ENOMEM;
752         }
753
754         lnet_ni_addref(ni);
755         cr->ksncr_ni   = ni;
756         cr->ksncr_sock = sock;
757
758         spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
759
760         list_add_tail(&cr->ksncr_list, &ksocknal_data.ksnd_connd_connreqs);
761         wake_up(&ksocknal_data.ksnd_connd_waitq);
762
763         spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
764         return 0;
765 }
766
767 static int
768 ksocknal_connecting(struct ksock_conn_cb *conn_cb, struct sockaddr *sa)
769 {
770         if (conn_cb &&
771             rpc_cmp_addr((struct sockaddr *)&conn_cb->ksnr_addr, sa))
772                 return conn_cb->ksnr_connecting;
773         return 0;
774 }
775
776 int
777 ksocknal_create_conn(struct lnet_ni *ni, struct ksock_conn_cb *conn_cb,
778                      struct socket *sock, int type)
779 {
780         rwlock_t *global_lock = &ksocknal_data.ksnd_global_lock;
781         LIST_HEAD(zombies);
782         struct lnet_process_id peerid;
783         u64 incarnation;
784         struct ksock_conn *conn;
785         struct ksock_conn *conn2;
786         struct ksock_peer_ni *peer_ni = NULL;
787         struct ksock_peer_ni *peer2;
788         struct ksock_sched *sched;
789         struct ksock_hello_msg *hello;
790         int cpt;
791         struct ksock_tx *tx;
792         struct ksock_tx *txtmp;
793         int rc;
794         int rc2;
795         int active;
796         int num_dup = 0;
797         char *warn = NULL;
798
799         active = (conn_cb != NULL);
800
801         LASSERT(active == (type != SOCKLND_CONN_NONE));
802
803         LIBCFS_ALLOC(conn, sizeof(*conn));
804         if (conn == NULL) {
805                 rc = -ENOMEM;
806                 goto failed_0;
807         }
808
809         conn->ksnc_peer = NULL;
810         conn->ksnc_conn_cb = NULL;
811         conn->ksnc_sock = sock;
812         /* 2 ref, 1 for conn, another extra ref prevents socket
813          * being closed before establishment of connection */
814         refcount_set(&conn->ksnc_sock_refcount, 2);
815         conn->ksnc_type = type;
816         ksocknal_lib_save_callback(sock, conn);
817         refcount_set(&conn->ksnc_conn_refcount, 1); /* 1 ref for me */
818
819         conn->ksnc_rx_ready = 0;
820         conn->ksnc_rx_scheduled = 0;
821
822         INIT_LIST_HEAD(&conn->ksnc_tx_queue);
823         conn->ksnc_tx_ready = 0;
824         conn->ksnc_tx_scheduled = 0;
825         conn->ksnc_tx_carrier = NULL;
826         atomic_set (&conn->ksnc_tx_nob, 0);
827
828         LIBCFS_ALLOC(hello, offsetof(struct ksock_hello_msg,
829                                      kshm_ips[LNET_INTERFACES_NUM]));
830         if (hello == NULL) {
831                 rc = -ENOMEM;
832                 goto failed_1;
833         }
834
835         /* stash conn's local and remote addrs */
836         rc = ksocknal_lib_get_conn_addrs(conn);
837         if (rc != 0)
838                 goto failed_1;
839
840         /* Find out/confirm peer_ni's NID and connection type and get the
841          * vector of interfaces she's willing to let me connect to.
842          * Passive connections use the listener timeout since the peer_ni sends
843          * eagerly
844          */
845
846         if (active) {
847                 peer_ni = conn_cb->ksnr_peer;
848                 LASSERT(ni == peer_ni->ksnp_ni);
849
850                 /* Active connection sends HELLO eagerly */
851                 hello->kshm_nips = 0;
852                 peerid = peer_ni->ksnp_id;
853
854                 write_lock_bh(global_lock);
855                 conn->ksnc_proto = peer_ni->ksnp_proto;
856                 write_unlock_bh(global_lock);
857
858                 if (conn->ksnc_proto == NULL) {
859                         conn->ksnc_proto = &ksocknal_protocol_v3x;
860 #if SOCKNAL_VERSION_DEBUG
861                         if (*ksocknal_tunables.ksnd_protocol == 2)
862                                 conn->ksnc_proto = &ksocknal_protocol_v2x;
863                         else if (*ksocknal_tunables.ksnd_protocol == 1)
864                                 conn->ksnc_proto = &ksocknal_protocol_v1x;
865 #endif
866                 }
867
868                 rc = ksocknal_send_hello(ni, conn, peerid.nid, hello);
869                 if (rc != 0)
870                         goto failed_1;
871         } else {
872                 peerid.nid = LNET_NID_ANY;
873                 peerid.pid = LNET_PID_ANY;
874
875                 /* Passive, get protocol from peer_ni */
876                 conn->ksnc_proto = NULL;
877         }
878
879         rc = ksocknal_recv_hello(ni, conn, hello, &peerid, &incarnation);
880         if (rc < 0)
881                 goto failed_1;
882
883         LASSERT(rc == 0 || active);
884         LASSERT(conn->ksnc_proto != NULL);
885         LASSERT(peerid.nid != LNET_NID_ANY);
886
887         cpt = lnet_cpt_of_nid(peerid.nid, ni);
888
889         if (active) {
890                 ksocknal_peer_addref(peer_ni);
891                 write_lock_bh(global_lock);
892         } else {
893                 peer_ni = ksocknal_create_peer(ni, peerid);
894                 if (IS_ERR(peer_ni)) {
895                         rc = PTR_ERR(peer_ni);
896                         goto failed_1;
897                 }
898
899                 write_lock_bh(global_lock);
900
901                 /* called with a ref on ni, so shutdown can't have started */
902                 LASSERT(atomic_read(&((struct ksock_net *)ni->ni_data)->ksnn_npeers) >= 0);
903
904                 peer2 = ksocknal_find_peer_locked(ni, peerid);
905                 if (peer2 == NULL) {
906                         /* NB this puts an "empty" peer_ni in the peer_ni
907                          * table (which takes my ref) */
908                         hash_add(ksocknal_data.ksnd_peers,
909                                  &peer_ni->ksnp_list, peerid.nid);
910                 } else {
911                         ksocknal_peer_decref(peer_ni);
912                         peer_ni = peer2;
913                 }
914
915                 /* +1 ref for me */
916                 ksocknal_peer_addref(peer_ni);
917                 peer_ni->ksnp_accepting++;
918
919                 /* Am I already connecting to this guy?  Resolve in
920                  * favour of higher NID...
921                  */
922                 if (peerid.nid < ni->ni_nid &&
923                     ksocknal_connecting(peer_ni->ksnp_conn_cb,
924                                         ((struct sockaddr *) &conn->ksnc_peeraddr))) {
925                         rc = EALREADY;
926                         warn = "connection race resolution";
927                         goto failed_2;
928                 }
929         }
930
931         if (peer_ni->ksnp_closing ||
932             (active && conn_cb->ksnr_deleted)) {
933                 /* peer_ni/conn_cb got closed under me */
934                 rc = -ESTALE;
935                 warn = "peer_ni/conn_cb removed";
936                 goto failed_2;
937         }
938
939         if (peer_ni->ksnp_proto == NULL) {
940                 /* Never connected before.
941                  * NB recv_hello may have returned EPROTO to signal my peer_ni
942                  * wants a different protocol than the one I asked for.
943                  */
944                 LASSERT(list_empty(&peer_ni->ksnp_conns));
945
946                 peer_ni->ksnp_proto = conn->ksnc_proto;
947                 peer_ni->ksnp_incarnation = incarnation;
948         }
949
950         if (peer_ni->ksnp_proto != conn->ksnc_proto ||
951             peer_ni->ksnp_incarnation != incarnation) {
952                 /* peer_ni rebooted or I've got the wrong protocol version */
953                 ksocknal_close_peer_conns_locked(peer_ni, NULL, 0);
954
955                 peer_ni->ksnp_proto = NULL;
956                 rc = ESTALE;
957                 warn = peer_ni->ksnp_incarnation != incarnation ?
958                         "peer_ni rebooted" :
959                         "wrong proto version";
960                 goto failed_2;
961         }
962
963         switch (rc) {
964         default:
965                 LBUG();
966         case 0:
967                 break;
968         case EALREADY:
969                 warn = "lost conn race";
970                 goto failed_2;
971         case EPROTO:
972                 warn = "retry with different protocol version";
973                 goto failed_2;
974         }
975
976         /* Refuse to duplicate an existing connection, unless this is a
977          * loopback connection */
978         if (!rpc_cmp_addr((struct sockaddr *)&conn->ksnc_peeraddr,
979                           (struct sockaddr *)&conn->ksnc_myaddr)) {
980                 list_for_each_entry(conn2, &peer_ni->ksnp_conns, ksnc_list) {
981                         if (!rpc_cmp_addr(
982                                     (struct sockaddr *)&conn2->ksnc_peeraddr,
983                                     (struct sockaddr *)&conn->ksnc_peeraddr) ||
984                             !rpc_cmp_addr(
985                                     (struct sockaddr *)&conn2->ksnc_myaddr,
986                                     (struct sockaddr *)&conn->ksnc_myaddr) ||
987                             conn2->ksnc_type != conn->ksnc_type)
988                                 continue;
989
990                         num_dup++;
991                         if (num_dup < *ksocknal_tunables.ksnd_conns_per_peer)
992                                 continue;
993
994                         /* Reply on a passive connection attempt so the peer_ni
995                          * realises we're connected.
996                          */
997                         LASSERT(rc == 0);
998                         if (!active)
999                                 rc = EALREADY;
1000
1001                         warn = "duplicate";
1002                         goto failed_2;
1003                 }
1004         }
1005         /* If the connection created by this route didn't bind to the IP
1006          * address the route connected to, the connection/route matching
1007          * code below probably isn't going to work.
1008          */
1009         if (active &&
1010             !rpc_cmp_addr((struct sockaddr *)&conn_cb->ksnr_addr,
1011                           (struct sockaddr *)&conn->ksnc_peeraddr)) {
1012                 CERROR("Route %s %pIS connected to %pIS\n",
1013                        libcfs_id2str(peer_ni->ksnp_id),
1014                        &conn_cb->ksnr_addr,
1015                        &conn->ksnc_peeraddr);
1016         }
1017
1018         /* Search for a conn_cb corresponding to the new connection and
1019          * create an association.  This allows incoming connections created
1020          * by conn_cbs in my peer_ni to match my own conn_cb entries so I don't
1021          * continually create duplicate conn_cbs.
1022          */
1023         conn_cb = peer_ni->ksnp_conn_cb;
1024
1025         if (conn_cb && rpc_cmp_addr((struct sockaddr *)&conn->ksnc_peeraddr,
1026                                     (struct sockaddr *)&conn_cb->ksnr_addr))
1027                 ksocknal_associate_cb_conn_locked(conn_cb, conn);
1028
1029         conn->ksnc_peer = peer_ni;                 /* conn takes my ref on peer_ni */
1030         peer_ni->ksnp_last_alive = ktime_get_seconds();
1031         peer_ni->ksnp_send_keepalive = 0;
1032         peer_ni->ksnp_error = 0;
1033
1034         sched = ksocknal_choose_scheduler_locked(cpt);
1035         if (!sched) {
1036                 CERROR("no schedulers available. node is unhealthy\n");
1037                 goto failed_2;
1038         }
1039         /*
1040          * The cpt might have changed if we ended up selecting a non cpt
1041          * native scheduler. So use the scheduler's cpt instead.
1042          */
1043         cpt = sched->kss_cpt;
1044         sched->kss_nconns++;
1045         conn->ksnc_scheduler = sched;
1046
1047         conn->ksnc_tx_last_post = ktime_get_seconds();
1048         /* Set the deadline for the outgoing HELLO to drain */
1049         conn->ksnc_tx_bufnob = sock->sk->sk_wmem_queued;
1050         conn->ksnc_tx_deadline = ktime_get_seconds() +
1051                                  ksocknal_timeout();
1052         smp_mb();   /* order with adding to peer_ni's conn list */
1053
1054         list_add(&conn->ksnc_list, &peer_ni->ksnp_conns);
1055         ksocknal_conn_addref(conn);
1056
1057         ksocknal_new_packet(conn, 0);
1058
1059         conn->ksnc_zc_capable = ksocknal_lib_zc_capable(conn);
1060
1061         /* Take packets blocking for this connection. */
1062         list_for_each_entry_safe(tx, txtmp, &peer_ni->ksnp_tx_queue, tx_list) {
1063                 if (conn->ksnc_proto->pro_match_tx(conn, tx, tx->tx_nonblk) ==
1064                     SOCKNAL_MATCH_NO)
1065                         continue;
1066
1067                 list_del(&tx->tx_list);
1068                 ksocknal_queue_tx_locked(tx, conn);
1069         }
1070
1071         write_unlock_bh(global_lock);
1072
1073         /* We've now got a new connection.  Any errors from here on are just
1074          * like "normal" comms errors and we close the connection normally.
1075          * NB (a) we still have to send the reply HELLO for passive
1076          *        connections,
1077          *    (b) normal I/O on the conn is blocked until I setup and call the
1078          *        socket callbacks.
1079          */
1080
1081         CDEBUG(D_NET, "New conn %s p %d.x %pIS -> %pISp"
1082                " incarnation:%lld sched[%d]\n",
1083                libcfs_id2str(peerid), conn->ksnc_proto->pro_version,
1084                &conn->ksnc_myaddr, &conn->ksnc_peeraddr,
1085                incarnation, cpt);
1086
1087         if (!active) {
1088                 hello->kshm_nips = 0;
1089                 rc = ksocknal_send_hello(ni, conn, peerid.nid, hello);
1090         }
1091
1092         LIBCFS_FREE(hello, offsetof(struct ksock_hello_msg,
1093                                     kshm_ips[LNET_INTERFACES_NUM]));
1094
1095         /* setup the socket AFTER I've received hello (it disables
1096          * SO_LINGER).  I might call back to the acceptor who may want
1097          * to send a protocol version response and then close the
1098          * socket; this ensures the socket only tears down after the
1099          * response has been sent.
1100          */
1101         if (rc == 0)
1102                 rc = ksocknal_lib_setup_sock(sock);
1103
1104         write_lock_bh(global_lock);
1105
1106         /* NB my callbacks block while I hold ksnd_global_lock */
1107         ksocknal_lib_set_callback(sock, conn);
1108
1109         if (!active)
1110                 peer_ni->ksnp_accepting--;
1111
1112         write_unlock_bh(global_lock);
1113
1114         if (rc != 0) {
1115                 write_lock_bh(global_lock);
1116                 if (!conn->ksnc_closing) {
1117                         /* could be closed by another thread */
1118                         ksocknal_close_conn_locked(conn, rc);
1119                 }
1120                 write_unlock_bh(global_lock);
1121         } else if (ksocknal_connsock_addref(conn) == 0) {
1122                 /* Allow I/O to proceed. */
1123                 ksocknal_read_callback(conn);
1124                 ksocknal_write_callback(conn);
1125                 ksocknal_connsock_decref(conn);
1126         }
1127
1128         ksocknal_connsock_decref(conn);
1129         ksocknal_conn_decref(conn);
1130         return rc;
1131
1132 failed_2:
1133
1134         if (!peer_ni->ksnp_closing &&
1135             list_empty(&peer_ni->ksnp_conns) &&
1136             peer_ni->ksnp_conn_cb == NULL) {
1137                 list_splice_init(&peer_ni->ksnp_tx_queue, &zombies);
1138                 ksocknal_unlink_peer_locked(peer_ni);
1139         }
1140
1141         write_unlock_bh(global_lock);
1142
1143         if (warn != NULL) {
1144                 if (rc < 0)
1145                         CERROR("Not creating conn %s type %d: %s\n",
1146                                libcfs_id2str(peerid), conn->ksnc_type, warn);
1147                 else
1148                         CDEBUG(D_NET, "Not creating conn %s type %d: %s\n",
1149                                libcfs_id2str(peerid), conn->ksnc_type, warn);
1150         }
1151
1152         if (!active) {
1153                 if (rc > 0) {
1154                         /* Request retry by replying with CONN_NONE
1155                          * ksnc_proto has been set already
1156                          */
1157                         conn->ksnc_type = SOCKLND_CONN_NONE;
1158                         hello->kshm_nips = 0;
1159                         ksocknal_send_hello(ni, conn, peerid.nid, hello);
1160                 }
1161
1162                 write_lock_bh(global_lock);
1163                 peer_ni->ksnp_accepting--;
1164                 write_unlock_bh(global_lock);
1165         }
1166
1167         /*
1168          * If we get here without an error code, just use -EALREADY.
1169          * Depending on how we got here, the error may be positive
1170          * or negative. Normalize the value for ksocknal_txlist_done().
1171          */
1172         rc2 = (rc == 0 ? -EALREADY : (rc > 0 ? -rc : rc));
1173         ksocknal_txlist_done(ni, &zombies, rc2);
1174         ksocknal_peer_decref(peer_ni);
1175
1176 failed_1:
1177         if (hello != NULL)
1178                 LIBCFS_FREE(hello, offsetof(struct ksock_hello_msg,
1179                                             kshm_ips[LNET_INTERFACES_NUM]));
1180
1181         LIBCFS_FREE(conn, sizeof(*conn));
1182
1183 failed_0:
1184         sock_release(sock);
1185
1186         return rc;
1187 }
1188
1189 void
1190 ksocknal_close_conn_locked(struct ksock_conn *conn, int error)
1191 {
1192         /* This just does the immmediate housekeeping, and queues the
1193          * connection for the reaper to terminate.
1194          * Caller holds ksnd_global_lock exclusively in irq context */
1195         struct ksock_peer_ni *peer_ni = conn->ksnc_peer;
1196         struct ksock_conn_cb *conn_cb;
1197         struct ksock_conn *conn2;
1198
1199         LASSERT(peer_ni->ksnp_error == 0);
1200         LASSERT(!conn->ksnc_closing);
1201         conn->ksnc_closing = 1;
1202
1203         /* ksnd_deathrow_conns takes over peer_ni's ref */
1204         list_del(&conn->ksnc_list);
1205
1206         conn_cb = conn->ksnc_conn_cb;
1207         if (conn_cb != NULL) {
1208                 /* dissociate conn from cb... */
1209                 LASSERT(!conn_cb->ksnr_deleted);
1210
1211                 /* connected bit is set only if all connections
1212                  * of the given type got created
1213                  */
1214                 if (ksocknal_get_conn_count_by_type(conn_cb, conn->ksnc_type) ==
1215                     *ksocknal_tunables.ksnd_conns_per_peer)
1216                         LASSERT((conn_cb->ksnr_connected &
1217                                 BIT(conn->ksnc_type)) != 0);
1218
1219                 list_for_each_entry(conn2, &peer_ni->ksnp_conns, ksnc_list) {
1220                         if (conn2->ksnc_conn_cb == conn_cb &&
1221                             conn2->ksnc_type == conn->ksnc_type)
1222                                 goto conn2_found;
1223                 }
1224                 conn_cb->ksnr_connected &= ~BIT(conn->ksnc_type);
1225 conn2_found:
1226                 conn->ksnc_conn_cb = NULL;
1227
1228                 /* drop conn's ref on conn_cb */
1229                 ksocknal_conn_cb_decref(conn_cb);
1230         }
1231
1232         if (list_empty(&peer_ni->ksnp_conns)) {
1233                 /* No more connections to this peer_ni */
1234
1235                 if (!list_empty(&peer_ni->ksnp_tx_queue)) {
1236                         struct ksock_tx *tx;
1237
1238                         LASSERT(conn->ksnc_proto == &ksocknal_protocol_v3x);
1239
1240                         /* throw them to the last connection...,
1241                          * these TXs will be send to /dev/null by scheduler */
1242                         list_for_each_entry(tx, &peer_ni->ksnp_tx_queue,
1243                                             tx_list)
1244                                 ksocknal_tx_prep(conn, tx);
1245
1246                         spin_lock_bh(&conn->ksnc_scheduler->kss_lock);
1247                         list_splice_init(&peer_ni->ksnp_tx_queue,
1248                                          &conn->ksnc_tx_queue);
1249                         spin_unlock_bh(&conn->ksnc_scheduler->kss_lock);
1250                 }
1251
1252                 /* renegotiate protocol version */
1253                 peer_ni->ksnp_proto = NULL;
1254                 /* stash last conn close reason */
1255                 peer_ni->ksnp_error = error;
1256
1257                 if (peer_ni->ksnp_conn_cb == NULL) {
1258                         /* I've just closed last conn belonging to a
1259                          * peer_ni with no connections to it
1260                          */
1261                         ksocknal_unlink_peer_locked(peer_ni);
1262                 }
1263         }
1264
1265         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
1266
1267         list_add_tail(&conn->ksnc_list, &ksocknal_data.ksnd_deathrow_conns);
1268         wake_up(&ksocknal_data.ksnd_reaper_waitq);
1269
1270         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
1271 }
1272
1273 void
1274 ksocknal_peer_failed(struct ksock_peer_ni *peer_ni)
1275 {
1276         bool notify = false;
1277         time64_t last_alive = 0;
1278
1279         /* There has been a connection failure or comms error; but I'll only
1280          * tell LNET I think the peer_ni is dead if it's to another kernel and
1281          * there are no connections or connection attempts in existence. */
1282
1283         read_lock(&ksocknal_data.ksnd_global_lock);
1284
1285         if ((peer_ni->ksnp_id.pid & LNET_PID_USERFLAG) == 0 &&
1286              list_empty(&peer_ni->ksnp_conns) &&
1287              peer_ni->ksnp_accepting == 0 &&
1288              !ksocknal_find_connecting_conn_cb_locked(peer_ni)) {
1289                 notify = true;
1290                 last_alive = peer_ni->ksnp_last_alive;
1291         }
1292
1293         read_unlock(&ksocknal_data.ksnd_global_lock);
1294
1295         if (notify)
1296                 lnet_notify(peer_ni->ksnp_ni, peer_ni->ksnp_id.nid,
1297                             false, false, last_alive);
1298 }
1299
1300 void
1301 ksocknal_finalize_zcreq(struct ksock_conn *conn)
1302 {
1303         struct ksock_peer_ni *peer_ni = conn->ksnc_peer;
1304         struct ksock_tx *tx;
1305         struct ksock_tx *tmp;
1306         LIST_HEAD(zlist);
1307
1308         /* NB safe to finalize TXs because closing of socket will
1309          * abort all buffered data */
1310         LASSERT(conn->ksnc_sock == NULL);
1311
1312         spin_lock(&peer_ni->ksnp_lock);
1313
1314         list_for_each_entry_safe(tx, tmp, &peer_ni->ksnp_zc_req_list,
1315                                  tx_zc_list) {
1316                 if (tx->tx_conn != conn)
1317                         continue;
1318
1319                 LASSERT(tx->tx_msg.ksm_zc_cookies[0] != 0);
1320
1321                 tx->tx_msg.ksm_zc_cookies[0] = 0;
1322                 tx->tx_zc_aborted = 1;  /* mark it as not-acked */
1323                 list_move(&tx->tx_zc_list, &zlist);
1324         }
1325
1326         spin_unlock(&peer_ni->ksnp_lock);
1327
1328         while ((tx = list_first_entry_or_null(&zlist, struct ksock_tx,
1329                                               tx_zc_list)) != NULL) {
1330                 list_del(&tx->tx_zc_list);
1331                 ksocknal_tx_decref(tx);
1332         }
1333 }
1334
1335 void
1336 ksocknal_terminate_conn(struct ksock_conn *conn)
1337 {
1338         /* This gets called by the reaper (guaranteed thread context) to
1339          * disengage the socket from its callbacks and close it.
1340          * ksnc_refcount will eventually hit zero, and then the reaper will
1341          * destroy it.
1342          */
1343         struct ksock_peer_ni *peer_ni = conn->ksnc_peer;
1344         struct ksock_sched *sched = conn->ksnc_scheduler;
1345         bool failed = false;
1346
1347         LASSERT(conn->ksnc_closing);
1348
1349         /* wake up the scheduler to "send" all remaining packets to /dev/null */
1350         spin_lock_bh(&sched->kss_lock);
1351
1352         /* a closing conn is always ready to tx */
1353         conn->ksnc_tx_ready = 1;
1354
1355         if (!conn->ksnc_tx_scheduled &&
1356             !list_empty(&conn->ksnc_tx_queue)) {
1357                 list_add_tail(&conn->ksnc_tx_list,
1358                               &sched->kss_tx_conns);
1359                 conn->ksnc_tx_scheduled = 1;
1360                 /* extra ref for scheduler */
1361                 ksocknal_conn_addref(conn);
1362
1363                 wake_up(&sched->kss_waitq);
1364         }
1365
1366         spin_unlock_bh(&sched->kss_lock);
1367
1368         /* serialise with callbacks */
1369         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1370
1371         ksocknal_lib_reset_callback(conn->ksnc_sock, conn);
1372
1373         /* OK, so this conn may not be completely disengaged from its
1374          * scheduler yet, but it _has_ committed to terminate...
1375          */
1376         conn->ksnc_scheduler->kss_nconns--;
1377
1378         if (peer_ni->ksnp_error != 0) {
1379                 /* peer_ni's last conn closed in error */
1380                 LASSERT(list_empty(&peer_ni->ksnp_conns));
1381                 failed = true;
1382                 peer_ni->ksnp_error = 0;     /* avoid multiple notifications */
1383         }
1384
1385         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1386
1387         if (failed)
1388                 ksocknal_peer_failed(peer_ni);
1389
1390         /* The socket is closed on the final put; either here, or in
1391          * ksocknal_{send,recv}msg().  Since we set up the linger2 option
1392          * when the connection was established, this will close the socket
1393          * immediately, aborting anything buffered in it. Any hung
1394          * zero-copy transmits will therefore complete in finite time.
1395          */
1396         ksocknal_connsock_decref(conn);
1397 }
1398
1399 void
1400 ksocknal_queue_zombie_conn(struct ksock_conn *conn)
1401 {
1402         /* Queue the conn for the reaper to destroy */
1403         LASSERT(refcount_read(&conn->ksnc_conn_refcount) == 0);
1404         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
1405
1406         list_add_tail(&conn->ksnc_list, &ksocknal_data.ksnd_zombie_conns);
1407         wake_up(&ksocknal_data.ksnd_reaper_waitq);
1408
1409         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
1410 }
1411
1412 void
1413 ksocknal_destroy_conn(struct ksock_conn *conn)
1414 {
1415         time64_t last_rcv;
1416
1417         /* Final coup-de-grace of the reaper */
1418         CDEBUG(D_NET, "connection %p\n", conn);
1419
1420         LASSERT(refcount_read(&conn->ksnc_conn_refcount) == 0);
1421         LASSERT(refcount_read(&conn->ksnc_sock_refcount) == 0);
1422         LASSERT(conn->ksnc_sock == NULL);
1423         LASSERT(conn->ksnc_conn_cb == NULL);
1424         LASSERT(!conn->ksnc_tx_scheduled);
1425         LASSERT(!conn->ksnc_rx_scheduled);
1426         LASSERT(list_empty(&conn->ksnc_tx_queue));
1427
1428         /* complete current receive if any */
1429         switch (conn->ksnc_rx_state) {
1430         case SOCKNAL_RX_LNET_PAYLOAD:
1431                 last_rcv = conn->ksnc_rx_deadline -
1432                            ksocknal_timeout();
1433                 CERROR("Completing partial receive from %s[%d], ip %pISp, with error, wanted: %d, left: %d, last alive is %lld secs ago\n",
1434                        libcfs_id2str(conn->ksnc_peer->ksnp_id), conn->ksnc_type,
1435                        &conn->ksnc_peeraddr,
1436                        conn->ksnc_rx_nob_wanted, conn->ksnc_rx_nob_left,
1437                        ktime_get_seconds() - last_rcv);
1438                 if (conn->ksnc_lnet_msg)
1439                         conn->ksnc_lnet_msg->msg_health_status =
1440                                 LNET_MSG_STATUS_REMOTE_ERROR;
1441                 lnet_finalize(conn->ksnc_lnet_msg, -EIO);
1442                 break;
1443         case SOCKNAL_RX_LNET_HEADER:
1444                 if (conn->ksnc_rx_started)
1445                         CERROR("Incomplete receive of lnet header from %s, ip %pISp, with error, protocol: %d.x.\n",
1446                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1447                                &conn->ksnc_peeraddr,
1448                                conn->ksnc_proto->pro_version);
1449                 break;
1450         case SOCKNAL_RX_KSM_HEADER:
1451                 if (conn->ksnc_rx_started)
1452                         CERROR("Incomplete receive of ksock message from %s, ip %pISp, with error, protocol: %d.x.\n",
1453                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1454                                &conn->ksnc_peeraddr,
1455                                conn->ksnc_proto->pro_version);
1456                 break;
1457         case SOCKNAL_RX_SLOP:
1458                 if (conn->ksnc_rx_started)
1459                         CERROR("Incomplete receive of slops from %s, ip %pISp, with error\n",
1460                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1461                                &conn->ksnc_peeraddr);
1462                break;
1463         default:
1464                 LBUG ();
1465                 break;
1466         }
1467
1468         ksocknal_peer_decref(conn->ksnc_peer);
1469
1470         LIBCFS_FREE (conn, sizeof (*conn));
1471 }
1472
1473 int
1474 ksocknal_close_peer_conns_locked(struct ksock_peer_ni *peer_ni,
1475                                  struct sockaddr *addr, int why)
1476 {
1477         struct ksock_conn *conn;
1478         struct ksock_conn *cnxt;
1479         int count = 0;
1480
1481         list_for_each_entry_safe(conn, cnxt, &peer_ni->ksnp_conns, ksnc_list) {
1482                 if (!addr ||
1483                     rpc_cmp_addr(addr,
1484                                  (struct sockaddr *)&conn->ksnc_peeraddr)) {
1485                         count++;
1486                         ksocknal_close_conn_locked(conn, why);
1487                 }
1488         }
1489
1490         return count;
1491 }
1492
1493 int
1494 ksocknal_close_conn_and_siblings(struct ksock_conn *conn, int why)
1495 {
1496         struct ksock_peer_ni *peer_ni = conn->ksnc_peer;
1497         int count;
1498
1499         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1500
1501         count = ksocknal_close_peer_conns_locked(
1502                 peer_ni, (struct sockaddr *)&conn->ksnc_peeraddr, why);
1503
1504         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1505
1506         return count;
1507 }
1508
1509 int
1510 ksocknal_close_matching_conns(struct lnet_process_id id, __u32 ipaddr)
1511 {
1512         struct ksock_peer_ni *peer_ni;
1513         struct hlist_node *pnxt;
1514         int lo;
1515         int hi;
1516         int i;
1517         int count = 0;
1518         struct sockaddr_in sa = {.sin_family = AF_INET};
1519
1520         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1521
1522         if (id.nid != LNET_NID_ANY) {
1523                 lo = hash_min(id.nid, HASH_BITS(ksocknal_data.ksnd_peers));
1524                 hi = lo;
1525         } else {
1526                 lo = 0;
1527                 hi = HASH_SIZE(ksocknal_data.ksnd_peers) - 1;
1528         }
1529
1530         sa.sin_addr.s_addr = htonl(ipaddr);
1531         for (i = lo; i <= hi; i++) {
1532                 hlist_for_each_entry_safe(peer_ni, pnxt,
1533                                           &ksocknal_data.ksnd_peers[i],
1534                                           ksnp_list) {
1535
1536                         if (!((id.nid == LNET_NID_ANY ||
1537                                id.nid == peer_ni->ksnp_id.nid) &&
1538                               (id.pid == LNET_PID_ANY ||
1539                                id.pid == peer_ni->ksnp_id.pid)))
1540                                 continue;
1541
1542                         count += ksocknal_close_peer_conns_locked(
1543                                 peer_ni,
1544                                 ipaddr ? (struct sockaddr *)&sa : NULL, 0);
1545                 }
1546         }
1547
1548         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1549
1550         /* wildcards always succeed */
1551         if (id.nid == LNET_NID_ANY || id.pid == LNET_PID_ANY || ipaddr == 0)
1552                 return 0;
1553
1554         return (count == 0 ? -ENOENT : 0);
1555 }
1556
1557 void
1558 ksocknal_notify_gw_down(lnet_nid_t gw_nid)
1559 {
1560         /* The router is telling me she's been notified of a change in
1561          * gateway state....
1562          */
1563         struct lnet_process_id id = {
1564                 .nid    = gw_nid,
1565                 .pid    = LNET_PID_ANY,
1566         };
1567
1568         CDEBUG(D_NET, "gw %s down\n", libcfs_nid2str(gw_nid));
1569
1570         /* If the gateway crashed, close all open connections... */
1571         ksocknal_close_matching_conns(id, 0);
1572         return;
1573
1574         /* We can only establish new connections
1575          * if we have autroutes, and these connect on demand. */
1576 }
1577
1578 static void
1579 ksocknal_push_peer(struct ksock_peer_ni *peer_ni)
1580 {
1581         int index;
1582         int i;
1583         struct ksock_conn *conn;
1584
1585         for (index = 0; ; index++) {
1586                 read_lock(&ksocknal_data.ksnd_global_lock);
1587
1588                 i = 0;
1589                 conn = NULL;
1590
1591                 list_for_each_entry(conn, &peer_ni->ksnp_conns, ksnc_list) {
1592                         if (i++ == index) {
1593                                 ksocknal_conn_addref(conn);
1594                                 break;
1595                         }
1596                 }
1597
1598                 read_unlock(&ksocknal_data.ksnd_global_lock);
1599
1600                 if (i <= index)
1601                         break;
1602
1603                 ksocknal_lib_push_conn (conn);
1604                 ksocknal_conn_decref(conn);
1605         }
1606 }
1607
1608 static int
1609 ksocknal_push(struct lnet_ni *ni, struct lnet_process_id id)
1610 {
1611         int lo;
1612         int hi;
1613         int bkt;
1614         int rc = -ENOENT;
1615
1616         if (id.nid != LNET_NID_ANY) {
1617                 lo = hash_min(id.nid, HASH_BITS(ksocknal_data.ksnd_peers));
1618                 hi = lo;
1619         } else {
1620                 lo = 0;
1621                 hi = HASH_SIZE(ksocknal_data.ksnd_peers) - 1;
1622         }
1623
1624         for (bkt = lo; bkt <= hi; bkt++) {
1625                 int peer_off; /* searching offset in peer_ni hash table */
1626
1627                 for (peer_off = 0; ; peer_off++) {
1628                         struct ksock_peer_ni *peer_ni;
1629                         int           i = 0;
1630
1631                         read_lock(&ksocknal_data.ksnd_global_lock);
1632                         hlist_for_each_entry(peer_ni,
1633                                              &ksocknal_data.ksnd_peers[bkt],
1634                                              ksnp_list) {
1635                                 if (!((id.nid == LNET_NID_ANY ||
1636                                        id.nid == peer_ni->ksnp_id.nid) &&
1637                                       (id.pid == LNET_PID_ANY ||
1638                                        id.pid == peer_ni->ksnp_id.pid)))
1639                                         continue;
1640
1641                                 if (i++ == peer_off) {
1642                                         ksocknal_peer_addref(peer_ni);
1643                                         break;
1644                                 }
1645                         }
1646                         read_unlock(&ksocknal_data.ksnd_global_lock);
1647
1648                         if (i <= peer_off) /* no match */
1649                                 break;
1650
1651                         rc = 0;
1652                         ksocknal_push_peer(peer_ni);
1653                         ksocknal_peer_decref(peer_ni);
1654                 }
1655         }
1656         return rc;
1657 }
1658
1659 int
1660 ksocknal_ctl(struct lnet_ni *ni, unsigned int cmd, void *arg)
1661 {
1662         struct lnet_process_id id = {0};
1663         struct libcfs_ioctl_data *data = arg;
1664         int rc;
1665
1666         switch(cmd) {
1667         case IOC_LIBCFS_GET_INTERFACE: {
1668                 struct ksock_net *net = ni->ni_data;
1669                 struct ksock_interface *iface;
1670                 struct sockaddr_in *sa;
1671
1672                 read_lock(&ksocknal_data.ksnd_global_lock);
1673
1674                 if (data->ioc_count >= 1) {
1675                         rc = -ENOENT;
1676                 } else {
1677                         rc = 0;
1678                         iface = &net->ksnn_interface;
1679
1680                         sa = (void *)&iface->ksni_addr;
1681                         if (sa->sin_family == AF_INET)
1682                                 data->ioc_u32[0] = ntohl(sa->sin_addr.s_addr);
1683                         else
1684                                 data->ioc_u32[0] = 0xFFFFFFFF;
1685                         data->ioc_u32[1] = iface->ksni_netmask;
1686                         data->ioc_u32[2] = iface->ksni_npeers;
1687                         data->ioc_u32[3] = iface->ksni_nroutes;
1688                 }
1689
1690                 read_unlock(&ksocknal_data.ksnd_global_lock);
1691                 return rc;
1692         }
1693
1694         case IOC_LIBCFS_GET_PEER: {
1695                 __u32            myip = 0;
1696                 __u32            ip = 0;
1697                 int              port = 0;
1698                 int              conn_count = 0;
1699                 int              share_count = 0;
1700
1701                 rc = ksocknal_get_peer_info(ni, data->ioc_count,
1702                                             &id, &myip, &ip, &port,
1703                                             &conn_count,  &share_count);
1704                 if (rc != 0)
1705                         return rc;
1706
1707                 data->ioc_nid    = id.nid;
1708                 data->ioc_count  = share_count;
1709                 data->ioc_u32[0] = ip;
1710                 data->ioc_u32[1] = port;
1711                 data->ioc_u32[2] = myip;
1712                 data->ioc_u32[3] = conn_count;
1713                 data->ioc_u32[4] = id.pid;
1714                 return 0;
1715         }
1716
1717         case IOC_LIBCFS_ADD_PEER: {
1718                 struct sockaddr_in sa = {.sin_family = AF_INET};
1719
1720                 id.nid = data->ioc_nid;
1721                 id.pid = LNET_PID_LUSTRE;
1722                 sa.sin_addr.s_addr = htonl(data->ioc_u32[0]);
1723                 sa.sin_port = htons(data->ioc_u32[1]);
1724                 return ksocknal_add_peer(ni, id, (struct sockaddr *)&sa);
1725         }
1726         case IOC_LIBCFS_DEL_PEER:
1727                 id.nid = data->ioc_nid;
1728                 id.pid = LNET_PID_ANY;
1729                 return ksocknal_del_peer (ni, id,
1730                                           data->ioc_u32[0]); /* IP */
1731
1732         case IOC_LIBCFS_GET_CONN: {
1733                 int           txmem;
1734                 int           rxmem;
1735                 int           nagle;
1736                 struct ksock_conn *conn = ksocknal_get_conn_by_idx(ni, data->ioc_count);
1737                 struct sockaddr_in *psa = (void *)&conn->ksnc_peeraddr;
1738                 struct sockaddr_in *mysa = (void *)&conn->ksnc_myaddr;
1739
1740                 if (conn == NULL)
1741                         return -ENOENT;
1742
1743                 ksocknal_lib_get_conn_tunables(conn, &txmem, &rxmem, &nagle);
1744
1745                 data->ioc_count  = txmem;
1746                 data->ioc_nid    = conn->ksnc_peer->ksnp_id.nid;
1747                 data->ioc_flags  = nagle;
1748                 if (psa->sin_family == AF_INET)
1749                         data->ioc_u32[0] = ntohl(psa->sin_addr.s_addr);
1750                 else
1751                         data->ioc_u32[0] = 0xFFFFFFFF;
1752                 data->ioc_u32[1] = rpc_get_port((struct sockaddr *)
1753                                                 &conn->ksnc_peeraddr);
1754                 if (mysa->sin_family == AF_INET)
1755                         data->ioc_u32[2] = ntohl(mysa->sin_addr.s_addr);
1756                 else
1757                         data->ioc_u32[2] = 0xFFFFFFFF;
1758                 data->ioc_u32[3] = conn->ksnc_type;
1759                 data->ioc_u32[4] = conn->ksnc_scheduler->kss_cpt;
1760                 data->ioc_u32[5] = rxmem;
1761                 data->ioc_u32[6] = conn->ksnc_peer->ksnp_id.pid;
1762                 ksocknal_conn_decref(conn);
1763                 return 0;
1764         }
1765
1766         case IOC_LIBCFS_CLOSE_CONNECTION:
1767                 id.nid = data->ioc_nid;
1768                 id.pid = LNET_PID_ANY;
1769                 return ksocknal_close_matching_conns (id,
1770                                                       data->ioc_u32[0]);
1771
1772         case IOC_LIBCFS_REGISTER_MYNID:
1773                 /* Ignore if this is a noop */
1774                 if (data->ioc_nid == ni->ni_nid)
1775                         return 0;
1776
1777                 CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID: %s(%s)\n",
1778                        libcfs_nid2str(data->ioc_nid),
1779                        libcfs_nid2str(ni->ni_nid));
1780                 return -EINVAL;
1781
1782         case IOC_LIBCFS_PUSH_CONNECTION:
1783                 id.nid = data->ioc_nid;
1784                 id.pid = LNET_PID_ANY;
1785                 return ksocknal_push(ni, id);
1786
1787         default:
1788                 return -EINVAL;
1789         }
1790         /* not reached */
1791 }
1792
1793 static void
1794 ksocknal_free_buffers (void)
1795 {
1796         LASSERT (atomic_read(&ksocknal_data.ksnd_nactive_txs) == 0);
1797
1798         if (ksocknal_data.ksnd_schedulers != NULL)
1799                 cfs_percpt_free(ksocknal_data.ksnd_schedulers);
1800
1801         spin_lock(&ksocknal_data.ksnd_tx_lock);
1802
1803         if (!list_empty(&ksocknal_data.ksnd_idle_noop_txs)) {
1804                 LIST_HEAD(zlist);
1805                 struct ksock_tx *tx;
1806
1807                 list_splice_init(&ksocknal_data.ksnd_idle_noop_txs, &zlist);
1808                 spin_unlock(&ksocknal_data.ksnd_tx_lock);
1809
1810                 while ((tx = list_first_entry_or_null(&zlist, struct ksock_tx,
1811                                                       tx_list)) != NULL) {
1812                         list_del(&tx->tx_list);
1813                         LIBCFS_FREE(tx, tx->tx_desc_size);
1814                 }
1815         } else {
1816                 spin_unlock(&ksocknal_data.ksnd_tx_lock);
1817         }
1818 }
1819
1820 static int ksocknal_get_link_status(struct net_device *dev)
1821 {
1822         int ret = -1;
1823
1824         LASSERT(dev);
1825
1826         if (!netif_running(dev))
1827                 ret = 0;
1828         /* Some devices may not be providing link settings */
1829         else if (dev->ethtool_ops->get_link)
1830                 ret = dev->ethtool_ops->get_link(dev);
1831
1832         return ret;
1833 }
1834
1835 static int
1836 ksocknal_handle_link_state_change(struct net_device *dev,
1837                                   unsigned char operstate)
1838 {
1839         struct lnet_ni *ni;
1840         struct ksock_net *net;
1841         struct ksock_net *cnxt;
1842         int ifindex;
1843         unsigned char link_down = !(operstate == IF_OPER_UP);
1844
1845         ifindex = dev->ifindex;
1846
1847         if (!ksocknal_data.ksnd_nnets)
1848                 goto out;
1849
1850         list_for_each_entry_safe(net, cnxt, &ksocknal_data.ksnd_nets,
1851                                  ksnn_list) {
1852                 if (net->ksnn_interface.ksni_index != ifindex)
1853                         continue;
1854                 ni = net->ksnn_ni;
1855                 if (link_down)
1856                         atomic_set(&ni->ni_fatal_error_on, link_down);
1857                 else
1858                         atomic_set(&ni->ni_fatal_error_on,
1859                                    (ksocknal_get_link_status(dev) == 0));
1860         }
1861 out:
1862         return 0;
1863 }
1864
1865
1866 /************************************
1867  * Net device notifier event handler
1868  ************************************/
1869 static int ksocknal_device_event(struct notifier_block *unused,
1870                                  unsigned long event, void *ptr)
1871 {
1872         struct net_device *dev = netdev_notifier_info_to_dev(ptr);
1873         unsigned char operstate;
1874
1875         operstate = dev->operstate;
1876
1877         switch (event) {
1878         case NETDEV_UP:
1879         case NETDEV_DOWN:
1880         case NETDEV_CHANGE:
1881                 ksocknal_handle_link_state_change(dev, operstate);
1882                 break;
1883         }
1884
1885         return NOTIFY_OK;
1886 }
1887
1888 static struct notifier_block ksocknal_notifier_block = {
1889         .notifier_call = ksocknal_device_event,
1890 };
1891
1892 static void
1893 ksocknal_base_shutdown(void)
1894 {
1895         struct ksock_sched *sched;
1896         struct ksock_peer_ni *peer_ni;
1897         int i;
1898
1899         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %lld\n",
1900                libcfs_kmem_read());
1901         LASSERT (ksocknal_data.ksnd_nnets == 0);
1902
1903         if (ksocknal_data.ksnd_init == SOCKNAL_INIT_ALL)
1904                 unregister_netdevice_notifier(&ksocknal_notifier_block);
1905
1906         switch (ksocknal_data.ksnd_init) {
1907         default:
1908                 LASSERT(0);
1909                 /* fallthrough */
1910
1911         case SOCKNAL_INIT_ALL:
1912         case SOCKNAL_INIT_DATA:
1913                 hash_for_each(ksocknal_data.ksnd_peers, i, peer_ni, ksnp_list)
1914                         LASSERT(0);
1915
1916                 LASSERT(list_empty(&ksocknal_data.ksnd_nets));
1917                 LASSERT(list_empty(&ksocknal_data.ksnd_enomem_conns));
1918                 LASSERT(list_empty(&ksocknal_data.ksnd_zombie_conns));
1919                 LASSERT(list_empty(&ksocknal_data.ksnd_connd_connreqs));
1920                 LASSERT(list_empty(&ksocknal_data.ksnd_connd_routes));
1921
1922                 if (ksocknal_data.ksnd_schedulers != NULL) {
1923                         cfs_percpt_for_each(sched, i,
1924                                             ksocknal_data.ksnd_schedulers) {
1925
1926                                 LASSERT(list_empty(&sched->kss_tx_conns));
1927                                 LASSERT(list_empty(&sched->kss_rx_conns));
1928                                 LASSERT(list_empty(&sched->kss_zombie_noop_txs));
1929                                 LASSERT(sched->kss_nconns == 0);
1930                         }
1931                 }
1932
1933                 /* flag threads to terminate; wake and wait for them to die */
1934                 ksocknal_data.ksnd_shuttingdown = 1;
1935                 wake_up_all(&ksocknal_data.ksnd_connd_waitq);
1936                 wake_up(&ksocknal_data.ksnd_reaper_waitq);
1937
1938                 if (ksocknal_data.ksnd_schedulers != NULL) {
1939                         cfs_percpt_for_each(sched, i,
1940                                             ksocknal_data.ksnd_schedulers)
1941                                         wake_up_all(&sched->kss_waitq);
1942                 }
1943
1944                 wait_var_event_warning(&ksocknal_data.ksnd_nthreads,
1945                                        atomic_read(&ksocknal_data.ksnd_nthreads) == 0,
1946                                        "waiting for %d threads to terminate\n",
1947                                        atomic_read(&ksocknal_data.ksnd_nthreads));
1948
1949                 ksocknal_free_buffers();
1950
1951                 ksocknal_data.ksnd_init = SOCKNAL_INIT_NOTHING;
1952                 break;
1953         }
1954
1955         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %lld\n",
1956                libcfs_kmem_read());
1957
1958         module_put(THIS_MODULE);
1959 }
1960
1961 static int
1962 ksocknal_base_startup(void)
1963 {
1964         struct ksock_sched *sched;
1965         int rc;
1966         int i;
1967
1968         LASSERT(ksocknal_data.ksnd_init == SOCKNAL_INIT_NOTHING);
1969         LASSERT(ksocknal_data.ksnd_nnets == 0);
1970
1971         memset(&ksocknal_data, 0, sizeof(ksocknal_data)); /* zero pointers */
1972
1973         hash_init(ksocknal_data.ksnd_peers);
1974
1975         rwlock_init(&ksocknal_data.ksnd_global_lock);
1976         INIT_LIST_HEAD(&ksocknal_data.ksnd_nets);
1977
1978         spin_lock_init(&ksocknal_data.ksnd_reaper_lock);
1979         INIT_LIST_HEAD(&ksocknal_data.ksnd_enomem_conns);
1980         INIT_LIST_HEAD(&ksocknal_data.ksnd_zombie_conns);
1981         INIT_LIST_HEAD(&ksocknal_data.ksnd_deathrow_conns);
1982         init_waitqueue_head(&ksocknal_data.ksnd_reaper_waitq);
1983
1984         spin_lock_init(&ksocknal_data.ksnd_connd_lock);
1985         INIT_LIST_HEAD(&ksocknal_data.ksnd_connd_connreqs);
1986         INIT_LIST_HEAD(&ksocknal_data.ksnd_connd_routes);
1987         init_waitqueue_head(&ksocknal_data.ksnd_connd_waitq);
1988
1989         spin_lock_init(&ksocknal_data.ksnd_tx_lock);
1990         INIT_LIST_HEAD(&ksocknal_data.ksnd_idle_noop_txs);
1991
1992         /* NB memset above zeros whole of ksocknal_data */
1993
1994         /* flag lists/ptrs/locks initialised */
1995         ksocknal_data.ksnd_init = SOCKNAL_INIT_DATA;
1996         if (!try_module_get(THIS_MODULE))
1997                 goto failed;
1998
1999         /* Create a scheduler block per available CPT */
2000         ksocknal_data.ksnd_schedulers = cfs_percpt_alloc(lnet_cpt_table(),
2001                                                          sizeof(*sched));
2002         if (ksocknal_data.ksnd_schedulers == NULL)
2003                 goto failed;
2004
2005         cfs_percpt_for_each(sched, i, ksocknal_data.ksnd_schedulers) {
2006                 int nthrs;
2007
2008                 /*
2009                  * make sure not to allocate more threads than there are
2010                  * cores/CPUs in teh CPT
2011                  */
2012                 nthrs = cfs_cpt_weight(lnet_cpt_table(), i);
2013                 if (*ksocknal_tunables.ksnd_nscheds > 0) {
2014                         nthrs = min(nthrs, *ksocknal_tunables.ksnd_nscheds);
2015                 } else {
2016                         /*
2017                          * max to half of CPUs, assume another half should be
2018                          * reserved for upper layer modules
2019                          */
2020                         nthrs = min(max(SOCKNAL_NSCHEDS, nthrs >> 1), nthrs);
2021                 }
2022
2023                 sched->kss_nthreads_max = nthrs;
2024                 sched->kss_cpt = i;
2025
2026                 spin_lock_init(&sched->kss_lock);
2027                 INIT_LIST_HEAD(&sched->kss_rx_conns);
2028                 INIT_LIST_HEAD(&sched->kss_tx_conns);
2029                 INIT_LIST_HEAD(&sched->kss_zombie_noop_txs);
2030                 init_waitqueue_head(&sched->kss_waitq);
2031         }
2032
2033         ksocknal_data.ksnd_connd_starting         = 0;
2034         ksocknal_data.ksnd_connd_failed_stamp     = 0;
2035         ksocknal_data.ksnd_connd_starting_stamp   = ktime_get_real_seconds();
2036         /* must have at least 2 connds to remain responsive to accepts while
2037          * connecting */
2038         if (*ksocknal_tunables.ksnd_nconnds < SOCKNAL_CONND_RESV + 1)
2039                 *ksocknal_tunables.ksnd_nconnds = SOCKNAL_CONND_RESV + 1;
2040
2041         if (*ksocknal_tunables.ksnd_nconnds_max <
2042             *ksocknal_tunables.ksnd_nconnds) {
2043                 ksocknal_tunables.ksnd_nconnds_max =
2044                         ksocknal_tunables.ksnd_nconnds;
2045         }
2046
2047         for (i = 0; i < *ksocknal_tunables.ksnd_nconnds; i++) {
2048                 spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
2049                 ksocknal_data.ksnd_connd_starting++;
2050                 spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
2051
2052                 rc = ksocknal_thread_start(ksocknal_connd,
2053                                            (void *)((uintptr_t)i),
2054                                            "socknal_cd%02d", i);
2055                 if (rc != 0) {
2056                         spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
2057                         ksocknal_data.ksnd_connd_starting--;
2058                         spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
2059                         CERROR("Can't spawn socknal connd: %d\n", rc);
2060                         goto failed;
2061                 }
2062         }
2063
2064         rc = ksocknal_thread_start(ksocknal_reaper, NULL, "socknal_reaper");
2065         if (rc != 0) {
2066                 CERROR ("Can't spawn socknal reaper: %d\n", rc);
2067                 goto failed;
2068         }
2069
2070         register_netdevice_notifier(&ksocknal_notifier_block);
2071
2072         /* flag everything initialised */
2073         ksocknal_data.ksnd_init = SOCKNAL_INIT_ALL;
2074
2075         return 0;
2076
2077  failed:
2078         ksocknal_base_shutdown();
2079         return -ENETDOWN;
2080 }
2081
2082 static int
2083 ksocknal_debug_peerhash(struct lnet_ni *ni)
2084 {
2085         struct ksock_peer_ni *peer_ni;
2086         int i;
2087
2088         read_lock(&ksocknal_data.ksnd_global_lock);
2089
2090         hash_for_each(ksocknal_data.ksnd_peers, i, peer_ni, ksnp_list) {
2091                 struct ksock_conn_cb *conn_cb;
2092                 struct ksock_conn *conn;
2093
2094                 if (peer_ni->ksnp_ni != ni)
2095                         continue;
2096
2097                 CWARN("Active peer_ni on shutdown: %s, ref %d, "
2098                       "closing %d, accepting %d, err %d, zcookie %llu, "
2099                       "txq %d, zc_req %d\n", libcfs_id2str(peer_ni->ksnp_id),
2100                       refcount_read(&peer_ni->ksnp_refcount),
2101                       peer_ni->ksnp_closing,
2102                       peer_ni->ksnp_accepting, peer_ni->ksnp_error,
2103                       peer_ni->ksnp_zc_next_cookie,
2104                       !list_empty(&peer_ni->ksnp_tx_queue),
2105                       !list_empty(&peer_ni->ksnp_zc_req_list));
2106
2107                 conn_cb = peer_ni->ksnp_conn_cb;
2108                 if (conn_cb) {
2109                         CWARN("ConnCB: ref %d, schd %d, conn %d, cnted %d, del %d\n",
2110                               refcount_read(&conn_cb->ksnr_refcount),
2111                               conn_cb->ksnr_scheduled, conn_cb->ksnr_connecting,
2112                               conn_cb->ksnr_connected, conn_cb->ksnr_deleted);
2113                 }
2114
2115                 list_for_each_entry(conn, &peer_ni->ksnp_conns, ksnc_list) {
2116                         CWARN("Conn: ref %d, sref %d, t %d, c %d\n",
2117                               refcount_read(&conn->ksnc_conn_refcount),
2118                               refcount_read(&conn->ksnc_sock_refcount),
2119                               conn->ksnc_type, conn->ksnc_closing);
2120                 }
2121                 break;
2122         }
2123
2124         read_unlock(&ksocknal_data.ksnd_global_lock);
2125         return 0;
2126 }
2127
2128 void
2129 ksocknal_shutdown(struct lnet_ni *ni)
2130 {
2131         struct ksock_net *net = ni->ni_data;
2132         struct lnet_process_id anyid = {
2133                 .nid = LNET_NID_ANY,
2134                 .pid = LNET_PID_ANY,
2135         };
2136
2137         LASSERT(ksocknal_data.ksnd_init == SOCKNAL_INIT_ALL);
2138         LASSERT(ksocknal_data.ksnd_nnets > 0);
2139
2140         /* prevent new peers */
2141         atomic_add(SOCKNAL_SHUTDOWN_BIAS, &net->ksnn_npeers);
2142
2143         /* Delete all peers */
2144         ksocknal_del_peer(ni, anyid, 0);
2145
2146         /* Wait for all peer_ni state to clean up */
2147         wait_var_event_warning(&net->ksnn_npeers,
2148                                atomic_read(&net->ksnn_npeers) ==
2149                                SOCKNAL_SHUTDOWN_BIAS,
2150                                "waiting for %d peers to disconnect\n",
2151                                ksocknal_debug_peerhash(ni) +
2152                                atomic_read(&net->ksnn_npeers) -
2153                                SOCKNAL_SHUTDOWN_BIAS);
2154
2155         LASSERT(net->ksnn_interface.ksni_npeers == 0);
2156         LASSERT(net->ksnn_interface.ksni_nroutes == 0);
2157
2158         list_del(&net->ksnn_list);
2159         LIBCFS_FREE(net, sizeof(*net));
2160
2161         ksocknal_data.ksnd_nnets--;
2162         if (ksocknal_data.ksnd_nnets == 0)
2163                 ksocknal_base_shutdown();
2164 }
2165
2166 static int
2167 ksocknal_search_new_ipif(struct ksock_net *net)
2168 {
2169         int new_ipif = 0;
2170         char *ifnam = &net->ksnn_interface.ksni_name[0];
2171         char *colon = strchr(ifnam, ':');
2172         bool found = false;
2173         struct ksock_net *tmp;
2174
2175         if (colon != NULL)
2176                 *colon = 0;
2177
2178         list_for_each_entry(tmp, &ksocknal_data.ksnd_nets, ksnn_list) {
2179                 char *ifnam2 = &tmp->ksnn_interface.ksni_name[0];
2180                 char *colon2 = strchr(ifnam2, ':');
2181
2182                 if (colon2 != NULL)
2183                         *colon2 = 0;
2184
2185                 found = strcmp(ifnam, ifnam2) == 0;
2186                 if (colon2 != NULL)
2187                         *colon2 = ':';
2188         }
2189
2190         new_ipif += !found;
2191         if (colon != NULL)
2192                 *colon = ':';
2193
2194         return new_ipif;
2195 }
2196
2197 static int
2198 ksocknal_start_schedulers(struct ksock_sched *sched)
2199 {
2200         int     nthrs;
2201         int     rc = 0;
2202         int     i;
2203
2204         if (sched->kss_nthreads == 0) {
2205                 if (*ksocknal_tunables.ksnd_nscheds > 0) {
2206                         nthrs = sched->kss_nthreads_max;
2207                 } else {
2208                         nthrs = cfs_cpt_weight(lnet_cpt_table(),
2209                                                sched->kss_cpt);
2210                         nthrs = min(max(SOCKNAL_NSCHEDS, nthrs >> 1), nthrs);
2211                         nthrs = min(SOCKNAL_NSCHEDS_HIGH, nthrs);
2212                 }
2213                 nthrs = min(nthrs, sched->kss_nthreads_max);
2214         } else {
2215                 LASSERT(sched->kss_nthreads <= sched->kss_nthreads_max);
2216                 /* increase two threads if there is new interface */
2217                 nthrs = min(2, sched->kss_nthreads_max - sched->kss_nthreads);
2218         }
2219
2220         for (i = 0; i < nthrs; i++) {
2221                 long id;
2222
2223                 id = KSOCK_THREAD_ID(sched->kss_cpt, sched->kss_nthreads + i);
2224                 rc = ksocknal_thread_start(ksocknal_scheduler, (void *)id,
2225                                            "socknal_sd%02d_%02d",
2226                                            sched->kss_cpt,
2227                                            (int)KSOCK_THREAD_SID(id));
2228                 if (rc == 0)
2229                         continue;
2230
2231                 CERROR("Can't spawn thread %d for scheduler[%d]: %d\n",
2232                        sched->kss_cpt, (int) KSOCK_THREAD_SID(id), rc);
2233                 break;
2234         }
2235
2236         sched->kss_nthreads += i;
2237         return rc;
2238 }
2239
2240 static int
2241 ksocknal_net_start_threads(struct ksock_net *net, __u32 *cpts, int ncpts)
2242 {
2243         int newif = ksocknal_search_new_ipif(net);
2244         int rc;
2245         int i;
2246
2247         if (ncpts > 0 && ncpts > cfs_cpt_number(lnet_cpt_table()))
2248                 return -EINVAL;
2249
2250         for (i = 0; i < ncpts; i++) {
2251                 struct ksock_sched *sched;
2252                 int cpt = (cpts == NULL) ? i : cpts[i];
2253
2254                 LASSERT(cpt < cfs_cpt_number(lnet_cpt_table()));
2255                 sched = ksocknal_data.ksnd_schedulers[cpt];
2256
2257                 if (!newif && sched->kss_nthreads > 0)
2258                         continue;
2259
2260                 rc = ksocknal_start_schedulers(sched);
2261                 if (rc != 0)
2262                         return rc;
2263         }
2264         return 0;
2265 }
2266
2267 int
2268 ksocknal_startup(struct lnet_ni *ni)
2269 {
2270         struct ksock_net *net;
2271         struct lnet_ioctl_config_lnd_cmn_tunables *net_tunables;
2272         struct ksock_interface *ksi = NULL;
2273         struct lnet_inetdev *ifaces = NULL;
2274         struct sockaddr_in *sa;
2275         int i = 0;
2276         int rc;
2277
2278         LASSERT (ni->ni_net->net_lnd == &the_ksocklnd);
2279         if (ksocknal_data.ksnd_init == SOCKNAL_INIT_NOTHING) {
2280                 rc = ksocknal_base_startup();
2281                 if (rc != 0)
2282                         return rc;
2283         }
2284         LIBCFS_ALLOC(net, sizeof(*net));
2285         if (net == NULL)
2286                 goto fail_0;
2287         net->ksnn_incarnation = ktime_get_real_ns();
2288         ni->ni_data = net;
2289         net_tunables = &ni->ni_net->net_tunables;
2290         if (net_tunables->lct_peer_timeout == -1)
2291                 net_tunables->lct_peer_timeout =
2292                         *ksocknal_tunables.ksnd_peertimeout;
2293
2294         if (net_tunables->lct_max_tx_credits == -1)
2295                 net_tunables->lct_max_tx_credits =
2296                         *ksocknal_tunables.ksnd_credits;
2297
2298         if (net_tunables->lct_peer_tx_credits == -1)
2299                 net_tunables->lct_peer_tx_credits =
2300                         *ksocknal_tunables.ksnd_peertxcredits;
2301
2302         if (net_tunables->lct_peer_tx_credits >
2303             net_tunables->lct_max_tx_credits)
2304                 net_tunables->lct_peer_tx_credits =
2305                         net_tunables->lct_max_tx_credits;
2306
2307         if (net_tunables->lct_peer_rtr_credits == -1)
2308                 net_tunables->lct_peer_rtr_credits =
2309                         *ksocknal_tunables.ksnd_peerrtrcredits;
2310
2311         rc = lnet_inet_enumerate(&ifaces, ni->ni_net_ns);
2312         if (rc < 0)
2313                 goto fail_1;
2314
2315         ksi = &net->ksnn_interface;
2316
2317         /* Use the first discovered interface or look in the list */
2318         if (ni->ni_interface) {
2319                 for (i = 0; i < rc; i++)
2320                         if (strcmp(ifaces[i].li_name, ni->ni_interface) == 0)
2321                                 break;
2322
2323                 /* ni_interfaces doesn't contain the interface we want */
2324                 if (i == rc) {
2325                         CERROR("ksocklnd: failed to find interface %s\n",
2326                                ni->ni_interface);
2327                         goto fail_1;
2328                 }
2329         }
2330
2331         ni->ni_dev_cpt = ifaces[i].li_cpt;
2332         sa = (void *)&ksi->ksni_addr;
2333         memset(sa, 0, sizeof(*sa));
2334         sa->sin_family = AF_INET;
2335         sa->sin_addr.s_addr = htonl(ifaces[i].li_ipaddr);
2336         ksi->ksni_index = ksocknal_ip2index((struct sockaddr *)sa, ni);
2337         ksi->ksni_netmask = ifaces[i].li_netmask;
2338         strlcpy(ksi->ksni_name, ifaces[i].li_name, sizeof(ksi->ksni_name));
2339
2340         /* call it before add it to ksocknal_data.ksnd_nets */
2341         rc = ksocknal_net_start_threads(net, ni->ni_cpts, ni->ni_ncpts);
2342         if (rc != 0)
2343                 goto fail_1;
2344
2345         LASSERT(ksi);
2346         LASSERT(ksi->ksni_addr.ss_family == AF_INET);
2347         ni->ni_nid = LNET_MKNID(
2348                 LNET_NIDNET(ni->ni_nid),
2349                 ntohl(((struct sockaddr_in *)
2350                        &ksi->ksni_addr)->sin_addr.s_addr));
2351         list_add(&net->ksnn_list, &ksocknal_data.ksnd_nets);
2352         net->ksnn_ni = ni;
2353         ksocknal_data.ksnd_nnets++;
2354
2355         return 0;
2356
2357 fail_1:
2358         LIBCFS_FREE(net, sizeof(*net));
2359 fail_0:
2360         if (ksocknal_data.ksnd_nnets == 0)
2361                 ksocknal_base_shutdown();
2362
2363         return -ENETDOWN;
2364 }
2365
2366
2367 static void __exit ksocklnd_exit(void)
2368 {
2369         lnet_unregister_lnd(&the_ksocklnd);
2370 }
2371
2372 static const struct lnet_lnd the_ksocklnd = {
2373         .lnd_type               = SOCKLND,
2374         .lnd_startup            = ksocknal_startup,
2375         .lnd_shutdown           = ksocknal_shutdown,
2376         .lnd_ctl                = ksocknal_ctl,
2377         .lnd_send               = ksocknal_send,
2378         .lnd_recv               = ksocknal_recv,
2379         .lnd_notify_peer_down   = ksocknal_notify_gw_down,
2380         .lnd_accept             = ksocknal_accept,
2381 };
2382
2383 static int __init ksocklnd_init(void)
2384 {
2385         int rc;
2386
2387         /* check ksnr_connected/connecting field large enough */
2388         BUILD_BUG_ON(SOCKLND_CONN_NTYPES > 4);
2389         BUILD_BUG_ON(SOCKLND_CONN_ACK != SOCKLND_CONN_BULK_IN);
2390
2391         rc = ksocknal_tunables_init();
2392         if (rc != 0)
2393                 return rc;
2394
2395         lnet_register_lnd(&the_ksocklnd);
2396
2397         return 0;
2398 }
2399
2400 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2401 MODULE_DESCRIPTION("TCP Socket LNet Network Driver");
2402 MODULE_VERSION("2.8.0");
2403 MODULE_LICENSE("GPL");
2404
2405 module_init(ksocklnd_init);
2406 module_exit(ksocklnd_exit);