/* Prepare the request */
imp = d->opd_obd->u.cli.cl_import;
LASSERT(imp);
+
+ if (OBD_FAIL_CHECK(OBD_FAIL_OSP_CHECK_ENOMEM))
+ RETURN(ERR_PTR(-ENOMEM));
+
req = ptlrpc_request_alloc(imp, format);
if (req == NULL)
RETURN(ERR_PTR(-ENOMEM));
* \param[in] h llog record
*
* \retval 0 on success
+ * \retval 1 on invalid record
* \retval negative negated errno on error
*/
static int osp_sync_new_setattr_job(struct osp_device *d,
ENTRY;
LASSERT(h->lrh_type == MDS_SETATTR64_REC);
+ if (OBD_FAIL_CHECK(OBD_FAIL_OSP_CHECK_INVALID_REC))
+ RETURN(1);
/* lsr_valid can only be 0 or have OBD_MD_{FLUID,FLGID} set,
* so no bits other than these should be set. */
if ((rec->lsr_valid & ~(OBD_MD_FLUID | OBD_MD_FLGID)) != 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);
+ /* return 1 on invalid record */
+ RETURN(1);
}
req = osp_sync_new_job(d, llh, h, OST_SETATTR, &RQF_OST_SETATTR);
body->oa.o_valid |= rec->lsr_valid;
osp_sync_send_new_rpc(d, req);
- RETURN(1);
+ RETURN(0);
}
/**
body->oa.o_valid |= OBD_MD_FLOBJCOUNT;
osp_sync_send_new_rpc(d, req);
- RETURN(1);
+ RETURN(0);
}
/**
* use OUT for OST as well, this will allow batching and better code
* unification.
*
- * \param[in] env LU environment provided by the caller
* \param[in] d OSP device
* \param[in] llh llog handle where the record is stored
* \param[in] h llog record
* \retval 0 on success
* \retval negative negated errno on error
*/
-static int osp_sync_new_unlink64_job(const struct lu_env *env,
- struct osp_device *d,
+static int osp_sync_new_unlink64_job(struct osp_device *d,
struct llog_handle *llh,
struct llog_rec_hdr *h)
{
body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
OBD_MD_FLOBJCOUNT;
osp_sync_send_new_rpc(d, req);
- RETURN(1);
+ RETURN(0);
}
/**
* \param[in] d OSP device
* \param[in] llh llog handle where the record is stored
* \param[in] rec llog record
- *
- * \retval 0 on success
- * \retval negative negated errno on error
*/
-static int osp_sync_process_record(const struct lu_env *env,
- struct osp_device *d,
- struct llog_handle *llh,
- struct llog_rec_hdr *rec)
+static void osp_sync_process_record(const struct lu_env *env,
+ struct osp_device *d,
+ struct llog_handle *llh,
+ struct llog_rec_hdr *rec)
{
+ struct llog_handle *cathandle = llh->u.phd.phd_cat_handle;
struct llog_cookie cookie;
int rc = 0;
+ ENTRY;
+
cookie.lgc_lgl = llh->lgh_id;
cookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
cookie.lgc_index = rec->lrh_index;
}
/* cancel any generation record */
- rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle,
- 1, &cookie);
+ rc = llog_cat_cancel_records(env, cathandle, 1, &cookie);
- return rc;
+ RETURN_EXIT;
}
/*
rc = osp_sync_new_unlink_job(d, llh, rec);
break;
case MDS_UNLINK64_REC:
- rc = osp_sync_new_unlink64_job(env, d, llh, rec);
+ rc = osp_sync_new_unlink64_job(d, llh, rec);
break;
case MDS_SETATTR64_REC:
rc = osp_sync_new_setattr_job(d, llh, rec);
default:
CERROR("%s: unknown record type: %x\n", d->opd_obd->obd_name,
rec->lrh_type);
- /* we should continue processing */
+ /* treat "unknown record type" as "invalid" */
+ rc = 1;
+ break;
}
- /* 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);
- LASSERT(rec->lrh_id <= d->opd_syn_last_committed_id);
- /*
- * 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) {
- d->opd_syn_last_processed_id = rec->lrh_id;
- wake_up(&d->opd_syn_barrier_waitq);
- }
+ spin_lock(&d->opd_syn_lock);
- d->opd_syn_changes--;
+ /* For all kinds of records, not matter successful or not,
+ * we should decrease changes and bump last_processed_id.
+ */
+ if (d->opd_syn_prev_done) {
+ LASSERT(d->opd_syn_changes > 0);
+ LASSERT(rec->lrh_id <= d->opd_syn_last_committed_id);
+ /* 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) {
+ d->opd_syn_last_processed_id = rec->lrh_id;
+ 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);
- spin_unlock(&d->opd_syn_lock);
- rc = 0;
- } else {
- spin_lock(&d->opd_syn_lock);
+ d->opd_syn_changes--;
+ }
+ if (rc != 0) {
d->opd_syn_rpc_in_flight--;
d->opd_syn_rpc_in_progress--;
- spin_unlock(&d->opd_syn_lock);
+ }
+ 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);
+
+ spin_unlock(&d->opd_syn_lock);
+
+ /* Delete the invalid record */
+ if (rc == 1) {
+ rc = llog_cat_cancel_records(env, cathandle, 1, &cookie);
+ if (rc != 0)
+ CERROR("%s: can't delete invalid record: "
+ "fid = "DFID", rec_id = %u, rc = %d\n",
+ d->opd_obd->obd_name,
+ PFID(lu_object_fid(&cathandle->lgh_obj->do_lu)),
+ rec->lrh_id, rc);
}
- CDEBUG(D_HA, "found record %x, %d, idx %u, id %u: %d\n",
- rec->lrh_type, rec->lrh_len, rec->lrh_index, rec->lrh_id, rc);
- return rc;
+ CDEBUG(D_HA, "found record %x, %d, idx %u, id %u\n",
+ rec->lrh_type, rec->lrh_len, rec->lrh_index, rec->lrh_id);
+
+ RETURN_EXIT;
}
/**
void *data)
{
struct osp_device *d = data;
- int rc;
do {
struct l_wait_info lwi = { 0 };
d->opd_syn_rpc_in_flight);
return 0;
}
-
- /*
- * try to send, in case of disconnection, suspend
- * processing till we can send this request
- */
- do {
- rc = osp_sync_process_record(env, d, llh, rec);
- /*
- * XXX: probably different handling is needed
- * for some bugs, like immediate exit or if
- * OSP gets inactive
- */
- if (rc) {
- CERROR("can't send: %d\n", rc);
- l_wait_event(d->opd_syn_waitq,
- !osp_sync_running(d) ||
- osp_sync_has_work(d),
- &lwi);
- }
- } while (rc != 0 && osp_sync_running(d));
-
+ osp_sync_process_record(env, d, llh, rec);
llh = NULL;
rec = NULL;
}