struct lu_attr *attr);
int ofd_object_ff_check(const struct lu_env *env, struct ofd_object *fo);
int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
- obd_id id, struct ofd_seq *oseq, int nr);
+ obd_id id, struct ofd_seq *oseq, int nr, int sync);
void ofd_object_put(const struct lu_env *env, struct ofd_object *fo);
int ofd_attr_set(const struct lu_env *env, struct ofd_object *fo,
struct ofd_thread_info *info;
obd_seq seq = oa->o_seq;
struct ofd_seq *oseq;
- int rc = 0, diff;
+ int rc = 0, diff;
+ int sync_trans = 0;
ENTRY;
diff = 1; /* shouldn't we create this right now? */
} else {
diff = oa->o_id - ofd_seq_last_oid(oseq);
+ /* Do sync create if the seq is about to used up */
+ if (fid_seq_is_idif(oa->o_seq) ||
+ fid_seq_is_mdt0(oa->o_seq)) {
+ if (unlikely(oa->o_id >= IDIF_MAX_OID - 1))
+ sync_trans = 1;
+ } else if (fid_seq_is_norm(oa->o_seq)) {
+ if (unlikely(oa->o_id >=
+ LUSTRE_DATA_SEQ_MAX_WIDTH - 1))
+ sync_trans = 1;
+ } else {
+ CERROR("%s : invalid o_seq "LPX64": rc = %d\n",
+ ofd_name(ofd), oa->o_seq, -EINVAL);
+ GOTO(out, rc = -EINVAL);
+ }
}
}
if (diff > 0) {
}
rc = ofd_precreate_objects(env, ofd, next_id,
- oseq, count);
+ oseq, count, sync_trans);
if (rc > 0) {
created += rc;
diff -= rc;
}
int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
- obd_id id, struct ofd_seq *oseq, int nr)
+ obd_id id, struct ofd_seq *oseq, int nr, int sync)
{
struct ofd_thread_info *info = ofd_info(env);
struct ofd_object *fo = NULL;
if (IS_ERR(th))
GOTO(out, rc = PTR_ERR(th));
+ th->th_sync |= sync;
+
rc = dt_declare_record_write(env, oseq->os_lastid_obj, sizeof(tmp),
info->fti_off, th);
if (rc)
}
}
+static int osp_fid_end_seq(const struct lu_env *env, struct lu_fid *fid)
+{
+ if (fid_is_idif(fid)) {
+ struct osp_thread_info *info = osp_env_info(env);
+ struct ost_id *oi = &info->osi_oi;
+
+ fid_ostid_pack(fid, oi);
+ return oi->oi_id == IDIF_MAX_OID;
+ } else {
+ return fid_oid(fid) == LUSTRE_DATA_SEQ_MAX_WIDTH;
+ }
+}
+
+static inline int osp_precreate_end_seq_nolock(const struct lu_env *env,
+ struct osp_device *osp)
+{
+ struct lu_fid *fid = &osp->opd_pre_last_created_fid;
+
+ return osp_fid_end_seq(env, fid);
+}
+
+static inline int osp_precreate_end_seq(const struct lu_env *env,
+ struct osp_device *osp)
+{
+ int rc;
+
+ spin_lock(&osp->opd_pre_lock);
+ rc = osp_precreate_end_seq_nolock(env, osp);
+ spin_unlock(&osp->opd_pre_lock);
+ return rc;
+}
+
static inline int osp_is_fid_client(struct osp_device *osp)
{
struct obd_import *imp = osp->opd_obd->u.cli.cl_import;
CDEBUG(D_INODE, "fid for osp_obj %p is "DFID"!\n", osp_obj, PFID(fid));
+ /* If the precreate ends, it means it will be ready to rollover to
+ * the new sequence soon, all the creation should be synchronized,
+ * otherwise during replay, the replay fid will be inconsistent with
+ * last_used/create fid */
+ if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
+ th->th_sync = 1;
+
/*
* it's OK if the import is inactive by this moment - id was created
* by OST earlier, we just need to maintain it consistently on the disk
return rc;
}
+static inline int osp_create_end_seq(const struct lu_env *env,
+ struct osp_device *osp)
+{
+ struct lu_fid *fid = &osp->opd_pre_used_fid;
+ int rc;
+
+ spin_lock(&osp->opd_pre_lock);
+ rc = osp_fid_end_seq(env, fid);
+ spin_unlock(&osp->opd_pre_lock);
+ return rc;
+}
+
/**
* Write fid into last_oid/last_seq file.
**/
RETURN(rc);
}
+int osp_precreate_rollover_new_seq(struct lu_env *env, struct osp_device *osp)
+{
+ struct lu_fid *fid = &osp_env_info(env)->osi_fid;
+ struct lu_fid *last_fid = &osp->opd_last_used_fid;
+ int rc;
+ ENTRY;
+
+ rc = seq_client_get_seq(env, osp->opd_obd->u.cli.cl_seq, &fid->f_seq);
+ if (rc != 0) {
+ CERROR("%s: alloc fid error: rc = %d\n",
+ osp->opd_obd->obd_name, rc);
+ RETURN(rc);
+ }
+
+ fid->f_oid = 1;
+ fid->f_ver = 0;
+ LASSERTF(fid_seq(fid) != fid_seq(last_fid),
+ "fid "DFID", last_fid "DFID"\n", PFID(fid),
+ PFID(last_fid));
+
+ rc = osp_write_last_oid_seq_files(env, osp, fid, 1);
+ if (rc != 0) {
+ CERROR("%s: Can not update oid/seq file: rc = %d\n",
+ osp->opd_obd->obd_name, rc);
+ RETURN(rc);
+ }
+
+ LCONSOLE_INFO("%s: update sequence from "LPX64" to "LPX64"\n",
+ osp->opd_obd->obd_name, fid_seq(last_fid),
+ fid_seq(fid));
+ /* Update last_xxx to the new seq */
+ spin_lock(&osp->opd_pre_lock);
+ osp->opd_last_used_fid = *fid;
+ osp->opd_gap_start_fid = *fid;
+ osp->opd_pre_used_fid = *fid;
+ osp->opd_pre_last_created_fid = *fid;
+ spin_unlock(&osp->opd_pre_lock);
+
+ RETURN(rc);
+}
+
/**
* alloc fids for precreation.
* rc = 0 Success, @grow is the count of real allocation.
if (osp_statfs_need_update(d))
osp_statfs_update(d);
+ /* To avoid handling different seq in precreate/orphan
+ * cleanup, it will hold precreate until current seq is
+ * used up. */
+ if (unlikely(osp_precreate_end_seq(&env, d) &&
+ !osp_create_end_seq(&env, d)))
+ continue;
+
+ if (unlikely(osp_precreate_end_seq(&env, d) &&
+ osp_create_end_seq(&env, d))) {
+ LCONSOLE_INFO("%s:"LPX64" is used up."
+ " Update to new seq\n",
+ d->opd_obd->obd_name,
+ fid_seq(&d->opd_pre_last_created_fid));
+ rc = osp_precreate_rollover_new_seq(&env, d);
+ if (rc)
+ continue;
+ }
+
if (osp_precreate_near_empty(&env, d)) {
rc = osp_precreate_send(&env, d);
/* osp_precreate_send() sets opd_pre_status
rc = 0;
/* XXX: don't wake up if precreation is in progress */
- if (osp_precreate_near_empty_nolock(env, d))
+ if (osp_precreate_near_empty_nolock(env, d) &&
+ !osp_precreate_end_seq_nolock(env, d))
cfs_waitq_signal(&d->opd_pre_waitq);
break;