+struct lod_recovery_data {
+ struct lod_device *lrd_lod;
+ struct lod_tgt_desc *lrd_ltd;
+ struct task_struct **lrd_task;
+ u32 lrd_idx;
+ struct lu_env lrd_env;
+ struct completion *lrd_started;
+};
+
+
+/**
+ * process update recovery record
+ *
+ * Add the update recovery recode to the update recovery list in
+ * lod_recovery_data. Then the recovery thread (target_recovery_thread)
+ * will redo these updates.
+ *
+ * \param[in]env execution environment
+ * \param[in]llh log handle of update record
+ * \param[in]rec update record to be replayed
+ * \param[in]data update recovery data which holds the necessary
+ * arguments for recovery (see struct lod_recovery_data)
+ *
+ * \retval 0 if the record is processed successfully.
+ * \retval negative errno if the record processing fails.
+ */
+static int lod_process_recovery_updates(const struct lu_env *env,
+ struct llog_handle *llh,
+ struct llog_rec_hdr *rec,
+ void *data)
+{
+ struct lod_recovery_data *lrd = data;
+ struct llog_cookie *cookie = &lod_env_info(env)->lti_cookie;
+ struct lu_target *lut;
+ u32 index = 0;
+
+ ENTRY;
+
+ if (!lrd->lrd_ltd) {
+ int rc;
+
+ rc = lodname2mdt_index(lod2obd(lrd->lrd_lod)->obd_name, &index);
+ if (rc != 0)
+ return rc;
+ } else {
+ index = lrd->lrd_ltd->ltd_index;
+ }
+
+ if (rec->lrh_len !=
+ llog_update_record_size((struct llog_update_record *)rec)) {
+ CERROR("%s: broken update record! index %u "DFID".%u: rc = %d\n",
+ lod2obd(lrd->lrd_lod)->obd_name, index,
+ PFID(&llh->lgh_id.lgl_oi.oi_fid), rec->lrh_index, -EIO);
+ return -EINVAL;
+ }
+
+ cookie->lgc_lgl = llh->lgh_id;
+ cookie->lgc_index = rec->lrh_index;
+ cookie->lgc_subsys = LLOG_UPDATELOG_ORIG_CTXT;
+
+ CDEBUG(D_HA, "%s: process recovery updates "DFID".%u\n",
+ lod2obd(lrd->lrd_lod)->obd_name,
+ PFID(&llh->lgh_id.lgl_oi.oi_fid), rec->lrh_index);
+ lut = lod2lu_dev(lrd->lrd_lod)->ld_site->ls_tgt;
+
+ if (lut->lut_obd->obd_stopping ||
+ lut->lut_obd->obd_abort_recovery)
+ return -ESHUTDOWN;
+
+ return insert_update_records_to_replay_list(lut->lut_tdtd,
+ (struct llog_update_record *)rec,
+ cookie, index);
+}
+
+/**
+ * recovery thread for update log
+ *
+ * Start recovery thread and prepare the sub llog, then it will retrieve
+ * the update records from the correpondent MDT and do recovery.
+ *
+ * \param[in] arg pointer to the recovery data
+ *
+ * \retval 0 if recovery succeeds
+ * \retval negative errno if recovery failed.
+ */
+static int lod_sub_recovery_thread(void *arg)
+{
+ struct lod_recovery_data *lrd = arg;
+ struct lod_device *lod = lrd->lrd_lod;
+ struct dt_device *dt;
+ struct llog_ctxt *ctxt = NULL;
+ struct lu_env *env = &lrd->lrd_env;
+ struct lu_target *lut;
+ struct lu_tgt_desc *mdt = NULL;
+ time64_t start;
+ int retries = 0;
+ int rc;
+
+ ENTRY;
+
+ lut = lod2lu_dev(lod)->ld_site->ls_tgt;
+ atomic_inc(&lut->lut_tdtd->tdtd_recovery_threads_count);
+ if (!lrd->lrd_ltd)
+ dt = lod->lod_child;
+ else
+ dt = lrd->lrd_ltd->ltd_tgt;
+
+ start = ktime_get_real_seconds();
+ complete(lrd->lrd_started);
+
+again:
+
+ if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_TGT_RECOVERY_CONNECT)) &&
+ lrd->lrd_ltd) {
+ OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_RECOVERY_CONNECT, cfs_fail_val);
+ rc = -EIO;
+ } else {
+ rc = lod_sub_prep_llog(env, lod, dt, lrd->lrd_idx);
+ }
+ if (!rc && !lod->lod_child->dd_rdonly) {
+ /* Process the recovery record */
+ ctxt = llog_get_context(dt->dd_lu_dev.ld_obd,
+ LLOG_UPDATELOG_ORIG_CTXT);
+ LASSERT(ctxt != NULL);
+ LASSERT(ctxt->loc_handle != NULL);
+
+ rc = llog_cat_process(env, ctxt->loc_handle,
+ lod_process_recovery_updates, lrd, 0, 0);
+ }
+
+ if (rc < 0) {
+ struct lu_device *top_device;
+
+ top_device = lod->lod_dt_dev.dd_lu_dev.ld_site->ls_top_dev;
+ /*
+ * Because the remote target might failover at the same time,
+ * let's retry here
+ */
+ if ((rc == -ETIMEDOUT || rc == -EAGAIN || rc == -EIO) &&
+ dt != lod->lod_child &&
+ !top_device->ld_obd->obd_abort_recovery &&
+ !top_device->ld_obd->obd_stopping) {
+ if (ctxt) {
+ if (ctxt->loc_handle)
+ llog_cat_close(env,
+ ctxt->loc_handle);
+ llog_ctxt_put(ctxt);
+ }
+ retries++;
+ CDEBUG(D_HA, "%s get update log failed %d, retry\n",
+ dt->dd_lu_dev.ld_obd->obd_name, rc);
+ goto again;
+ }
+
+ CERROR("%s get update log failed: rc = %d\n",
+ dt->dd_lu_dev.ld_obd->obd_name, rc);
+ llog_ctxt_put(ctxt);
+
+ spin_lock(&top_device->ld_obd->obd_dev_lock);
+ if (!top_device->ld_obd->obd_abort_recovery &&
+ !top_device->ld_obd->obd_stopping)
+ top_device->ld_obd->obd_abort_recovery = 1;
+ spin_unlock(&top_device->ld_obd->obd_dev_lock);
+
+ GOTO(out, rc);
+ }
+ llog_ctxt_put(ctxt);
+
+ CDEBUG(D_HA, "%s retrieved update log, duration %lld, retries %d\n",
+ dt->dd_lu_dev.ld_obd->obd_name, ktime_get_real_seconds() - start,
+ retries);
+
+ spin_lock(&lod->lod_lock);
+ if (!lrd->lrd_ltd)
+ lod->lod_child_got_update_log = 1;
+ else
+ lrd->lrd_ltd->ltd_got_update_log = 1;
+
+ if (!lod->lod_child_got_update_log) {
+ spin_unlock(&lod->lod_lock);
+ GOTO(out, rc = 0);
+ }
+
+ lod_foreach_mdt(lod, mdt) {
+ if (!mdt->ltd_got_update_log) {
+ spin_unlock(&lod->lod_lock);
+ GOTO(out, rc = 0);
+ }
+ }
+ lut->lut_tdtd->tdtd_replay_ready = 1;
+ spin_unlock(&lod->lod_lock);
+
+ CDEBUG(D_HA, "%s got update logs from all MDTs.\n",
+ lut->lut_obd->obd_name);
+ wake_up(&lut->lut_obd->obd_next_transno_waitq);
+ EXIT;
+
+out:
+ atomic_dec(&lut->lut_tdtd->tdtd_recovery_threads_count);
+ wake_up(&lut->lut_tdtd->tdtd_recovery_threads_waitq);
+ if (xchg(lrd->lrd_task, NULL) == NULL)
+ /* Someone is waiting for us to finish, need
+ * to synchronize cleanly.
+ */
+ wait_var_event(lrd, kthread_should_stop());
+ lu_env_fini(env);
+ OBD_FREE_PTR(lrd);
+ return 0;
+}
+
+/**
+ * finish sub llog context
+ *
+ * Stop update recovery thread for the sub device, then cleanup the
+ * correspondent llog ctxt.
+ *
+ * \param[in] env execution environment
+ * \param[in] lod lod device to do update recovery
+ * \param[in] thread recovery thread on this sub device
+ */
+void lod_sub_fini_llog(const struct lu_env *env,
+ struct dt_device *dt, struct task_struct **thread)
+{
+ struct obd_device *obd;
+ struct llog_ctxt *ctxt;
+ struct task_struct *task = NULL;
+
+ ENTRY;
+
+ obd = dt->dd_lu_dev.ld_obd;
+ CDEBUG(D_INFO, "%s: finish sub llog\n", obd->obd_name);
+ /* Wait for recovery thread to complete */
+ if (thread)
+ task = xchg(thread, NULL);
+ if (task)
+ kthread_stop(task);
+
+ ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
+ if (!ctxt)
+ RETURN_EXIT;
+
+ if (ctxt->loc_handle)
+ llog_cat_close(env, ctxt->loc_handle);
+
+ llog_cleanup(env, ctxt);
+
+ RETURN_EXIT;
+}
+