socklnd.c

非常经典的一个分布式系统

        int                 best_netmatch;
        int                 best_npeers;

        /* CAVEAT EMPTOR: We do all our interface matching with an
         * exclusive hold of global lock at IRQ priority.  We're only
         * expecting to be dealing with small numbers of interfaces, so the
         * O(n**3)-ness shouldn't matter */

        /* Also note that I'm not going to return more than n_peerips
         * interfaces, even if I have more myself */

        write_lock_bh (global_lock);

        LASSERT (n_peerips <= LNET_MAX_INTERFACES);
        LASSERT (net->ksnn_ninterfaces <= LNET_MAX_INTERFACES);

        /* Only match interfaces for additional connections 
         * if I have > 1 interface */
        n_ips = (net->ksnn_ninterfaces < 2) ? 0 :
                MIN(n_peerips, net->ksnn_ninterfaces);

        for (i = 0; peer->ksnp_n_passive_ips < n_ips; i++) {
                /*              ^ yes really... */

                /* If we have any new interfaces, first tick off all the
                 * peer IPs that match old interfaces, then choose new
                 * interfaces to match the remaining peer IPS.
                 * We don't forget interfaces we've stopped using; we might
                 * start using them again... */

                if (i < peer->ksnp_n_passive_ips) {
                        /* Old interface. */
                        ip = peer->ksnp_passive_ips[i];
                        best_iface = ksocknal_ip2iface(peer->ksnp_ni, ip);

                        /* peer passive ips are kept up to date */
                        LASSERT(best_iface != NULL);
                } else {
                        /* choose a new interface */
                        LASSERT (i == peer->ksnp_n_passive_ips);

                        best_iface = NULL;
                        best_netmatch = 0;
                        best_npeers = 0;

                        for (j = 0; j < net->ksnn_ninterfaces; j++) {
                                iface = &net->ksnn_interfaces[j];
                                ip = iface->ksni_ipaddr;

                                for (k = 0; k < peer->ksnp_n_passive_ips; k++)
                                        if (peer->ksnp_passive_ips[k] == ip)
                                                break;

                                if (k < peer->ksnp_n_passive_ips) /* using it already */
                                        continue;

                                k = ksocknal_match_peerip(iface, peerips, n_peerips);
                                xor = (ip ^ peerips[k]);
                                this_netmatch = ((xor & iface->ksni_netmask) == 0) ? 1 : 0;

                                if (!(best_iface == NULL ||
                                      best_netmatch < this_netmatch ||
                                      (best_netmatch == this_netmatch &&
                                       best_npeers > iface->ksni_npeers)))
                                        continue;

                                best_iface = iface;
                                best_netmatch = this_netmatch;
                                best_npeers = iface->ksni_npeers;
                        }

                        best_iface->ksni_npeers++;
                        ip = best_iface->ksni_ipaddr;
                        peer->ksnp_passive_ips[i] = ip;
                        peer->ksnp_n_passive_ips = i+1;
                }

                LASSERT (best_iface != NULL);

                /* mark the best matching peer IP used */
                j = ksocknal_match_peerip(best_iface, peerips, n_peerips);
                peerips[j] = 0;
        }

        /* Overwrite input peer IP addresses */
        memcpy(peerips, peer->ksnp_passive_ips, n_ips * sizeof(*peerips));

        write_unlock_bh (global_lock);

        return (n_ips);
}

void
ksocknal_create_routes(ksock_peer_t *peer, int port,
                       __u32 *peer_ipaddrs, int npeer_ipaddrs)
{
        ksock_route_t      *newroute = NULL;
        rwlock_t           *global_lock = &ksocknal_data.ksnd_global_lock;
        lnet_ni_t          *ni = peer->ksnp_ni;
        ksock_net_t        *net = ni->ni_data;
        struct list_head   *rtmp;
        ksock_route_t      *route;
        ksock_interface_t  *iface;
        ksock_interface_t  *best_iface;
        int                 best_netmatch;
        int                 this_netmatch;
        int                 best_nroutes;
        int                 i;
        int                 j;

        /* CAVEAT EMPTOR: We do all our interface matching with an
         * exclusive hold of global lock at IRQ priority.  We're only
         * expecting to be dealing with small numbers of interfaces, so the
         * O(n**3)-ness here shouldn't matter */

        write_lock_bh (global_lock);

        if (net->ksnn_ninterfaces < 2) {
                /* Only create additional connections 
                 * if I have > 1 interface */
                write_unlock_bh (global_lock);
                return;
        }
        
        LASSERT (npeer_ipaddrs <= LNET_MAX_INTERFACES);

        for (i = 0; i < npeer_ipaddrs; i++) {
                if (newroute != NULL) {
                        newroute->ksnr_ipaddr = peer_ipaddrs[i];
                } else {
                        write_unlock_bh (global_lock);

                        newroute = ksocknal_create_route(peer_ipaddrs[i], port);
                        if (newroute == NULL)
                                return;

                        write_lock_bh (global_lock);
                }

                if (peer->ksnp_closing) {
                        /* peer got closed under me */
                        break;
                }

                /* Already got a route? */
                route = NULL;
                list_for_each(rtmp, &peer->ksnp_routes) {
                        route = list_entry(rtmp, ksock_route_t, ksnr_list);

                        if (route->ksnr_ipaddr == newroute->ksnr_ipaddr)
                                break;

                        route = NULL;
                }
                if (route != NULL)
                        continue;

                best_iface = NULL;
                best_nroutes = 0;
                best_netmatch = 0;

                LASSERT (net->ksnn_ninterfaces <= LNET_MAX_INTERFACES);

                /* Select interface to connect from */
                for (j = 0; j < net->ksnn_ninterfaces; j++) {
                        iface = &net->ksnn_interfaces[j];

                        /* Using this interface already? */
                        list_for_each(rtmp, &peer->ksnp_routes) {
                                route = list_entry(rtmp, ksock_route_t, ksnr_list);

                                if (route->ksnr_myipaddr == iface->ksni_ipaddr)
                                        break;

                                route = NULL;
                        }
                        if (route != NULL)
                                continue;

                        this_netmatch = (((iface->ksni_ipaddr ^
                                           newroute->ksnr_ipaddr) &
                                           iface->ksni_netmask) == 0) ? 1 : 0;

                        if (!(best_iface == NULL ||
                              best_netmatch < this_netmatch ||
                              (best_netmatch == this_netmatch &&
                               best_nroutes > iface->ksni_nroutes)))
                                continue;

                        best_iface = iface;
                        best_netmatch = this_netmatch;
                        best_nroutes = iface->ksni_nroutes;
                }

                if (best_iface == NULL)
                        continue;

                newroute->ksnr_myipaddr = best_iface->ksni_ipaddr;
                best_iface->ksni_nroutes++;

                ksocknal_add_route_locked(peer, newroute);
                newroute = NULL;
        }

        write_unlock_bh (global_lock);
        if (newroute != NULL)
                ksocknal_route_decref(newroute);
}

int
ksocknal_accept (lnet_ni_t *ni, cfs_socket_t *sock)
{
        ksock_connreq_t    *cr;
        int                 rc;
        __u32               peer_ip;
        int                 peer_port;

        rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
        LASSERT (rc == 0);                      /* we succeeded before */

        LIBCFS_ALLOC(cr, sizeof(*cr));
        if (cr == NULL) {
                LCONSOLE_ERROR_MSG(0x12f, "Dropping connection request from "
                                   "%u.%u.%u.%u: memory exhausted\n",
                                   HIPQUAD(peer_ip));
                return -ENOMEM;
        }

        lnet_ni_addref(ni);
        cr->ksncr_ni   = ni;
        cr->ksncr_sock = sock;

        spin_lock_bh (&ksocknal_data.ksnd_connd_lock);

        list_add_tail(&cr->ksncr_list, &ksocknal_data.ksnd_connd_connreqs);
        cfs_waitq_signal(&ksocknal_data.ksnd_connd_waitq);
                        
        spin_unlock_bh (&ksocknal_data.ksnd_connd_lock);
        return 0;
}

int
ksocknal_connecting (ksock_peer_t *peer, __u32 ipaddr) 
{
        ksock_route_t   *route;
        
        list_for_each_entry (route, &peer->ksnp_routes, ksnr_list) {

                if (route->ksnr_ipaddr == ipaddr)
                        return route->ksnr_connecting;
        }
        return 0;
}

int
ksocknal_create_conn (lnet_ni_t *ni, ksock_route_t *route, 
                      cfs_socket_t *sock, int type)
{
        rwlock_t          *global_lock = &ksocknal_data.ksnd_global_lock;
        CFS_LIST_HEAD     (zombies);
        lnet_process_id_t  peerid;
        struct list_head  *tmp;
        __u64              incarnation;
        ksock_conn_t      *conn;
        ksock_conn_t      *conn2;
        ksock_peer_t      *peer = NULL;
        ksock_peer_t      *peer2;
        ksock_sched_t     *sched;
        ksock_hello_msg_t *hello;
        unsigned int       irq;
        ksock_tx_t        *tx;
        int                rc;
        int                active;
        char              *warn = NULL;

        active = (route != NULL);

        LASSERT (active == (type != SOCKLND_CONN_NONE));

        irq = ksocknal_lib_sock_irq (sock);

        LIBCFS_ALLOC(conn, sizeof(*conn));
        if (conn == NULL) {
                rc = -ENOMEM;
                goto failed_0;
        }

        memset (conn, 0, sizeof (*conn));
        conn->ksnc_peer = NULL;
        conn->ksnc_route = NULL;
        conn->ksnc_sock = sock;
        atomic_set (&conn->ksnc_sock_refcount, 1); /* 1 ref for conn */
        conn->ksnc_type = type;
        ksocknal_lib_save_callback(sock, conn);
        atomic_set (&conn->ksnc_conn_refcount, 1); /* 1 ref for me */

        conn->ksnc_zc_capable = ksocknal_lib_zc_capable(sock);
        conn->ksnc_rx_ready = 0;
        conn->ksnc_rx_scheduled = 0;

        CFS_INIT_LIST_HEAD (&conn->ksnc_tx_queue);
        conn->ksnc_tx_ready = 0;
        conn->ksnc_tx_scheduled = 0;
        conn->ksnc_tx_mono = NULL;
        atomic_set (&conn->ksnc_tx_nob, 0);

        LIBCFS_ALLOC(hello, offsetof(ksock_hello_msg_t,
                                     kshm_ips[LNET_MAX_INTERFACES]));
        if (hello == NULL) {
                rc = -ENOMEM;
                goto failed_1;
        }

        /* stash conn's local and remote addrs */
        rc = ksocknal_lib_get_conn_addrs (conn);
        if (rc != 0)
                goto failed_1;

        /* Find out/confirm peer's NID and connection type and get the
         * vector of interfaces she's willing to let me connect to.
         * Passive connections use the listener timeout since the peer sends
         * eagerly */

        if (active) {
                peer = route->ksnr_peer;
                LASSERT(ni == peer->ksnp_ni);

                /* Active connection sends HELLO eagerly */
                hello->kshm_nips = ksocknal_local_ipvec(ni, hello->kshm_ips);
                peerid = peer->ksnp_id;

                write_lock_bh(global_lock);
                conn->ksnc_proto = peer->ksnp_proto;
                write_unlock_bh(global_lock);
                
                if (conn->ksnc_proto == NULL) {
                        conn->ksnc_proto = &ksocknal_protocol_v2x;
#if SOCKNAL_VERSION_DEBUG
                        if (*ksocknal_tunables.ksnd_protocol != 2)
                                conn->ksnc_proto = &ksocknal_protocol_v1x;
#endif
                }

                rc = ksocknal_send_hello (ni, conn, peerid.nid, hello);
                if (rc != 0)
                        goto failed_1;
        } else {
                peerid.nid = LNET_NID_ANY;
                peerid.pid = LNET_PID_ANY;

                /* Passive, get protocol from peer */
                conn->ksnc_proto = NULL;
        }

        rc = ksocknal_recv_hello (ni, conn, hello, &peerid, &incarnation);
        if (rc < 0)
                goto failed_1;

        LASSERT (rc == 0 || active);
        LASSERT (conn->ksnc_proto != NULL);
        LASSERT (peerid.nid != LNET_NID_ANY);

        if (active) {
                ksocknal_peer_addref(peer);
                write_lock_bh (global_lock);
        } else {
                rc = ksocknal_create_peer(&peer, ni, peerid);
                if (rc != 0)
                        goto failed_1;

                write_lock_bh (global_lock);

                /* called with a ref on ni, so shutdown can't have started */
                LASSERT (((ksock_net_t *) ni->ni_data)->ksnn_shutdown == 0);