+/**
+ * This function is called if this client was notified for target restarting
+ * by the MGS. A CONFIG_READ RPC is going to send to fetch recovery logs.
+ */
+static int mgc_process_recover_log(struct obd_device *obd,
+ struct config_llog_data *cld)
+{
+ struct ptlrpc_request *req = NULL;
+ struct config_llog_instance *cfg = &cld->cld_cfg;
+ struct mgs_config_body *body;
+ struct mgs_config_res *res;
+ struct ptlrpc_bulk_desc *desc;
+ cfs_page_t **pages;
+ int nrpages;
+ bool eof = true;
+ bool mne_swab = false;
+ int i;
+ int ealen;
+ int rc;
+ ENTRY;
+
+ /* allocate buffer for bulk transfer.
+ * if this is the first time for this mgs to read logs,
+ * CONFIG_READ_NRPAGES_INIT will be used since it will read all logs
+ * once; otherwise, it only reads increment of logs, this should be
+ * small and CONFIG_READ_NRPAGES will be used.
+ */
+ nrpages = CONFIG_READ_NRPAGES;
+ if (cfg->cfg_last_idx == 0) /* the first time */
+ nrpages = CONFIG_READ_NRPAGES_INIT;
+
+ OBD_ALLOC(pages, sizeof(*pages) * nrpages);
+ if (pages == NULL)
+ GOTO(out, rc = -ENOMEM);
+
+ for (i = 0; i < nrpages; i++) {
+ pages[i] = cfs_alloc_page(CFS_ALLOC_STD);
+ if (pages[i] == NULL)
+ GOTO(out, rc = -ENOMEM);
+ }
+
+again:
+ LASSERT(cld_is_recover(cld));
+ LASSERT(cfs_mutex_is_locked(&cld->cld_lock));
+ req = ptlrpc_request_alloc(class_exp2cliimp(cld->cld_mgcexp),
+ &RQF_MGS_CONFIG_READ);
+ if (req == NULL)
+ GOTO(out, rc = -ENOMEM);
+
+ rc = ptlrpc_request_pack(req, LUSTRE_MGS_VERSION, MGS_CONFIG_READ);
+ if (rc)
+ GOTO(out, rc);
+
+ /* pack request */
+ body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
+ LASSERT(body != NULL);
+ LASSERT(sizeof(body->mcb_name) > strlen(cld->cld_logname));
+ strncpy(body->mcb_name, cld->cld_logname, sizeof(body->mcb_name));
+ body->mcb_offset = cfg->cfg_last_idx + 1;
+ body->mcb_type = cld->cld_type;
+ body->mcb_bits = CFS_PAGE_SHIFT;
+ body->mcb_units = nrpages;
+
+ /* allocate bulk transfer descriptor */
+ desc = ptlrpc_prep_bulk_imp(req, nrpages, BULK_PUT_SINK,
+ MGS_BULK_PORTAL);
+ if (desc == NULL)
+ GOTO(out, rc = -ENOMEM);
+
+ for (i = 0; i < nrpages; i++)
+ ptlrpc_prep_bulk_page(desc, pages[i], 0, CFS_PAGE_SIZE);
+
+ ptlrpc_request_set_replen(req);
+ rc = ptlrpc_queue_wait(req);
+ if (rc)
+ GOTO(out, rc);
+
+ res = req_capsule_server_get(&req->rq_pill, &RMF_MGS_CONFIG_RES);
+ if (res->mcr_size < res->mcr_offset)
+ GOTO(out, rc = -EINVAL);
+
+ /* always update the index even though it might have errors with
+ * handling the recover logs */
+ cfg->cfg_last_idx = res->mcr_offset;
+ eof = res->mcr_offset == res->mcr_size;
+
+ CDEBUG(D_INFO, "Latest version "LPD64", more %d.\n",
+ res->mcr_offset, eof == false);
+
+ ealen = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, 0);
+ if (ealen < 0)
+ GOTO(out, rc = ealen);
+
+ if (ealen > nrpages << CFS_PAGE_SHIFT)
+ GOTO(out, rc = -EINVAL);
+
+ if (ealen == 0) { /* no logs transferred */
+ if (!eof)
+ rc = -EINVAL;
+ GOTO(out, rc);
+ }
+
+ mne_swab = !!ptlrpc_rep_need_swab(req);
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 6, 50, 0)
+ /* This import flag means the server did an extra swab of IR MNE
+ * records (fixed in LU-1252), reverse it here if needed. LU-1644 */
+ if (unlikely(req->rq_import->imp_need_mne_swab))
+ mne_swab = !mne_swab;
+#else
+#warning "LU-1644: Remove old OBD_CONNECT_MNE_SWAB fixup and exp_need_mne_swab"
+#endif
+
+ for (i = 0; i < nrpages && ealen > 0; i++) {
+ int rc2;
+ void *ptr;
+
+ ptr = cfs_kmap(pages[i]);
+ rc2 = mgc_apply_recover_logs(obd, cld, res->mcr_offset, ptr,
+ min_t(int, ealen, CFS_PAGE_SIZE),
+ mne_swab);
+ cfs_kunmap(pages[i]);
+ if (rc2 < 0) {
+ CWARN("Process recover log %s error %d\n",
+ cld->cld_logname, rc2);
+ break;
+ }
+
+ ealen -= CFS_PAGE_SIZE;
+ }
+
+out:
+ if (req)
+ ptlrpc_req_finished(req);
+
+ if (rc == 0 && !eof)
+ goto again;
+
+ if (pages) {
+ for (i = 0; i < nrpages; i++) {
+ if (pages[i] == NULL)
+ break;
+ cfs_free_page(pages[i]);
+ }
+ OBD_FREE(pages, sizeof(*pages) * nrpages);
+ }
+ return rc;
+}
+
+/* identical to mgs_log_is_empty */
+static int mgc_llog_is_empty(struct obd_device *obd, struct llog_ctxt *ctxt,
+ char *name)
+{
+ struct lvfs_run_ctxt saved;
+ struct llog_handle *llh;
+ int rc = 0;
+
+ push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+ rc = llog_create(NULL, ctxt, &llh, NULL, name);
+ if (rc == 0) {
+ llog_init_handle(NULL, llh, LLOG_F_IS_PLAIN, NULL);
+ rc = llog_get_size(llh);
+ llog_close(NULL, llh);
+ }
+ pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+ /* header is record 1 */
+ return (rc <= 1);
+}
+
+static int mgc_copy_handler(const struct lu_env *env, struct llog_handle *llh,
+ struct llog_rec_hdr *rec, void *data)