genops.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,319 行 · 第 1/3 页

                }
        }
        spin_unlock(&obd_dev_lock);

        return NULL;
}

struct obd_device *class_find_client_notype(struct obd_uuid *tgt_uuid,
                                            struct obd_uuid *grp_uuid)
{
        struct obd_device *obd;

        obd = class_find_client_obd(tgt_uuid, LUSTRE_MDC_NAME, NULL);
        if (!obd)
                obd = class_find_client_obd(tgt_uuid, LUSTRE_OSC_NAME,
                                            grp_uuid);
        return obd;
}

/* Iterate the obd_device list looking devices have grp_uuid. Start
   searching at *next, and if a device is found, the next index to look
   at is saved in *next. If next is NULL, then the first matching device
   will always be returned. */
struct obd_device * class_devices_in_group(struct obd_uuid *grp_uuid, int *next)
{
        int i;

        if (next == NULL)
                i = 0;
        else if (*next >= 0 && *next < class_devno_max())
                i = *next;
        else
                return NULL;

        spin_lock(&obd_dev_lock);
        for (; i < class_devno_max(); i++) {
                struct obd_device *obd = class_num2obd(i);
                if (obd == NULL)
                        continue;
                if (obd_uuid_equals(grp_uuid, &obd->obd_uuid)) {
                        if (next != NULL)
                                *next = i+1;
                        spin_unlock(&obd_dev_lock);
                        return obd;
                }
        }
        spin_unlock(&obd_dev_lock);

        return NULL;
}


void obd_cleanup_caches(void)
{
        int rc;

        ENTRY;
        if (obd_device_cachep) {
                rc = cfs_mem_cache_destroy(obd_device_cachep);
                LASSERTF(rc == 0, "Cannot destropy ll_obd_device_cache: rc %d\n", rc);
                obd_device_cachep = NULL;
        }
        if (obdo_cachep) {
                rc = cfs_mem_cache_destroy(obdo_cachep);
                LASSERTF(rc == 0, "Cannot destory ll_obdo_cache\n");
                obdo_cachep = NULL;
        }
        if (import_cachep) {
                rc = cfs_mem_cache_destroy(import_cachep);
                LASSERTF(rc == 0, "Cannot destory ll_import_cache\n");
                import_cachep = NULL;
        }
        EXIT;
}

int obd_init_caches(void)
{
        ENTRY;

        LASSERT(obd_device_cachep == NULL);
        obd_device_cachep = cfs_mem_cache_create("ll_obd_dev_cache",
                                                 sizeof(struct obd_device), 
                                                 0, 0);
        if (!obd_device_cachep)
                GOTO(out, -ENOMEM);

        LASSERT(obdo_cachep == NULL);
        obdo_cachep = cfs_mem_cache_create("ll_obdo_cache", sizeof(struct obdo),
                                           0, 0);
        if (!obdo_cachep)
                GOTO(out, -ENOMEM);

        LASSERT(import_cachep == NULL);
        import_cachep = cfs_mem_cache_create("ll_import_cache",
                                             sizeof(struct obd_import),
                                             0, 0);
        if (!import_cachep)
                GOTO(out, -ENOMEM);

        RETURN(0);
 out:
        obd_cleanup_caches();
        RETURN(-ENOMEM);

}

/* map connection to client */
struct obd_export *class_conn2export(struct lustre_handle *conn)
{
        struct obd_export *export;
        ENTRY;

        if (!conn) {
                CDEBUG(D_CACHE, "looking for null handle\n");
                RETURN(NULL);
        }

        if (conn->cookie == -1) {  /* this means assign a new connection */
                CDEBUG(D_CACHE, "want a new connection\n");
                RETURN(NULL);
        }

        CDEBUG(D_INFO, "looking for export cookie "LPX64"\n", conn->cookie);
        export = class_handle2object(conn->cookie);
        RETURN(export);
}

struct obd_device *class_exp2obd(struct obd_export *exp)
{
        if (exp)
                return exp->exp_obd;
        return NULL;
}

struct obd_device *class_conn2obd(struct lustre_handle *conn)
{
        struct obd_export *export;
        export = class_conn2export(conn);
        if (export) {
                struct obd_device *obd = export->exp_obd;
                class_export_put(export);
                return obd;
        }
        return NULL;
}

struct obd_import *class_exp2cliimp(struct obd_export *exp)
{
        struct obd_device *obd = exp->exp_obd;
        if (obd == NULL)
                return NULL;
        return obd->u.cli.cl_import;
}

struct obd_import *class_conn2cliimp(struct lustre_handle *conn)
{
        struct obd_device *obd = class_conn2obd(conn);
        if (obd == NULL)
                return NULL;
        return obd->u.cli.cl_import;
}

/* Export management functions */
static void export_handle_addref(void *export)
{
        class_export_get(export);
}

void __class_export_put(struct obd_export *exp)
{
        if (atomic_dec_and_test(&exp->exp_refcount)) {
                LASSERT (list_empty(&exp->exp_obd_chain));

                CDEBUG(D_IOCTL, "final put %p/%s\n",
                       exp, exp->exp_client_uuid.uuid);
        
                spin_lock(&obd_zombie_impexp_lock);
                list_add(&exp->exp_obd_chain, &obd_zombie_exports);
                spin_unlock(&obd_zombie_impexp_lock);

                if (obd_zombie_impexp_notify != NULL)
                        obd_zombie_impexp_notify();
        }
}
EXPORT_SYMBOL(__class_export_put);

void class_export_destroy(struct obd_export *exp)
{
        struct obd_device *obd = exp->exp_obd;

        LASSERT (atomic_read(&exp->exp_refcount) == 0);

        CDEBUG(D_IOCTL, "destroying export %p/%s\n", exp,
               exp->exp_client_uuid.uuid);

        LASSERT(obd != NULL);

        /* "Local" exports (lctl, LOV->{mdc,osc}) have no connection. */
        if (exp->exp_connection)
                ptlrpc_put_connection_superhack(exp->exp_connection);

        LASSERT(list_empty(&exp->exp_outstanding_replies));
        obd_destroy_export(exp);

        OBD_FREE_RCU(exp, sizeof(*exp), &exp->exp_handle);
        class_decref(obd);
}

/* Creates a new export, adds it to the hash table, and returns a
 * pointer to it. The refcount is 2: one for the hash reference, and
 * one for the pointer returned by this function. */
struct obd_export *class_new_export(struct obd_device *obd,
                                    struct obd_uuid *cluuid)
{
        struct obd_export *export;
        int rc = 0;

        OBD_ALLOC(export, sizeof(*export));
        if (!export)
                return ERR_PTR(-ENOMEM);

        export->exp_conn_cnt = 0;
        atomic_set(&export->exp_refcount, 2);
        atomic_set(&export->exp_rpc_count, 0);
        export->exp_obd = obd;
        CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies);
        /* XXX this should be in LDLM init */
        CFS_INIT_LIST_HEAD(&export->exp_ldlm_data.led_held_locks);
        spin_lock_init(&export->exp_ldlm_data.led_lock);

        CFS_INIT_LIST_HEAD(&export->exp_handle.h_link);
        class_handle_hash(&export->exp_handle, export_handle_addref);
        export->exp_last_request_time = cfs_time_current_sec();
        spin_lock_init(&export->exp_lock);
        INIT_HLIST_NODE(&export->exp_uuid_hash);
        INIT_HLIST_NODE(&export->exp_nid_hash);

        export->exp_client_uuid = *cluuid;
        obd_init_export(export);

        if (!obd_uuid_equals(cluuid, &obd->obd_uuid)) {
               rc = lustre_hash_additem_unique(obd->obd_uuid_hash_body, cluuid, 
                                               &export->exp_uuid_hash);
               if (rc != 0) {
                       CWARN("%s: denying duplicate export for %s\n",
                             obd->obd_name, cluuid->uuid);
                       class_handle_unhash(&export->exp_handle);
                       OBD_FREE_PTR(export);
                       return ERR_PTR(-EALREADY);
               }
        }

        spin_lock(&obd->obd_dev_lock);
        LASSERT(!obd->obd_stopping); /* shouldn't happen, but might race */
        class_incref(obd);
        list_add(&export->exp_obd_chain, &export->exp_obd->obd_exports);
        list_add_tail(&export->exp_obd_chain_timed,
                      &export->exp_obd->obd_exports_timed);
        export->exp_obd->obd_num_exports++;
        spin_unlock(&obd->obd_dev_lock);

        return export;
}
EXPORT_SYMBOL(class_new_export);

void class_unlink_export(struct obd_export *exp)
{
        class_handle_unhash(&exp->exp_handle);

        spin_lock(&exp->exp_obd->obd_dev_lock);
        /* delete an uuid-export hashitem from hashtables */
        if (!hlist_unhashed(&exp->exp_uuid_hash)) {
                lustre_hash_delitem(exp->exp_obd->obd_uuid_hash_body, 
                                    &exp->exp_client_uuid, &exp->exp_uuid_hash);
        }
        list_del_init(&exp->exp_obd_chain);
        list_del_init(&exp->exp_obd_chain_timed);
        exp->exp_obd->obd_num_exports--;
        spin_unlock(&exp->exp_obd->obd_dev_lock);

        class_export_put(exp);
}
EXPORT_SYMBOL(class_unlink_export);

/* Import management functions */
static void import_handle_addref(void *import)
{
        class_import_get(import);
}

struct obd_import *class_import_get(struct obd_import *import)
{
        LASSERT(atomic_read(&import->imp_refcount) >= 0);
        LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a);
        atomic_inc(&import->imp_refcount);
        CDEBUG(D_INFO, "import %p refcount=%d\n", import,
               atomic_read(&import->imp_refcount));
        return import;
}
EXPORT_SYMBOL(class_import_get);

void class_import_put(struct obd_import *import)
{
        ENTRY;

        CDEBUG(D_INFO, "import %p refcount=%d\n", import,
               atomic_read(&import->imp_refcount) - 1);

        LASSERT(atomic_read(&import->imp_refcount) > 0);
        LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a);
        LASSERT(list_empty(&import->imp_zombie_chain));

        if (atomic_dec_and_test(&import->imp_refcount)) {

                CDEBUG(D_INFO, "final put import %p\n", import);
                
                spin_lock(&obd_zombie_impexp_lock);
                list_add(&import->imp_zombie_chain, &obd_zombie_imports);
                spin_unlock(&obd_zombie_impexp_lock);

                if (obd_zombie_impexp_notify != NULL)
                        obd_zombie_impexp_notify();
        }

        EXIT;
}

void class_import_destroy(struct obd_import *import)
{
        ENTRY;
        
        CDEBUG(D_IOCTL, "destroying import %p\n", import);

        LASSERT(atomic_read(&import->imp_refcount) == 0);

        ptlrpc_put_connection_superhack(import->imp_connection);

        while (!list_empty(&import->imp_conn_list)) {
                struct obd_import_conn *imp_conn;

                imp_conn = list_entry(import->imp_conn_list.next,
                                      struct obd_import_conn, oic_item);
                list_del(&imp_conn->oic_item);
                ptlrpc_put_connection_superhack(imp_conn->oic_conn);
                OBD_FREE(imp_conn, sizeof(*imp_conn));
        }

        class_decref(import->imp_obd);
        OBD_FREE_RCU(import, sizeof(*import), &import->imp_handle);
        EXIT;
}
EXPORT_SYMBOL(class_import_put);

static void init_imp_at(struct imp_at *at) {
        int i;
        at_init(&at->iat_net_latency, 0, 0);
        for (i = 0; i < IMP_AT_MAX_PORTALS; i++) {
                /* max service estimates are tracked on the server side, so
                   don't use the AT history here, just use the last reported
                   val. (But keep hist for proc histogram, worst_ever) */
                at_init(&at->iat_service_estimate[i], INITIAL_CONNECT_TIMEOUT,
                        AT_FLG_NOHIST);
        }
}

struct obd_import *class_new_import(struct obd_device *obd)
{
        struct obd_import *imp;

        OBD_ALLOC(imp, sizeof(*imp));
        if (imp == NULL)
                return NULL;

        CFS_INIT_LIST_HEAD(&imp->imp_zombie_chain);
        CFS_INIT_LIST_HEAD(&imp->imp_replay_list);
        CFS_INIT_LIST_HEAD(&imp->imp_sending_list);
        CFS_INIT_LIST_HEAD(&imp->imp_delayed_list);
        spin_lock_init(&imp->imp_lock);
        imp->imp_last_success_conn = 0;
        imp->imp_state = LUSTRE_IMP_NEW;
        imp->imp_obd = class_incref(obd);
        cfs_waitq_init(&imp->imp_recovery_waitq);

        atomic_set(&imp->imp_refcount, 2);
        atomic_set(&imp->imp_inflight, 0);
        atomic_set(&imp->imp_replay_inflight, 0);
        atomic_set(&imp->imp_inval_count, 0);
        CFS_INIT_LIST_HEAD(&imp->imp_conn_list);
        CFS_INIT_LIST_HEAD(&imp->imp_handle.h_link);
        class_handle_hash(&imp->imp_handle, import_handle_addref);
        init_imp_at(&imp->imp_at);

#ifdef HAVE_DEFAULT_V2_CONNECT
        /* the default magic is V2, will be used in connect RPC, and
         * then adjusted according to the flags in request/reply. */
        imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
#else
        /* the default magic is V1, will be used in connect RPC, and
         * then adjusted according to the flags in request/reply. */
        imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V1;
#endif

        return imp;
}
EXPORT_SYMBOL(class_new_import);

void class_destroy_import(struct obd_import *import)
{
        LASSERT(import != NULL);
        LASSERT(import != LP_POISON);

        class_handle_unhash(&import->imp_handle);

        spin_lock(&import->imp_lock);
        import->imp_generation++;
        spin_unlock(&import->imp_lock);

        class_import_put(import);
}
EXPORT_SYMBOL(class_destroy_import);

/* A connection defines an export context in which preallocation can
   be managed. This releases the export pointer reference, and returns
   the export handle, so the export refcount is 1 when this function
   returns. */
int class_connect(struct lustre_handle *conn, struct obd_device *obd,
                  struct obd_uuid *cluuid)
{
        struct obd_export *export;
        LASSERT(conn != NULL);
        LASSERT(obd != NULL);
        LASSERT(cluuid != NULL);
        ENTRY;

        export = class_new_export(obd, cluuid);
        if (IS_ERR(export))
                RETURN(PTR_ERR(export));

        conn->cookie = export->exp_handle.h_cookie;
        class_export_put(export);