+ lmv_activate_target(lmv, tgt, 0);
+ tgt->ltd_exp = NULL;
+ RETURN(0);
+}
+
+static int lmv_disconnect(struct obd_export *exp)
+{
+ struct obd_device *obd = class_exp2obd(exp);
+#ifdef __KERNEL__
+ struct proc_dir_entry *lmv_proc_dir;
+#endif
+ struct lmv_obd *lmv = &obd->u.lmv;
+ int rc;
+ int i;
+ ENTRY;
+
+ if (!lmv->tgts)
+ goto out_local;
+
+ /*
+ * Only disconnect the underlying layers on the final disconnect.
+ */
+ lmv->refcount--;
+ if (lmv->refcount != 0)
+ goto out_local;
+
+ for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
+ if (lmv->tgts[i].ltd_exp == NULL)
+ continue;
+ lmv_disconnect_mdc(obd, &lmv->tgts[i]);
+ }
+
+#ifdef __KERNEL__
+ lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
+ if (lmv_proc_dir) {
+ lprocfs_remove(&lmv_proc_dir);
+ } else {
+ CERROR("/proc/fs/lustre/%s/%s/target_obds missing\n",
+ obd->obd_type->typ_name, obd->obd_name);
+ }
+#endif
+
+out_local:
+ /*
+ * This is the case when no real connection is established by
+ * lmv_check_connect().
+ */
+ if (!lmv->connected)
+ class_export_put(exp);
+ rc = class_disconnect(exp);
+ if (lmv->refcount == 0)
+ lmv->connected = 0;
+ RETURN(rc);
+}
+
+static int lmv_iocontrol(unsigned int cmd, struct obd_export *exp,
+ int len, void *karg, void *uarg)
+{
+ struct obd_device *obddev = class_exp2obd(exp);
+ struct lmv_obd *lmv = &obddev->u.lmv;
+ int i = 0;
+ int rc = 0;
+ int set = 0;
+ int count = lmv->desc.ld_tgt_count;
+ ENTRY;
+
+ if (count == 0)
+ RETURN(-ENOTTY);
+
+ switch (cmd) {
+ case IOC_OBD_STATFS: {
+ struct obd_ioctl_data *data = karg;
+ struct obd_device *mdc_obd;
+ struct obd_statfs stat_buf = {0};
+ __u32 index;
+
+ memcpy(&index, data->ioc_inlbuf2, sizeof(__u32));
+ LASSERT(data->ioc_plen1 == sizeof(struct obd_statfs));
+
+ if ((index >= count))
+ RETURN(-ENODEV);
+
+ if (!lmv->tgts[index].ltd_active)
+ RETURN(-ENODATA);
+
+ mdc_obd = class_exp2obd(lmv->tgts[index].ltd_exp);
+ if (!mdc_obd)
+ RETURN(-EINVAL);
+
+ rc = obd_statfs(mdc_obd, &stat_buf,
+ cfs_time_current_64() - HZ, 0);
+ if (rc)
+ RETURN(rc);
+ if (copy_to_user(data->ioc_pbuf1, &stat_buf, data->ioc_plen1))
+ RETURN(-EFAULT);
+ if (copy_to_user(data->ioc_pbuf2, obd2cli_tgt(mdc_obd),
+ data->ioc_plen2))
+ RETURN(-EFAULT);
+ break;
+ }
+ case OBD_IOC_QUOTACTL: {
+ struct if_quotactl *qctl = karg;
+ struct lmv_tgt_desc *tgt = NULL;
+ struct obd_quotactl *oqctl;
+
+ if (qctl->qc_valid == QC_MDTIDX) {
+ if (qctl->qc_idx < 0 || count <= qctl->qc_idx)
+ RETURN(-EINVAL);
+
+ tgt = &lmv->tgts[qctl->qc_idx];
+ if (!tgt->ltd_exp)
+ RETURN(-EINVAL);
+ } else if (qctl->qc_valid == QC_UUID) {
+ for (i = 0; i < count; i++) {
+ tgt = &lmv->tgts[i];
+ if (!obd_uuid_equals(&tgt->ltd_uuid,
+ &qctl->obd_uuid))
+ continue;
+
+ if (tgt->ltd_exp == NULL)
+ RETURN(-EINVAL);
+
+ break;
+ }
+ } else {
+ RETURN(-EINVAL);
+ }
+
+ if (i >= count)
+ RETURN(-EAGAIN);
+
+ LASSERT(tgt && tgt->ltd_exp);
+ OBD_ALLOC_PTR(oqctl);
+ if (!oqctl)
+ RETURN(-ENOMEM);
+
+ QCTL_COPY(oqctl, qctl);
+ rc = obd_quotactl(tgt->ltd_exp, oqctl);
+ if (rc == 0) {
+ QCTL_COPY(qctl, oqctl);
+ qctl->qc_valid = QC_MDTIDX;
+ qctl->obd_uuid = tgt->ltd_uuid;
+ }
+ OBD_FREE_PTR(oqctl);
+ break;
+ }
+ case OBD_IOC_CHANGELOG_CLEAR: {
+ struct ioc_changelog_clear *icc = karg;
+
+ if (icc->icc_mdtindex >= count)
+ RETURN(-ENODEV);
+
+ rc = obd_iocontrol(cmd, lmv->tgts[icc->icc_mdtindex].ltd_exp,
+ sizeof(*icc), icc, NULL);
+ break;
+ }
+
+ default : {
+ for (i = 0; i < count; i++) {
+ int err;
+
+ if (lmv->tgts[i].ltd_exp == NULL)
+ continue;
+
+ err = obd_iocontrol(cmd, lmv->tgts[i].ltd_exp, len,
+ karg, uarg);
+ if (err == -ENODATA && cmd == OBD_IOC_POLL_QUOTACHECK) {
+ RETURN(err);
+ } else if (err) {
+ if (lmv->tgts[i].ltd_active) {
+ CERROR("error: iocontrol MDC %s on MDT"
+ "idx %d cmd %x: err = %d\n",
+ lmv->tgts[i].ltd_uuid.uuid,
+ i, cmd, err);
+ if (!rc)
+ rc = err;
+ }
+ } else
+ set = 1;
+ }
+ if (!set && !rc)
+ rc = -EIO;
+ }
+ }
+ RETURN(rc);
+}
+
+static int lmv_all_chars_policy(int count, const char *name,
+ int len)
+{
+ unsigned int c = 0;
+
+ while (len > 0)
+ c += name[--len];
+ c = c % count;
+ return c;
+}
+
+static int lmv_nid_policy(struct lmv_obd *lmv)
+{
+ struct obd_import *imp;
+ __u32 id;
+
+ /*
+ * XXX: To get nid we assume that underlying obd device is mdc.
+ */
+ imp = class_exp2cliimp(lmv->tgts[0].ltd_exp);
+ id = imp->imp_connection->c_self ^ (imp->imp_connection->c_self >> 32);
+ return id % lmv->desc.ld_tgt_count;
+}
+
+static int lmv_choose_mds(struct lmv_obd *lmv, struct md_op_data *op_data,
+ placement_policy_t placement)
+{
+ switch (placement) {
+ case PLACEMENT_CHAR_POLICY:
+ return lmv_all_chars_policy(lmv->desc.ld_tgt_count,
+ op_data->op_name,
+ op_data->op_namelen);
+ case PLACEMENT_NID_POLICY:
+ return lmv_nid_policy(lmv);
+
+ default:
+ break;
+ }
+
+ CERROR("Unsupported placement policy %x\n", placement);
+ return -EINVAL;
+}
+
+/**
+ * This is _inode_ placement policy function (not name).
+ */
+static int lmv_placement_policy(struct obd_device *obd,
+ struct md_op_data *op_data,
+ mdsno_t *mds)
+{
+ struct lmv_obd *lmv = &obd->u.lmv;
+ struct lmv_object *obj;
+ int rc;
+ ENTRY;
+
+ LASSERT(mds != NULL);
+
+ if (lmv->desc.ld_tgt_count == 1) {
+ *mds = 0;
+ RETURN(0);
+ }
+
+ /*
+ * Allocate new fid on target according to operation type and parent
+ * home mds.
+ */
+ obj = lmv_object_find(obd, &op_data->op_fid1);
+ if (obj != NULL || op_data->op_name == NULL ||
+ op_data->op_opc != LUSTRE_OPC_MKDIR) {
+ /*
+ * Allocate fid for non-dir or for null name or for case parent
+ * dir is split.
+ */
+ if (obj) {
+ lmv_object_put(obj);
+
+ /*
+ * If we have this flag turned on, and we see that
+ * parent dir is split, this means, that caller did not
+ * notice split yet. This is race and we would like to
+ * let caller know that.
+ */
+ if (op_data->op_bias & MDS_CHECK_SPLIT)
+ RETURN(-ERESTART);
+ }
+
+ /*
+ * Allocate new fid on same mds where parent fid is located and
+ * where operation will be sent. In case of split dir, ->op_fid1
+ * and ->op_mds here will contain fid and mds of slave directory
+ * object (assigned by caller).
+ */
+ *mds = op_data->op_mds;
+ rc = 0;
+ } else {
+ /*
+ * Parent directory is not split and we want to create a
+ * directory in it. Let's calculate where to place it according
+ * to operation data @op_data.
+ */
+ *mds = lmv_choose_mds(lmv, op_data, lmv->lmv_placement);
+ rc = 0;
+ }
+
+ if (rc) {
+ CERROR("Can't choose MDS, err = %d\n", rc);
+ } else {
+ LASSERT(*mds < lmv->desc.ld_tgt_count);
+ }
+
+ RETURN(rc);
+}
+
+int __lmv_fid_alloc(struct lmv_obd *lmv, struct lu_fid *fid,
+ mdsno_t mds)
+{
+ struct lmv_tgt_desc *tgt;
+ int rc;
+ ENTRY;
+
+ tgt = lmv_get_target(lmv, mds);
+
+ /*
+ * New seq alloc and FLD setup should be atomic. Otherwise we may find
+ * on server that seq in new allocated fid is not yet known.
+ */
+ down(&tgt->ltd_fid_sem);
+
+ if (!tgt->ltd_active)
+ GOTO(out, rc = -ENODEV);
+
+ /*
+ * Asking underlaying tgt layer to allocate new fid.
+ */
+ rc = obd_fid_alloc(tgt->ltd_exp, fid, NULL);
+ if (rc > 0) {
+ LASSERT(fid_is_sane(fid));
+ rc = 0;
+ }
+
+ EXIT;
+out:
+ up(&tgt->ltd_fid_sem);
+ return rc;
+}
+
+int lmv_fid_alloc(struct obd_export *exp, struct lu_fid *fid,
+ struct md_op_data *op_data)
+{
+ struct obd_device *obd = class_exp2obd(exp);
+ struct lmv_obd *lmv = &obd->u.lmv;
+ mdsno_t mds;
+ int rc;
+ ENTRY;
+
+ LASSERT(op_data != NULL);
+ LASSERT(fid != NULL);
+
+ rc = lmv_placement_policy(obd, op_data, &mds);
+ if (rc) {
+ CERROR("Can't get target for allocating fid, "
+ "rc %d\n", rc);
+ RETURN(rc);
+ }
+
+ rc = __lmv_fid_alloc(lmv, fid, mds);
+ if (rc) {
+ CERROR("Can't alloc new fid, rc %d\n", rc);
+ RETURN(rc);
+ }
+
+ RETURN(rc);
+}
+
+static int lmv_fid_delete(struct obd_export *exp, const struct lu_fid *fid)
+{
+ ENTRY;
+ LASSERT(exp != NULL && fid != NULL);
+ if (lmv_object_delete(exp, fid)) {
+ CDEBUG(D_INODE, "Object "DFID" is destroyed.\n",
+ PFID(fid));
+ }
+ RETURN(0);
+}
+
+static int lmv_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
+{
+ struct lmv_obd *lmv = &obd->u.lmv;
+ struct lprocfs_static_vars lvars;
+ struct lmv_desc *desc;
+ int rc;
+ int i = 0;
+ ENTRY;
+
+ if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
+ CERROR("LMV setup requires a descriptor\n");