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