iiblnd.c

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

        int                    *rcp = arg;
        FSTATUS                 frc = qry_result->Status;
        SERVICE_RECORD_RESULTS *svc_rslt;
        IB_SERVICE_RECORD      *svc;
        lnet_nid_t              nid;

        if (frc != FSUCCESS || qry_result->ResultDataSize == 0) {
                CERROR("Error checking advert: status %d data size %d\n",
                       frc, qry_result->ResultDataSize);
                *rcp = -EIO;
                goto out;
        }

        svc_rslt = (SERVICE_RECORD_RESULTS *)qry_result->QueryResult;

        if (svc_rslt->NumServiceRecords < 1) {
                CERROR("Check advert: %d records\n",
                       svc_rslt->NumServiceRecords);
                *rcp = -ENOENT;
                goto out;
        }

        svc = &svc_rslt->ServiceRecords[0];
        nid = le64_to_cpu(*kibnal_service_nid_field(svc));
        
        CDEBUG(D_NET, "Check advert: %s "LPX64" "LPX64":%04x\n",
               libcfs_nid2str(nid), svc->RID.ServiceID, 
               svc->RID.ServiceGID.Type.Global.InterfaceID, 
               svc->RID.ServiceP_Key);

        if (nid != kibnal_data.kib_ni->ni_nid) {
                CERROR("Check advert: Bad NID %s (%s expected)\n",
                       libcfs_nid2str(nid),
                       libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
                *rcp = -EINVAL;
                goto out;
        }

        if (svc->RID.ServiceID != *kibnal_tunables.kib_service_number) {
                CERROR("Check advert: Bad ServiceID "LPX64" (%x expected)\n",
                       svc->RID.ServiceID,
                       *kibnal_tunables.kib_service_number);
                *rcp = -EINVAL;
                goto out;
        }

        if (svc->RID.ServiceGID.Type.Global.InterfaceID != 
            kibnal_data.kib_port_guid) {
                CERROR("Check advert: Bad GUID "LPX64" ("LPX64" expected)\n",
                       svc->RID.ServiceGID.Type.Global.InterfaceID,
                       kibnal_data.kib_port_guid);
                *rcp = -EINVAL;
                goto out;
        }

        if (svc->RID.ServiceP_Key != kibnal_data.kib_port_pkey) {
                CERROR("Check advert: Bad PKEY %04x (%04x expected)\n",
                       svc->RID.ServiceP_Key, kibnal_data.kib_port_pkey);
                *rcp = -EINVAL;
                goto out;
        }

        CDEBUG(D_NET, "Check advert OK\n");
        *rcp = 0;
                
 out:
        up (&kibnal_data.kib_listener_signal);                
}

int
kibnal_check_advert (void)
{
        /* single-threaded */
        static QUERY               qry;

        FSTATUS                    frc;
        int                        rc;

        memset (&qry, 0, sizeof(qry));
        qry.InputType = InputTypeServiceRecord;
        qry.OutputType = OutputTypeServiceRecord;
        kibnal_set_service_keys(&qry.InputValue.ServiceRecordValue.ServiceRecord,
                                kibnal_data.kib_ni->ni_nid);
        qry.InputValue.ServiceRecordValue.ComponentMask = KIBNAL_SERVICE_KEY_MASK;

        frc = iba_sd_query_port_fabric_info(kibnal_data.kib_sd, 
                                            kibnal_data.kib_port_guid,
                                            &qry, 
                                            kibnal_service_query_done,
                                            &kibnal_data.kib_sdretry, 
                                            &rc);
        if (frc != FPENDING) {
                CERROR ("Immediate error %d checking SM service\n", frc);
                return -EIO;
        }
        
        down (&kibnal_data.kib_listener_signal);
        
        if (rc != 0)
                CERROR ("Error %d checking SM service\n", rc);
        return rc;
}
#else
int
kibnal_check_advert(void)
{
        return 0;
}
#endif

void 
kibnal_fill_fod(FABRIC_OPERATION_DATA *fod, FABRIC_OPERATION_TYPE type)
{
        IB_SERVICE_RECORD     *svc;

        memset (fod, 0, sizeof(*fod));
        fod->Type = type;

        svc = &fod->Value.ServiceRecordValue.ServiceRecord;
        svc->RID.ServiceID = *kibnal_tunables.kib_service_number;
        svc->RID.ServiceGID.Type.Global.InterfaceID = kibnal_data.kib_port_guid;
        svc->RID.ServiceGID.Type.Global.SubnetPrefix = DEFAULT_SUBNET_PREFIX;
        svc->RID.ServiceP_Key = kibnal_data.kib_port_pkey;
        svc->ServiceLease = 0xffffffff;

        kibnal_set_service_keys(svc, kibnal_data.kib_ni->ni_nid);
}

void
kibnal_service_setunset_done (void *arg, FABRIC_OPERATION_DATA *fod,
                              FSTATUS frc, uint32 madrc)
{
        *(FSTATUS *)arg = frc;
        up (&kibnal_data.kib_listener_signal);
}

int
kibnal_advertise (void)
{
        /* Single threaded here */
        static FABRIC_OPERATION_DATA fod;

        IB_SERVICE_RECORD *svc = &fod.Value.ServiceRecordValue.ServiceRecord;
        FSTATUS            frc;
        FSTATUS            frc2;

        if (strlen(*kibnal_tunables.kib_service_name) >=
            sizeof(svc->ServiceName)) {
                CERROR("Service name '%s' too long (%d chars max)\n",
                       *kibnal_tunables.kib_service_name,
                       (int)sizeof(svc->ServiceName) - 1);
                return -EINVAL;
        }

        kibnal_fill_fod(&fod, FabOpSetServiceRecord);

        CDEBUG(D_NET, "Advertising service id "LPX64" %s:%s\n", 
               svc->RID.ServiceID, svc->ServiceName, 
               libcfs_nid2str(le64_to_cpu(*kibnal_service_nid_field(svc))));

        frc = iba_sd_port_fabric_operation(kibnal_data.kib_sd,
                                           kibnal_data.kib_port_guid,
                                           &fod, 
                                           kibnal_service_setunset_done, 
                                           &kibnal_data.kib_sdretry,
                                           &frc2);

        if (frc != FSUCCESS && frc != FPENDING) {
                CERROR ("Immediate error %d advertising NID %s\n",
                        frc, libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
                return -EIO;
        }

        down (&kibnal_data.kib_listener_signal);

        frc = frc2;
        if (frc == FSUCCESS)
                return 0;
        
        CERROR ("Error %d advertising %s\n",
                frc, libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
        return -EIO;
}

void
kibnal_unadvertise (int expect_success)
{
        /* single threaded */
        static FABRIC_OPERATION_DATA fod;

        IB_SERVICE_RECORD *svc = &fod.Value.ServiceRecordValue.ServiceRecord;
        FSTATUS            frc;
        FSTATUS            frc2;

        LASSERT (kibnal_data.kib_ni->ni_nid != LNET_NID_ANY);

        kibnal_fill_fod(&fod, FabOpDeleteServiceRecord);

        CDEBUG(D_NET, "Unadvertising service %s:%s\n",
               svc->ServiceName, 
               libcfs_nid2str(le64_to_cpu(*kibnal_service_nid_field(svc))));
        
        frc = iba_sd_port_fabric_operation(kibnal_data.kib_sd,
                                           kibnal_data.kib_port_guid,
                                           &fod, 
                                           kibnal_service_setunset_done, 
                                           &kibnal_data.kib_sdretry, 
                                           &frc2);
        if (frc != FSUCCESS && frc != FPENDING) {
                CERROR ("Immediate error %d unadvertising NID %s\n",
                        frc, libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
                return;
        }

        down (&kibnal_data.kib_listener_signal);

        CDEBUG(D_NET, "Unadvertise rc: %d\n", frc2);

        if ((frc2 == FSUCCESS) == !!expect_success)
                return;

        if (expect_success)
                CERROR("Error %d unadvertising NID %s\n",
                       frc2, libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
        else
                CWARN("Removed conflicting NID %s\n",
                      libcfs_nid2str(kibnal_data.kib_ni->ni_nid));
}

void
kibnal_stop_listener(int normal_shutdown)
{
        /* NB this also disables peer creation and destroys all existing
         * peers */
        IB_HANDLE      cep = kibnal_data.kib_listener_cep;
        unsigned long  flags;
        FSTATUS        frc;

        LASSERT (cep != NULL);

        kibnal_unadvertise(normal_shutdown);

        frc = iba_cm_cancel(cep);
        if (frc != FSUCCESS && frc != FPENDING)
                CERROR ("Error %d stopping listener\n", frc);

        down(&kibnal_data.kib_listener_signal);

        frc = iba_cm_destroy_cep(cep);
        if (frc != FSUCCESS)
                CERROR ("Error %d destroying listener CEP\n", frc);

        write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
        /* This assignment disables peer creation */
        kibnal_data.kib_listener_cep = NULL;
        write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);

        /* Start to tear down any peers created while the listener was
         * running */
        kibnal_del_peer(LNET_NID_ANY);
}

int
kibnal_start_listener(void)
{
        /* NB this also enables peer creation */

        IB_HANDLE      cep;
        CM_LISTEN_INFO info;
        unsigned long  flags;
        int            rc;
        FSTATUS        frc;

        LASSERT (kibnal_data.kib_listener_cep == NULL);
        init_MUTEX_LOCKED (&kibnal_data.kib_listener_signal);

        cep = kibnal_create_cep(LNET_NID_ANY);
        if (cep == NULL)
                return -ENOMEM;

        memset (&info, 0, sizeof(info));
        info.ListenAddr.EndPt.SID = *kibnal_tunables.kib_service_number;

        frc = iba_cm_listen(cep, &info, kibnal_listen_callback, NULL);
        if (frc != FSUCCESS && frc != FPENDING) {
                CERROR ("iba_cm_listen error: %d\n", frc);

                iba_cm_destroy_cep(cep);
                return -EIO;
        }

        write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
        /* This assignment enables peer creation */
        kibnal_data.kib_listener_cep = cep;
        write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);

        rc = kibnal_advertise();
        if (rc == 0)
                rc = kibnal_check_advert();

        if (rc == 0)
                return 0;

        kibnal_stop_listener(0);
        return rc;
}

int
kibnal_create_peer (kib_peer_t **peerp, lnet_nid_t nid)
{
        kib_peer_t    *peer;
        unsigned long  flags;
        int            rc;

        LASSERT (nid != LNET_NID_ANY);

        LIBCFS_ALLOC (peer, sizeof (*peer));
        if (peer == NULL) {
                CERROR("Cannot allocate peer\n");
                return -ENOMEM;
        }

        memset(peer, 0, sizeof(*peer));         /* zero flags etc */

        peer->ibp_nid = nid;
        atomic_set (&peer->ibp_refcount, 1);    /* 1 ref for caller */

        INIT_LIST_HEAD (&peer->ibp_list);       /* not in the peer table yet */
        INIT_LIST_HEAD (&peer->ibp_conns);
        INIT_LIST_HEAD (&peer->ibp_tx_queue);

        peer->ibp_error = 0;
        peer->ibp_last_alive = cfs_time_current();
        peer->ibp_reconnect_interval = 0;       /* OK to connect at any time */

        write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
        
        if (atomic_read(&kibnal_data.kib_npeers) >=
            *kibnal_tunables.kib_concurrent_peers) {
                rc = -EOVERFLOW;        /* !! but at least it distinguishes */
        } else if (kibnal_data.kib_listener_cep == NULL) {
                rc = -ESHUTDOWN;        /* shutdown has started */
        } else {
                rc = 0;
                /* npeers only grows with the global lock held */
                atomic_inc(&kibnal_data.kib_npeers);
        }
        
        write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);

        if (rc != 0) {
                CERROR("Can't create peer: %s\n", 
                       (rc == -ESHUTDOWN) ? "shutting down" : 
                       "too many peers");
                LIBCFS_FREE(peer, sizeof(*peer));
        } else {
                *peerp = peer;
        }
        
        return rc;
}

void
kibnal_destroy_peer (kib_peer_t *peer)
{

        LASSERT (atomic_read (&peer->ibp_refcount) == 0);
        LASSERT (peer->ibp_persistence == 0);
        LASSERT (!kibnal_peer_active(peer));
        LASSERT (!kibnal_peer_connecting(peer));
        LASSERT (list_empty (&peer->ibp_conns));
        LASSERT (list_empty (&peer->ibp_tx_queue));

        LIBCFS_FREE (peer, sizeof (*peer));

        /* NB a peer's connections keep a reference on their peer until