obd = class_name2obd(next);
if (obd == NULL) {
CERROR("%s: can't locate next device: %s\n",
- m->ofd_dt_dev.dd_lu_dev.ld_obd->obd_name, next);
+ ofd_name(m), next);
GOTO(out, rc = -ENOTCONN);
}
rc = obd_connect(NULL, exp, obd, &obd->obd_uuid, data, NULL);
if (rc) {
CERROR("%s: cannot connect to next dev %s: rc = %d\n",
- m->ofd_dt_dev.dd_lu_dev.ld_obd->obd_name, next, rc);
+ ofd_name(m), next, rc);
GOTO(out, rc);
}
LASSERT(m->ofd_osd_exp);
obd_disconnect(m->ofd_osd_exp);
- m->ofd_osd = NULL;
EXIT;
}
ENTRY;
- OBD_SLAB_ALLOC_PTR_GFP(of, ofd_object_kmem, __GFP_IO);
+ OBD_SLAB_ALLOC_PTR_GFP(of, ofd_object_kmem, GFP_NOFS);
if (of != NULL) {
struct lu_object *o;
struct lu_object_header *h;
extern int ost_handle(struct ptlrpc_request *req);
+static int ofd_lfsck_out_notify(const struct lu_env *env, void *data,
+ enum lfsck_events event)
+{
+ struct ofd_device *ofd = data;
+ struct obd_device *obd = ofd_obd(ofd);
+
+ switch (event) {
+ case LE_LASTID_REBUILDING:
+ CWARN("%s: Found crashed LAST_ID, deny creating new OST-object "
+ "on the device until the LAST_ID rebuilt successfully.\n",
+ obd->obd_name);
+ down_write(&ofd->ofd_lastid_rwsem);
+ ofd->ofd_lastid_rebuilding = 1;
+ up_write(&ofd->ofd_lastid_rwsem);
+ break;
+ case LE_LASTID_REBUILT: {
+ down_write(&ofd->ofd_lastid_rwsem);
+ ofd_seqs_free(env, ofd);
+ ofd->ofd_lastid_rebuilding = 0;
+ ofd->ofd_lastid_gen++;
+ up_write(&ofd->ofd_lastid_rwsem);
+ break;
+ }
+ default:
+ CERROR("%s: unknown lfsck event: rc = %d\n",
+ ofd_name(ofd), event);
+ return -EINVAL;
+ }
+
+ return 0;
+}
+
static int ofd_prepare(const struct lu_env *env, struct lu_device *pdev,
struct lu_device *dev)
{
struct ofd_device *ofd = ofd_dev(dev);
struct obd_device *obd = ofd_obd(ofd);
struct lu_device *next = &ofd->ofd_osd->dd_lu_dev;
- struct lfsck_start_param lsp;
int rc;
ENTRY;
if (rc != 0)
RETURN(rc);
- rc = lfsck_register(env, ofd->ofd_osd, &ofd->ofd_dt_dev, false);
+ rc = lfsck_register(env, ofd->ofd_osd, ofd->ofd_osd, obd,
+ ofd_lfsck_out_notify, ofd, false);
if (rc != 0) {
CERROR("%s: failed to initialize lfsck: rc = %d\n",
obd->obd_name, rc);
RETURN(rc);
}
- lsp.lsp_start = NULL;
- lsp.lsp_namespace = ofd->ofd_namespace;
- rc = lfsck_start(env, ofd->ofd_osd, &lsp);
- if (rc != 0) {
- CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n",
- obd->obd_name, rc);
- rc = 0;
- }
+ rc = lfsck_register_namespace(env, ofd->ofd_osd, ofd->ofd_namespace);
+ /* The LFSCK instance is registered just now, so it must be there when
+ * register the namespace to such instance. */
+ LASSERTF(rc == 0, "register namespace failed: rc = %d\n", rc);
target_recovery_init(&ofd->ofd_lut, tgt_request_handle);
LASSERT(obd->obd_no_conn);
LASSERT(exp != NULL);
skip_orphan = !!(exp_connect_flags(exp) & OBD_CONNECT_SKIP_ORPHAN);
+ if (OBD_FAIL_CHECK(OBD_FAIL_OST_NODESTROY))
+ goto done;
+
LCONSOLE(D_INFO, "%s: deleting orphan objects from "DOSTID
" to "DOSTID"\n", ofd_name(ofd), seq, end_id + 1, seq, last);
GOTO(out_put, rc);
rc = ofd_destroy_by_fid(env, ofd, fid, 1);
- if (rc != 0 && rc != -ENOENT) /* this is pretty fatal... */
- CEMERG("%s: error destroying precreated id "DFID
- ": rc = %d\n", ofd_name(ofd), PFID(fid), rc);
+ if (rc != 0 && rc != -ENOENT && rc != -ESTALE &&
+ likely(rc != -EREMCHG && rc != -EINPROGRESS))
+ /* this is pretty fatal... */
+ CEMERG("%s: error destroying precreated id "
+ DFID": rc = %d\n",
+ ofd_name(ofd), PFID(fid), rc);
oid--;
if (!skip_orphan) {
CDEBUG(D_HA, "%s: after destroy: set last_id to "DOSTID"\n",
ofd_name(ofd), seq, oid);
+done:
if (!skip_orphan) {
+ ofd_seq_last_oid_set(oseq, oid);
rc = ofd_seq_last_oid_write(env, ofd, oseq);
} else {
/* don't reuse orphan object, return last used objid */
struct ost_body *repbody;
const struct obdo *oa = &tsi->tsi_ost_body->oa;
struct obdo *rep_oa;
- struct ofd_device *ofd = ofd_exp(tsi->tsi_exp);
+ struct obd_export *exp = tsi->tsi_exp;
+ struct ofd_device *ofd = ofd_exp(exp);
obd_seq seq = ostid_seq(&oa->o_oi);
obd_id oid = ostid_id(&oa->o_oi);
struct ofd_seq *oseq;
if (repbody == NULL)
RETURN(-ENOMEM);
+ down_read(&ofd->ofd_lastid_rwsem);
+ /* Currently, for safe, we do not distinguish which LAST_ID is broken,
+ * we may do that in the future.
+ * Return -ENOSPC until the LAST_ID rebuilt. */
+ if (unlikely(ofd->ofd_lastid_rebuilding))
+ GOTO(out_sem, rc = -ENOSPC);
+
rep_oa = &repbody->oa;
rep_oa->o_oi = oa->o_oi;
if (IS_ERR(oseq)) {
CERROR("%s: Can't find FID Sequence "LPX64": rc = %ld\n",
ofd_name(ofd), seq, PTR_ERR(oseq));
- RETURN(-EINVAL);
+ GOTO(out_sem, rc = -EINVAL);
}
if ((oa->o_valid & OBD_MD_FLFLAGS) &&
/* former ofd_handle_precreate */
if ((oa->o_valid & OBD_MD_FLFLAGS) &&
(oa->o_flags & OBD_FL_DELORPHAN)) {
+ exp->exp_filter_data.fed_lastid_gen = ofd->ofd_lastid_gen;
+
/* destroy orphans */
if (lustre_msg_get_conn_cnt(tgt_ses_req(tsi)->rq_reqmsg) <
- tsi->tsi_exp->exp_conn_cnt) {
+ exp->exp_conn_cnt) {
CERROR("%s: dropping old orphan cleanup request\n",
ofd_name(ofd));
GOTO(out_nolock, rc = 0);
/* FIXME: should reset precreate_next_id on MDS */
rc = 0;
} else if (diff < 0) {
- rc = ofd_orphans_destroy(tsi->tsi_env, tsi->tsi_exp,
+ rc = ofd_orphans_destroy(tsi->tsi_env, exp,
ofd, rep_oa);
oseq->os_destroys_in_progress = 0;
} else {
oseq->os_destroys_in_progress = 0;
}
} else {
+ if (unlikely(exp->exp_filter_data.fed_lastid_gen !=
+ ofd->ofd_lastid_gen)) {
+ ofd_obd_disconnect(exp);
+ GOTO(out_nolock, rc = -ENOTCONN);
+ }
+
mutex_lock(&oseq->os_create_lock);
if (lustre_msg_get_conn_cnt(tgt_ses_req(tsi)->rq_reqmsg) <
- tsi->tsi_exp->exp_conn_cnt) {
+ exp->exp_conn_cnt) {
CERROR("%s: dropping old precreate request\n",
ofd_name(ofd));
GOTO(out, rc = 0);
ostid_set_id(&rep_oa->o_oi, ofd_seq_last_oid(oseq));
}
EXIT;
- ofd_counter_incr(tsi->tsi_exp, LPROC_OFD_STATS_CREATE,
+ ofd_counter_incr(exp, LPROC_OFD_STATS_CREATE,
tsi->tsi_jobid, 1);
out:
mutex_unlock(&oseq->os_create_lock);
rep_oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
ofd_seq_put(tsi->tsi_env, oseq);
+
+out_sem:
+ up_read(&ofd->ofd_lastid_rwsem);
return rc;
}
}
rc = tgt_sync(tsi->tsi_env, tsi->tsi_tgt,
- fo != NULL ? ofd_object_child(fo) : NULL);
+ fo != NULL ? ofd_object_child(fo) : NULL,
+ repbody->oa.o_size, repbody->oa.o_blocks);
if (rc)
GOTO(put, rc);
}
rc = ofd_object_punch(tsi->tsi_env, fo, start, end, &info->fti_attr,
- ff);
+ ff, (struct obdo *)oa);
if (rc)
GOTO(out_put, rc);
.tos_hs = tgt_dlm_handlers
},
{
- .tos_opc_start = UPDATE_OBJ,
- .tos_opc_end = UPDATE_LAST_OPC,
+ .tos_opc_start = OUT_UPDATE_FIRST_OPC,
+ .tos_opc_end = OUT_UPDATE_LAST_OPC,
.tos_hs = tgt_out_handlers
},
{
.tos_hs = seq_handlers
},
{
+ .tos_opc_start = LFSCK_FIRST_OPC,
+ .tos_opc_end = LFSCK_LAST_OPC,
+ .tos_hs = tgt_lfsck_handlers
+ },
+ {
.tos_hs = NULL
}
};
m->ofd_tot_granted = 0;
m->ofd_tot_pending = 0;
m->ofd_seq_count = 0;
+ init_waitqueue_head(&m->ofd_inconsistency_thread.t_ctl_waitq);
+ INIT_LIST_HEAD(&m->ofd_inconsistency_list);
+ spin_lock_init(&m->ofd_inconsistency_lock);
spin_lock_init(&m->ofd_batch_lock);
rwlock_init(&obd->u.filter.fo_sptlrpc_lock);
sptlrpc_rule_set_init(&obd->u.filter.fo_sptlrpc_rset);
+ init_rwsem(&m->ofd_lastid_rwsem);
obd->u.filter.fo_fl_oss_capa = 0;
CFS_INIT_LIST_HEAD(&obd->u.filter.fo_capa_keys);
if (rc)
GOTO(err_fini_lut, rc);
+ rc = ofd_start_inconsistency_verification_thread(m);
+ if (rc != 0)
+ GOTO(err_fini_fs, rc);
+
RETURN(0);
+
+err_fini_fs:
+ ofd_fs_cleanup(env, m);
err_fini_lut:
tgt_fini(env, &m->ofd_lut);
err_free_ns:
static void ofd_fini(const struct lu_env *env, struct ofd_device *m)
{
- struct obd_device *obd = ofd_obd(m);
- struct lu_device *d = &m->ofd_dt_dev.dd_lu_dev;
+ struct obd_device *obd = ofd_obd(m);
+ struct lu_device *d = &m->ofd_dt_dev.dd_lu_dev;
+ struct lfsck_stop stop;
- lfsck_stop(env, m->ofd_osd, true);
- lfsck_degister(env, m->ofd_osd);
+ stop.ls_status = LS_PAUSED;
+ stop.ls_flags = 0;
+ lfsck_stop(env, m->ofd_osd, &stop);
target_recovery_fini(obd);
obd_exports_barrier(obd);
obd_zombie_barrier();
tgt_fini(env, &m->ofd_lut);
+ ofd_stop_inconsistency_verification_thread(m);
+ lfsck_degister(env, m->ofd_osd);
ofd_fs_cleanup(env, m);
ofd_free_capa_keys(m);
ofd_stack_fini(env, m, &m->ofd_dt_dev.dd_lu_dev);
ofd_procfs_fini(m);
LASSERT(atomic_read(&d->ld_ref) == 0);
- server_put_mount(obd->obd_name);
+ server_put_mount(obd->obd_name, true);
EXIT;
}
lprocfs_ofd_init_vars(&lvars);
- rc = class_register_type(&ofd_obd_ops, NULL, NULL,
+ rc = class_register_type(&ofd_obd_ops, NULL, true, NULL,
#ifndef HAVE_ONLY_PROCFS_SEQ
- lvars.module_vars,
+ lvars.module_vars,
#endif
- LUSTRE_OST_NAME, &ofd_device_type);
+ LUSTRE_OST_NAME, &ofd_device_type);
return rc;
}