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