socklnd.h

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

        __u32               ksnc_rx_csum;       /* partial checksum for incoming data */
        void               *ksnc_cookie;        /* rx lnet_finalize passthru arg */
        ksock_msg_t         ksnc_msg;           /* incoming message buffer:
                                                 * V2.x message takes the whole struct
                                                 * V1.x message is a bare lnet_hdr_t, it's stored
                                                 * in ksnc_msg.ksm_u.lnetmsg */

        /* WRITER */
        struct list_head    ksnc_tx_list;       /* where I enq waiting for output space */
        struct list_head    ksnc_tx_queue;      /* packets waiting to be sent */
        ksock_tx_t         *ksnc_tx_mono;       /* V2.x only, next mono-packet, mono-packet is :
                                                 * a. lnet packet without piggyback
                                                 * b. noop ZC-ACK packet */
        cfs_time_t          ksnc_tx_deadline;   /* when (in jiffies) tx times out */
        int                 ksnc_tx_bufnob;     /* send buffer marker */
        atomic_t            ksnc_tx_nob;        /* # bytes queued */
        int                 ksnc_tx_ready;      /* write space */
        int                 ksnc_tx_scheduled;  /* being progressed */
        
#if !SOCKNAL_SINGLE_FRAG_RX
        struct iovec        ksnc_rx_scratch_iov[LNET_MAX_IOV];
#endif
#if !SOCKNAL_SINGLE_FRAG_TX
        struct iovec        ksnc_tx_scratch_iov[LNET_MAX_IOV];
#endif
} ksock_conn_t;

typedef struct ksock_route
{
        struct list_head    ksnr_list;          /* chain on peer route list */
        struct list_head    ksnr_connd_list;    /* chain on ksnr_connd_routes */
        struct ksock_peer  *ksnr_peer;          /* owning peer */
        atomic_t            ksnr_refcount;      /* # users */
        cfs_time_t          ksnr_timeout;       /* when (in jiffies) reconnection can happen next */
        cfs_duration_t      ksnr_retry_interval; /* how long between retries */
        __u32               ksnr_myipaddr;      /* my IP */
        __u32               ksnr_ipaddr;        /* IP address to connect to */
        int                 ksnr_port;          /* port to connect to */
        unsigned int        ksnr_scheduled:1;   /* scheduled for attention */
        unsigned int        ksnr_connecting:1;  /* connection establishment in progress */
        unsigned int        ksnr_connected:4;   /* connections established by type */
        unsigned int        ksnr_deleted:1;     /* been removed from peer? */
        unsigned int        ksnr_share_count;   /* created explicitly? */
        int                 ksnr_conn_count;    /* # conns established by this route */
} ksock_route_t;

typedef struct ksock_peer
{
        struct list_head    ksnp_list;          /* stash on global peer list */
        lnet_process_id_t   ksnp_id;            /* who's on the other end(s) */
        atomic_t            ksnp_refcount;      /* # users */
        int                 ksnp_sharecount;    /* lconf usage counter */
        int                 ksnp_closing;       /* being closed */
        int                 ksnp_accepting;     /* # passive connections pending */
        int                 ksnp_error;         /* errno on closing last conn */
        __u64               ksnp_zc_next_cookie;/* ZC completion cookie */
        __u64               ksnp_incarnation;   /* latest known peer incarnation */
        struct ksock_proto *ksnp_proto;         /* latest known peer protocol */
        struct list_head    ksnp_conns;         /* all active connections */
        struct list_head    ksnp_routes;        /* routes */
        struct list_head    ksnp_tx_queue;      /* waiting packets */
        spinlock_t          ksnp_lock;          /* serialize, NOT safe in g_lock */
        struct list_head    ksnp_zc_req_list;   /* zero copy requests wait for ACK  */
        cfs_time_t          ksnp_last_alive;    /* when (in jiffies) I was last alive */
        lnet_ni_t          *ksnp_ni;            /* which network */
        int                 ksnp_n_passive_ips; /* # of... */
        __u32               ksnp_passive_ips[LNET_MAX_INTERFACES]; /* preferred local interfaces */
} ksock_peer_t;

typedef struct ksock_connreq
{
        struct list_head    ksncr_list;         /* stash on ksnd_connd_connreqs */
        lnet_ni_t          *ksncr_ni;           /* chosen NI */
        cfs_socket_t       *ksncr_sock;         /* accepted socket */
} ksock_connreq_t;

extern ksock_nal_data_t ksocknal_data;
extern ksock_tunables_t ksocknal_tunables;

typedef struct ksock_proto
{
        int     pro_version;                                                /* version number of protocol */
        int     (*pro_send_hello)(ksock_conn_t *, ksock_hello_msg_t *);     /* handshake function */
        int     (*pro_recv_hello)(ksock_conn_t *, ksock_hello_msg_t *, int);/* handshake function */
        void    (*pro_pack)(ksock_tx_t *);                                  /* message pack */
        void    (*pro_unpack)(ksock_msg_t *);                               /* message unpack */
} ksock_proto_t;

extern ksock_proto_t ksocknal_protocol_v1x;
extern ksock_proto_t ksocknal_protocol_v2x;

#define KSOCK_PROTO_V1_MAJOR    LNET_PROTO_TCP_VERSION_MAJOR
#define KSOCK_PROTO_V1_MINOR    LNET_PROTO_TCP_VERSION_MINOR
#define KSOCK_PROTO_V1          KSOCK_PROTO_V1_MAJOR

static inline int
ksocknal_route_mask(void) 
{
        if (!*ksocknal_tunables.ksnd_typed_conns)
                return (1 << SOCKLND_CONN_ANY);
        
        return ((1 << SOCKLND_CONN_CONTROL) |
                (1 << SOCKLND_CONN_BULK_IN) |
                (1 << SOCKLND_CONN_BULK_OUT));
}

static inline struct list_head *
ksocknal_nid2peerlist (lnet_nid_t nid)
{
        unsigned int hash = ((unsigned int)nid) % ksocknal_data.ksnd_peer_hash_size;

        return (&ksocknal_data.ksnd_peers [hash]);
}

static inline void
ksocknal_conn_addref (ksock_conn_t *conn)
{
        LASSERT (atomic_read(&conn->ksnc_conn_refcount) > 0);
        atomic_inc(&conn->ksnc_conn_refcount);
}

extern void ksocknal_queue_zombie_conn (ksock_conn_t *conn);

static inline void
ksocknal_conn_decref (ksock_conn_t *conn)
{
        LASSERT (atomic_read(&conn->ksnc_conn_refcount) > 0);
        if (atomic_dec_and_test(&conn->ksnc_conn_refcount))
                ksocknal_queue_zombie_conn(conn);
}

static inline int
ksocknal_connsock_addref (ksock_conn_t *conn)
{
        int   rc = -ESHUTDOWN;

        read_lock (&ksocknal_data.ksnd_global_lock);
        if (!conn->ksnc_closing) {
                LASSERT (atomic_read(&conn->ksnc_sock_refcount) > 0);
                atomic_inc(&conn->ksnc_sock_refcount);
                rc = 0;
        }
        read_unlock (&ksocknal_data.ksnd_global_lock);

        return (rc);
}

static inline void
ksocknal_connsock_decref (ksock_conn_t *conn)
{
        LASSERT (atomic_read(&conn->ksnc_sock_refcount) > 0);
        if (atomic_dec_and_test(&conn->ksnc_sock_refcount)) {
                LASSERT (conn->ksnc_closing);
                libcfs_sock_release(conn->ksnc_sock);
                conn->ksnc_sock = NULL;
        }
}

static inline void
ksocknal_tx_addref (ksock_tx_t *tx)
{
        LASSERT (atomic_read(&tx->tx_refcount) > 0);
        atomic_inc(&tx->tx_refcount);
}

extern void ksocknal_tx_done (lnet_ni_t *ni, ksock_tx_t *tx);

static inline void
ksocknal_tx_decref (ksock_tx_t *tx)
{
        LASSERT (atomic_read(&tx->tx_refcount) > 0);
        if (atomic_dec_and_test(&tx->tx_refcount))
                ksocknal_tx_done(NULL, tx);
}

static inline void
ksocknal_route_addref (ksock_route_t *route)
{
        LASSERT (atomic_read(&route->ksnr_refcount) > 0);
        atomic_inc(&route->ksnr_refcount);
}

extern void ksocknal_destroy_route (ksock_route_t *route);

static inline void
ksocknal_route_decref (ksock_route_t *route)
{
        LASSERT (atomic_read (&route->ksnr_refcount) > 0);
        if (atomic_dec_and_test(&route->ksnr_refcount))
                ksocknal_destroy_route (route);
}

static inline void
ksocknal_peer_addref (ksock_peer_t *peer)
{
        LASSERT (atomic_read (&peer->ksnp_refcount) > 0);
        atomic_inc(&peer->ksnp_refcount);
}

extern void ksocknal_destroy_peer (ksock_peer_t *peer);

static inline void
ksocknal_peer_decref (ksock_peer_t *peer)
{
        LASSERT (atomic_read (&peer->ksnp_refcount) > 0);
        if (atomic_dec_and_test(&peer->ksnp_refcount))
                ksocknal_destroy_peer (peer);
}

int ksocknal_startup (lnet_ni_t *ni);
void ksocknal_shutdown (lnet_ni_t *ni);
int ksocknal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg);
int ksocknal_send (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg);
int ksocknal_recv(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, 
                  int delayed, unsigned int niov, 
                  struct iovec *iov, lnet_kiov_t *kiov,
                  unsigned int offset, unsigned int mlen, unsigned int rlen);
int ksocknal_accept(lnet_ni_t *ni, cfs_socket_t *sock);

extern int ksocknal_add_peer(lnet_ni_t *ni, lnet_process_id_t id, __u32 ip, int port);
extern ksock_peer_t *ksocknal_find_peer_locked (lnet_ni_t *ni, lnet_process_id_t id);
extern ksock_peer_t *ksocknal_find_peer (lnet_ni_t *ni, lnet_process_id_t id);
extern void ksocknal_peer_failed (ksock_peer_t *peer);
extern int ksocknal_create_conn (lnet_ni_t *ni, ksock_route_t *route,
                                 cfs_socket_t *sock, int type);
extern void ksocknal_close_conn_locked (ksock_conn_t *conn, int why);
extern void ksocknal_terminate_conn (ksock_conn_t *conn);
extern void ksocknal_destroy_conn (ksock_conn_t *conn);
extern int  ksocknal_close_peer_conns_locked (ksock_peer_t *peer,
                                              __u32 ipaddr, int why);
extern int ksocknal_close_conn_and_siblings (ksock_conn_t *conn, int why);
extern int ksocknal_close_matching_conns (lnet_process_id_t id, __u32 ipaddr);

extern void ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn);
extern void ksocknal_txlist_done (lnet_ni_t *ni, struct list_head *txlist, int error);
extern void ksocknal_notify (lnet_ni_t *ni, lnet_nid_t gw_nid, int alive);
extern int ksocknal_thread_start (int (*fn)(void *arg), void *arg);
extern void ksocknal_thread_fini (void);
extern ksock_route_t *ksocknal_find_connecting_route_locked (ksock_peer_t *peer);
extern int ksocknal_new_packet (ksock_conn_t *conn, int skip);
extern int ksocknal_scheduler (void *arg);
extern int ksocknal_connd (void *arg);
extern int ksocknal_reaper (void *arg);
extern int ksocknal_send_hello (lnet_ni_t *ni, ksock_conn_t *conn,
                                lnet_nid_t peer_nid, ksock_hello_msg_t *hello);
extern int ksocknal_recv_hello (lnet_ni_t *ni, ksock_conn_t *conn, 
                                ksock_hello_msg_t *hello, lnet_process_id_t *id,
                                __u64 *incarnation);
extern void ksocknal_read_callback(ksock_conn_t *conn);
extern void ksocknal_write_callback(ksock_conn_t *conn);

extern int ksocknal_lib_zc_capable(cfs_socket_t *sock);
extern void ksocknal_lib_save_callback(cfs_socket_t *sock, ksock_conn_t *conn);
extern void ksocknal_lib_set_callback(cfs_socket_t *sock,  ksock_conn_t *conn);
extern void ksocknal_lib_reset_callback(cfs_socket_t *sock, ksock_conn_t *conn);
extern void ksocknal_lib_push_conn (ksock_conn_t *conn);
extern void ksocknal_lib_bind_irq (unsigned int irq);
extern int ksocknal_lib_get_conn_addrs (ksock_conn_t *conn);
extern unsigned int ksocknal_lib_sock_irq (cfs_socket_t *sock);
extern int ksocknal_lib_setup_sock (cfs_socket_t *so);
extern int ksocknal_lib_send_iov (ksock_conn_t *conn, ksock_tx_t *tx);
extern int ksocknal_lib_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx);
extern void ksocknal_lib_eager_ack (ksock_conn_t *conn);
extern int ksocknal_lib_recv_iov (ksock_conn_t *conn);
extern int ksocknal_lib_recv_kiov (ksock_conn_t *conn);
extern int ksocknal_lib_get_conn_tunables (ksock_conn_t *conn, int *txmem, 
                                           int *rxmem, int *nagle);

extern int ksocknal_lib_tunables_init(void);
extern void ksocknal_lib_tunables_fini(void);

extern void ksocknal_lib_csum_tx(ksock_tx_t *tx);