usocklnd.c

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2007 Cluster File Systems, Inc.
 *   Author: Maxim Patlasov <[e-mail]>
 *
 *   This file is part of the Lustre file system, http://www.lustre.org
 *   Lustre is a trademark of Cluster File Systems, Inc.
 *
 */

#include "usocklnd.h"
#include <sys/time.h>

lnd_t the_tcplnd = {
        .lnd_type      = SOCKLND,
        .lnd_startup   = usocklnd_startup,
        .lnd_shutdown  = usocklnd_shutdown,
        .lnd_send      = usocklnd_send,
        .lnd_recv      = usocklnd_recv,
        .lnd_accept    = usocklnd_accept,
};

usock_data_t usock_data;
usock_tunables_t usock_tuns = {
        .ut_timeout         = 50,
        .ut_poll_timeout    = 1,
        .ut_fair_limit      = 1,
        .ut_npollthreads    = 0,
        .ut_min_bulk        = 1<<10,
        .ut_txcredits       = 256,
        .ut_peertxcredits   = 8,
        .ut_socknagle       = 0,
        .ut_sockbufsiz      = 0,
};

#define MAX_REASONABLE_TIMEOUT 36000 /* 10 hours */
#define MAX_REASONABLE_NPT 1000

int
usocklnd_validate_tunables()
{
        if (usock_tuns.ut_timeout <= 0 ||
            usock_tuns.ut_timeout > MAX_REASONABLE_TIMEOUT) {
                CERROR("USOCK_TIMEOUT: %d is out of reasonable limits\n",
                       usock_tuns.ut_timeout);
                return -1;
        }
                
        if (usock_tuns.ut_poll_timeout <= 0 ||
            usock_tuns.ut_poll_timeout > MAX_REASONABLE_TIMEOUT) {
                CERROR("USOCK_POLL_TIMEOUT: %d is out of reasonable limits\n",
                       usock_tuns.ut_poll_timeout);
                return -1;
        }

        if (usock_tuns.ut_fair_limit <= 0) {
                CERROR("Invalid USOCK_FAIR_LIMIT: %d (should be >0)\n",
                       usock_tuns.ut_fair_limit);
                return -1;
        }
                
        if (usock_tuns.ut_npollthreads < 0 ||
            usock_tuns.ut_npollthreads > MAX_REASONABLE_NPT) {
                CERROR("USOCK_NPOLLTHREADS: %d is out of reasonable limits\n",
                       usock_tuns.ut_npollthreads);
                return -1;
        }

        if (usock_tuns.ut_txcredits <= 0) {
                CERROR("USOCK_TXCREDITS: %d should be positive\n",
                       usock_tuns.ut_txcredits);
                return -1;
        }
                
        if (usock_tuns.ut_peertxcredits <= 0) {
                CERROR("USOCK_PEERTXCREDITS: %d should be positive\n",
                       usock_tuns.ut_peertxcredits);
                return -1;
        }

        if (usock_tuns.ut_peertxcredits > usock_tuns.ut_txcredits) {
                CERROR("USOCK_PEERTXCREDITS: %d should not be greater"
                       " than USOCK_TXCREDITS: %d\n",
                       usock_tuns.ut_peertxcredits, usock_tuns.ut_txcredits);
                return -1;
        }

        if (usock_tuns.ut_socknagle != 0 &&
            usock_tuns.ut_socknagle != 1) {
                CERROR("USOCK_SOCKNAGLE: %d should be 0 or 1\n",
                       usock_tuns.ut_socknagle);
                return -1;
        }
        
        if (usock_tuns.ut_sockbufsiz < 0) {
                CERROR("USOCK_SOCKBUFSIZ: %d should be 0 or positive\n",
                       usock_tuns.ut_sockbufsiz);
                return -1;
        }

        return 0;
}

void
usocklnd_release_poll_states(int n)
{
        int i;
        
        for (i = 0; i < n; i++) {
                usock_pollthread_t *pt = &usock_data.ud_pollthreads[i];
                
                close(pt->upt_notifier_fd);
                close(pt->upt_pollfd[0].fd);

                pthread_mutex_destroy(&pt->upt_pollrequests_lock);
                cfs_fini_completion(&pt->upt_completion);
                
                LIBCFS_FREE (pt->upt_pollfd,
                             sizeof(struct pollfd) * pt->upt_npollfd);
                LIBCFS_FREE (pt->upt_idx2conn,
                              sizeof(usock_conn_t *) * pt->upt_npollfd);
                LIBCFS_FREE (pt->upt_fd2idx,
                              sizeof(int) * pt->upt_nfd2idx);                
        }
}

int
usocklnd_update_tunables()
{
        int rc;
        
        rc = cfs_parse_int_tunable(&usock_tuns.ut_timeout,
                                      "USOCK_TIMEOUT");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_poll_timeout,
                                      "USOCK_POLL_TIMEOUT");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_npollthreads,
                                      "USOCK_NPOLLTHREADS");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_fair_limit,
                                      "USOCK_FAIR_LIMIT");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_min_bulk,
                                      "USOCK_MIN_BULK");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_txcredits,
                                      "USOCK_TXCREDITS");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_peertxcredits,
                                      "USOCK_PEERTXCREDITS");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_socknagle,
                                      "USOCK_SOCKNAGLE");
        if (rc)
                return rc;

        rc = cfs_parse_int_tunable(&usock_tuns.ut_sockbufsiz,
                                      "USOCK_SOCKBUFSIZ");
        if (rc)
                return rc;

        if (usocklnd_validate_tunables())
                return -EINVAL;
        
        if (usock_tuns.ut_npollthreads == 0) {
                usock_tuns.ut_npollthreads = cfs_online_cpus();

                if (usock_tuns.ut_npollthreads <= 0) {
                        CERROR("Cannot find out the number of online CPUs\n");
                        return -EINVAL;
                }
        }
        
        return 0;
}


int
usocklnd_base_startup()
{
        usock_pollthread_t *pt;
        int                 i;
        int                 rc;
        
        rc = usocklnd_update_tunables();
        if (rc)
                return rc;
        
        usock_data.ud_npollthreads = usock_tuns.ut_npollthreads;

        LIBCFS_ALLOC (usock_data.ud_pollthreads,
                      usock_data.ud_npollthreads *
                      sizeof(usock_pollthread_t));
        if (usock_data.ud_pollthreads == NULL)
                return -ENOMEM;

        /* Initialize poll thread state structures */
        for (i = 0; i < usock_data.ud_npollthreads; i++) {
                int notifier[2];

                pt = &usock_data.ud_pollthreads[i];

                rc = -ENOMEM;
                
                LIBCFS_ALLOC (pt->upt_pollfd,
                              sizeof(struct pollfd) * UPT_START_SIZ);
                if (pt->upt_pollfd == NULL)
                        goto base_startup_failed_0;
                
                LIBCFS_ALLOC (pt->upt_idx2conn,
                              sizeof(usock_conn_t *) * UPT_START_SIZ);
                if (pt->upt_idx2conn == NULL)
                        goto base_startup_failed_1;

                LIBCFS_ALLOC (pt->upt_fd2idx,
                              sizeof(int) * UPT_START_SIZ);
                if (pt->upt_fd2idx == NULL)
                        goto base_startup_failed_2;                
                
                memset(pt->upt_fd2idx, 0,
                       sizeof(int) * UPT_START_SIZ);                       
                
                LIBCFS_ALLOC (pt->upt_skip,
                              sizeof(int) * UPT_START_SIZ);
                if (pt->upt_skip == NULL)
                        goto base_startup_failed_3;

                pt->upt_npollfd = pt->upt_nfd2idx = UPT_START_SIZ;

                rc = libcfs_socketpair(notifier);
                if (rc != 0)
                        goto base_startup_failed_4;

                pt->upt_notifier_fd = notifier[0];

                pt->upt_pollfd[0].fd = notifier[1];
                pt->upt_pollfd[0].events = POLLIN;
                pt->upt_pollfd[0].revents = 0;

                pt->upt_nfds = 1;
                pt->upt_idx2conn[0] = NULL;

                pt->upt_errno = 0;
                CFS_INIT_LIST_HEAD (&pt->upt_pollrequests);
                CFS_INIT_LIST_HEAD (&pt->upt_stale_list);
                pthread_mutex_init(&pt->upt_pollrequests_lock, NULL);
                cfs_init_completion(&pt->upt_completion);
        }

        /* Initialize peer hash list */        
        for (i = 0; i < UD_PEER_HASH_SIZE; i++)
                CFS_INIT_LIST_HEAD(&usock_data.ud_peers[i]);
        
        pthread_rwlock_init(&usock_data.ud_peers_lock, NULL);

        /* Spawn poll threads */
        for (i = 0; i < usock_data.ud_npollthreads; i++) {
                rc = cfs_create_thread(usocklnd_poll_thread,