socklnd_cb.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"

ksock_tx_t *
ksocknal_alloc_tx (int size) 
{
        ksock_tx_t *tx = NULL;

        if (size == KSOCK_NOOP_TX_SIZE) {
                /* searching for a noop tx in free list */
                spin_lock(&ksocknal_data.ksnd_tx_lock);

                if (!list_empty(&ksocknal_data.ksnd_idle_noop_txs)) {
                        tx = list_entry(ksocknal_data.ksnd_idle_noop_txs.next,
                                        ksock_tx_t, tx_list);
                        LASSERT(tx->tx_desc_size == size);
                        list_del(&tx->tx_list);
                }
                             
                spin_unlock(&ksocknal_data.ksnd_tx_lock);
        }
                
        if (tx == NULL)
                LIBCFS_ALLOC(tx, size);

        if (tx == NULL) 
                return NULL;

        atomic_set(&tx->tx_refcount, 1);
        tx->tx_desc_size = size;
        atomic_inc(&ksocknal_data.ksnd_nactive_txs);

        return tx;
}

void
ksocknal_free_tx (ksock_tx_t *tx)
{
        atomic_dec(&ksocknal_data.ksnd_nactive_txs);

        if (tx->tx_desc_size == KSOCK_NOOP_TX_SIZE) {
                /* it's a noop tx */
                spin_lock(&ksocknal_data.ksnd_tx_lock);

                list_add(&tx->tx_list, &ksocknal_data.ksnd_idle_noop_txs);

                spin_unlock(&ksocknal_data.ksnd_tx_lock);
        } else {
                LIBCFS_FREE(tx, tx->tx_desc_size);
        }
}

void
ksocknal_init_msg(ksock_msg_t *msg, int type)
{
        msg->ksm_type           = type;
        msg->ksm_csum           = 0;
        msg->ksm_zc_req_cookie  = 0;
        msg->ksm_zc_ack_cookie  = 0;
}

int
ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
{ 
        struct iovec  *iov = tx->tx_iov;
        int    nob;
        int    rc;

        LASSERT (tx->tx_niov > 0);

        /* Never touch tx->tx_iov inside ksocknal_lib_send_iov() */
        rc = ksocknal_lib_send_iov(conn, tx);

        if (rc <= 0)                            /* sent nothing? */ 
                return (rc);

        nob = rc; 
        LASSERT (nob <= tx->tx_resid); 
        tx->tx_resid -= nob;

        /* "consume" iov */ 
        do { 
                LASSERT (tx->tx_niov > 0); 

                if (nob < iov->iov_len) { 
                        iov->iov_base = (void *)(((unsigned long)(iov->iov_base)) + nob); 
                        iov->iov_len -= nob; 
                        return (rc); 
                } 

                nob -= iov->iov_len; 
                tx->tx_iov = ++iov; 
                tx->tx_niov--; 
        } while (nob != 0);

        return (rc);
}

int
ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
{ 
        lnet_kiov_t    *kiov = tx->tx_kiov;
        int     nob;
        int     rc;

        LASSERT (tx->tx_niov == 0); 
        LASSERT (tx->tx_nkiov > 0);

        /* Never touch tx->tx_kiov inside ksocknal_lib_send_kiov() */
        rc = ksocknal_lib_send_kiov(conn, tx);

        if (rc <= 0)                            /* sent nothing? */ 
                return (rc); 
        
        nob = rc; 
        LASSERT (nob <= tx->tx_resid); 
        tx->tx_resid -= nob; 

        /* "consume" kiov */ 
        do { 
                LASSERT(tx->tx_nkiov > 0); 

                if (nob < kiov->kiov_len) { 
                        kiov->kiov_offset += nob; 
                        kiov->kiov_len -= nob; 
                        return rc; 
                } 

                nob -= kiov->kiov_len; 
                tx->tx_kiov = ++kiov; 
                tx->tx_nkiov--; 
        } while (nob != 0);

        return (rc);
}

int
ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
{
        int      rc;
        int      bufnob;
        
        if (ksocknal_data.ksnd_stall_tx != 0) {
                cfs_pause(cfs_time_seconds(ksocknal_data.ksnd_stall_tx));
        }

        LASSERT (tx->tx_resid != 0);

        rc = ksocknal_connsock_addref(conn);
        if (rc != 0) {
                LASSERT (conn->ksnc_closing);
                return (-ESHUTDOWN);
        }

        do {
                if (ksocknal_data.ksnd_enomem_tx > 0) {
                        /* testing... */
                        ksocknal_data.ksnd_enomem_tx--;
                        rc = -EAGAIN;
                } else if (tx->tx_niov != 0) {
                        rc = ksocknal_send_iov (conn, tx);
                } else {
                        rc = ksocknal_send_kiov (conn, tx);
                }

                bufnob = SOCK_WMEM_QUEUED(conn->ksnc_sock);
                if (rc > 0)                     /* sent something? */
                        conn->ksnc_tx_bufnob += rc; /* account it */
                
                if (bufnob < conn->ksnc_tx_bufnob) {
                        /* allocated send buffer bytes < computed; infer
                         * something got ACKed */
                        conn->ksnc_tx_deadline = 
                                cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
                        conn->ksnc_peer->ksnp_last_alive = cfs_time_current();
                        conn->ksnc_tx_bufnob = bufnob;
                        mb();
                }

                if (rc <= 0) { /* Didn't write anything? */
                        ksock_sched_t *sched;

                        if (rc == 0) /* some stacks return 0 instead of -EAGAIN */
                                rc = -EAGAIN;

                        if (rc != -EAGAIN)
                                break;

                        /* Check if EAGAIN is due to memory pressure */

                        sched = conn->ksnc_scheduler;
                        spin_lock_bh (&sched->kss_lock);
                                
                        if (!SOCK_TEST_NOSPACE(conn->ksnc_sock) &&
                            !conn->ksnc_tx_ready) {
                                /* SOCK_NOSPACE is set when the socket fills
                                 * and cleared in the write_space callback
                                 * (which also sets ksnc_tx_ready).  If
                                 * SOCK_NOSPACE and ksnc_tx_ready are BOTH
                                 * zero, I didn't fill the socket and
                                 * write_space won't reschedule me, so I
                                 * return -ENOMEM to get my caller to retry
                                 * after a timeout */
                                rc = -ENOMEM;
                        }

                        spin_unlock_bh (&sched->kss_lock);
                        break;
                }

                /* socket's wmem_queued now includes 'rc' bytes */
                atomic_sub (rc, &conn->ksnc_tx_nob);
                rc = 0;

        } while (tx->tx_resid != 0);

        ksocknal_connsock_decref(conn);
        return (rc);
}

int
ksocknal_recv_iov (ksock_conn_t *conn)
{ 
        struct iovec *iov = conn->ksnc_rx_iov;
        int     nob;
        int     rc;

        LASSERT (conn->ksnc_rx_niov > 0);

        /* Never touch conn->ksnc_rx_iov or change connection 
         * status inside ksocknal_lib_recv_iov */
        rc = ksocknal_lib_recv_iov(conn); 

        if (rc <= 0) 
                return (rc); 

        /* received something... */ 
        nob = rc; 
        
        conn->ksnc_peer->ksnp_last_alive = cfs_time_current(); 
        conn->ksnc_rx_deadline = 
                cfs_time_shift(*ksocknal_tunables.ksnd_timeout); 
        mb();                           /* order with setting rx_started */ 
        conn->ksnc_rx_started = 1; 
        
        conn->ksnc_rx_nob_wanted -= nob; 
        conn->ksnc_rx_nob_left -= nob;

        do { 
                LASSERT (conn->ksnc_rx_niov > 0); 

                if (nob < iov->iov_len) { 
                        iov->iov_len -= nob; 
                        iov->iov_base = (void *)(((unsigned long)iov->iov_base) + nob); 
                        return (-EAGAIN); 
                } 

                nob -= iov->iov_len; 
                conn->ksnc_rx_iov = ++iov; 
                conn->ksnc_rx_niov--; 
        } while (nob != 0);

        return (rc);
}

int
ksocknal_recv_kiov (ksock_conn_t *conn)
{
        lnet_kiov_t   *kiov = conn->ksnc_rx_kiov;
        int     nob;
        int     rc;
        LASSERT (conn->ksnc_rx_nkiov > 0);

        /* Never touch conn->ksnc_rx_kiov or change connection 
         * status inside ksocknal_lib_recv_iov */
        rc = ksocknal_lib_recv_kiov(conn); 
        
        if (rc <= 0) 
                return (rc); 
        
        /* received something... */ 
        nob = rc; 

        conn->ksnc_peer->ksnp_last_alive = cfs_time_current(); 
        conn->ksnc_rx_deadline = 
                cfs_time_shift(*ksocknal_tunables.ksnd_timeout); 
        mb();                           /* order with setting rx_started */ 
        conn->ksnc_rx_started = 1;

        conn->ksnc_rx_nob_wanted -= nob; 
        conn->ksnc_rx_nob_left -= nob; 
        
        do { 
                LASSERT (conn->ksnc_rx_nkiov > 0); 

                if (nob < kiov->kiov_len) { 
                        kiov->kiov_offset += nob; 
                        kiov->kiov_len -= nob; 
                        return -EAGAIN; 
                } 

                nob -= kiov->kiov_len; 
                conn->ksnc_rx_kiov = ++kiov; 
                conn->ksnc_rx_nkiov--; 
        } while (nob != 0);

        return 1;
}

int
ksocknal_receive (ksock_conn_t *conn) 
{
        /* Return 1 on success, 0 on EOF, < 0 on error.
         * Caller checks ksnc_rx_nob_wanted to determine
         * progress/completion. */
        int     rc;
        ENTRY;
        
        if (ksocknal_data.ksnd_stall_rx != 0) {
                cfs_pause(cfs_time_seconds (ksocknal_data.ksnd_stall_rx));
        }

        rc = ksocknal_connsock_addref(conn);
        if (rc != 0) {
                LASSERT (conn->ksnc_closing);
                return (-ESHUTDOWN);
        }

        for (;;) {
                if (conn->ksnc_rx_niov != 0)
                        rc = ksocknal_recv_iov (conn);
                else
                        rc = ksocknal_recv_kiov (conn);

                if (rc <= 0) {
                        /* error/EOF or partial receive */
                        if (rc == -EAGAIN) {
                                rc = 1;