socklnd.h

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *   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 Lustre, http://www.lustre.org
 *
 *   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.
 *
 */

#define DEBUG_PORTAL_ALLOC
#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif

#define DEBUG_SUBSYSTEM S_LND

#if defined(__linux__)
#include "socklnd_lib-linux.h"
#elif defined(__APPLE__)
#include "socklnd_lib-darwin.h"
#elif defined(__WINNT__)
#include "socklnd_lib-winnt.h"
#else
#error Unsupported Operating System
#endif

#include <libcfs/kp30.h>
#include <lnet/lnet.h>
#include <lnet/lib-lnet.h>
#include <lnet/socklnd.h>

#define SOCKNAL_PEER_HASH_SIZE  101             /* # peer lists */
#define SOCKNAL_RESCHED         100             /* # scheduler loops before reschedule */
#define SOCKNAL_ENOMEM_RETRY    CFS_TICK        /* jiffies between retries */

#define SOCKNAL_ROUND_ROBIN     0               /* round robin / load balance */

#define SOCKNAL_SINGLE_FRAG_TX      0           /* disable multi-fragment sends */
#define SOCKNAL_SINGLE_FRAG_RX      0           /* disable multi-fragment receives */

#define SOCKNAL_VERSION_DEBUG       0           /* enable protocol version debugging */

/* risk kmap deadlock on multi-frag I/O (backs off to single-frag if disabled).
 * no risk if we're not running on a CONFIG_HIGHMEM platform. */
#ifdef CONFIG_HIGHMEM
# define SOCKNAL_RISK_KMAP_DEADLOCK  0
#else
# define SOCKNAL_RISK_KMAP_DEADLOCK  1
#endif

typedef struct                                  /* per scheduler state */
{
        spinlock_t        kss_lock;             /* serialise */
        struct list_head  kss_rx_conns;         /* conn waiting to be read */
        struct list_head  kss_tx_conns;         /* conn waiting to be written */
        struct list_head  kss_zombie_noop_txs;  /* zombie noop tx list */
        cfs_waitq_t       kss_waitq;            /* where scheduler sleeps */
        int               kss_nconns;           /* # connections assigned to this scheduler */
} ksock_sched_t;

typedef struct
{
        unsigned int      ksni_valid:1;         /* been set yet? */
        unsigned int      ksni_bound:1;         /* bound to a cpu yet? */
        unsigned int      ksni_sched:6;         /* which scheduler (assumes < 64) */
} ksock_irqinfo_t;

typedef struct                                  /* in-use interface */
{
        __u32             ksni_ipaddr;          /* interface's IP address */
        __u32             ksni_netmask;         /* interface's network mask */
        int               ksni_nroutes;         /* # routes using (active) */
        int               ksni_npeers;          /* # peers using (passive) */
        char              ksni_name[16];        /* interface name */
} ksock_interface_t;

typedef struct
{
        int              *ksnd_timeout;         /* "stuck" socket timeout (seconds) */
        int              *ksnd_nconnds;         /* # connection daemons */
        int              *ksnd_min_reconnectms; /* first connection retry after (ms)... */
        int              *ksnd_max_reconnectms; /* ...exponentially increasing to this */
        int              *ksnd_eager_ack;       /* make TCP ack eagerly? */
        int              *ksnd_typed_conns;     /* drive sockets by type? */
        int              *ksnd_min_bulk;        /* smallest "large" message */
        int              *ksnd_tx_buffer_size;  /* socket tx buffer size */
        int              *ksnd_rx_buffer_size;  /* socket rx buffer size */
        int              *ksnd_nagle;           /* enable NAGLE? */
        int              *ksnd_keepalive_idle;  /* # idle secs before 1st probe */
        int              *ksnd_keepalive_count; /* # probes */
        int              *ksnd_keepalive_intvl; /* time between probes */
        int              *ksnd_credits;         /* # concurrent sends */
        int              *ksnd_peercredits;     /* # concurrent sends to 1 peer */
        int              *ksnd_enable_csum;     /* enable check sum */
        int              *ksnd_inject_csum_error; /* set non-zero to inject checksum error */
        unsigned int     *ksnd_zc_min_frag;     /* minimum zero copy frag size */
#ifdef CPU_AFFINITY
        int              *ksnd_irq_affinity;    /* enable IRQ affinity? */
#endif
#ifdef SOCKNAL_BACKOFF
        int              *ksnd_backoff_init;    /* initial TCP backoff */
        int              *ksnd_backoff_max;     /* maximum TCP backoff */
#endif
#if SOCKNAL_VERSION_DEBUG
        int              *ksnd_protocol;        /* protocol version */
#endif
#if defined(CONFIG_SYSCTL) && !CFS_SYSFS_MODULE_PARM
        cfs_sysctl_table_header_t *ksnd_sysctl;   /* sysctl interface */
#endif
} ksock_tunables_t;

typedef struct
{
        __u64             ksnn_incarnation;     /* my epoch */
        spinlock_t        ksnn_lock;            /* serialise */
        int               ksnn_npeers;          /* # peers */
        int               ksnn_shutdown;        /* shutting down? */
        int               ksnn_ninterfaces;     /* IP interfaces */
        ksock_interface_t ksnn_interfaces[LNET_MAX_INTERFACES];
} ksock_net_t;

typedef struct
{
        int               ksnd_init;            /* initialisation state */
        int               ksnd_nnets;           /* # networks set up */

        rwlock_t          ksnd_global_lock;     /* stabilize peer/conn ops */
        struct list_head *ksnd_peers;           /* hash table of all my known peers */
        int               ksnd_peer_hash_size;  /* size of ksnd_peers */

        int               ksnd_nthreads;        /* # live threads */
        int               ksnd_shuttingdown;    /* tell threads to exit */
        int               ksnd_nschedulers;     /* # schedulers */
        ksock_sched_t    *ksnd_schedulers;      /* their state */

        atomic_t          ksnd_nactive_txs;     /* #active txs */

        struct list_head  ksnd_deathrow_conns;  /* conns to close: reaper_lock*/
        struct list_head  ksnd_zombie_conns;    /* conns to free: reaper_lock */
        struct list_head  ksnd_enomem_conns;    /* conns to retry: reaper_lock*/
        cfs_waitq_t       ksnd_reaper_waitq;    /* reaper sleeps here */
        cfs_time_t        ksnd_reaper_waketime; /* when reaper will wake */
        spinlock_t        ksnd_reaper_lock;     /* serialise */

        int               ksnd_enomem_tx;       /* test ENOMEM sender */
        int               ksnd_stall_tx;        /* test sluggish sender */
        int               ksnd_stall_rx;        /* test sluggish receiver */

        struct list_head  ksnd_connd_connreqs;  /* incoming connection requests */
        struct list_head  ksnd_connd_routes;    /* routes waiting to be connected */
        cfs_waitq_t       ksnd_connd_waitq;     /* connds sleep here */
        int               ksnd_connd_connecting;/* # connds connecting */
        spinlock_t        ksnd_connd_lock;      /* serialise */

        struct list_head  ksnd_idle_noop_txs;   /* list head for freed noop tx */
        spinlock_t        ksnd_tx_lock;         /* serialise, NOT safe in g_lock */

        ksock_irqinfo_t   ksnd_irqinfo[NR_IRQS];/* irq->scheduler lookup */

} ksock_nal_data_t;

#define SOCKNAL_INIT_NOTHING    0
#define SOCKNAL_INIT_DATA       1
#define SOCKNAL_INIT_ALL        2

/* A packet just assembled for transmission is represented by 1 or more
 * struct iovec fragments (the first frag contains the portals header),
 * followed by 0 or more lnet_kiov_t fragments.
 *
 * On the receive side, initially 1 struct iovec fragment is posted for
 * receive (the header).  Once the header has been received, the payload is
 * received into either struct iovec or lnet_kiov_t fragments, depending on
 * what the header matched or whether the message needs forwarding. */

struct ksock_conn;                              /* forward ref */
struct ksock_peer;                              /* forward ref */
struct ksock_route;                             /* forward ref */
struct ksock_proto;                             /* forward ref */

typedef struct                                  /* transmit packet */
{
        struct list_head        tx_list;        /* queue on conn for transmission etc */
        struct list_head        tx_zc_list;     /* queue on peer for ZC request */
        atomic_t                tx_refcount;    /* tx reference count */
        int                     tx_nob;         /* # packet bytes */
        int                     tx_resid;       /* residual bytes */
        int                     tx_niov;        /* # packet iovec frags */
        struct iovec           *tx_iov;         /* packet iovec frags */
        int                     tx_nkiov;       /* # packet page frags */
        unsigned int            tx_checked_zc;  /* Have I checked if I should ZC? */
        lnet_kiov_t            *tx_kiov;        /* packet page frags */
        struct ksock_conn      *tx_conn;        /* owning conn */
        lnet_msg_t             *tx_lnetmsg;     /* lnet message for lnet_finalize() */
        ksock_msg_t             tx_msg;         /* socklnd message buffer */
        int                     tx_desc_size;   /* size of this descriptor */
        union {
                struct {
                        struct iovec iov;       /* virt hdr */
                        lnet_kiov_t  kiov[0];   /* paged payload */
                }                  paged;
                struct {
                        struct iovec iov[1];    /* virt hdr + payload */
                }                  virt;
        }                       tx_frags;
} ksock_tx_t;

#define KSOCK_NOOP_TX_SIZE      offsetof(ksock_tx_t, tx_frags.paged.kiov[0])

/* network zero copy callback descriptor embedded in ksock_tx_t */

/* space for the rx frag descriptors; we either read a single contiguous
 * header, or up to LNET_MAX_IOV frags of payload of either type. */
typedef union {
        struct iovec     iov[LNET_MAX_IOV];
        lnet_kiov_t      kiov[LNET_MAX_IOV];
} ksock_rxiovspace_t;

#define SOCKNAL_RX_KSM_HEADER   1               /* reading ksock message header */
#define SOCKNAL_RX_LNET_HEADER  2               /* reading lnet message header */
#define SOCKNAL_RX_PARSE        3               /* Calling lnet_parse() */
#define SOCKNAL_RX_PARSE_WAIT   4               /* waiting to be told to read the body */
#define SOCKNAL_RX_LNET_PAYLOAD 5               /* reading lnet payload (to deliver here) */
#define SOCKNAL_RX_SLOP         6               /* skipping body */

typedef struct ksock_conn
{
        struct ksock_peer  *ksnc_peer;          /* owning peer */
        struct ksock_route *ksnc_route;         /* owning route */
        struct list_head    ksnc_list;          /* stash on peer's conn list */
        cfs_socket_t       *ksnc_sock;          /* actual socket */
        void               *ksnc_saved_data_ready; /* socket's original data_ready() callback */
        void               *ksnc_saved_write_space; /* socket's original write_space() callback */
        atomic_t            ksnc_conn_refcount; /* conn refcount */
        atomic_t            ksnc_sock_refcount; /* sock refcount */
        ksock_sched_t	   *ksnc_scheduler;     /* who schedules this connection */
        __u32               ksnc_myipaddr;      /* my IP */
        __u32               ksnc_ipaddr;        /* peer's IP */
        int                 ksnc_port;          /* peer's port */
        int                 ksnc_type:3;        /* type of connection, should be signed value */
        int                 ksnc_closing:1;     /* being shut down */
        int                 ksnc_flip:1;        /* flip or not, only for V2.x */
        int                 ksnc_zc_capable:1;  /* enable to ZC */
        struct ksock_proto *ksnc_proto;         /* protocol for the connection */

        /* reader */
        struct list_head    ksnc_rx_list;       /* where I enq waiting input or a forwarding descriptor */
        cfs_time_t          ksnc_rx_deadline;   /* when (in jiffies) receive times out */
        __u8                ksnc_rx_started;    /* started receiving a message */
        __u8                ksnc_rx_ready;      /* data ready to read */
        __u8                ksnc_rx_scheduled;  /* being progressed */
        __u8                ksnc_rx_state;      /* what is being read */
        int                 ksnc_rx_nob_left;   /* # bytes to next hdr/body  */
        int                 ksnc_rx_nob_wanted; /* bytes actually wanted */
        int                 ksnc_rx_niov;       /* # iovec frags */
        struct iovec       *ksnc_rx_iov;        /* the iovec frags */
        int                 ksnc_rx_nkiov;      /* # page frags */
        lnet_kiov_t        *ksnc_rx_kiov;       /* the page frags */
        ksock_rxiovspace_t  ksnc_rx_iov_space;  /* space for frag descriptors */