X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosp%2Fosp_sync.c;h=8713ff0e32cf871af8eee1a3e92325b65b87a4b1;hb=446eb3dda264c24acdeeea3158bba55a16c8c5e1;hp=fa90527ad12068476431e42d24b0037a5ba3a736;hpb=57d8d7f152b1a1cabc103193f859bd927b3fbe9c;p=fs%2Flustre-release.git

diff --git a/lustre/osp/osp_sync.c b/lustre/osp/osp_sync.c
index fa90527..8713ff0e3 100644
--- a/lustre/osp/osp_sync.c
+++ b/lustre/osp/osp_sync.c
@@ -41,9 +41,6 @@
  * Author: Mikhail Pershin <mike.pershin@intel.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 #define DEBUG_SUBSYSTEM S_MDS
 
 #include <lustre_log.h>
@@ -131,7 +128,7 @@ static inline int osp_sync_has_work(struct osp_device *d)
 		return 1;
 
 	/* has remotely committed? */
-	if (!cfs_list_empty(&d->opd_syn_committed_there))
+	if (!list_empty(&d->opd_syn_committed_there))
 		return 1;
 
 	return 0;
@@ -140,7 +137,7 @@ static inline int osp_sync_has_work(struct osp_device *d)
 #define osp_sync_check_for_work(d)                      \
 {                                                       \
 	if (osp_sync_has_work(d)) {                     \
-		cfs_waitq_signal(&d->opd_syn_waitq);    \
+		wake_up(&d->opd_syn_waitq);    \
 	}                                               \
 }
 
@@ -154,6 +151,8 @@ static inline int osp_sync_can_process_new(struct osp_device *d,
 {
 	LASSERT(d);
 
+	if (unlikely(atomic_read(&d->opd_syn_barrier) > 0))
+		return 0;
 	if (!osp_sync_low_in_progress(d))
 		return 0;
 	if (!osp_sync_low_in_flight(d))
@@ -238,6 +237,7 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
 		LASSERT(attr);
 		osi->osi_setattr.lsr_uid = attr->la_uid;
 		osi->osi_setattr.lsr_gid = attr->la_gid;
+		osi->osi_setattr.lsr_valid = attr->la_valid;
 		break;
 	default:
 		LBUG();
@@ -253,7 +253,7 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
 	if (ctxt == NULL)
 		RETURN(-ENOMEM);
 	rc = llog_add(env, ctxt->loc_handle, &osi->osi_hdr, &osi->osi_cookie,
-		      NULL, th);
+		      th);
 	llog_ctxt_put(ctxt);
 
 	CDEBUG(D_OTHER, "%s: new record "DOSTID":%lu/%lu: %d\n",
@@ -324,16 +324,16 @@ static void osp_sync_request_commit_cb(struct ptlrpc_request *req)
 
 	LASSERT(d);
 	LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-	LASSERT(cfs_list_empty(&req->rq_exp_list));
+	LASSERT(list_empty(&req->rq_exp_list));
 
 	ptlrpc_request_addref(req);
 
 	spin_lock(&d->opd_syn_lock);
-	cfs_list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+	list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
 	spin_unlock(&d->opd_syn_lock);
 
 	/* XXX: some batching wouldn't hurt */
-	cfs_waitq_signal(&d->opd_syn_waitq);
+	wake_up(&d->opd_syn_waitq);
 }
 
 static int osp_sync_interpret(const struct lu_env *env,
@@ -347,7 +347,7 @@ static int osp_sync_interpret(const struct lu_env *env,
 	LASSERT(d);
 
 	CDEBUG(D_HA, "reply req %p/%d, rc %d, transno %u\n", req,
-	       cfs_atomic_read(&req->rq_refcount),
+	       atomic_read(&req->rq_refcount),
 	       rc, (unsigned) req->rq_transno);
 	LASSERT(rc || req->rq_transno);
 
@@ -357,15 +357,15 @@ static int osp_sync_interpret(const struct lu_env *env,
 		 * but object doesn't exist anymore - cancell llog record
 		 */
 		LASSERT(req->rq_transno == 0);
-		LASSERT(cfs_list_empty(&req->rq_exp_list));
+		LASSERT(list_empty(&req->rq_exp_list));
 
 		ptlrpc_request_addref(req);
 
 		spin_lock(&d->opd_syn_lock);
-		cfs_list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+		list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
 		spin_unlock(&d->opd_syn_lock);
 
-		cfs_waitq_signal(&d->opd_syn_waitq);
+		wake_up(&d->opd_syn_waitq);
 	} else if (rc) {
 		struct obd_import *imp = req->rq_import;
 		/*
@@ -386,8 +386,9 @@ static int osp_sync_interpret(const struct lu_env *env,
 			spin_unlock(&d->opd_syn_lock);
 		}
 
-		cfs_waitq_signal(&d->opd_syn_waitq);
-	} else if (unlikely(d->opd_pre_status == -ENOSPC)) {
+		wake_up(&d->opd_syn_waitq);
+	} else if (d->opd_pre != NULL &&
+		   unlikely(d->opd_pre_status == -ENOSPC)) {
 		/*
 		 * if current status is -ENOSPC (lack of free space on OST)
 		 * then we should poll OST immediately once object destroy
@@ -400,6 +401,8 @@ static int osp_sync_interpret(const struct lu_env *env,
 	spin_lock(&d->opd_syn_lock);
 	d->opd_syn_rpc_in_flight--;
 	spin_unlock(&d->opd_syn_lock);
+	if (unlikely(atomic_read(&d->opd_syn_barrier) > 0))
+		wake_up(&d->opd_syn_barrier_waitq);
 	CDEBUG(D_OTHER, "%s: %d in flight, %d in progress\n",
 	       d->opd_obd->obd_name, d->opd_syn_rpc_in_flight,
 	       d->opd_syn_rpc_in_progress);
@@ -457,7 +460,7 @@ static struct ptlrpc_request *osp_sync_new_job(struct osp_device *d,
 	body->oa.o_lcookie.lgc_lgl = llh->lgh_id;
 	body->oa.o_lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
 	body->oa.o_lcookie.lgc_index = h->lrh_index;
-	CFS_INIT_LIST_HEAD(&req->rq_exp_list);
+	INIT_LIST_HEAD(&req->rq_exp_list);
 	req->rq_svc_thread = (void *) OSP_JOB_MAGIC;
 
 	req->rq_interpret_reply = osp_sync_interpret;
@@ -480,6 +483,15 @@ static int osp_sync_new_setattr_job(struct osp_device *d,
 	ENTRY;
 	LASSERT(h->lrh_type == MDS_SETATTR64_REC);
 
+	/* lsr_valid can only be 0 or LA_UID/GID set */
+	if ((rec->lsr_valid & ~(LA_UID | LA_GID)) != 0) {
+		CERROR("%s: invalid setattr record, lsr_valid:"LPU64"\n",
+		       d->opd_obd->obd_name, rec->lsr_valid);
+		/* return 0 so that sync thread can continue processing
+		 * other records. */
+		RETURN(0);
+	}
+
 	req = osp_sync_new_job(d, llh, h, OST_SETATTR, &RQF_OST_SETATTR);
 	if (IS_ERR(req))
 		RETURN(PTR_ERR(req));
@@ -489,14 +501,22 @@ static int osp_sync_new_setattr_job(struct osp_device *d,
 	body->oa.o_oi = rec->lsr_oi;
 	body->oa.o_uid = rec->lsr_uid;
 	body->oa.o_gid = rec->lsr_gid;
-	body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
-			   OBD_MD_FLUID | OBD_MD_FLGID;
+	body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID;
+	/* old setattr record (prior 2.6.0) doesn't have 'valid' stored,
+	 * we assume that both UID and GID are valid in that case. */
+	if (rec->lsr_valid == 0) {
+		body->oa.o_valid |= (OBD_MD_FLUID | OBD_MD_FLGID);
+	} else {
+		if (rec->lsr_valid & LA_UID)
+			body->oa.o_valid |= OBD_MD_FLUID;
+		if (rec->lsr_valid & LA_GID)
+			body->oa.o_valid |= OBD_MD_FLGID;
+	}
 
 	osp_sync_send_new_rpc(d, req);
-	RETURN(0);
+	RETURN(1);
 }
 
-/* Old records may be in old format, so we handle that too */
 static int osp_sync_new_unlink_job(struct osp_device *d,
 				   struct llog_handle *llh,
 				   struct llog_rec_hdr *h)
@@ -522,36 +542,107 @@ static int osp_sync_new_unlink_job(struct osp_device *d,
 		body->oa.o_valid |= OBD_MD_FLOBJCOUNT;
 
 	osp_sync_send_new_rpc(d, req);
-	RETURN(0);
+	RETURN(1);
 }
 
-static int osp_sync_new_unlink64_job(struct osp_device *d,
+static int osp_prep_unlink_update_req(const struct lu_env *env,
+				      struct osp_device *osp,
+				      struct llog_handle *llh,
+				      struct llog_rec_hdr *h,
+				      struct ptlrpc_request **reqp)
+{
+	struct llog_unlink64_rec	*rec = (struct llog_unlink64_rec *)h;
+	struct dt_update_request	*update = NULL;
+	struct ptlrpc_request		*req;
+	const char			*buf;
+	struct llog_cookie		lcookie;
+	int				size;
+	int				rc;
+	ENTRY;
+
+	update = out_create_update_req(&osp->opd_dt_dev);
+	if (IS_ERR(update))
+		RETURN(PTR_ERR(update));
+
+	/* This can only happens for unlink slave directory, so decrease
+	 * ref for ".." and "." */
+	rc = out_insert_update(env, update, OUT_REF_DEL, &rec->lur_fid, 0,
+			       NULL, NULL);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	rc = out_insert_update(env, update, OUT_REF_DEL, &rec->lur_fid, 0,
+			       NULL, NULL);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	lcookie.lgc_lgl = llh->lgh_id;
+	lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
+	lcookie.lgc_index = h->lrh_index;
+	size = sizeof(lcookie);
+	buf = (const char *)&lcookie;
+
+	rc = out_insert_update(env, update, OUT_DESTROY, &rec->lur_fid, 1,
+			       &size, &buf);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
+				 update->dur_req, &req);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	INIT_LIST_HEAD(&req->rq_exp_list);
+	req->rq_svc_thread = (void *)OSP_JOB_MAGIC;
+
+	req->rq_interpret_reply = osp_sync_interpret;
+	req->rq_commit_cb = osp_sync_request_commit_cb;
+	req->rq_cb_data = osp;
+
+	ptlrpc_request_set_replen(req);
+	*reqp = req;
+out:
+	if (update != NULL)
+		out_destroy_update_req(update);
+
+	RETURN(rc);
+}
+
+static int osp_sync_new_unlink64_job(const struct lu_env *env,
+				     struct osp_device *d,
 				     struct llog_handle *llh,
 				     struct llog_rec_hdr *h)
 {
 	struct llog_unlink64_rec	*rec = (struct llog_unlink64_rec *)h;
-	struct ptlrpc_request		*req;
+	struct ptlrpc_request		*req = NULL;
 	struct ost_body			*body;
 	int				 rc;
 
 	ENTRY;
 	LASSERT(h->lrh_type == MDS_UNLINK64_REC);
 
-	req = osp_sync_new_job(d, llh, h, OST_DESTROY, &RQF_OST_DESTROY);
-	if (IS_ERR(req))
-		RETURN(PTR_ERR(req));
-
-	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
-	if (body == NULL)
-		RETURN(-EFAULT);
-	rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
-	if (rc < 0)
-		RETURN(rc);
-	body->oa.o_misc = rec->lur_count;
-	body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID | OBD_MD_FLOBJCOUNT;
+	if (d->opd_connect_mdt) {
+		rc = osp_prep_unlink_update_req(env, d, llh, h, &req);
+		if (rc != 0)
+			RETURN(rc);
+	} else {
+		req = osp_sync_new_job(d, llh, h, OST_DESTROY,
+				       &RQF_OST_DESTROY);
+		if (IS_ERR(req))
+			RETURN(PTR_ERR(req));
 
+		body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
+		if (body == NULL)
+			RETURN(-EFAULT);
+		rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
+		if (rc < 0)
+			RETURN(rc);
+		body->oa.o_misc = rec->lur_count;
+		body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
+				   OBD_MD_FLOBJCOUNT;
+	}
 	osp_sync_send_new_rpc(d, req);
-	RETURN(0);
+	RETURN(1);
 }
 
 static int osp_sync_process_record(const struct lu_env *env,
@@ -602,18 +693,19 @@ static int osp_sync_process_record(const struct lu_env *env,
 		rc = osp_sync_new_unlink_job(d, llh, rec);
 		break;
 	case MDS_UNLINK64_REC:
-		rc = osp_sync_new_unlink64_job(d, llh, rec);
+		rc = osp_sync_new_unlink64_job(env, d, llh, rec);
 		break;
 	case MDS_SETATTR64_REC:
 		rc = osp_sync_new_setattr_job(d, llh, rec);
 		break;
 	default:
-		CERROR("unknown record type: %x\n", rec->lrh_type);
-		       rc = -EINVAL;
-		       break;
+		CERROR("%s: unknown record type: %x\n", d->opd_obd->obd_name,
+		       rec->lrh_type);
+		/* we should continue processing */
 	}
 
-	if (likely(rc == 0)) {
+	/* rc > 0 means sync RPC being added to the queue */
+	if (likely(rc > 0)) {
 		spin_lock(&d->opd_syn_lock);
 		if (d->opd_syn_prev_done) {
 			LASSERT(d->opd_syn_changes > 0);
@@ -622,8 +714,10 @@ static int osp_sync_process_record(const struct lu_env *env,
 			 * NOTE: it's possible to meet same id if
 			 * OST stores few stripes of same file
 			 */
-			if (rec->lrh_id > d->opd_syn_last_processed_id)
+			if (rec->lrh_id > d->opd_syn_last_processed_id) {
 				d->opd_syn_last_processed_id = rec->lrh_id;
+				wake_up(&d->opd_syn_barrier_waitq);
+			}
 
 			d->opd_syn_changes--;
 		}
@@ -631,6 +725,7 @@ static int osp_sync_process_record(const struct lu_env *env,
 		       d->opd_obd->obd_name, d->opd_syn_rpc_in_flight,
 		       d->opd_syn_rpc_in_progress);
 		spin_unlock(&d->opd_syn_lock);
+		rc = 0;
 	} else {
 		spin_lock(&d->opd_syn_lock);
 		d->opd_syn_rpc_in_flight--;
@@ -652,12 +747,12 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	struct ptlrpc_request	*req, *tmp;
 	struct llog_ctxt	*ctxt;
 	struct llog_handle	*llh;
-	cfs_list_t		 list;
+	struct list_head	 list;
 	int			 rc, done = 0;
 
 	ENTRY;
 
-	if (cfs_list_empty(&d->opd_syn_committed_there))
+	if (list_empty(&d->opd_syn_committed_there))
 		return;
 
 	/*
@@ -667,7 +762,7 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	 * notice: we do this upon commit as well because some backends
 	 * (like DMU) do not release space right away.
 	 */
-	if (unlikely(d->opd_pre_status == -ENOSPC))
+	if (d->opd_pre != NULL && unlikely(d->opd_pre_status == -ENOSPC))
 		osp_statfs_need_now(d);
 
 	/*
@@ -682,24 +777,43 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	llh = ctxt->loc_handle;
 	LASSERT(llh);
 
-	CFS_INIT_LIST_HEAD(&list);
+	INIT_LIST_HEAD(&list);
 	spin_lock(&d->opd_syn_lock);
-	cfs_list_splice(&d->opd_syn_committed_there, &list);
-	CFS_INIT_LIST_HEAD(&d->opd_syn_committed_there);
+	list_splice(&d->opd_syn_committed_there, &list);
+	INIT_LIST_HEAD(&d->opd_syn_committed_there);
 	spin_unlock(&d->opd_syn_lock);
 
-	cfs_list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
-		LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-		cfs_list_del_init(&req->rq_exp_list);
-
-		body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
-		LASSERT(body);
+	list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
+		struct llog_cookie *lcookie = NULL;
 
+		LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
+		list_del_init(&req->rq_exp_list);
+
+		if (d->opd_connect_mdt) {
+			struct object_update_request *ureq;
+			struct object_update *update;
+			ureq = req_capsule_client_get(&req->rq_pill,
+						      &RMF_OUT_UPDATE);
+			LASSERT(ureq != NULL &&
+				ureq->ourq_magic == UPDATE_REQUEST_MAGIC);
+
+			/* 1st/2nd is for decref . and .., 3rd one is for
+			 * destroy, where the log cookie is stored.
+			 * See osp_prep_unlink_update_req */
+			update = object_update_request_get(ureq, 2, NULL);
+			LASSERT(update != NULL);
+			lcookie = object_update_param_get(update, 0, NULL);
+			LASSERT(lcookie != NULL);
+		} else {
+			body = req_capsule_client_get(&req->rq_pill,
+						      &RMF_OST_BODY);
+			LASSERT(body);
+			lcookie = &body->oa.o_lcookie;
+		}
 		/* import can be closing, thus all commit cb's are
 		 * called we can check committness directly */
 		if (req->rq_transno <= imp->imp_peer_committed_transno) {
-			rc = llog_cat_cancel_records(env, llh, 1,
-						     &body->oa.o_lcookie);
+			rc = llog_cat_cancel_records(env, llh, 1, lcookie);
 			if (rc)
 				CERROR("%s: can't cancel record: %d\n",
 				       obd->obd_name, rc);
@@ -726,7 +840,7 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	/* wake up the thread if requested to stop:
 	 * it might be waiting for in-progress to complete */
 	if (unlikely(osp_sync_running(d) == 0))
-		cfs_waitq_signal(&d->opd_syn_waitq);
+		wake_up(&d->opd_syn_waitq);
 
 	EXIT;
 }
@@ -795,7 +909,7 @@ static int osp_sync_process_queues(const struct lu_env *env,
 		l_wait_event(d->opd_syn_waitq,
 			     !osp_sync_running(d) ||
 			     osp_sync_can_process_new(d, rec) ||
-			     !cfs_list_empty(&d->opd_syn_committed_there),
+			     !list_empty(&d->opd_syn_committed_there),
 			     &lwi);
 	} while (1);
 }
@@ -823,7 +937,6 @@ static int osp_sync_thread(void *_arg)
 	struct llog_handle	*llh;
 	struct lu_env		 env;
 	int			 rc, count;
-	char			 pname[16];
 
 	ENTRY;
 
@@ -834,13 +947,10 @@ static int osp_sync_thread(void *_arg)
 		RETURN(rc);
 	}
 
-	sprintf(pname, "osp-syn-%u", d->opd_index);
-	cfs_daemonize(pname);
-
 	spin_lock(&d->opd_syn_lock);
 	thread->t_flags = SVC_RUNNING;
 	spin_unlock(&d->opd_syn_lock);
-	cfs_waitq_signal(&thread->t_ctl_waitq);
+	wake_up(&thread->t_ctl_waitq);
 
 	ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 	if (ctxt == NULL) {
@@ -881,7 +991,7 @@ static int osp_sync_thread(void *_arg)
 		LASSERTF(count < 10, "%s: %d %d %sempty\n",
 			 d->opd_obd->obd_name, d->opd_syn_rpc_in_progress,
 			 d->opd_syn_rpc_in_flight,
-			 cfs_list_empty(&d->opd_syn_committed_there) ? "" :"!");
+			 list_empty(&d->opd_syn_committed_there) ? "" : "!");
 
 	}
 
@@ -890,14 +1000,15 @@ static int osp_sync_thread(void *_arg)
 	if (rc)
 		CERROR("can't cleanup llog: %d\n", rc);
 out:
-	thread->t_flags = SVC_STOPPED;
-
-	cfs_waitq_signal(&thread->t_ctl_waitq);
 	LASSERTF(d->opd_syn_rpc_in_progress == 0,
 		 "%s: %d %d %sempty\n",
 		 d->opd_obd->obd_name, d->opd_syn_rpc_in_progress,
 		 d->opd_syn_rpc_in_flight,
-		 cfs_list_empty(&d->opd_syn_committed_there) ? "" : "!");
+		 list_empty(&d->opd_syn_committed_there) ? "" : "!");
+
+	thread->t_flags = SVC_STOPPED;
+
+	wake_up(&thread->t_ctl_waitq);
 
 	lu_env_fini(&env);
 
@@ -906,11 +1017,12 @@ out:
 
 static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 {
-	struct osp_thread_info *osi = osp_env_info(env);
-	struct llog_handle     *lgh = NULL;
-	struct obd_device      *obd = d->opd_obd;
-	struct llog_ctxt       *ctxt;
-	int                     rc;
+	struct osp_thread_info	*osi = osp_env_info(env);
+	struct lu_fid		*fid = &osi->osi_fid;
+	struct llog_handle	*lgh = NULL;
+	struct obd_device	*obd = d->opd_obd;
+	struct llog_ctxt	*ctxt;
+	int			rc;
 
 	ENTRY;
 
@@ -922,8 +1034,13 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 	OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
 	obd->obd_lvfs_ctxt.dt = d->opd_storage;
 
+	if (d->opd_connect_mdt)
+		lu_local_obj_fid(fid, SLAVE_LLOG_CATALOGS_OID);
+	else
+		lu_local_obj_fid(fid, LLOG_CATALOGS_OID);
+
 	rc = llog_osd_get_cat_list(env, d->opd_storage, d->opd_index, 1,
-				   &osi->osi_cid);
+				   &osi->osi_cid, fid);
 	if (rc) {
 		CERROR("%s: can't get id from catalogs: rc = %d\n",
 		       obd->obd_name, rc);
@@ -968,7 +1085,7 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 		GOTO(out_close, rc);
 
 	rc = llog_osd_put_cat_list(env, d->opd_storage, d->opd_index, 1,
-				   &osi->osi_cid);
+				   &osi->osi_cid, fid);
 	if (rc)
 		GOTO(out_close, rc);
 
@@ -985,8 +1102,7 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 	memcpy(&osi->osi_gen.lgr_gen, &d->opd_syn_generation,
 	       sizeof(osi->osi_gen.lgr_gen));
 
-	rc = llog_cat_add(env, lgh, &osi->osi_gen.lgr_hdr, &osi->osi_cookie,
-			  NULL);
+	rc = llog_cat_add(env, lgh, &osi->osi_gen.lgr_hdr, &osi->osi_cookie);
 	if (rc < 0)
 		GOTO(out_close, rc);
 	llog_ctxt_put(ctxt);
@@ -1003,7 +1119,8 @@ static void osp_sync_llog_fini(const struct lu_env *env, struct osp_device *d)
 	struct llog_ctxt *ctxt;
 
 	ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
-	llog_cat_close(env, ctxt->loc_handle);
+	if (ctxt != NULL)
+		llog_cat_close(env, ctxt->loc_handle);
 	llog_cleanup(env, ctxt);
 }
 
@@ -1013,6 +1130,7 @@ static void osp_sync_llog_fini(const struct lu_env *env, struct osp_device *d)
 int osp_sync_init(const struct lu_env *env, struct osp_device *d)
 {
 	struct l_wait_info	 lwi = { 0 };
+	struct task_struct	*task;
 	int			 rc;
 
 	ENTRY;
@@ -1037,13 +1155,16 @@ int osp_sync_init(const struct lu_env *env, struct osp_device *d)
 	d->opd_syn_max_rpc_in_flight = OSP_MAX_IN_FLIGHT;
 	d->opd_syn_max_rpc_in_progress = OSP_MAX_IN_PROGRESS;
 	spin_lock_init(&d->opd_syn_lock);
-	cfs_waitq_init(&d->opd_syn_waitq);
-	cfs_waitq_init(&d->opd_syn_thread.t_ctl_waitq);
-	CFS_INIT_LIST_HEAD(&d->opd_syn_committed_there);
-
-	rc = cfs_create_thread(osp_sync_thread, d, 0);
-	if (rc < 0) {
-		CERROR("%s: can't start sync thread: rc = %d\n",
+	init_waitqueue_head(&d->opd_syn_waitq);
+	init_waitqueue_head(&d->opd_syn_barrier_waitq);
+	init_waitqueue_head(&d->opd_syn_thread.t_ctl_waitq);
+	INIT_LIST_HEAD(&d->opd_syn_committed_there);
+
+	task = kthread_run(osp_sync_thread, d, "osp-syn-%u-%u",
+			   d->opd_index, d->opd_group);
+	if (IS_ERR(task)) {
+		rc = PTR_ERR(task);
+		CERROR("%s: cannot start sync thread: rc = %d\n",
 		       d->opd_obd->obd_name, rc);
 		GOTO(err_llog, rc);
 	}
@@ -1066,8 +1187,8 @@ int osp_sync_fini(struct osp_device *d)
 	ENTRY;
 
 	thread->t_flags = SVC_STOPPING;
-	cfs_waitq_signal(&d->opd_syn_waitq);
-	cfs_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);
+	wake_up(&d->opd_syn_waitq);
+	wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);
 
 	/*
 	 * unregister transaction callbacks only when sync thread
@@ -1079,7 +1200,8 @@ int osp_sync_fini(struct osp_device *d)
 }
 
 static DEFINE_MUTEX(osp_id_tracker_sem);
-static CFS_LIST_HEAD(osp_id_tracker_list);
+static struct list_head osp_id_tracker_list =
+		LIST_HEAD_INIT(osp_id_tracker_list);
 
 static void osp_sync_tracker_commit_cb(struct thandle *th, void *cookie)
 {
@@ -1099,10 +1221,10 @@ static void osp_sync_tracker_commit_cb(struct thandle *th, void *cookie)
 		       tr->otr_committed_id, txn->oti_current_id);
 		tr->otr_committed_id = txn->oti_current_id;
 
-		cfs_list_for_each_entry(d, &tr->otr_wakeup_list,
-					opd_syn_ontrack) {
+		list_for_each_entry(d, &tr->otr_wakeup_list,
+				    opd_syn_ontrack) {
 			d->opd_syn_last_committed_id = tr->otr_committed_id;
-			cfs_waitq_signal(&d->opd_syn_waitq);
+			wake_up(&d->opd_syn_waitq);
 		}
 	}
 	spin_unlock(&tr->otr_lock);
@@ -1116,13 +1238,13 @@ static int osp_sync_id_traction_init(struct osp_device *d)
 	LASSERT(d);
 	LASSERT(d->opd_storage);
 	LASSERT(d->opd_syn_tracker == NULL);
-	CFS_INIT_LIST_HEAD(&d->opd_syn_ontrack);
+	INIT_LIST_HEAD(&d->opd_syn_ontrack);
 
 	mutex_lock(&osp_id_tracker_sem);
-	cfs_list_for_each_entry(tr, &osp_id_tracker_list, otr_list) {
+	list_for_each_entry(tr, &osp_id_tracker_list, otr_list) {
 		if (tr->otr_dev == d->opd_storage) {
-			LASSERT(cfs_atomic_read(&tr->otr_refcount));
-			cfs_atomic_inc(&tr->otr_refcount);
+			LASSERT(atomic_read(&tr->otr_refcount));
+			atomic_inc(&tr->otr_refcount);
 			d->opd_syn_tracker = tr;
 			found = tr;
 			break;
@@ -1138,9 +1260,9 @@ static int osp_sync_id_traction_init(struct osp_device *d)
 			tr->otr_dev = d->opd_storage;
 			tr->otr_next_id = 1;
 			tr->otr_committed_id = 0;
-			cfs_atomic_set(&tr->otr_refcount, 1);
-			CFS_INIT_LIST_HEAD(&tr->otr_wakeup_list);
-			cfs_list_add(&tr->otr_list, &osp_id_tracker_list);
+			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;
@@ -1170,10 +1292,10 @@ static void osp_sync_id_traction_fini(struct osp_device *d)
 	osp_sync_remove_from_tracker(d);
 
 	mutex_lock(&osp_id_tracker_sem);
-	if (cfs_atomic_dec_and_test(&tr->otr_refcount)) {
+	if (atomic_dec_and_test(&tr->otr_refcount)) {
 		dt_txn_callback_del(d->opd_storage, &tr->otr_tx_cb);
-		LASSERT(cfs_list_empty(&tr->otr_wakeup_list));
-		cfs_list_del(&tr->otr_list);
+		LASSERT(list_empty(&tr->otr_wakeup_list));
+		list_del(&tr->otr_list);
 		OBD_FREE_PTR(tr);
 		d->opd_syn_tracker = NULL;
 	}
@@ -1206,8 +1328,8 @@ static __u32 osp_sync_id_get(struct osp_device *d, __u32 id)
 		id = tr->otr_next_id++;
 	if (id > d->opd_syn_last_used_id)
 		d->opd_syn_last_used_id = id;
-	if (cfs_list_empty(&d->opd_syn_ontrack))
-		cfs_list_add(&d->opd_syn_ontrack, &tr->otr_wakeup_list);
+	if (list_empty(&d->opd_syn_ontrack))
+		list_add(&d->opd_syn_ontrack, &tr->otr_wakeup_list);
 	spin_unlock(&tr->otr_lock);
 	CDEBUG(D_OTHER, "new id %u\n", (unsigned) id);
 
@@ -1221,11 +1343,11 @@ static void osp_sync_remove_from_tracker(struct osp_device *d)
 	tr = d->opd_syn_tracker;
 	LASSERT(tr);
 
-	if (cfs_list_empty(&d->opd_syn_ontrack))
+	if (list_empty(&d->opd_syn_ontrack))
 		return;
 
 	spin_lock(&tr->otr_lock);
-	cfs_list_del_init(&d->opd_syn_ontrack);
+	list_del_init(&d->opd_syn_ontrack);
 	spin_unlock(&tr->otr_lock);
 }