ralnd.c

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

kranal_free_acceptsock (kra_acceptsock_t *ras)
{
        libcfs_sock_release(ras->ras_sock);
        LIBCFS_FREE(ras, sizeof(*ras));
}

int
kranal_accept (lnet_ni_t *ni, struct socket *sock)
{
        kra_acceptsock_t  *ras;
        int                rc;
        __u32              peer_ip;
        int                peer_port;
        unsigned long      flags;

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

        LIBCFS_ALLOC(ras, sizeof(*ras));
        if (ras == NULL) {
                CERROR("ENOMEM allocating connection request from "
                       "%u.%u.%u.%u\n", HIPQUAD(peer_ip));
                return -ENOMEM;
        }

        ras->ras_sock = sock;

        spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);

        list_add_tail(&ras->ras_list, &kranal_data.kra_connd_acceptq);
        wake_up(&kranal_data.kra_connd_waitq);

        spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
        return 0;
}

int
kranal_create_peer (kra_peer_t **peerp, lnet_nid_t nid)
{
        kra_peer_t    *peer;
        unsigned long  flags;

        LASSERT (nid != LNET_NID_ANY);

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

        memset(peer, 0, sizeof(*peer));         /* zero flags etc */

        peer->rap_nid = nid;
        atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */

        INIT_LIST_HEAD(&peer->rap_list);
        INIT_LIST_HEAD(&peer->rap_connd_list);
        INIT_LIST_HEAD(&peer->rap_conns);
        INIT_LIST_HEAD(&peer->rap_tx_queue);

        peer->rap_reconnect_interval = 0;       /* OK to connect at any time */

        write_lock_irqsave(&kranal_data.kra_global_lock, flags);

        if (kranal_data.kra_nonewpeers) {
                /* shutdown has started already */
                write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
                
                LIBCFS_FREE(peer, sizeof(*peer));
                CERROR("Can't create peer: network shutdown\n");
                return -ESHUTDOWN;
        }

        atomic_inc(&kranal_data.kra_npeers);

        write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);

        *peerp = peer;
        return 0;
}

void
kranal_destroy_peer (kra_peer_t *peer)
{
        CDEBUG(D_NET, "peer %s %p deleted\n", 
               libcfs_nid2str(peer->rap_nid), peer);

        LASSERT (atomic_read(&peer->rap_refcount) == 0);
        LASSERT (peer->rap_persistence == 0);
        LASSERT (!kranal_peer_active(peer));
        LASSERT (!peer->rap_connecting);
        LASSERT (list_empty(&peer->rap_conns));
        LASSERT (list_empty(&peer->rap_tx_queue));
        LASSERT (list_empty(&peer->rap_connd_list));

        LIBCFS_FREE(peer, sizeof(*peer));

        /* NB a peer's connections 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. */
        atomic_dec(&kranal_data.kra_npeers);
}

kra_peer_t *
kranal_find_peer_locked (lnet_nid_t nid)
{
        struct list_head *peer_list = kranal_nid2peerlist(nid);
        struct list_head *tmp;
        kra_peer_t       *peer;

        list_for_each (tmp, peer_list) {

                peer = list_entry(tmp, kra_peer_t, rap_list);

                LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
                         !list_empty(&peer->rap_conns));  /* active conn */

                if (peer->rap_nid != nid)
                        continue;

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

kra_peer_t *
kranal_find_peer (lnet_nid_t nid)
{
        kra_peer_t     *peer;

        read_lock(&kranal_data.kra_global_lock);
        peer = kranal_find_peer_locked(nid);
        if (peer != NULL)                       /* +1 ref for caller? */
                kranal_peer_addref(peer);
        read_unlock(&kranal_data.kra_global_lock);

        return peer;
}

void
kranal_unlink_peer_locked (kra_peer_t *peer)
{
        LASSERT (peer->rap_persistence == 0);
        LASSERT (list_empty(&peer->rap_conns));

        LASSERT (kranal_peer_active(peer));
        list_del_init(&peer->rap_list);

        /* lose peerlist's ref */
        kranal_peer_decref(peer);
}

int
kranal_get_peer_info (int index, lnet_nid_t *nidp, __u32 *ipp, int *portp,
                      int *persistencep)
{
        kra_peer_t        *peer;
        struct list_head  *ptmp;
        int                i;

        read_lock(&kranal_data.kra_global_lock);

        for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {

                list_for_each(ptmp, &kranal_data.kra_peers[i]) {

                        peer = list_entry(ptmp, kra_peer_t, rap_list);
                        LASSERT (peer->rap_persistence > 0 ||
                                 !list_empty(&peer->rap_conns));

                        if (index-- > 0)
                                continue;

                        *nidp = peer->rap_nid;
                        *ipp = peer->rap_ip;
                        *portp = peer->rap_port;
                        *persistencep = peer->rap_persistence;

                        read_unlock(&kranal_data.kra_global_lock);
                        return 0;
                }
        }

        read_unlock(&kranal_data.kra_global_lock);
        return -ENOENT;
}

int
kranal_add_persistent_peer (lnet_nid_t nid, __u32 ip, int port)
{
        unsigned long      flags;
        kra_peer_t        *peer;
        kra_peer_t        *peer2;
        int                rc;

        if (nid == LNET_NID_ANY)
                return -EINVAL;

        rc = kranal_create_peer(&peer, nid);
        if (rc != 0)
                return rc;

        write_lock_irqsave(&kranal_data.kra_global_lock, flags);

        peer2 = kranal_find_peer_locked(nid);
        if (peer2 != NULL) {
                kranal_peer_decref(peer);
                peer = peer2;
        } else {
                /* peer table takes existing ref on peer */
                list_add_tail(&peer->rap_list,
                              kranal_nid2peerlist(nid));
        }

        peer->rap_ip = ip;
        peer->rap_port = port;
        peer->rap_persistence++;

        write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
        return 0;
}

void
kranal_del_peer_locked (kra_peer_t *peer)
{
        struct list_head *ctmp;
        struct list_head *cnxt;
        kra_conn_t       *conn;

        peer->rap_persistence = 0;

        if (list_empty(&peer->rap_conns)) {
                kranal_unlink_peer_locked(peer);
        } else {
                list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
                        conn = list_entry(ctmp, kra_conn_t, rac_list);

                        kranal_close_conn_locked(conn, 0);
                }
                /* peer unlinks itself when last conn is closed */
        }
}

int
kranal_del_peer (lnet_nid_t nid)
{
        unsigned long      flags;
        struct list_head  *ptmp;
        struct list_head  *pnxt;
        kra_peer_t        *peer;
        int                lo;
        int                hi;
        int                i;
        int                rc = -ENOENT;

        write_lock_irqsave(&kranal_data.kra_global_lock, flags);

        if (nid != LNET_NID_ANY)
                lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
        else {
                lo = 0;
                hi = kranal_data.kra_peer_hash_size - 1;
        }

        for (i = lo; i <= hi; i++) {
                list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
                        peer = list_entry(ptmp, kra_peer_t, rap_list);
                        LASSERT (peer->rap_persistence > 0 ||
                                 !list_empty(&peer->rap_conns));

                        if (!(nid == LNET_NID_ANY || peer->rap_nid == nid))
                                continue;

                        kranal_del_peer_locked(peer);
                        rc = 0;         /* matched something */
                }
        }

        write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);

        return rc;
}

kra_conn_t *
kranal_get_conn_by_idx (int index)
{
        kra_peer_t        *peer;
        struct list_head  *ptmp;
        kra_conn_t        *conn;
        struct list_head  *ctmp;
        int                i;

        read_lock (&kranal_data.kra_global_lock);

        for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
                list_for_each (ptmp, &kranal_data.kra_peers[i]) {

                        peer = list_entry(ptmp, kra_peer_t, rap_list);
                        LASSERT (peer->rap_persistence > 0 ||
                                 !list_empty(&peer->rap_conns));

                        list_for_each (ctmp, &peer->rap_conns) {
                                if (index-- > 0)
                                        continue;

                                conn = list_entry(ctmp, kra_conn_t, rac_list);
                                CDEBUG(D_NET, "++conn[%p] -> %s (%d)\n", conn, 
                                       libcfs_nid2str(conn->rac_peer->rap_nid),
                                       atomic_read(&conn->rac_refcount));
                                atomic_inc(&conn->rac_refcount);
                                read_unlock(&kranal_data.kra_global_lock);
                                return conn;
                        }
                }
        }

        read_unlock(&kranal_data.kra_global_lock);
        return NULL;
}

int
kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
{
        kra_conn_t         *conn;
        struct list_head   *ctmp;
        struct list_head   *cnxt;
        int                 count = 0;

        list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
                conn = list_entry(ctmp, kra_conn_t, rac_list);

                count++;
                kranal_close_conn_locked(conn, why);
        }

        return count;
}

int
kranal_close_matching_conns (lnet_nid_t nid)
{
        unsigned long       flags;
        kra_peer_t         *peer;
        struct list_head   *ptmp;
        struct list_head   *pnxt;
        int                 lo;
        int                 hi;
        int                 i;
        int                 count = 0;

        write_lock_irqsave(&kranal_data.kra_global_lock, flags);

        if (nid != LNET_NID_ANY)
                lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
        else {
                lo = 0;
                hi = kranal_data.kra_peer_hash_size - 1;
        }

        for (i = lo; i <= hi; i++) {
                list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {

                        peer = list_entry(ptmp, kra_peer_t, rap_list);
                        LASSERT (peer->rap_persistence > 0 ||
                                 !list_empty(&peer->rap_conns));

                        if (!(nid == LNET_NID_ANY || nid == peer->rap_nid))
                                continue;

                        count += kranal_close_peer_conns_locked(peer, 0);
                }
        }

        write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);

        /* wildcards always succeed */
        if (nid == LNET_NID_ANY)
                return 0;

        return (count == 0) ? -ENOENT : 0;
}

int
kranal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
{
        struct libcfs_ioctl_data *data = arg;
        int                       rc = -EINVAL;

        LASSERT (ni == kranal_data.kra_ni);

        switch(cmd) {
        case IOC_LIBCFS_GET_PEER: {
                lnet_nid_t   nid = 0;
                __u32       ip = 0;
                int         port = 0;
                int         share_count = 0;

                rc = kranal_get_peer_info(data->ioc_count,
                                          &nid, &ip, &port, &share_count);
                data->ioc_nid    = nid;
                data->ioc_count  = share_count;
                data->ioc_u32[0] = ip;
                data->ioc_u32[1] = port;
                break;
        }
        case IOC_LIBCFS_ADD_PEER: {
                rc = kranal_add_persistent_peer(data->ioc_nid,
                                                data->ioc_u32[0], /* IP */
                                                data->ioc_u32[1]); /* port */
                break;
        }
        case IOC_LIBCFS_DEL_PEER: {
                rc = kranal_del_peer(data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_GET_CONN: {
                kra_conn_t *conn = kranal_get_conn_by_idx(data->ioc_count);

                if (conn == NULL)
                        rc = -ENOENT;
                else {
                        rc = 0;
                        data->ioc_nid    = conn->rac_peer->rap_nid;
                        data->ioc_u32[0] = conn->rac_device->rad_id;
                        kranal_conn_decref(conn);
                }
                break;
        }
        case IOC_LIBCFS_CLOSE_CONNECTION: {
                rc = kranal_close_matching_conns(data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_REGISTER_MYNID: {
                /* Ignore if this is a noop */