#include "mdt_internal.h"
static int mdt_server_data_update(const struct lu_env *env,
- struct mdt_device *mdt);
+ struct mdt_device *mdt,
+ int need_sync);
struct lu_buf *mdt_buf(const struct lu_env *env, void *area, ssize_t len)
{
LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
LASSERT(th != NULL);
- rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA);
+ rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA, 1);
if (rc == buf->lb_len)
rc = 0;
else if (rc >= 0)
rc = -EFAULT;
return rc;
}
-/* only one record write */
-enum {
- MDT_TXN_LAST_RCVD_WRITE_CREDITS = 3
-};
+static inline int mdt_trans_credit_get(const struct lu_env *env,
+ struct mdt_device *mdt,
+ enum mdt_txn_op op)
+{
+ struct dt_device *dev = mdt->mdt_bottom;
+ int cr;
+ switch (op) {
+ case MDT_TXN_CAPA_KEYS_WRITE_OP:
+ case MDT_TXN_LAST_RCVD_WRITE_OP:
+ cr = dev->dd_ops->dt_credit_get(env,
+ dev,
+ DTO_WRITE_BLOCK);
+ break;
+ default:
+ LBUG();
+ }
+ return cr;
+}
+
+void mdt_trans_credit_init(const struct lu_env *env,
+ struct mdt_device *mdt,
+ enum mdt_txn_op op)
+{
+ struct mdt_thread_info *mti;
+ struct txn_param *p;
+ int cr;
+
+ mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
+ p = &mti->mti_txn_param;
+
+ cr = mdt_trans_credit_get(env, mdt, op);
+ txn_param_init(p, cr);
+}
struct thandle* mdt_trans_start(const struct lu_env *env,
- struct mdt_device *mdt, int credits)
+ struct mdt_device *mdt)
{
struct mdt_thread_info *mti;
struct txn_param *p;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
p = &mti->mti_txn_param;
- txn_param_init(p, credits);
/* export can require sync operations */
if (mti->mti_exp != NULL)
return rc;
}
+static void mdt_client_cb(const struct mdt_device *mdt, __u64 transno,
+ void *data, int err)
+{
+ struct obd_device *obd = mdt2obd_dev(mdt);
+ target_client_add_cb(obd, transno, data, err);
+}
+
static inline int mdt_last_rcvd_header_write(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ int need_sync)
{
struct mdt_thread_info *mti;
struct thandle *th;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ if (mti->mti_exp) {
+ spin_lock(&mti->mti_exp->exp_lock);
+ mti->mti_exp->exp_need_sync = need_sync;
+ spin_unlock(&mti->mti_exp->exp_lock);
+ }
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
mti->mti_off = 0;
lsd_cpu_to_le(&mdt->mdt_lsd, &mti->mti_lsd);
+ if (need_sync && mti->mti_exp)
+ mdt_trans_add_cb(th, mdt_client_cb, mti->mti_exp);
+
rc = mdt_record_write(env, mdt->mdt_last_rcvd,
- mdt_buf_const(env, &mti->mti_lsd, sizeof(mti->mti_lsd)),
+ mdt_buf_const(env, &mti->mti_lsd,
+ sizeof(mti->mti_lsd)),
&mti->mti_off, th);
mdt_trans_stop(env, mdt, th);
return rc;
}
-
static int mdt_clients_data_init(const struct lu_env *env,
struct mdt_device *mdt,
unsigned long last_size)
{
struct lr_server_data *lsd = &mdt->mdt_lsd;
struct lsd_client_data *lcd = NULL;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
loff_t off;
int cl_idx;
int rc = 0;
}
static int mdt_server_data_init(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ struct lustre_sb_info *lsi)
{
struct lr_server_data *lsd = &mdt->mdt_lsd;
struct lsd_client_data *lcd = NULL;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct mdt_thread_info *mti;
struct dt_object *obj;
struct lu_attr *la;
+ struct lustre_disk_data *ldd;
unsigned long last_rcvd_size;
__u64 mount_count;
int rc;
}
mount_count = lsd->lsd_mount_count;
+ ldd = lsi->lsi_ldd;
+
+ if (ldd->ldd_flags & LDD_F_IAM_DIR)
+ lsd->lsd_feature_incompat |= OBD_INCOMPAT_IAM_DIR;
+
lsd->lsd_feature_compat = OBD_COMPAT_MDT;
+ lsd->lsd_feature_incompat |= OBD_INCOMPAT_FID;
spin_lock(&mdt->mdt_transno_lock);
mdt->mdt_last_transno = lsd->lsd_last_transno;
lsd->lsd_mount_count = mdt->mdt_mount_count;
/* save it, so mount count and last_transno is current */
- rc = mdt_server_data_update(env, mdt);
+ rc = mdt_server_data_update(env, mdt, (mti->mti_exp &&
+ mti->mti_exp->exp_need_sync));
if (rc)
GOTO(err_client, rc);
}
static int mdt_server_data_update(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ int need_sync)
{
int rc = 0;
ENTRY;
* mdt->mdt_last_rcvd may be NULL that time.
*/
if (mdt->mdt_last_rcvd != NULL)
- rc = mdt_last_rcvd_header_write(env, mdt);
+ rc = mdt_last_rcvd_header_write(env, mdt, need_sync);
RETURN(rc);
}
-void mdt_cb_new_client(const struct mdt_device *mdt, __u64 transno,
- void *data, int err)
-{
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
-
- target_client_add_cb(obd, transno, data, err);
-}
-
int mdt_client_new(const struct lu_env *env, struct mdt_device *mdt)
{
unsigned long *bitmap = mdt->mdt_client_bitmap;
struct mdt_export_data *med;
struct lsd_client_data *lcd;
struct lr_server_data *lsd = &mdt->mdt_lsd;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct thandle *th;
loff_t off;
int rc;
init_mutex(&med->med_lcd_lock);
LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
- /* write new client data */
+
+ /* Write new client data. */
off = med->med_lr_off;
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
- /* until this operations will be committed the sync is needed for this
- * export */
- mdt_trans_add_cb(th, mdt_cb_new_client, mti->mti_exp);
+ /*
+ * Until this operations will be committed the sync is needed
+ * for this export. This should be done _after_ starting the
+ * transaction so that many connecting clients will not bring
+ * server down with lots of sync writes.
+ */
+ mdt_trans_add_cb(th, mdt_client_cb, mti->mti_exp);
spin_lock(&mti->mti_exp->exp_lock);
mti->mti_exp->exp_need_sync = 1;
spin_unlock(&mti->mti_exp->exp_lock);
struct mdt_thread_info *mti;
struct mdt_export_data *med;
unsigned long *bitmap = mdt->mdt_client_bitmap;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct lr_server_data *lsd = &mdt->mdt_lsd;
int rc = 0;
ENTRY;
struct mdt_thread_info *mti;
struct mdt_export_data *med;
struct lsd_client_data *lcd;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
- struct thandle *th;
- loff_t off;
- int rc = 0;
+ struct obd_device *obd = mdt2obd_dev(mdt);
+ struct obd_export *exp;
+ struct thandle *th;
+ int need_sync;
+ loff_t off;
+ int rc = 0;
ENTRY;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
LASSERT(mti != NULL);
- med = &mti->mti_exp->exp_mdt_data;
+ exp = mti->mti_exp;
+ med = &exp->exp_mdt_data;
lcd = med->med_lcd;
if (!lcd)
RETURN(0);
/* XXX: If lcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
- if (!strcmp(med->med_lcd->lcd_uuid, obd->obd_uuid.uuid))
+ if (!strcmp(lcd->lcd_uuid, obd->obd_uuid.uuid))
GOTO(free, 0);
CDEBUG(D_INFO, "freeing client at idx %u, offset %lld\n",
LBUG();
}
+ /* Don't force sync on disconnect if aborting recovery,
+ * or it does num_clients * num_osts. b=17194 */
+ need_sync = (!exp->exp_libclient || exp->exp_need_sync) &&
+ !(exp->exp_flags & OBD_OPT_ABORT_RECOV);
+
/*
* This may be called from difficult reply handler path and
* mdt->mdt_last_rcvd may be NULL that time.
*/
if (mdt->mdt_last_rcvd != NULL) {
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+
+ spin_lock(&exp->exp_lock);
+ exp->exp_need_sync = need_sync;
+ spin_unlock(&exp->exp_lock);
+
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
GOTO(free, rc = PTR_ERR(th));
+ if (need_sync) {
+ /*
+ * Until this operations will be committed the sync
+ * is needed for this export.
+ */
+ mdt_trans_add_cb(th, mdt_client_cb, exp);
+ }
+
mutex_down(&med->med_lcd_lock);
memset(lcd, 0, sizeof *lcd);
}
CDEBUG(rc == 0 ? D_INFO : D_ERROR, "Zeroing out client idx %u in "
- "%s rc %d\n", med->med_lr_idx, LAST_RCVD, rc);
+ "%s %ssync rc %d\n", med->med_lr_idx, LAST_RCVD,
+ need_sync ? "" : "a", rc);
spin_lock(&mdt->mdt_client_bitmap_lock);
clear_bit(med->med_lr_idx, mdt->mdt_client_bitmap);
spin_unlock(&mdt->mdt_client_bitmap_lock);
- /*
- * Make sure the server's last_transno is up to date. Do this after the
- * client is freed so we know all the client's transactions have been
- * committed.
+ /*
+ * Make sure the server's last_transno is up to date. Do this
+ * after the client is freed so we know all the client's
+ * transactions have been committed.
*/
- mdt_server_data_update(env, mdt);
+ mdt_server_data_update(env, mdt, need_sync);
+
EXIT;
free:
OBD_FREE_PTR(lcd);
*/
if (mti->mti_transno == 0 &&
*transno_p == mdt->mdt_last_transno)
- mdt_server_data_update(mti->mti_env, mdt);
+ mdt_server_data_update(mti->mti_env, mdt,
+ (mti->mti_exp &&
+ mti->mti_exp->exp_need_sync));
*transno_p = mti->mti_transno;
static int mdt_txn_start_cb(const struct lu_env *env,
struct txn_param *param, void *cookie)
{
- param->tp_credits += MDT_TXN_LAST_RCVD_WRITE_CREDITS;
+ struct mdt_device *mdt = cookie;
+
+ param->tp_credits += mdt_trans_credit_get(env, mdt,
+ MDT_TXN_LAST_RCVD_WRITE_OP);
return 0;
}
if (mti->mti_has_trans) {
/* XXX: currently there are allowed cases, but the wrong cases
* are also possible, so better check is needed here */
- CDEBUG(D_INFO, "More than one transaction "LPU64"\n", mti->mti_transno);
+ CDEBUG(D_INFO, "More than one transaction "LPU64"\n",
+ mti->mti_transno);
return 0;
}
return mdt_last_rcvd_update(mti, txn);
}
-/* commit callback, need to update last_commited value */
+/* commit callback, need to update last_committed value */
static int mdt_txn_commit_cb(const struct lu_env *env,
struct thandle *txn, void *cookie)
{
struct mdt_device *mdt = cookie;
- struct obd_device *obd = md2lu_dev(&mdt->mdt_md_dev)->ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct mdt_txn_info *txi;
int i;
}
int mdt_fs_setup(const struct lu_env *env, struct mdt_device *mdt,
- struct obd_device *obd)
+ struct obd_device *obd,
+ struct lustre_sb_info *lsi)
{
struct lu_fid fid;
struct dt_object *o;
dt_txn_callback_add(mdt->mdt_bottom, &mdt->mdt_txn_cb);
- o = dt_store_open(env, mdt->mdt_bottom, LAST_RCVD, &fid);
+ o = dt_store_open(env, mdt->mdt_bottom, "", LAST_RCVD, &fid);
if (!IS_ERR(o)) {
mdt->mdt_last_rcvd = o;
- rc = mdt_server_data_init(env, mdt);
+ rc = mdt_server_data_init(env, mdt, lsi);
if (rc)
GOTO(put_last_rcvd, rc);
} else {
RETURN(rc);
}
- o = dt_store_open(env, mdt->mdt_bottom, CAPA_KEYS, &fid);
+ o = dt_store_open(env, mdt->mdt_bottom, "", CAPA_KEYS, &fid);
if (!IS_ERR(o)) {
mdt->mdt_ck_obj = o;
rc = mdt_capa_keys_init(env, mdt);
libcfs_nid2str(exp->exp_connection->c_peer.nid));
for (i = 0; i < oldrep->rs_nlocks; i++)
- ptlrpc_save_lock(req,
- &oldrep->rs_locks[i],
- oldrep->rs_modes[i]);
+ ptlrpc_save_lock(req, &oldrep->rs_locks[i],
+ oldrep->rs_modes[i], 0);
oldrep->rs_nlocks = 0;
DEBUG_REQ(D_HA, req, "stole locks for");
+ spin_lock(&oldrep->rs_lock);
ptlrpc_schedule_difficult_reply (oldrep);
+ spin_unlock(&oldrep->rs_lock);
spin_unlock (&svc->srv_lock);
break;
req->rq_status = rc;
body->valid |= OBD_MD_MDS;
}
- mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr, mdt_object_fid(child));
+ mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr,
+ mdt_object_fid(child));
mdt_object_put(mti->mti_env, child);
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff