diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index 4f8a0ce81be482e959be1f88b700ed91e1e428d3..7d8e702d76d73cd17a0ada00bd66841bf3ff4738 100644
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -1886,11 +1886,6 @@ struct thandle {
* top thandle here for now, will fix it when we have better
* callback mechanism */
struct thandle *th_top;
- /** context for this transaction, tag is LCT_TX_HANDLE */
- struct lu_context th_ctx;
-
- /** additional tags (layers can add in declare) */
- __u32 th_tags;
/** the last operation result in this transaction.
* this value is used in recovery */
diff --git a/lustre/include/lu_object.h b/lustre/include/lu_object.h
index ab41873bea3c375e7ec6d7e7e1eeac4392e93259..ae5bb3dde4c828992f4c3c718bdddc890d4c970a 100644
--- a/lustre/include/lu_object.h
+++ b/lustre/include/lu_object.h
@@ -997,10 +997,6 @@ enum lu_context_tag {
* Thread on dt server
*/
LCT_DT_THREAD = 1 << 1,
- /**
- * Context for transaction handle
- */
- LCT_TX_HANDLE = 1 << 2,
/**
* Thread on client
*/
diff --git a/lustre/osd-ldiskfs/osd_handler.c b/lustre/osd-ldiskfs/osd_handler.c
index a35f498dd2de370bd25b56df1f83ff029055a0e0..6041d8a37b9b8310ae0d5180a305c75909d7466d 100644
--- a/lustre/osd-ldiskfs/osd_handler.c
+++ b/lustre/osd-ldiskfs/osd_handler.c
@@ -1601,8 +1601,6 @@ static void osd_trans_commit_cb(struct super_block *sb,
lu_device_put(lud);
th->th_dev = NULL;
- lu_context_exit(&th->th_ctx);
- lu_context_fini(&th->th_ctx);
OBD_FREE_PTR(oh);
}
@@ -1640,7 +1638,6 @@ static struct thandle *osd_trans_create(const struct lu_env *env,
th = &oh->ot_super;
th->th_dev = d;
th->th_result = 0;
- th->th_tags = LCT_TX_HANDLE;
oh->ot_credits = 0;
INIT_LIST_HEAD(&oh->ot_commit_dcb_list);
INIT_LIST_HEAD(&oh->ot_stop_dcb_list);
@@ -1774,8 +1771,6 @@ static int osd_trans_start(const struct lu_env *env, struct dt_device *d,
if (!IS_ERR(jh)) {
oh->ot_handle = jh;
LASSERT(oti->oti_txns == 0);
- lu_context_init(&th->th_ctx, th->th_tags);
- lu_context_enter(&th->th_ctx);
lu_device_get(&d->dd_lu_dev);
lu_ref_add_at(&d->dd_lu_dev.ld_reference, &oh->ot_dev_link,
diff --git a/lustre/osd-zfs/osd_handler.c b/lustre/osd-zfs/osd_handler.c
index 598003108defd0dfb34288c8b2d9f07b0600fa86..35df6b75cfe8d2483027b2da75a9d5c890e13743 100644
--- a/lustre/osd-zfs/osd_handler.c
+++ b/lustre/osd-zfs/osd_handler.c
@@ -169,8 +169,6 @@ static void osd_trans_commit_cb(void *cb_data, int error)
lu_device_put(lud);
th->th_dev = NULL;
- lu_context_exit(&th->th_ctx);
- lu_context_fini(&th->th_ctx);
OBD_FREE_PTR(oh);
EXIT;
@@ -228,8 +226,6 @@ static int osd_trans_start(const struct lu_env *env, struct dt_device *d,
/* add commit callback */
dmu_tx_callback_register(oh->ot_tx, osd_trans_commit_cb, oh);
oh->ot_assigned = 1;
- lu_context_init(&th->th_ctx, th->th_tags);
- lu_context_enter(&th->th_ctx);
lu_device_get(&d->dd_lu_dev);
}
@@ -367,7 +363,6 @@ static struct thandle *osd_trans_create(const struct lu_env *env,
th = &oh->ot_super;
th->th_dev = dt;
th->th_result = 0;
- th->th_tags = LCT_TX_HANDLE;
RETURN(th);
}
diff --git a/lustre/osp/lproc_osp.c b/lustre/osp/lproc_osp.c
index 5fc7a028d352820de426bb183bd39c55331512d1..4fad7945c060b6aedc8ac3ccb4b157c4840971d1 100644
--- a/lustre/osp/lproc_osp.c
+++ b/lustre/osp/lproc_osp.c
@@ -875,6 +875,26 @@ osp_reserved_mb_low_seq_write(struct file *file, const char __user *buffer,
}
LPROC_SEQ_FOPS(osp_reserved_mb_low);
+static ssize_t
+lprocfs_force_sync_seq_write(struct file *file, const char __user *buffer,
+ size_t count, loff_t *off)
+{
+ struct seq_file *m = file->private_data;
+ struct obd_device *dev = m->private;
+ struct dt_device *dt = lu2dt_dev(dev->obd_lu_dev);
+ struct lu_env env;
+ int rc;
+
+ rc = lu_env_init(&env, LCT_LOCAL);
+ if (rc)
+ return rc;
+ rc = dt_sync(&env, dt);
+ lu_env_fini(&env);
+
+ return rc == 0 ? count : rc;
+}
+LPROC_SEQ_FOPS_WO_TYPE(osp, force_sync);
+
static struct lprocfs_vars lprocfs_osp_obd_vars[] = {
{ .name = "uuid",
.fops = &osp_uuid_fops },
@@ -929,6 +949,8 @@ static struct lprocfs_vars lprocfs_osp_obd_vars[] = {
.fops = &osp_reserved_mb_high_fops },
{ .name = "reserved_mb_low",
.fops = &osp_reserved_mb_low_fops },
+ { .name = "force_sync",
+ .fops = &osp_force_sync_fops },
/* for compatibility reasons */
{ .name = "destroys_in_flight",
diff --git a/lustre/osp/osp_dev.c b/lustre/osp/osp_dev.c
index 982a46f68eb174252317e666485e2b80d1d33675..d7848e2423939f26ab62663fc635dd759297c426 100644
--- a/lustre/osp/osp_dev.c
+++ b/lustre/osp/osp_dev.c
@@ -797,19 +797,24 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
struct osp_device *d = dt2osp_dev(dev);
cfs_time_t expire;
struct l_wait_info lwi = { 0 };
- unsigned long id, old;
- int rc = 0;
- unsigned long start = cfs_time_current();
+ int recs, rc = 0;
+ unsigned long start = cfs_time_current();
+ __u64 old;
+
ENTRY;
/* No Sync between MDTs yet. */
if (d->opd_connect_mdt)
RETURN(0);
+ recs = atomic_read(&d->opd_sync_changes);
+ old = atomic64_read(&d->opd_sync_processed_recs);
+
+ osp_sync_force(env, dt2osp_dev(dev));
+
if (unlikely(d->opd_imp_active == 0))
RETURN(-ENOTCONN);
- id = d->opd_sync_last_used_id;
down_write(&d->opd_async_updates_rwsem);
CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
@@ -825,26 +830,23 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
if (rc != 0)
GOTO(out, rc);
- CDEBUG(D_CACHE, "%s: id: used %lu, processed %llu\n",
- d->opd_obd->obd_name, id, d->opd_sync_last_processed_id);
-
- /* wait till all-in-line are processed */
- while (d->opd_sync_last_processed_id < id) {
-
- old = d->opd_sync_last_processed_id;
+ CDEBUG(D_CACHE, "%s: processed %lu\n", d->opd_obd->obd_name,
+ atomic64_read(&d->opd_sync_processed_recs));
+ while (atomic64_read(&d->opd_sync_processed_recs) < old + recs) {
+ __u64 last = atomic64_read(&d->opd_sync_processed_recs);
/* make sure the connection is fine */
expire = cfs_time_shift(obd_timeout);
lwi = LWI_TIMEOUT(expire - cfs_time_current(),
osp_sync_timeout, d);
l_wait_event(d->opd_sync_barrier_waitq,
- d->opd_sync_last_processed_id >= id,
- &lwi);
+ atomic64_read(&d->opd_sync_processed_recs)
+ >= old + recs, &lwi);
- if (d->opd_sync_last_processed_id >= id)
+ if (atomic64_read(&d->opd_sync_processed_recs) >= old + recs)
break;
- if (d->opd_sync_last_processed_id != old) {
+ if (atomic64_read(&d->opd_sync_processed_recs) != last) {
/* some progress have been made,
* keep trying... */
continue;
@@ -1344,8 +1346,12 @@ static struct lu_device *osp_device_fini(const struct lu_env *env,
osp->opd_async_requests = NULL;
}
- if (osp->opd_storage_exp)
+ if (osp->opd_storage_exp) {
+ /* wait for the commit callbacks to complete */
+ wait_event(osp->opd_sync_waitq,
+ atomic_read(&osp->opd_commits_registered) == 0);
obd_disconnect(osp->opd_storage_exp);
+ }
if (osp->opd_symlink)
lprocfs_remove(&osp->opd_symlink);
@@ -1869,7 +1875,7 @@ struct lu_context_key osp_thread_key = {
LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
struct lu_context_key osp_txn_key = {
- .lct_tags = LCT_OSP_THREAD | LCT_TX_HANDLE,
+ .lct_tags = LCT_OSP_THREAD,
.lct_init = osp_txn_key_init,
.lct_fini = osp_txn_key_fini
};
diff --git a/lustre/osp/osp_internal.h b/lustre/osp/osp_internal.h
index 6a95d9aca588727726e4fdc21e9cd6e7f6c2a89c..74dc8f8bfa75f46f9a8f8ef7ca858fe9542cdf5a 100644
--- a/lustre/osp/osp_internal.h
+++ b/lustre/osp/osp_internal.h
@@ -224,17 +224,20 @@ struct osp_device {
/* osd api's commit cb control structure */
struct dt_txn_callback opd_sync_txn_cb;
/* last used change number -- semantically similar to transno */
- __u64 opd_sync_last_used_id;
+ unsigned long opd_sync_last_used_id;
/* last committed change number -- semantically similar to
* last_committed */
__u64 opd_sync_last_committed_id;
- /* last processed (taken from llog) id */
- volatile __u64 opd_sync_last_processed_id;
- struct osp_id_tracker *opd_sync_tracker;
- struct list_head opd_sync_ontrack;
+ /* last processed catalog index */
+ int opd_sync_last_catalog_idx;
+ /* number of processed records */
+ atomic64_t opd_sync_processed_recs;
/* stop processing new requests until barrier=0 */
atomic_t opd_sync_barrier;
wait_queue_head_t opd_sync_barrier_waitq;
+ /* last generated id */
+ cfs_time_t opd_sync_next_commit_cb;
+ atomic_t opd_commits_registered;
/*
* statfs related fields: OSP maintains it on its own
@@ -812,6 +815,11 @@ int osp_sync_add(const struct lu_env *env, struct osp_object *o,
int osp_sync_init(const struct lu_env *env, struct osp_device *d);
int osp_sync_fini(struct osp_device *d);
void osp_sync_check_for_work(struct osp_device *osp);
+void osp_sync_force(const struct lu_env *env, struct osp_device *d);
+int osp_sync_add_commit_cb(const struct lu_env *env, struct osp_device *d,
+ struct thandle *th);
+int osp_sync_add_commit_cb_1s(const struct lu_env *env, struct osp_device *d,
+ struct thandle *th);
/* lwp_dev.c */
extern struct obd_ops lwp_obd_device_ops;
diff --git a/lustre/osp/osp_precreate.c b/lustre/osp/osp_precreate.c
index e3d2ee1ceba926258335b685baa328f30fe25eb2..cc40e98213a4b847d08e39403dfae02938e77481 100644
--- a/lustre/osp/osp_precreate.c
+++ b/lustre/osp/osp_precreate.c
@@ -29,7 +29,7 @@
* This file is part of Lustre, http://www.lustre.org/
* Lustre is a trademark of Sun Microsystems, Inc.
*
- * lustre/osp/osp_sync.c
+ * lustre/osp/osp_precreate.c
*
* Lustre OST Proxy Device
*
@@ -168,7 +168,7 @@ out:
*
* \param[in] d OSP device
*/
-static int osp_statfs_update(struct osp_device *d)
+static int osp_statfs_update(const struct lu_env *env, struct osp_device *d)
{
struct ptlrpc_request *req;
struct obd_import *imp;
@@ -208,6 +208,29 @@ static int osp_statfs_update(struct osp_device *d)
ptlrpcd_add_req(req);
+ /* we still want to sync changes if no new changes are coming */
+ if (cfs_time_before(cfs_time_current(), d->opd_sync_next_commit_cb))
+ GOTO(out, rc);
+
+ if (atomic_read(&d->opd_sync_changes)) {
+ struct thandle *th;
+
+ th = dt_trans_create(env, d->opd_storage);
+ if (IS_ERR(th)) {
+ CERROR("%s: can't sync\n", d->opd_obd->obd_name);
+ GOTO(out, rc);
+ }
+ rc = dt_trans_start_local(env, d->opd_storage, th);
+ if (rc == 0) {
+ CDEBUG(D_OTHER, "%s: sync forced, %d changes\n",
+ d->opd_obd->obd_name,
+ atomic_read(&d->opd_sync_changes));
+ osp_sync_add_commit_cb_1s(env, d, th);
+ dt_trans_stop(env, d->opd_storage, th);
+ }
+ }
+
+out:
RETURN(0);
}
@@ -1193,7 +1216,7 @@ static int osp_precreate_thread(void *_arg)
continue;
}
- if (osp_statfs_update(d)) {
+ if (osp_statfs_update(&env, d)) {
l_wait_event(d->opd_pre_waitq,
!osp_precreate_running(d), &lwi2);
continue;
@@ -1226,7 +1249,7 @@ static int osp_precreate_thread(void *_arg)
break;
if (osp_statfs_need_update(d))
- if (osp_statfs_update(d))
+ if (osp_statfs_update(&env, d))
break;
/* To avoid handling different seq in precreate/orphan
@@ -1359,7 +1382,7 @@ int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d)
{
struct l_wait_info lwi;
cfs_time_t expire = cfs_time_shift(obd_timeout);
- int precreated, rc;
+ int precreated, rc, synced = 0;
ENTRY;
@@ -1418,9 +1441,11 @@ int osp_precreate_reserve(const struct lu_env *env, struct osp_device *d)
* wait till that is done - some space might be released
*/
if (unlikely(rc == -ENOSPC)) {
- if (atomic_read(&d->opd_sync_changes)) {
+ if (atomic_read(&d->opd_sync_changes) && synced == 0) {
/* force local commit to release space */
dt_commit_async(env, d->opd_storage);
+ osp_sync_force(env, d);
+ synced = 1;
}
if (atomic_read(&d->opd_sync_rpcs_in_progress)) {
/* just wait till destroys are done */
diff --git a/lustre/osp/osp_sync.c b/lustre/osp/osp_sync.c
index aef7c79724f11c7f46ce9a583749d57a3a9bc445..8223366d96516b7f65779a1f583227b9ca7e4d2c 100644
--- a/lustre/osp/osp_sync.c
+++ b/lustre/osp/osp_sync.c
@@ -44,11 +44,6 @@
#include <lustre_update.h>
#include "osp_internal.h"
-static int osp_sync_id_traction_init(struct osp_device *d);
-static void osp_sync_id_traction_fini(struct osp_device *d);
-static __u64 osp_sync_id_get(struct osp_device *d, __u64 id);
-static void osp_sync_remove_from_tracker(struct osp_device *d);
-
/*
* this is a components of OSP implementing synchronization between MDS and OST
* it llogs all interesting changes (currently it's uig/gid change and object
@@ -128,9 +123,8 @@ static inline int osp_sync_stopped(struct osp_device *d)
*/
static inline int osp_sync_has_new_job(struct osp_device *d)
{
- return ((d->opd_sync_last_processed_id < d->opd_sync_last_used_id) &&
- (d->opd_sync_last_processed_id < d->opd_sync_last_committed_id))
- || (d->opd_sync_prev_done == 0);
+ return atomic_read(&d->opd_sync_changes) > 0 ||
+ d->opd_sync_prev_done == 0;
}
static inline int osp_sync_in_flight_conflict(struct osp_device *d,
@@ -259,6 +253,7 @@ static inline __u64 osp_sync_correct_id(struct osp_device *d,
return correct_id;
}
+
/**
* Check and return ready-for-new status.
*
@@ -292,8 +287,10 @@ static inline int osp_sync_can_process_new(struct osp_device *d,
return 1;
if (atomic_read(&d->opd_sync_changes) == 0)
return 0;
- if (rec == NULL ||
- osp_sync_correct_id(d, rec) <= d->opd_sync_last_committed_id)
+ if (rec == NULL)
+ return 1;
+ /* notice "<" not "<=" */
+ if (osp_sync_correct_id(d, rec) < d->opd_sync_last_committed_id)
return 1;
return 0;
}
@@ -342,9 +339,6 @@ int osp_sync_declare_add(const struct lu_env *env, struct osp_object *o,
LBUG();
}
- /* we want ->dt_trans_start() to allocate per-thandle structure */
- storage_th->th_tags |= LCT_OSP_THREAD;
-
ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
LASSERT(ctxt);
@@ -355,6 +349,28 @@ int osp_sync_declare_add(const struct lu_env *env, struct osp_object *o,
RETURN(rc);
}
+/* add the commit callback every second */
+int osp_sync_add_commit_cb_1s(const struct lu_env *env, struct osp_device *d,
+ struct thandle *th)
+{
+ int add = 0;
+
+ /* fast path */
+ if (cfs_time_before(cfs_time_current(), d->opd_sync_next_commit_cb))
+ return 0;
+
+ spin_lock(&d->opd_sync_lock);
+ if (cfs_time_aftereq(cfs_time_current(), d->opd_sync_next_commit_cb))
+ add = 1;
+ d->opd_sync_next_commit_cb = cfs_time_shift(1);
+ spin_unlock(&d->opd_sync_lock);
+
+ if (add == 0)
+ return 0;
+ return osp_sync_add_commit_cb(env, d, th);
+}
+
+
/**
* Generate a llog record for a given change.
*
@@ -383,7 +399,6 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
{
struct osp_thread_info *osi = osp_env_info(env);
struct llog_ctxt *ctxt;
- struct osp_txn_info *txn;
struct thandle *storage_th;
int rc;
@@ -422,11 +437,13 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
LBUG();
}
- txn = osp_txn_info(&storage_th->th_ctx);
- LASSERT(txn);
+ /* we keep the same id, but increment it when the callback
+ * is registered, so that all records upto the one taken
+ * by the callback are subject to processing */
+ spin_lock(&d->opd_sync_lock);
+ osi->osi_hdr.lrh_id = d->opd_sync_last_used_id;
+ spin_unlock(&d->opd_sync_lock);
- txn->oti_current_id = osp_sync_id_get(d, txn->oti_current_id);
- osi->osi_hdr.lrh_id = (txn->oti_current_id & 0xffffffffULL);
ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
if (ctxt == NULL)
RETURN(-ENOMEM);
@@ -443,6 +460,9 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
osi->osi_cookie.lgc_index, rc);
atomic_inc(&d->opd_sync_changes);
}
+
+ rc = osp_sync_add_commit_cb_1s(env, d, th);
+
/* return 0 always here, error case just cause no llog record */
RETURN(0);
}
@@ -929,23 +949,11 @@ static void osp_sync_process_record(const struct lu_env *env,
* we should decrease changes and bump last_processed_id.
*/
if (d->opd_sync_prev_done) {
- __u64 correct_id = osp_sync_correct_id(d, rec);
LASSERT(atomic_read(&d->opd_sync_changes) > 0);
- LASSERT(correct_id <= d->opd_sync_last_committed_id);
- /* NOTE: it's possible to meet same id if
- * OST stores few stripes of same file
- */
- while (1) {
- /* another thread may be trying to set new value */
- rmb();
- if (correct_id > d->opd_sync_last_processed_id) {
- d->opd_sync_last_processed_id = correct_id;
- wake_up(&d->opd_sync_barrier_waitq);
- } else
- break;
- }
atomic_dec(&d->opd_sync_changes);
+ wake_up(&d->opd_sync_barrier_waitq);
}
+ atomic64_inc(&d->opd_sync_processed_recs);
if (rc != 0) {
atomic_dec(&d->opd_sync_rpcs_in_flight);
atomic_dec(&d->opd_sync_rpcs_in_progress);
@@ -1125,9 +1133,6 @@ static int osp_sync_process_queues(const struct lu_env *env,
rec = NULL;
}
- if (d->opd_sync_last_processed_id == d->opd_sync_last_used_id)
- osp_sync_remove_from_tracker(d);
-
l_wait_event(d->opd_sync_waitq,
!osp_sync_running(d) ||
osp_sync_can_process_new(d, rec) ||
@@ -1432,10 +1437,6 @@ int osp_sync_init(const struct lu_env *env, struct osp_device *d)
if (d->opd_storage->dd_rdonly)
RETURN(0);
- rc = osp_sync_id_traction_init(d);
- if (rc)
- RETURN(rc);
-
/*
* initialize llog storing changes
*/
@@ -1465,7 +1466,6 @@ int osp_sync_init(const struct lu_env *env, struct osp_device *d)
err_llog:
osp_sync_llog_fini(env, d);
err_id:
- osp_sync_id_traction_fini(d);
return rc;
}
@@ -1490,222 +1490,93 @@ int osp_sync_fini(struct osp_device *d)
wait_event(thread->t_ctl_waitq, thread_is_stopped(thread));
}
- /*
- * unregister transaction callbacks only when sync thread
- * has finished operations with llog
- */
- osp_sync_id_traction_fini(d);
-
RETURN(0);
}
-static DEFINE_MUTEX(osp_id_tracker_sem);
-static struct list_head osp_id_tracker_list =
- LIST_HEAD_INIT(osp_id_tracker_list);
+struct osp_last_committed_cb {
+ struct dt_txn_commit_cb ospc_cb;
+ struct osp_device *ospc_dev;
+ __u64 ospc_transno;
+};
-/**
- * OSD commit callback.
- *
- * The function is used as a local OSD commit callback to track the highest
- * committed llog record id. see osp_sync_id_traction_init() for the details.
- *
- * \param[in] th local transaction handle committed
- * \param[in] cookie commit callback data (our private structure)
- */
-static void osp_sync_tracker_commit_cb(struct thandle *th, void *cookie)
+void osp_sync_local_commit_cb(struct lu_env *env, struct thandle *th,
+ struct dt_txn_commit_cb *dcb, int err)
{
- struct osp_id_tracker *tr = cookie;
- struct osp_device *d;
- struct osp_txn_info *txn;
+ struct osp_last_committed_cb *cb;
+ struct osp_device *d;
- LASSERT(tr);
+ cb = container_of0(dcb, struct osp_last_committed_cb, ospc_cb);
+ d = cb->ospc_dev;
- txn = osp_txn_info(&th->th_ctx);
- if (txn == NULL || txn->oti_current_id < tr->otr_committed_id)
- return;
+ CDEBUG(D_HA, "%s: %llu committed\n", d->opd_obd->obd_name,
+ cb->ospc_transno);
- spin_lock(&tr->otr_lock);
- if (likely(txn->oti_current_id > tr->otr_committed_id)) {
- CDEBUG(D_OTHER, "committed: %llu -> %llu\n",
- tr->otr_committed_id, txn->oti_current_id);
- tr->otr_committed_id = txn->oti_current_id;
+ spin_lock(&d->opd_sync_lock);
+ if (cb->ospc_transno > d->opd_sync_last_committed_id)
+ d->opd_sync_last_committed_id = cb->ospc_transno;
+ spin_unlock(&d->opd_sync_lock);
- list_for_each_entry(d, &tr->otr_wakeup_list,
- opd_sync_ontrack) {
- d->opd_sync_last_committed_id = tr->otr_committed_id;
- wake_up(&d->opd_sync_waitq);
- }
- }
- spin_unlock(&tr->otr_lock);
+ osp_sync_check_for_work(d);
+ lu_device_put(osp2lu_dev(d));
+ if (atomic_dec_and_test(&d->opd_commits_registered))
+ wake_up(&d->opd_sync_waitq);
+
+ OBD_FREE_PTR(cb);
}
-/**
- * Initialize commit tracking mechanism.
- *
- * Some setups may have thousands of OSTs and each will be represented by OSP.
- * Meaning order of magnitute many more changes to apply every second. In order
- * to keep the number of commit callbacks low this mechanism was introduced.
- * The mechanism is very similar to transno used by MDT service: it's an single
- * ID stream which can be assigned by any OSP to its llog records. The tricky
- * part is that ID is stored in per-transaction data and re-used by all the OSPs
- * involved in that transaction. Then all these OSPs are woken up utilizing a single OSD commit callback.
- *
- * The function initializes the data used by the tracker described above.
- * A singler tracker per OSD device is created.
- *
- * \param[in] d OSP device
- *
- * \retval 0 on success
- * \retval negative negated errno on error
- */
-static int osp_sync_id_traction_init(struct osp_device *d)
+int osp_sync_add_commit_cb(const struct lu_env *env, struct osp_device *d,
+ struct thandle *th)
{
- struct osp_id_tracker *tr, *found = NULL;
- int rc = 0;
+ struct osp_last_committed_cb *cb;
+ struct dt_txn_commit_cb *dcb;
+ int rc = 0;
+
+ OBD_ALLOC_PTR(cb);
+ if (cb == NULL)
+ return -ENOMEM;
+ cb->ospc_dev = d;
+ dcb = &cb->ospc_cb;
+ dcb->dcb_func = osp_sync_local_commit_cb;
+ spin_lock(&d->opd_sync_lock);
+ cb->ospc_transno = ++d->opd_sync_last_used_id;
+ spin_unlock(&d->opd_sync_lock);
- LASSERT(d);
- LASSERT(d->opd_storage);
- LASSERT(d->opd_sync_tracker == NULL);
- INIT_LIST_HEAD(&d->opd_sync_ontrack);
-
- mutex_lock(&osp_id_tracker_sem);
- list_for_each_entry(tr, &osp_id_tracker_list, otr_list) {
- if (tr->otr_dev == d->opd_storage) {
- LASSERT(atomic_read(&tr->otr_refcount));
- atomic_inc(&tr->otr_refcount);
- d->opd_sync_tracker = tr;
- found = tr;
- break;
- }
- }
+ rc = dt_trans_cb_add(th, dcb);
+ CDEBUG(D_HA, "%s: add commit cb at %llu, next at %llu, rc = %d\n",
+ d->opd_obd->obd_name, (unsigned long long) cfs_time_current(),
+ (unsigned long long) d->opd_sync_next_commit_cb, rc);
- if (found == NULL) {
- rc = -ENOMEM;
- OBD_ALLOC_PTR(tr);
- if (tr) {
- d->opd_sync_tracker = tr;
- spin_lock_init(&tr->otr_lock);
- tr->otr_dev = d->opd_storage;
- tr->otr_next_id = 1;
- tr->otr_committed_id = 0;
- atomic_set(&tr->otr_refcount, 1);
- INIT_LIST_HEAD(&tr->otr_wakeup_list);
- list_add(&tr->otr_list, &osp_id_tracker_list);
- tr->otr_tx_cb.dtc_txn_commit =
- osp_sync_tracker_commit_cb;
- tr->otr_tx_cb.dtc_cookie = tr;
- tr->otr_tx_cb.dtc_tag = LCT_MD_THREAD;
- dt_txn_callback_add(d->opd_storage, &tr->otr_tx_cb);
- rc = 0;
- }
- }
- mutex_unlock(&osp_id_tracker_sem);
+ if (likely(rc == 0)) {
+ lu_device_get(osp2lu_dev(d));
+ atomic_inc(&d->opd_commits_registered);
+ } else
+ OBD_FREE_PTR(cb);
return rc;
}
-/**
- * Release commit tracker.
- *
- * Decrease a refcounter on the tracker used by the given OSP device \a d.
- * If no more users left, then the tracker is released.
- *
- * \param[in] d OSP device
+/*
+ * generate an empty transaction and hook the commit callback in
+ * then force transaction commit
*/
-static void osp_sync_id_traction_fini(struct osp_device *d)
+void osp_sync_force(const struct lu_env *env, struct osp_device *d)
{
- struct osp_id_tracker *tr;
-
- ENTRY;
+ struct thandle *th;
+ int rc;
- LASSERT(d);
- tr = d->opd_sync_tracker;
- if (tr == NULL) {
- EXIT;
+ th = dt_trans_create(env, d->opd_storage);
+ if (IS_ERR(th)) {
+ CERROR("%s: can't sync\n", d->opd_obd->obd_name);
return;
}
-
- osp_sync_remove_from_tracker(d);
-
- mutex_lock(&osp_id_tracker_sem);
- if (atomic_dec_and_test(&tr->otr_refcount)) {
- dt_txn_callback_del(d->opd_storage, &tr->otr_tx_cb);
- LASSERT(list_empty(&tr->otr_wakeup_list));
- list_del(&tr->otr_list);
- OBD_FREE_PTR(tr);
- d->opd_sync_tracker = NULL;
- }
- mutex_unlock(&osp_id_tracker_sem);
-
- EXIT;
-}
-
-/**
- * Generate a new ID on a tracker.
- *
- * Generates a new ID using the tracker associated with the given OSP device
- * \a d, if the given ID \a id is non-zero. Unconditially adds OSP device to
- * the wakeup list, so OSP won't miss when a transaction using the ID is
- * committed.
- *
- * \param[in] d OSP device
- * \param[in] id 0 or ID generated previously
- *
- * \retval ID the caller should use
- */
-static __u64 osp_sync_id_get(struct osp_device *d, __u64 id)
-{
- struct osp_id_tracker *tr;
-
- tr = d->opd_sync_tracker;
- LASSERT(tr);
-
- /* XXX: we can improve this introducing per-cpu preallocated ids? */
- spin_lock(&tr->otr_lock);
- if (OBD_FAIL_CHECK(OBD_FAIL_MDS_TRACK_OVERFLOW))
- tr->otr_next_id = 0xfffffff0;
-
- if (unlikely(tr->otr_next_id <= d->opd_sync_last_used_id)) {
- spin_unlock(&tr->otr_lock);
- CERROR("%s: next %llu, last synced %llu\n",
- d->opd_obd->obd_name, tr->otr_next_id,
- d->opd_sync_last_used_id);
- LBUG();
+ rc = dt_trans_start_local(env, d->opd_storage, th);
+ if (rc == 0) {
+ CDEBUG(D_OTHER, "%s: sync forced, %d changes\n",
+ d->opd_obd->obd_name,
+ atomic_read(&d->opd_sync_changes));
+ rc = osp_sync_add_commit_cb(env, d, th);
+ dt_trans_stop(env, d->opd_storage, th);
}
- if (id == 0)
- id = tr->otr_next_id++;
- if (id > d->opd_sync_last_used_id)
- d->opd_sync_last_used_id = id;
- if (list_empty(&d->opd_sync_ontrack))
- list_add(&d->opd_sync_ontrack, &tr->otr_wakeup_list);
- spin_unlock(&tr->otr_lock);
- CDEBUG(D_OTHER, "new id %llu\n", id);
-
- return id;
+ dt_commit_async(env, d->opd_storage);
}
-
-/**
- * Stop to propagate commit status to OSP.
- *
- * If the OSP does not have any llog records she's waiting to commit, then
- * it is possible to unsubscribe from wakeups from the tracking using this
- * method.
- *
- * \param[in] d OSP device not willing to wakeup
- */
-static void osp_sync_remove_from_tracker(struct osp_device *d)
-{
- struct osp_id_tracker *tr;
-
- tr = d->opd_sync_tracker;
- LASSERT(tr);
-
- if (list_empty(&d->opd_sync_ontrack))
- return;
-
- spin_lock(&tr->otr_lock);
- list_del_init(&d->opd_sync_ontrack);
- spin_unlock(&tr->otr_lock);
-}
-
diff --git a/lustre/osp/osp_trans.c b/lustre/osp/osp_trans.c
index 4c620572fff245d25cc1f746a2a8bb5404b265bd..3e53062c7ec345b2f921b4b85fc1b55df37415b4 100644
--- a/lustre/osp/osp_trans.c
+++ b/lustre/osp/osp_trans.c
@@ -975,7 +975,6 @@ struct thandle *osp_trans_create(const struct lu_env *env, struct dt_device *d)
oth->ot_magic = OSP_THANDLE_MAGIC;
th = &oth->ot_super;
th->th_dev = d;
- th->th_tags = LCT_TX_HANDLE;
atomic_set(&oth->ot_refcount, 1);
INIT_LIST_HEAD(&oth->ot_commit_dcb_list);
diff --git a/lustre/target/update_records.c b/lustre/target/update_records.c
index a64dc38c3913093b7968a5ed2f50d050ca51f715..a36d5545255071f7d511fd70ef4670dd23ec4aba 100644
--- a/lustre/target/update_records.c
+++ b/lustre/target/update_records.c
@@ -1217,7 +1217,7 @@ static void update_key_fini(const struct lu_context *ctx,
LU_KEY_INIT(update, struct update_thread_info);
/* context key: update_thread_key */
LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
- LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
+ LCT_DT_THREAD | LCT_LOCAL);
EXPORT_SYMBOL(update_thread_key);
LU_KEY_INIT_GENERIC(update);
diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 76153b14ea85deb9a7b83a1ec37f7ee1d43b76e1..6c3e41438347c0849488e490317f505ff7c25011 100644
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -615,8 +615,6 @@ top_trans_create(const struct lu_env *env, struct dt_device *master_dev)
child_th->th_top = &top_th->tt_super;
child_th->th_wait_submit = 1;
top_th->tt_master_sub_thandle = child_th;
-
- top_th->tt_super.th_tags |= child_th->th_tags;
}
return &top_th->tt_super;
}
@@ -795,7 +793,6 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
top_th->tt_master_sub_thandle->th_sync = th->th_sync;
if (th->th_local)
top_th->tt_master_sub_thandle->th_local = th->th_local;
- top_th->tt_master_sub_thandle->th_tags = th->th_tags;
rc = dt_trans_start(env, top_th->tt_master_sub_thandle->th_dev,
top_th->tt_master_sub_thandle);
RETURN(rc);
@@ -813,7 +810,6 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
st->st_sub_th->th_sync = th->th_sync;
if (th->th_local)
st->st_sub_th->th_local = th->th_local;
- st->st_sub_th->th_tags = th->th_tags;
rc = dt_trans_start(env, st->st_sub_th->th_dev,
st->st_sub_th);
if (rc != 0)
@@ -951,7 +947,6 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
top_th->tt_master_sub_thandle->th_sync = th->th_sync;
if (th->th_local)
top_th->tt_master_sub_thandle->th_local = th->th_local;
- top_th->tt_master_sub_thandle->th_tags = th->th_tags;
rc = dt_trans_stop(env, master_dev,
top_th->tt_master_sub_thandle);
OBD_FREE_PTR(top_th);
@@ -1011,7 +1006,6 @@ stop_master_trans:
master_st->st_sub_th->th_local = th->th_local;
if (th->th_sync)
master_st->st_sub_th->th_sync = th->th_sync;
- master_st->st_sub_th->th_tags = th->th_tags;
master_st->st_sub_th->th_result = th->th_result;
rc = dt_trans_stop(env, master_st->st_dt, master_st->st_sub_th);
/* If it does not write_updates, then we call submit callback
@@ -1075,7 +1069,6 @@ stop_other_trans:
st->st_sub_th->th_sync = th->th_sync;
if (th->th_local)
st->st_sub_th->th_local = th->th_local;
- st->st_sub_th->th_tags = th->th_tags;
st->st_sub_th->th_result = th->th_result;
rc = dt_trans_stop(env, st->st_sub_th->th_dev,
st->st_sub_th);
diff --git a/lustre/tests/sanity.sh b/lustre/tests/sanity.sh
index 3420651e62c3020c5ec8f52de8f7f74792990a30..1d2f84d09942b1d91970b628c592a1daf4e8f767 100755
--- a/lustre/tests/sanity.sh
+++ b/lustre/tests/sanity.sh
@@ -14175,9 +14175,10 @@ test_239() {
mkdir -p $DIR/$tdir
createmany -o $DIR/$tdir/f- 5000
unlinkmany $DIR/$tdir/f- 5000
- do_nodes $list "lctl set_param -n osp*.*.sync_changes 1"
- changes=$(do_nodes $list "lctl get_param -n osc.*MDT*.sync_changes \
- osc.*MDT*.sync_in_flight" | calc_sum)
+ [ $(lustre_version_code $SINGLEMDS) -gt $(version_code 2.10.1) ] &&
+ do_nodes $list "lctl set_param -n osp.*.force_sync=1"
+ changes=$(do_nodes $list "lctl get_param -n osp.*MDT*.sync_changes \
+ osp.*MDT*.sync_in_flight" | calc_sum)
[ "$changes" -eq 0 ] || error "$changes not synced"
}
run_test 239 "osp_sync test"
diff --git a/lustre/tests/test-framework.sh b/lustre/tests/test-framework.sh
index becadcb91c4c21cc5b65296fd3a08040f88298a4..f4dbdc2e6f86f10ed0d44822d433ba3815987cc8 100755
--- a/lustre/tests/test-framework.sh
+++ b/lustre/tests/test-framework.sh
@@ -2517,7 +2517,7 @@ wait_update_facet() {
sync_all_data() {
do_nodes $(comma_list $(mdts_nodes)) \
- "lctl set_param -n osd*.*MDT*.force_sync=1"
+ "lctl set_param -n os[cd]*.*MDT*.force_sync=1"
do_nodes $(comma_list $(osts_nodes)) \
"lctl set_param -n osd*.*OS*.force_sync=1" 2>&1 |
grep -v 'Found no match'
@@ -2558,7 +2558,7 @@ wait_delete_completed_mds() {
mds2sync=$(comma_list $mds2sync)
# sync MDS transactions
- do_nodes $mds2sync "$LCTL set_param -n osd*.*MD*.force_sync 1"
+ do_nodes $mds2sync "$LCTL set_param -n os[cd]*.*MD*.force_sync 1"
# wait till all changes are sent and commmitted by OSTs
# for ldiskfs space is released upon execution, but DMU