echo_client.c

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

        einfo.ei_mode = mode;
        einfo.ei_cb_bl = echo_ldlm_callback;
        einfo.ei_cb_cp = ldlm_completion_ast;
        einfo.ei_cb_gl = NULL;
        einfo.ei_cbdata = eco;

        oinfo.oi_policy = ecl->ecl_policy;
        oinfo.oi_lockh = &ecl->ecl_lock_handle;
        oinfo.oi_md = eco->eco_lsm;
        rc = obd_enqueue(ec->ec_exp, &oinfo, &einfo, NULL);
        if (rc != 0)
                goto failed_1;

        CDEBUG(D_INFO, "enqueue handle "LPX64"\n", ecl->ecl_lock_handle.cookie);

        /* NB ecl takes object ref from echo_get_object() above */
        spin_lock(&ec->ec_lock);

        list_add(&ecl->ecl_exp_chain, &exp->exp_ec_data.eced_locks);
        ulh->cookie = ecl->ecl_cookie = ec->ec_unique++;

        spin_unlock(&ec->ec_lock);

        oa->o_valid |= OBD_MD_FLHANDLE;
        return 0;

 failed_1:
        OBD_FREE (ecl, sizeof (*ecl));
 failed_0:
        echo_put_object (eco);
        return (rc);
}

static int
echo_client_cancel(struct obd_export *exp, struct obdo *oa)
{
        struct obd_device      *obd = exp->exp_obd;
        struct echo_client_obd *ec = &obd->u.echo_client;
        struct lustre_handle   *ulh = &oa->o_handle;
        struct ec_lock         *ecl = NULL;
        int                     found = 0;
        struct list_head       *el;
        int                     rc;

        if ((oa->o_valid & OBD_MD_FLHANDLE) == 0)
                return -EINVAL;

        spin_lock (&ec->ec_lock);

        list_for_each (el, &exp->exp_ec_data.eced_locks) {
                ecl = list_entry (el, struct ec_lock, ecl_exp_chain);
                found = (ecl->ecl_cookie == ulh->cookie);
                if (found) {
                        list_del (&ecl->ecl_exp_chain);
                        break;
                }
        }

        spin_unlock (&ec->ec_lock);

        if (!found)
                return (-ENOENT);

        rc = obd_cancel(ec->ec_exp, ecl->ecl_object->eco_lsm, ecl->ecl_mode,
                        &ecl->ecl_lock_handle);

        echo_put_object (ecl->ecl_object);
        OBD_FREE (ecl, sizeof (*ecl));

        return rc;
}

static int
echo_client_iocontrol(unsigned int cmd, struct obd_export *exp,
                      int len, void *karg, void *uarg)
{
        struct obd_device      *obd;
        struct echo_client_obd *ec;
        struct ec_object       *eco;
        struct obd_ioctl_data  *data = karg;
        struct obd_trans_info   dummy_oti;
        struct oti_req_ack_lock *ack_lock;
        struct obdo            *oa;
        int                     rw = OBD_BRW_READ;
        int                     rc = 0;
        int                     i;
        ENTRY;

        unlock_kernel();

        memset(&dummy_oti, 0, sizeof(dummy_oti));

        obd = exp->exp_obd;
        ec = &obd->u.echo_client;

        switch (cmd) {
        case OBD_IOC_CREATE:                    /* may create echo object */
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);

                rc = echo_create_object (obd, 1, &data->ioc_obdo1,
                                         data->ioc_pbuf1, data->ioc_plen1,
                                         &dummy_oti);
                GOTO(out, rc);

        case OBD_IOC_DESTROY:
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);
                rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
                if (rc == 0) {
                        oa = &data->ioc_obdo1;
                        oa->o_gr = FILTER_GROUP_ECHO;
                        oa->o_valid |= OBD_MD_FLGROUP;
                        rc = obd_destroy(ec->ec_exp, oa, eco->eco_lsm,
                                         &dummy_oti, NULL);
                        if (rc == 0)
                                eco->eco_deleted = 1;
                        echo_put_object(eco);
                }
                GOTO(out, rc);

        case OBD_IOC_GETATTR:
                rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
                if (rc == 0) {
                        struct obd_info oinfo = { { { 0 } } };
                        oinfo.oi_md = eco->eco_lsm;
                        oinfo.oi_oa = &data->ioc_obdo1;
                        rc = obd_getattr(ec->ec_exp, &oinfo);
                        echo_put_object(eco);
                }
                GOTO(out, rc);

        case OBD_IOC_SETATTR:
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);

                rc = echo_get_object (&eco, obd, &data->ioc_obdo1);
                if (rc == 0) {
                        struct obd_info oinfo = { { { 0 } } };
                        oinfo.oi_oa = &data->ioc_obdo1;
                        oinfo.oi_md = eco->eco_lsm;

                        rc = obd_setattr(ec->ec_exp, &oinfo, NULL);
                        echo_put_object(eco);
                }
                GOTO(out, rc);

        case OBD_IOC_BRW_WRITE:
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);

                rw = OBD_BRW_WRITE;
                /* fall through */
        case OBD_IOC_BRW_READ:
                rc = echo_client_brw_ioctl(rw, exp, data);
                GOTO(out, rc);

        case ECHO_IOC_GET_STRIPE:
                rc = echo_get_object(&eco, obd, &data->ioc_obdo1);
                if (rc == 0) {
                        rc = echo_copyout_lsm(eco->eco_lsm, data->ioc_pbuf1,
                                              data->ioc_plen1);
                        echo_put_object(eco);
                }
                GOTO(out, rc);

        case ECHO_IOC_SET_STRIPE:
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);

                if (data->ioc_pbuf1 == NULL) {  /* unset */
                        rc = echo_get_object(&eco, obd, &data->ioc_obdo1);
                        if (rc == 0) {
                                eco->eco_deleted = 1;
                                echo_put_object(eco);
                        }
                } else {
                        rc = echo_create_object(obd, 0, &data->ioc_obdo1,
                                                data->ioc_pbuf1,
                                                data->ioc_plen1, &dummy_oti);
                }
                GOTO (out, rc);

        case ECHO_IOC_ENQUEUE:
                if (!capable (CAP_SYS_ADMIN))
                        GOTO (out, rc = -EPERM);

                rc = echo_client_enqueue(exp, &data->ioc_obdo1,
                                         data->ioc_conn1, /* lock mode */
                                   data->ioc_offset, data->ioc_count);/*extent*/
                GOTO (out, rc);

        case ECHO_IOC_CANCEL:
                rc = echo_client_cancel(exp, &data->ioc_obdo1);
                GOTO (out, rc);

        default:
                CERROR ("echo_ioctl(): unrecognised ioctl %#x\n", cmd);
                GOTO (out, rc = -ENOTTY);
        }

        EXIT;
 out:

        /* XXX this should be in a helper also called by target_send_reply */
        for (ack_lock = dummy_oti.oti_ack_locks, i = 0; i < 4; 
             i++, ack_lock++) {
                if (!ack_lock->mode)
                        break;
                ldlm_lock_decref(&ack_lock->lock, ack_lock->mode);
        }

        lock_kernel();

        return rc;
}

static int
echo_client_setup(struct obd_device *obddev, obd_count len, void *buf)
{
        struct lustre_cfg* lcfg = buf;
        struct echo_client_obd *ec = &obddev->u.echo_client;
        struct obd_device *tgt;
        struct lustre_handle conn = {0, };
        struct obd_uuid echo_uuid = { "ECHO_UUID" };
        struct obd_connect_data *ocd = NULL;
        int rc;
        ENTRY;

        if (lcfg->lcfg_bufcount < 2 || LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
                CERROR("requires a TARGET OBD name\n");
                RETURN(-EINVAL);
        }

        tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
        if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
                CERROR("device not attached or not set up (%s)\n",
                       lustre_cfg_string(lcfg, 1));
                RETURN(-EINVAL);
        }

        spin_lock_init (&ec->ec_lock);
        CFS_INIT_LIST_HEAD (&ec->ec_objects);
        ec->ec_unique = 0;

        OBD_ALLOC(ocd, sizeof(*ocd));
        if (ocd == NULL) {
                CERROR("Can't alloc ocd connecting to %s\n",
                       lustre_cfg_string(lcfg, 1));
                return -ENOMEM;
        }

        ocd->ocd_connect_flags = OBD_CONNECT_VERSION | OBD_CONNECT_REQPORTAL;
        ocd->ocd_version = LUSTRE_VERSION_CODE;

        rc = obd_connect(&conn, tgt, &echo_uuid, ocd, NULL);

        OBD_FREE(ocd, sizeof(*ocd));

        if (rc != 0) {
                CERROR("fail to connect to device %s\n",
                       lustre_cfg_string(lcfg, 1));
                return (rc);
        }
        ec->ec_exp = class_conn2export(&conn);

        RETURN(rc);
}

static int echo_client_cleanup(struct obd_device *obddev)
{
        struct list_head       *el;
        struct ec_object       *eco;
        struct echo_client_obd *ec = &obddev->u.echo_client;
        int rc;
        ENTRY;

        if (!list_empty(&obddev->obd_exports)) {
                CERROR("still has clients!\n");
                RETURN(-EBUSY);
        }

        /* XXX assuming sole access */
        while (!list_empty(&ec->ec_objects)) {
                el = ec->ec_objects.next;
                eco = list_entry(el, struct ec_object, eco_obj_chain);

                LASSERT(eco->eco_refcount == 0);
                eco->eco_refcount = 1;
                eco->eco_deleted = 1;
                echo_put_object(eco);
        }

        rc = obd_disconnect(ec->ec_exp);
        if (rc != 0)
                CERROR("fail to disconnect device: %d\n", rc);

        RETURN(rc);
}

static int echo_client_connect(struct lustre_handle *conn,
                               struct obd_device *src, struct obd_uuid *cluuid,
                               struct obd_connect_data *data, void *localdata)
{
        struct obd_export *exp;
        int                rc;

        ENTRY;
        rc = class_connect(conn, src, cluuid);
        if (rc == 0) {
                exp = class_conn2export(conn);
                CFS_INIT_LIST_HEAD(&exp->exp_ec_data.eced_locks);
                class_export_put(exp);
        }

        RETURN (rc);
}

static int echo_client_disconnect(struct obd_export *exp)
{
        struct obd_device      *obd;
        struct echo_client_obd *ec;
        struct ec_lock         *ecl;
        int                     rc;
        ENTRY;

        if (exp == NULL)
                GOTO(out, rc = -EINVAL);

        obd = exp->exp_obd;
        ec = &obd->u.echo_client;

        /* no more contention on export's lock list */
        while (!list_empty (&exp->exp_ec_data.eced_locks)) {
                ecl = list_entry (exp->exp_ec_data.eced_locks.next,
                                  struct ec_lock, ecl_exp_chain);
                list_del (&ecl->ecl_exp_chain);

                rc = obd_cancel(ec->ec_exp, ecl->ecl_object->eco_lsm,
                                 ecl->ecl_mode, &ecl->ecl_lock_handle);

                CDEBUG (D_INFO, "Cancel lock on object "LPX64" on disconnect "
                        "(%d)\n", ecl->ecl_object->eco_id, rc);

                echo_put_object (ecl->ecl_object);
                OBD_FREE (ecl, sizeof (*ecl));
        }

        rc = class_disconnect(exp);
        GOTO(out, rc);
 out:
        return rc;
}

static struct obd_ops echo_obd_ops = {
        .o_owner       = THIS_MODULE,
        .o_setup       = echo_client_setup,
        .o_cleanup     = echo_client_cleanup,
        .o_iocontrol   = echo_client_iocontrol,
        .o_connect     = echo_client_connect,
        .o_disconnect  = echo_client_disconnect
};

int echo_client_init(void)
{
        struct lprocfs_static_vars lvars = { 0 };

        lprocfs_echo_init_vars(&lvars);
        return class_register_type(&echo_obd_ops, lvars.module_vars,
                                   LUSTRE_ECHO_CLIENT_NAME);
}

void echo_client_exit(void)
{
        class_unregister_type(LUSTRE_ECHO_CLIENT_NAME);
}