ptllnd.c

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

        /* checksum must be computed with
         * 1) ptlm_cksum zero and
         * 2) BEFORE anything gets modified/flipped
         */
        msg_cksum = flip ? __swab32(msg->ptlm_cksum) : msg->ptlm_cksum;
        msg->ptlm_cksum = 0;
        if (msg_cksum != 0 &&
            msg_cksum != kptllnd_cksum(msg, hdr_size)) {
                CERROR("Bad checksum\n");
                return -EPROTO;
        }

        msg->ptlm_version = msg_version;
        msg->ptlm_cksum = msg_cksum;
        
        if (flip) {
                /* These two are 1 byte long so we don't swap them
                   But check this assumtion*/
                CLASSERT (sizeof(msg->ptlm_type) == 1);
                CLASSERT (sizeof(msg->ptlm_credits) == 1);
                /* src & dst stamps are opaque cookies */
                __swab32s(&msg->ptlm_nob);
                __swab64s(&msg->ptlm_srcnid);
                __swab64s(&msg->ptlm_dstnid);
                __swab32s(&msg->ptlm_srcpid);
                __swab32s(&msg->ptlm_dstpid);
        }

        if (msg->ptlm_nob != nob) {
                CERROR("msg_nob corrupt: got 0x%08x, wanted %08x\n",
                       msg->ptlm_nob, nob);
                return -EPROTO;
        }

        switch(msg->ptlm_type)
        {
        case PTLLND_MSG_TYPE_PUT:
        case PTLLND_MSG_TYPE_GET:
                if (nob < hdr_size + sizeof(kptl_rdma_msg_t)) {
                        CERROR("Short rdma request: got %d, want %d\n",
                               nob, hdr_size + (int)sizeof(kptl_rdma_msg_t));
                        return -EPROTO;
                }

                if (flip)
                        __swab64s(&msg->ptlm_u.rdma.kptlrm_matchbits);

                if (msg->ptlm_u.rdma.kptlrm_matchbits < PTL_RESERVED_MATCHBITS) {
                        CERROR("Bad matchbits "LPX64"\n",
                               msg->ptlm_u.rdma.kptlrm_matchbits);
                        return -EPROTO;
                }
                break;

        case PTLLND_MSG_TYPE_IMMEDIATE:
                if (nob < offsetof(kptl_msg_t, 
                                   ptlm_u.immediate.kptlim_payload)) {
                        CERROR("Short immediate: got %d, want %d\n", nob,
                               (int)offsetof(kptl_msg_t, 
                                             ptlm_u.immediate.kptlim_payload));
                        return -EPROTO;
                }
                /* Do nothing */
                break;
                        
        case PTLLND_MSG_TYPE_NOOP:
        case PTLLND_MSG_TYPE_NAK:
                /* Do nothing */
                break;

        case PTLLND_MSG_TYPE_HELLO:
                if (nob < hdr_size + sizeof(kptl_hello_msg_t)) {
                        CERROR("Short hello: got %d want %d\n",
                               nob, hdr_size + (int)sizeof(kptl_hello_msg_t));
                        return -EPROTO;
                }
                if (flip) {
                        __swab64s(&msg->ptlm_u.hello.kptlhm_matchbits);
                        __swab32s(&msg->ptlm_u.hello.kptlhm_max_msg_size);
                }
                break;

        default:
                CERROR("Bad message type: 0x%02x\n", (__u32)msg->ptlm_type);
                return -EPROTO;
        }

        return 0;
}

int
kptllnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
{
        struct libcfs_ioctl_data *data = arg;
        int          rc = -EINVAL;

        CDEBUG(D_NET, ">>> kptllnd_ctl cmd=%u arg=%p\n", cmd, arg);

        /*
         * Validate that the context block is actually
         * pointing to this interface
         */
        LASSERT (ni == kptllnd_data.kptl_ni);

        switch(cmd) {
        case IOC_LIBCFS_DEL_PEER: {
                lnet_process_id_t id;
                
                id.nid = data->ioc_nid;
                id.pid = data->ioc_u32[1];
                
                rc = kptllnd_peer_del(id);
                break;
        }

        case IOC_LIBCFS_GET_PEER: {
                lnet_process_id_t   id = {.nid = LNET_NID_ANY,
                                          .pid = LNET_PID_ANY};
                __u64               incarnation = 0;
                __u64               next_matchbits = 0;
                __u64               last_matchbits_seen = 0;
                int                 state = 0;
                int                 sent_hello = 0;
                int                 refcount = 0;
                int                 nsendq = 0;
                int                 nactiveq = 0;
                int                 credits = 0;
                int                 outstanding_credits = 0;

                rc = kptllnd_get_peer_info(data->ioc_count, &id,
                                           &state, &sent_hello,
                                           &refcount, &incarnation,
                                           &next_matchbits, &last_matchbits_seen,
                                           &nsendq, &nactiveq,
                                           &credits, &outstanding_credits);
                /* wince... */
                data->ioc_nid = id.nid;
                data->ioc_net = state;
                data->ioc_flags  = sent_hello;
                data->ioc_count = refcount;
                data->ioc_u64[0] = incarnation;
                data->ioc_u32[0] = (__u32)next_matchbits;
                data->ioc_u32[1] = (__u32)(next_matchbits >> 32);
                data->ioc_u32[2] = (__u32)last_matchbits_seen;
                data->ioc_u32[3] = (__u32)(last_matchbits_seen >> 32);
                data->ioc_u32[4] = id.pid;
                data->ioc_u32[5] = (nsendq << 16) | nactiveq;
                data->ioc_u32[6] = (credits << 16) | outstanding_credits;
                break;
        }
                
        default:
                rc=-EINVAL;
                break;
        }
        CDEBUG(D_NET, "<<< kptllnd_ctl rc=%d\n", rc);
        return rc;
}

int
kptllnd_startup (lnet_ni_t *ni)
{
        int             rc;
        int             i;
        int             spares;
        struct timeval  tv;
        ptl_err_t       ptl_rc;

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

        if (kptllnd_data.kptl_init != PTLLND_INIT_NOTHING) {
                CERROR("Only 1 instance supported\n");
                return -EPERM;
        }

        if (*kptllnd_tunables.kptl_max_procs_per_node < 1) {
                CERROR("max_procs_per_node must be >= 1\n");
                return -EINVAL;
        }

        *kptllnd_tunables.kptl_max_msg_size &= ~7;
        if (*kptllnd_tunables.kptl_max_msg_size < PTLLND_MIN_BUFFER_SIZE)
                *kptllnd_tunables.kptl_max_msg_size = PTLLND_MIN_BUFFER_SIZE;

        CLASSERT ((PTLLND_MIN_BUFFER_SIZE & 7) == 0);
        CLASSERT (sizeof(kptl_msg_t) <= PTLLND_MIN_BUFFER_SIZE);

        /*
         * zero pointers, flags etc
         * put everything into a known state.
         */
        memset (&kptllnd_data, 0, sizeof (kptllnd_data));
        kptllnd_data.kptl_eqh = PTL_INVALID_HANDLE;
        kptllnd_data.kptl_nih = PTL_INVALID_HANDLE;

        /*
         * Setup the sched locks/lists/waitq
         */
        spin_lock_init(&kptllnd_data.kptl_sched_lock);
        init_waitqueue_head(&kptllnd_data.kptl_sched_waitq);
        INIT_LIST_HEAD(&kptllnd_data.kptl_sched_txq);
        INIT_LIST_HEAD(&kptllnd_data.kptl_sched_rxq);
        INIT_LIST_HEAD(&kptllnd_data.kptl_sched_rxbq);

        /* init kptl_ptlid2str_lock before any call to kptllnd_ptlid2str */
        spin_lock_init(&kptllnd_data.kptl_ptlid2str_lock);

        /*
         * Setup the tx locks/lists
         */
        spin_lock_init(&kptllnd_data.kptl_tx_lock);
        INIT_LIST_HEAD(&kptllnd_data.kptl_idle_txs);
        atomic_set(&kptllnd_data.kptl_ntx, 0);

        /*
         * Uptick the module reference count
         */
        PORTAL_MODULE_USE;

        /*
         * Setup pointers between the ni and context data block
         */
        kptllnd_data.kptl_ni = ni;
        ni->ni_data = &kptllnd_data;

        /*
         * Setup Credits
         */
        ni->ni_maxtxcredits = *kptllnd_tunables.kptl_credits;
        ni->ni_peertxcredits = *kptllnd_tunables.kptl_peercredits;

        kptllnd_data.kptl_expected_peers =
                *kptllnd_tunables.kptl_max_nodes *
                *kptllnd_tunables.kptl_max_procs_per_node;
        
        /*
         * Initialize the Network interface instance
         * We use the default because we don't have any
         * way to choose a better interface.
         * Requested and actual limits are ignored.
         */
        ptl_rc = PtlNIInit(
#ifdef _USING_LUSTRE_PORTALS_
                PTL_IFACE_DEFAULT,
#else
                CRAY_KERN_NAL,
#endif
                *kptllnd_tunables.kptl_pid, NULL, NULL,
                &kptllnd_data.kptl_nih);

        /*
         * Note: PTL_IFACE_DUP simply means that the requested
         * interface was already inited and that we're sharing it.
         * Which is ok.
         */
        if (ptl_rc != PTL_OK && ptl_rc != PTL_IFACE_DUP) {
                CERROR ("PtlNIInit: error %s(%d)\n",
                        kptllnd_errtype2str(ptl_rc), ptl_rc);
                rc = -EINVAL;
                goto failed;
        }

        /* NB eq size irrelevant if using a callback */
        ptl_rc = PtlEQAlloc(kptllnd_data.kptl_nih,
                            8,                       /* size */
                            kptllnd_eq_callback,     /* handler callback */
                            &kptllnd_data.kptl_eqh); /* output handle */
        if (ptl_rc != PTL_OK) {
                CERROR("PtlEQAlloc failed %s(%d)\n",
                       kptllnd_errtype2str(ptl_rc), ptl_rc);
                rc = -ENOMEM;
                goto failed;
        }

        /*
         * Fetch the lower NID
         */
        ptl_rc = PtlGetId(kptllnd_data.kptl_nih,
                          &kptllnd_data.kptl_portals_id);
        if (ptl_rc != PTL_OK) {
                CERROR ("PtlGetID: error %s(%d)\n",
                        kptllnd_errtype2str(ptl_rc), ptl_rc);
                rc = -EINVAL;
                goto failed;
        }

        if (kptllnd_data.kptl_portals_id.pid != *kptllnd_tunables.kptl_pid) {
                /* The kernel ptllnd must have the expected PID */
                CERROR("Unexpected PID: %u (%u expected)\n",
                       kptllnd_data.kptl_portals_id.pid,
                       *kptllnd_tunables.kptl_pid);
                rc = -EINVAL;
                goto failed;
        }

        ni->ni_nid = kptllnd_ptl2lnetnid(kptllnd_data.kptl_portals_id.nid);