usocklnd.c

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

                                       &usock_data.ud_pollthreads[i]);
                if (rc) {
                        usocklnd_base_shutdown(i);
                        return rc;
                }
        }
        
        usock_data.ud_state = UD_STATE_INITIALIZED;
        
        return 0;

  base_startup_failed_4:
        LIBCFS_FREE (pt->upt_skip, sizeof(int) * UPT_START_SIZ);
  base_startup_failed_3:
        LIBCFS_FREE (pt->upt_fd2idx, sizeof(int) * UPT_START_SIZ);
  base_startup_failed_2:
        LIBCFS_FREE (pt->upt_idx2conn, sizeof(usock_conn_t *) * UPT_START_SIZ);
  base_startup_failed_1:
        LIBCFS_FREE (pt->upt_pollfd, sizeof(struct pollfd) * UPT_START_SIZ);
  base_startup_failed_0:
        LASSERT(rc != 0);
        usocklnd_release_poll_states(i);
        LIBCFS_FREE (usock_data.ud_pollthreads,
                     usock_data.ud_npollthreads *
                     sizeof(usock_pollthread_t));
        return rc;
}

void
usocklnd_base_shutdown(int n)
{
        int i;
        
        usock_data.ud_shutdown = 1;
        for (i = 0; i < n; i++) {
                usock_pollthread_t *pt = &usock_data.ud_pollthreads[i];
                usocklnd_wakeup_pollthread(i);
                cfs_wait_for_completion(&pt->upt_completion);
        }

        pthread_rwlock_destroy(&usock_data.ud_peers_lock);

        usocklnd_release_poll_states(usock_data.ud_npollthreads);

        LIBCFS_FREE (usock_data.ud_pollthreads,
                     usock_data.ud_npollthreads *
                     sizeof(usock_pollthread_t));
        
        usock_data.ud_state = UD_STATE_INIT_NOTHING;
}

__u64
usocklnd_new_incarnation()
{
        struct timeval tv;
        int            rc = gettimeofday(&tv, NULL);
        LASSERT (rc == 0);
        return (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
}

static int
usocklnd_assign_ni_nid(lnet_ni_t *ni)
{
        int   rc;
        int   up;
        __u32 ipaddr;
        
        /* Find correct IP-address and update ni_nid with it.
         * Two cases are supported:
         * 1) no explicit interfaces are defined. NID will be assigned to
         * first non-lo interface that is up;
         * 2) exactly one explicit interface is defined. For example,
         * LNET_NETWORKS='tcp(eth0)' */     

        if (ni->ni_interfaces[0] == NULL) {
                char **names;
                int    i, n;
            
                n = libcfs_ipif_enumerate(&names);
                if (n <= 0) {
                        CERROR("Can't enumerate interfaces: %d\n", n);
                        return -1;
                }

                for (i = 0; i < n; i++) {
   
                        if (!strcmp(names[i], "lo")) /* skip the loopback IF */
                                continue;
                    
                        rc = libcfs_ipif_query(names[i], &up, &ipaddr);
                        if (rc != 0) {
                                CWARN("Can't get interface %s info: %d\n",
                                      names[i], rc);
                                continue;
                        }
                    
                        if (!up) {
                                CWARN("Ignoring interface %s (down)\n",
                                      names[i]);
                            continue;
                        }
                        
                        break;      /* one address is quite enough */
                }
            
                libcfs_ipif_free_enumeration(names, n);

                if (i >= n) {
                        CERROR("Can't find any usable interfaces\n");
                        return -1;
                }

                CDEBUG(D_NET, "No explicit interfaces defined. "
                       "%u.%u.%u.%u used\n", HIPQUAD(ipaddr));
        } else {
                if (ni->ni_interfaces[1] != NULL) {
                        CERROR("only one explicit interface is allowed\n");
                        return -1;
                }

                rc = libcfs_ipif_query(ni->ni_interfaces[0], &up, &ipaddr);
                if (rc != 0) {
                        CERROR("Can't get interface %s info: %d\n",
                               ni->ni_interfaces[0], rc);
                        return -1;
                }

                if (!up) {
                        CERROR("Explicit interface defined: %s but is down\n",
                               ni->ni_interfaces[0]);
                        return -1;
                }
                
                CDEBUG(D_NET, "Explicit interface defined: %s. "
                       "%u.%u.%u.%u used\n",
                       ni->ni_interfaces[0], HIPQUAD(ipaddr));
                
        }
        
        ni->ni_nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), ipaddr);

        return 0;
}

int
usocklnd_startup(lnet_ni_t *ni)
{
        int          rc;
        usock_net_t *net;

        if (usock_data.ud_state == UD_STATE_INIT_NOTHING) {
                rc = usocklnd_base_startup();
                if (rc != 0)
                        return rc;
        }

        LIBCFS_ALLOC(net, sizeof(*net));
        if (net == NULL)
                goto startup_failed_0;

        memset(net, 0, sizeof(*net));
        net->un_incarnation = usocklnd_new_incarnation();
        pthread_mutex_init(&net->un_lock, NULL);
        pthread_cond_init(&net->un_cond, NULL);

        ni->ni_data = net;

        if (!(the_lnet.ln_pid & LNET_PID_USERFLAG)) {
                rc = usocklnd_assign_ni_nid(ni);
                if (rc != 0)
                        goto startup_failed_1;
        }

        LASSERT (ni->ni_lnd == &the_tcplnd);

        ni->ni_maxtxcredits = usock_tuns.ut_txcredits;
        ni->ni_peertxcredits = usock_tuns.ut_peertxcredits;
    
        usock_data.ud_nets_count++;
        return 0;

 startup_failed_1:
        pthread_mutex_destroy(&net->un_lock);
        pthread_cond_destroy(&net->un_cond);
        LIBCFS_FREE(net, sizeof(*net));
 startup_failed_0:
        if (usock_data.ud_nets_count == 0)
                usocklnd_base_shutdown(usock_data.ud_npollthreads);

        return -ENETDOWN;
}

void
usocklnd_shutdown(lnet_ni_t *ni)
{
        usock_net_t *net = ni->ni_data;

        net->un_shutdown = 1;

        usocklnd_del_all_peers(ni);        

        /* Wait for all peer state to clean up */
        pthread_mutex_lock(&net->un_lock);
        while (net->un_peercount != 0) 
                pthread_cond_wait(&net->un_cond, &net->un_lock);
        pthread_mutex_unlock(&net->un_lock);
        
        /* Release usock_net_t structure */
        pthread_mutex_destroy(&net->un_lock);
        pthread_cond_destroy(&net->un_cond);
        LIBCFS_FREE(net, sizeof(*net));

        usock_data.ud_nets_count--;
        if (usock_data.ud_nets_count == 0)
                usocklnd_base_shutdown(usock_data.ud_npollthreads);
}

void
usocklnd_del_all_peers(lnet_ni_t *ni)
{
        struct list_head  *ptmp;
        struct list_head  *pnxt;
        usock_peer_t      *peer;
        int                i;

        pthread_rwlock_wrlock(&usock_data.ud_peers_lock);

        for (i = 0; i < UD_PEER_HASH_SIZE; i++) {
                list_for_each_safe (ptmp, pnxt, &usock_data.ud_peers[i]) {
                        peer = list_entry (ptmp, usock_peer_t, up_list);
                        
                        if (peer->up_ni != ni)
                                continue;

                        usocklnd_del_peer_and_conns(peer);
                }
        }

        pthread_rwlock_unlock(&usock_data.ud_peers_lock);
        
        /* wakeup all threads */
        for (i = 0; i < usock_data.ud_npollthreads; i++)
                usocklnd_wakeup_pollthread(i);
}

void
usocklnd_del_peer_and_conns(usock_peer_t *peer)
{
        /* peer cannot disappear because it's still in hash list */

        pthread_mutex_lock(&peer->up_lock);
        /* content of conn[] array cannot change now */
        usocklnd_del_conns_locked(peer);
        pthread_mutex_unlock(&peer->up_lock);

        /* peer hash list is still protected by the caller */
        list_del(&peer->up_list);

        usocklnd_peer_decref(peer); /* peer isn't in hash list anymore */
}

void
usocklnd_del_conns_locked(usock_peer_t *peer)
{
        int i;
        
        for (i=0; i < N_CONN_TYPES; i++) {
                usock_conn_t *conn = peer->up_conns[i];
                if (conn != NULL)
                        usocklnd_conn_kill(conn);                 
        }       
}