socklnd.c

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
 *   Author: Zach Brown <[e-mail]>
 *   Author: Peter J. Braam <[e-mail]>
 *   Author: Phil Schwan <[e-mail]>
 *   Author: Eric Barton <[e-mail]>
 *
 *   This file is part of Portals, http://www.sf.net/projects/sandiaportals/
 *
 *   Portals is free software; you can redistribute it and/or
 *   modify it under the terms of version 2 of the GNU General Public
 *   License as published by the Free Software Foundation.
 *
 *   Portals is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Portals; if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

#include "socklnd.h"

lnd_t the_ksocklnd = {
        .lnd_type       = SOCKLND,
        .lnd_startup    = ksocknal_startup,
        .lnd_shutdown   = ksocknal_shutdown,
        .lnd_ctl        = ksocknal_ctl,
        .lnd_send       = ksocknal_send,
        .lnd_recv       = ksocknal_recv,
        .lnd_notify     = ksocknal_notify,
        .lnd_accept     = ksocknal_accept,
};

ksock_nal_data_t        ksocknal_data;

ksock_interface_t *
ksocknal_ip2iface(lnet_ni_t *ni, __u32 ip)
{
        ksock_net_t       *net = ni->ni_data;
        int                i;
        ksock_interface_t *iface;

        for (i = 0; i < net->ksnn_ninterfaces; i++) {
                LASSERT(i < LNET_MAX_INTERFACES);
                iface = &net->ksnn_interfaces[i];

                if (iface->ksni_ipaddr == ip)
                        return (iface);
        }

        return (NULL);
}

ksock_route_t *
ksocknal_create_route (__u32 ipaddr, int port)
{
        ksock_route_t *route;

        LIBCFS_ALLOC (route, sizeof (*route));
        if (route == NULL)
                return (NULL);

        atomic_set (&route->ksnr_refcount, 1);
        route->ksnr_peer = NULL;
        route->ksnr_retry_interval = 0;         /* OK to connect at any time */
        route->ksnr_ipaddr = ipaddr;
        route->ksnr_port = port;
        route->ksnr_scheduled = 0;
        route->ksnr_connecting = 0;
        route->ksnr_connected = 0;
        route->ksnr_deleted = 0;
        route->ksnr_conn_count = 0;
        route->ksnr_share_count = 0;

        return (route);
}

void
ksocknal_destroy_route (ksock_route_t *route)
{
        LASSERT (atomic_read(&route->ksnr_refcount) == 0);

        if (route->ksnr_peer != NULL)
                ksocknal_peer_decref(route->ksnr_peer);

        LIBCFS_FREE (route, sizeof (*route));
}

int
ksocknal_create_peer (ksock_peer_t **peerp, lnet_ni_t *ni, lnet_process_id_t id)
{
        ksock_net_t   *net = ni->ni_data;
        ksock_peer_t  *peer;

        LASSERT (id.nid != LNET_NID_ANY);
        LASSERT (id.pid != LNET_PID_ANY);
        LASSERT (!in_interrupt());

        LIBCFS_ALLOC (peer, sizeof (*peer));
        if (peer == NULL)
                return -ENOMEM;

        memset (peer, 0, sizeof (*peer));       /* NULL pointers/clear flags etc */

        peer->ksnp_ni = ni;
        peer->ksnp_id = id;
        atomic_set (&peer->ksnp_refcount, 1);   /* 1 ref for caller */
        peer->ksnp_closing = 0;
        peer->ksnp_accepting = 0;
        peer->ksnp_zc_next_cookie = 1;
        peer->ksnp_proto = NULL;
        CFS_INIT_LIST_HEAD (&peer->ksnp_conns);
        CFS_INIT_LIST_HEAD (&peer->ksnp_routes);
        CFS_INIT_LIST_HEAD (&peer->ksnp_tx_queue);
        CFS_INIT_LIST_HEAD (&peer->ksnp_zc_req_list);
        spin_lock_init(&peer->ksnp_lock);

        spin_lock_bh (&net->ksnn_lock);

        if (net->ksnn_shutdown) {
                spin_unlock_bh (&net->ksnn_lock);
                
                LIBCFS_FREE(peer, sizeof(*peer));
                CERROR("Can't create peer: network shutdown\n");
                return -ESHUTDOWN;
        }

        net->ksnn_npeers++;

        spin_unlock_bh (&net->ksnn_lock);

        *peerp = peer;
        return 0;
}

void
ksocknal_destroy_peer (ksock_peer_t *peer)
{
        ksock_net_t    *net = peer->ksnp_ni->ni_data;

        CDEBUG (D_NET, "peer %s %p deleted\n", 
                libcfs_id2str(peer->ksnp_id), peer);

        LASSERT (atomic_read (&peer->ksnp_refcount) == 0);
        LASSERT (peer->ksnp_accepting == 0);
        LASSERT (list_empty (&peer->ksnp_conns));
        LASSERT (list_empty (&peer->ksnp_routes));
        LASSERT (list_empty (&peer->ksnp_tx_queue));
        LASSERT (list_empty (&peer->ksnp_zc_req_list));

        LIBCFS_FREE (peer, sizeof (*peer));

        /* NB a peer's connections and routes keep a reference on their peer
         * until they are destroyed, so we can be assured that _all_ state to
         * do with this peer has been cleaned up when its refcount drops to
         * zero. */
        spin_lock_bh (&net->ksnn_lock);
        net->ksnn_npeers--;
        spin_unlock_bh (&net->ksnn_lock);
}

ksock_peer_t *
ksocknal_find_peer_locked (lnet_ni_t *ni, lnet_process_id_t id)
{
        struct list_head *peer_list = ksocknal_nid2peerlist(id.nid);
        struct list_head *tmp;
        ksock_peer_t     *peer;

        list_for_each (tmp, peer_list) {

                peer = list_entry (tmp, ksock_peer_t, ksnp_list);

                LASSERT (!peer->ksnp_closing);

                if (peer->ksnp_ni != ni)
                        continue;

                if (peer->ksnp_id.nid != id.nid ||
                    peer->ksnp_id.pid != id.pid)
                        continue;

                CDEBUG(D_NET, "got peer [%p] -> %s (%d)\n",
                       peer, libcfs_id2str(id), 
                       atomic_read(&peer->ksnp_refcount));
                return (peer);
        }
        return (NULL);
}

ksock_peer_t *
ksocknal_find_peer (lnet_ni_t *ni, lnet_process_id_t id)
{
        ksock_peer_t     *peer;

        read_lock (&ksocknal_data.ksnd_global_lock);
        peer = ksocknal_find_peer_locked (ni, id);
        if (peer != NULL)                       /* +1 ref for caller? */
                ksocknal_peer_addref(peer);
        read_unlock (&ksocknal_data.ksnd_global_lock);

        return (peer);
}

void
ksocknal_unlink_peer_locked (ksock_peer_t *peer)
{
        int                i;
        __u32              ip;

        for (i = 0; i < peer->ksnp_n_passive_ips; i++) {
                LASSERT (i < LNET_MAX_INTERFACES);
                ip = peer->ksnp_passive_ips[i];

                ksocknal_ip2iface(peer->ksnp_ni, ip)->ksni_npeers--;
        }

        LASSERT (list_empty(&peer->ksnp_conns));
        LASSERT (list_empty(&peer->ksnp_routes));
        LASSERT (!peer->ksnp_closing);
        peer->ksnp_closing = 1;
        list_del (&peer->ksnp_list);
        /* lose peerlist's ref */
        ksocknal_peer_decref(peer);
}

int
ksocknal_get_peer_info (lnet_ni_t *ni, int index, 
                        lnet_process_id_t *id, __u32 *myip, __u32 *peer_ip, int *port,
                        int *conn_count, int *share_count)
{
        ksock_peer_t      *peer;
        struct list_head  *ptmp;
        ksock_route_t     *route;
        struct list_head  *rtmp;
        int                i;
        int                j;
        int                rc = -ENOENT;

        read_lock (&ksocknal_data.ksnd_global_lock);

        for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++) {

                list_for_each (ptmp, &ksocknal_data.ksnd_peers[i]) {
                        peer = list_entry (ptmp, ksock_peer_t, ksnp_list);

                        if (peer->ksnp_ni != ni)
                                continue;

                        if (peer->ksnp_n_passive_ips == 0 &&
                            list_empty(&peer->ksnp_routes)) {
                                if (index-- > 0)
                                        continue;

                                *id = peer->ksnp_id;
                                *myip = 0;
                                *peer_ip = 0;
                                *port = 0;
                                *conn_count = 0;
                                *share_count = 0;
                                rc = 0;
                                goto out;
                        }

                        for (j = 0; j < peer->ksnp_n_passive_ips; j++) {
                                if (index-- > 0)
                                        continue;

                                *id = peer->ksnp_id;
                                *myip = peer->ksnp_passive_ips[j];
                                *peer_ip = 0;
                                *port = 0;
                                *conn_count = 0;
                                *share_count = 0;
                                rc = 0;
                                goto out;
                        }

                        list_for_each (rtmp, &peer->ksnp_routes) {
                                if (index-- > 0)
                                        continue;

                                route = list_entry(rtmp, ksock_route_t,
                                                   ksnr_list);

                                *id = peer->ksnp_id;
                                *myip = route->ksnr_myipaddr;
                                *peer_ip = route->ksnr_ipaddr;
                                *port = route->ksnr_port;
                                *conn_count = route->ksnr_conn_count;
                                *share_count = route->ksnr_share_count;
                                rc = 0;
                                goto out;
                        }
                }
        }
 out:
        read_unlock (&ksocknal_data.ksnd_global_lock);
        return (rc);
}

void
ksocknal_associate_route_conn_locked(ksock_route_t *route, ksock_conn_t *conn)
{
        ksock_peer_t      *peer = route->ksnr_peer;
        int                type = conn->ksnc_type;
        ksock_interface_t *iface;

        conn->ksnc_route = route;
        ksocknal_route_addref(route);

        if (route->ksnr_myipaddr != conn->ksnc_myipaddr) {
                if (route->ksnr_myipaddr == 0) {
                        /* route wasn't bound locally yet (the initial route) */
                        CDEBUG(D_NET, "Binding %s %u.%u.%u.%u to %u.%u.%u.%u\n",
                               libcfs_id2str(peer->ksnp_id),
                               HIPQUAD(route->ksnr_ipaddr),
                               HIPQUAD(conn->ksnc_myipaddr));
                } else {
                        CDEBUG(D_NET, "Rebinding %s %u.%u.%u.%u from "
                               "%u.%u.%u.%u to %u.%u.%u.%u\n",
                               libcfs_id2str(peer->ksnp_id),
                               HIPQUAD(route->ksnr_ipaddr),
                               HIPQUAD(route->ksnr_myipaddr),
                               HIPQUAD(conn->ksnc_myipaddr));

                        iface = ksocknal_ip2iface(route->ksnr_peer->ksnp_ni,
                                                  route->ksnr_myipaddr);
                        if (iface != NULL)
                                iface->ksni_nroutes--;
                }
                route->ksnr_myipaddr = conn->ksnc_myipaddr;
                iface = ksocknal_ip2iface(route->ksnr_peer->ksnp_ni,
                                          route->ksnr_myipaddr);
                if (iface != NULL)
                        iface->ksni_nroutes++;
        }

        route->ksnr_connected |= (1<<type);
        route->ksnr_conn_count++;

        /* Successful connection => further attempts can
         * proceed immediately */
        route->ksnr_retry_interval = 0;
}

void
ksocknal_add_route_locked (ksock_peer_t *peer, ksock_route_t *route)
{
        struct list_head  *tmp;
        ksock_conn_t      *conn;
        ksock_route_t     *route2;

        LASSERT (!peer->ksnp_closing);
        LASSERT (route->ksnr_peer == NULL);
        LASSERT (!route->ksnr_scheduled);
        LASSERT (!route->ksnr_connecting);
        LASSERT (route->ksnr_connected == 0);

        /* LASSERT(unique) */
        list_for_each(tmp, &peer->ksnp_routes) {
                route2 = list_entry(tmp, ksock_route_t, ksnr_list);

                if (route2->ksnr_ipaddr == route->ksnr_ipaddr) {
                        CERROR ("Duplicate route %s %u.%u.%u.%u\n",
                                libcfs_id2str(peer->ksnp_id), 
                                HIPQUAD(route->ksnr_ipaddr));
                        LBUG();
                }
        }