Viewing: update_recovery.c
// SPDX-License-Identifier: GPL-2.0
/*
* Copyright (c) 2015, 2017, Intel Corporation.
*/
/*
* This file implement the methods to handle the update recovery.
*
* During DNE recovery, the recovery thread will redo the operation according
* to the transaction no, and these replay are either from client replay req
* or update replay records(for distribute transaction) in the update log.
* For distribute transaction replay, the replay thread will call
* distribute_txn_replay_handle() to handle the updates.
*
* After the Master MDT restarts, it will retrieve the update records from all
* of MDTs, for each distributed operation, it will check updates on all MDTs,
* if some updates records are missing on some MDTs, the replay thread will redo
* updates on these MDTs.
*
* Author: Di Wang <di.wang@intel.com>
*/
#define DEBUG_SUBSYSTEM S_CLASS
#include <lu_target.h>
#include <lustre_obdo.h>
#include <lustre_update.h>
#include <lustre_swab.h>
#include <md_object.h>
#include <obd.h>
#include <obd_class.h>
#include "tgt_internal.h"
/**
* dtrq_lookup() - Lookup distribute_txn_replay req
* @tdtd: distribute_txn_data, which holds the replay list.
* @batchid: batchid used by lookup.
*
* Lookup distribute_txn_replay in the replay list by batchid.
* It is assumed the list has been locked before calling this function.
*
* Return:
* * pointer of the replay if succeeds.
* * %NULL if can not find it.
*/
static struct distribute_txn_replay_req *
dtrq_lookup(struct target_distribute_txn_data *tdtd, __u64 batchid)
{
struct distribute_txn_replay_req *tmp;
struct distribute_txn_replay_req *dtrq = NULL;
list_for_each_entry(tmp, &tdtd->tdtd_replay_list, dtrq_list) {
if (tmp->dtrq_batchid == batchid) {
dtrq = tmp;
break;
}
}
return dtrq;
}
/**
* dtrq_insert() - insert distribute txn replay req
* @tdtd: target distribute txn data where replay list is
* @new: distribute txn replay to be inserted
*
* Insert distribute txn replay to the replay list, and it assumes the
* list has been looked. Note: the replay list is a sorted list, which
* is sorted by master transno. It is assumed the replay list has been
* locked before calling this function.
*
* Return:
* * %0 if insertion succeeds
* * %EEXIST if the dtrq already exists
*/
static int dtrq_insert(struct target_distribute_txn_data *tdtd,
struct distribute_txn_replay_req *new)
{
struct distribute_txn_replay_req *iter;
/* Check if the dtrq has been added to the list */
iter = dtrq_lookup(tdtd, new->dtrq_batchid);
if (iter != NULL)
return -EEXIST;
list_for_each_entry_reverse(iter, &tdtd->tdtd_replay_list, dtrq_list) {
if (iter->dtrq_master_transno > new->dtrq_master_transno)
continue;
/* If there are mulitple replay req with same transno, then
* sort them with batchid */
if (iter->dtrq_master_transno == new->dtrq_master_transno &&
iter->dtrq_batchid > new->dtrq_batchid)
continue;
list_add(&new->dtrq_list, &iter->dtrq_list);
break;
}
if (list_empty(&new->dtrq_list))
list_add(&new->dtrq_list, &tdtd->tdtd_replay_list);
return 0;
}
/**
* dtrq_create() - create distribute txn replay req
* @tdtd: target distribute txn data where replay list is.
* @lur: update records from the update log.
*
* Allocate distribute txn replay req according to the update records.
*
* Return:
* * pointer of distribute txn replay req if the creation succeeds.
* * %NULL if the creation fails.
*/
static struct distribute_txn_replay_req *
dtrq_create(struct target_distribute_txn_data *tdtd,
struct llog_update_record *lur)
{
struct distribute_txn_replay_req *new;
OBD_ALLOC_PTR(new);
if (new == NULL)
RETURN(ERR_PTR(-ENOMEM));
new->dtrq_lur_size = llog_update_record_size(lur);
OBD_ALLOC_LARGE(new->dtrq_lur, new->dtrq_lur_size);
if (new->dtrq_lur == NULL) {
OBD_FREE_PTR(new);
RETURN(ERR_PTR(-ENOMEM));
}
memcpy(new->dtrq_lur, lur, new->dtrq_lur_size);
/* If the transno in the update record is 0, it means the
* update are from master MDT, and it will use the master
* last committed transno as its master transno. Later, if
* the update records are gotten from slave MDTs, then these
* transno will be replaced.
* See insert_update_records_to_replay_list(). */
if (lur->lur_update_rec.ur_master_transno == 0) {
new->dtrq_lur->lur_update_rec.ur_master_transno =
tdtd->tdtd_lut->lut_obd->obd_last_committed;
new->dtrq_master_transno =
tdtd->tdtd_lut->lut_obd->obd_last_committed;
} else {
new->dtrq_master_transno =
lur->lur_update_rec.ur_master_transno;
}
new->dtrq_batchid = lur->lur_update_rec.ur_batchid;
spin_lock_init(&new->dtrq_sub_list_lock);
INIT_LIST_HEAD(&new->dtrq_sub_list);
INIT_LIST_HEAD(&new->dtrq_list);
RETURN(new);
}
/**
* dtrq_sub_lookup() - Lookup distribute sub replay
* @dtrq: the distribute txn replay req to lookup
* @mdt_index: the mdt_index as the key of lookup
*
* Lookup distribute sub replay in the sub list of distribute_txn_replay by
* mdt_index.
*
* Return:
* * the pointer of sub replay if it can be found.
* * %NULL if it can not find.
*/
struct distribute_txn_replay_req_sub *
dtrq_sub_lookup(struct distribute_txn_replay_req *dtrq, __u32 mdt_index)
{
struct distribute_txn_replay_req_sub *dtrqs = NULL;
struct distribute_txn_replay_req_sub *tmp;
list_for_each_entry(tmp, &dtrq->dtrq_sub_list, dtrqs_list) {
if (tmp->dtrqs_mdt_index == mdt_index) {
dtrqs = tmp;
break;
}
}
return dtrqs;
}
/**
* dtrq_sub_add_cookie() - Try to add cookie to sub distribute txn request
* @dtrqs: sub replay req where cookies to be added.
* @cookie: cookie to be added.
*
* Check if the update log cookie has been added to the request, if not,
* add it to the dtrqs_cookie_list.
*
* Return:
* * %0 if the cookie is adding succeeds.
* * %negative errno if adding fails.
*/
static int dtrq_sub_add_cookie(struct distribute_txn_replay_req_sub *dtrqs,
struct llog_cookie *cookie)
{
struct sub_thandle_cookie *new;
OBD_ALLOC_PTR(new);
if (new == NULL)
return -ENOMEM;
INIT_LIST_HEAD(&new->stc_list);
new->stc_cookie = *cookie;
/* Note: only single thread will access one sub_request each time,
* so no need lock here */
list_add(&new->stc_list, &dtrqs->dtrqs_cookie_list);
return 0;
}
/**
* dtrq_sub_create_and_insert() - Insert distribute txn sub req replay
* @dtrq: d to be added
* @cookie: the cookie of the update record
* @mdt_index: the mdt_index of the update record
*
* Allocate sub replay req and insert distribute txn replay list.
*
* Return:
* * %0 if the adding succeeds.
* * %negative errno if the adding fails.
*/
static int
dtrq_sub_create_and_insert(struct distribute_txn_replay_req *dtrq,
struct llog_cookie *cookie,
__u32 mdt_index)
{
struct distribute_txn_replay_req_sub *dtrqs = NULL;
struct distribute_txn_replay_req_sub *new;
int rc;
ENTRY;
spin_lock(&dtrq->dtrq_sub_list_lock);
dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
spin_unlock(&dtrq->dtrq_sub_list_lock);
if (dtrqs != NULL) {
rc = dtrq_sub_add_cookie(dtrqs, cookie);
RETURN(0);
}
OBD_ALLOC_PTR(new);
if (new == NULL)
RETURN(-ENOMEM);
INIT_LIST_HEAD(&new->dtrqs_list);
INIT_LIST_HEAD(&new->dtrqs_cookie_list);
new->dtrqs_mdt_index = mdt_index;
spin_lock(&dtrq->dtrq_sub_list_lock);
dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
if (dtrqs == NULL) {
list_add(&new->dtrqs_list, &dtrq->dtrq_sub_list);
dtrqs = new;
} else {
OBD_FREE_PTR(new);
}
spin_unlock(&dtrq->dtrq_sub_list_lock);
rc = dtrq_sub_add_cookie(dtrqs, cookie);
RETURN(rc);
}
/**
* dtrq_append_updates() - append updates to the current replay updates
* @dtrq: the update replay request where the new update records will be added.
* @record: the new update record.
*
* Append more updates to the existent replay update. And this is only
* used when combining mulitple updates into one large updates during
* replay.
*
* Return:
* * %0 if appending succeeds.
* * %negative errno if appending fails.
*/
static int dtrq_append_updates(struct distribute_txn_replay_req *dtrq,
struct update_records *record)
{
struct llog_update_record *new_lur;
size_t lur_size = dtrq->dtrq_lur_size;
void *ptr;
ENTRY;
/* Because several threads might retrieve the same records from
* different targets, and we only need one copy of records. So
* we will check if the records is in the next one, if not, just
* skip it */
spin_lock(&dtrq->dtrq_sub_list_lock);
if (dtrq->dtrq_lur->lur_update_rec.ur_index + 1 != record->ur_index) {
spin_unlock(&dtrq->dtrq_sub_list_lock);
RETURN(0);
}
dtrq->dtrq_lur->lur_update_rec.ur_index++;
spin_unlock(&dtrq->dtrq_sub_list_lock);
lur_size += update_records_size(record);
OBD_ALLOC_LARGE(new_lur, lur_size);
if (new_lur == NULL) {
spin_lock(&dtrq->dtrq_sub_list_lock);
dtrq->dtrq_lur->lur_update_rec.ur_index--;
spin_unlock(&dtrq->dtrq_sub_list_lock);
RETURN(-ENOMEM);
}
/* Copy the old and new records to the new allocated buffer */
memcpy(new_lur, dtrq->dtrq_lur, dtrq->dtrq_lur_size);
ptr = (char *)&new_lur->lur_update_rec +
update_records_size(&new_lur->lur_update_rec);
memcpy(ptr, &record->ur_ops,
update_records_size(record) -
offsetof(struct update_records, ur_ops));
new_lur->lur_update_rec.ur_update_count += record->ur_update_count;
new_lur->lur_update_rec.ur_param_count += record->ur_param_count;
new_lur->lur_hdr.lrh_len = llog_update_record_size(new_lur);
/* Replace the records */
OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
dtrq->dtrq_lur = new_lur;
dtrq->dtrq_lur_size = lur_size;
dtrq->dtrq_lur->lur_update_rec.ur_flags = record->ur_flags;
update_records_dump(&new_lur->lur_update_rec, D_INFO, true);
RETURN(0);
}
/**
* insert_update_records_to_replay_list() - Insert update records to the replay
* list.
* @tdtd: distribute txn replay data where the replay list is.
* @lur: the update record
* @cookie: cookie of the record
* @mdt_index: mdt index of the record
*
* Allocate distribute txn replay req and insert it into the replay
* list, then insert the update records into the replay req.
*
* Return:
* * %0 if the adding succeeds.
* * %negative errno if the adding fails.
*/
int
insert_update_records_to_replay_list(struct target_distribute_txn_data *tdtd,
struct llog_update_record *lur,
struct llog_cookie *cookie,
__u32 mdt_index)
{
struct distribute_txn_replay_req *dtrq;
struct update_records *record = &lur->lur_update_rec;
bool replace_record = false;
int rc = 0;
ENTRY;
CDEBUG(D_HA, "%s: insert record batchid = %llu transno = %llu mdt_index %u\n",
tdtd->tdtd_lut->lut_obd->obd_name,
record->ur_batchid, record->ur_master_transno, mdt_index);
/* Update batchid if necessary */
spin_lock(&tdtd->tdtd_batchid_lock);
if (record->ur_batchid >= tdtd->tdtd_batchid) {
CDEBUG(D_HA, "%s update batchid from %llu to %llu\n",
tdtd->tdtd_lut->lut_obd->obd_name,
tdtd->tdtd_batchid, record->ur_batchid);
tdtd->tdtd_batchid = record->ur_batchid + 1;
}
spin_unlock(&tdtd->tdtd_batchid_lock);
again:
spin_lock(&tdtd->tdtd_replay_list_lock);
/* First try to build the replay update request with the records */
dtrq = dtrq_lookup(tdtd, record->ur_batchid);
if (dtrq == NULL) {
spin_unlock(&tdtd->tdtd_replay_list_lock);
dtrq = dtrq_create(tdtd, lur);
if (IS_ERR(dtrq))
RETURN(PTR_ERR(dtrq));
spin_lock(&tdtd->tdtd_replay_list_lock);
rc = dtrq_insert(tdtd, dtrq);
if (rc < 0) {
spin_unlock(&tdtd->tdtd_replay_list_lock);
dtrq_destroy(dtrq);
if (rc == -EEXIST)
goto again;
return rc;
}
} else {
/* If the master transno in update header is not
* matched with the one in the record, then it means
* the dtrq is originally created by master record,
* so we need update master transno and reposition
* the dtrq(by master transno) in the list and also
* replace update record */
if (record->ur_master_transno != 0 &&
dtrq->dtrq_master_transno != record->ur_master_transno &&
dtrq->dtrq_lur != NULL) {
list_del_init(&dtrq->dtrq_list);
dtrq->dtrq_lur->lur_update_rec.ur_master_transno =
record->ur_master_transno;
dtrq->dtrq_master_transno = record->ur_master_transno;
replace_record = true;
/* try to insert again */
rc = dtrq_insert(tdtd, dtrq);
if (rc < 0) {
spin_unlock(&tdtd->tdtd_replay_list_lock);
dtrq_destroy(dtrq);
return rc;
}
}
}
spin_unlock(&tdtd->tdtd_replay_list_lock);
/* Because there should be only thread access the update record, so
* we do not need lock here */
if (replace_record) {
/* Replace the update record and master transno */
OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
dtrq->dtrq_lur = NULL;
dtrq->dtrq_lur_size = llog_update_record_size(lur);
OBD_ALLOC_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
if (dtrq->dtrq_lur == NULL)
return -ENOMEM;
memcpy(dtrq->dtrq_lur, lur, dtrq->dtrq_lur_size);
}
/* This is a partial update records, let's try to append
* the record to the current replay request */
if (record->ur_flags & UPDATE_RECORD_CONTINUE)
rc = dtrq_append_updates(dtrq, record);
/* Then create and add sub update request */
rc = dtrq_sub_create_and_insert(dtrq, cookie, mdt_index);
RETURN(rc);
}
EXPORT_SYMBOL(insert_update_records_to_replay_list);
/**
* dtrq_list_dump() - Dump updates of distribute txns.
* @tdtd: distribute txn data where all of distribute txn are listed.
* @mask: debug mask
*
* Output all of recovery updates in the distribute txn list to the
* debug log.
*/
void dtrq_list_dump(struct target_distribute_txn_data *tdtd, unsigned int mask)
{
struct distribute_txn_replay_req *dtrq;
spin_lock(&tdtd->tdtd_replay_list_lock);
list_for_each_entry(dtrq, &tdtd->tdtd_replay_list, dtrq_list)
update_records_dump(&dtrq->dtrq_lur->lur_update_rec, mask,
false);
spin_unlock(&tdtd->tdtd_replay_list_lock);
}
EXPORT_SYMBOL(dtrq_list_dump);
/**
* dtrq_destroy() - Destroy distribute txn replay req
* @dtrq: distribute txn replqy req to be destroyed.
*
* Destroy distribute txn replay req and all of subs.
*/
void dtrq_destroy(struct distribute_txn_replay_req *dtrq)
{
struct distribute_txn_replay_req_sub *dtrqs;
struct distribute_txn_replay_req_sub *tmp;
LASSERT(list_empty(&dtrq->dtrq_list));
CDEBUG(D_HA, "destroy x%llu t%llu\n", dtrq->dtrq_xid,
dtrq->dtrq_master_transno);
spin_lock(&dtrq->dtrq_sub_list_lock);
list_for_each_entry_safe(dtrqs, tmp, &dtrq->dtrq_sub_list, dtrqs_list) {
struct sub_thandle_cookie *stc;
struct sub_thandle_cookie *tmp;
list_del(&dtrqs->dtrqs_list);
list_for_each_entry_safe(stc, tmp, &dtrqs->dtrqs_cookie_list,
stc_list) {
list_del(&stc->stc_list);
OBD_FREE_PTR(stc);
}
OBD_FREE_PTR(dtrqs);
}
spin_unlock(&dtrq->dtrq_sub_list_lock);
if (dtrq->dtrq_lur != NULL)
OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
OBD_FREE_PTR(dtrq);
}
EXPORT_SYMBOL(dtrq_destroy);
/**
* dtrq_list_destroy() - Destroy all of replay req.
* @tdtd: target distribute txn data where the replay list is.
*
* Destroy all of replay req in the replay list.
*/
void dtrq_list_destroy(struct target_distribute_txn_data *tdtd)
{
struct distribute_txn_replay_req *dtrq;
struct distribute_txn_replay_req *tmp;
spin_lock(&tdtd->tdtd_replay_list_lock);
list_for_each_entry_safe(dtrq, tmp, &tdtd->tdtd_replay_list,
dtrq_list) {
list_del_init(&dtrq->dtrq_list);
dtrq_destroy(dtrq);
}
list_for_each_entry_safe(dtrq, tmp, &tdtd->tdtd_replay_finish_list,
dtrq_list) {
list_del_init(&dtrq->dtrq_list);
dtrq_destroy(dtrq);
}
spin_unlock(&tdtd->tdtd_replay_list_lock);
}
EXPORT_SYMBOL(dtrq_list_destroy);
/**
* distribute_txn_get_next_req() - Get next req in the replay list
* @tdtd: distribute txn data where the replay list is
*
* Get next req needs to be replayed, since it is a sorted list
* (by master MDT transno)
*
* Return the pointer of update recovery header
*/
struct distribute_txn_replay_req *
distribute_txn_get_next_req(struct target_distribute_txn_data *tdtd)
{
struct distribute_txn_replay_req *dtrq = NULL;
spin_lock(&tdtd->tdtd_replay_list_lock);
if (!list_empty(&tdtd->tdtd_replay_list)) {
dtrq = list_first_entry(&tdtd->tdtd_replay_list,
struct distribute_txn_replay_req,
dtrq_list);
list_del_init(&dtrq->dtrq_list);
}
spin_unlock(&tdtd->tdtd_replay_list_lock);
return dtrq;
}
EXPORT_SYMBOL(distribute_txn_get_next_req);
/**
* distribute_txn_get_next_transno() - Get next transno in the replay list
* @tdtd: distribute txn data where the replay list is
*
* Get next transno in the replay list, because this is the sorted
* list, so it will return the transno of next req in the list.
*
* Returns the transno of next update in the list
*/
__u64 distribute_txn_get_next_transno(struct target_distribute_txn_data *tdtd)
{
struct distribute_txn_replay_req *dtrq = NULL;
__u64 transno = 0;
spin_lock(&tdtd->tdtd_replay_list_lock);
if (!list_empty(&tdtd->tdtd_replay_list)) {
dtrq = list_first_entry(&tdtd->tdtd_replay_list,
struct distribute_txn_replay_req,
dtrq_list);
transno = dtrq->dtrq_master_transno;
}
spin_unlock(&tdtd->tdtd_replay_list_lock);
CDEBUG(D_HA, "%s: Next update transno %llu\n",
tdtd->tdtd_lut->lut_obd->obd_name, transno);
return transno;
}
EXPORT_SYMBOL(distribute_txn_get_next_transno);
struct distribute_txn_replay_req *
distribute_txn_lookup_finish_list(struct target_distribute_txn_data *tdtd,
__u64 transno)
{
struct distribute_txn_replay_req *dtrq = NULL;
struct distribute_txn_replay_req *iter;
spin_lock(&tdtd->tdtd_replay_list_lock);
list_for_each_entry(iter, &tdtd->tdtd_replay_finish_list, dtrq_list) {
if (iter->dtrq_master_transno == transno) {
dtrq = iter;
break;
}
}
spin_unlock(&tdtd->tdtd_replay_list_lock);
return dtrq;
}
bool is_req_replayed_by_update(struct ptlrpc_request *req)
{
struct lu_target *tgt = class_exp2tgt(req->rq_export);
struct distribute_txn_replay_req *dtrq;
if (tgt->lut_tdtd == NULL)
return false;
dtrq = distribute_txn_lookup_finish_list(tgt->lut_tdtd,
lustre_msg_get_transno(req->rq_reqmsg));
if (dtrq == NULL)
return false;
return true;
}
EXPORT_SYMBOL(is_req_replayed_by_update);
/**
* update_is_committed() - Check if the update of one object is committed
* @env: execution environment
* @dtrq: replay request
* @dt_obj: object for the update
* @top_th: top thandle
* @st: sub thandle which the update belongs to
*
* Check whether the update for the object is committed by checking whether
* the correspondent sub exists in the replay req. If it is committed, mark
* the committed flag in correspondent the sub thandle.
*
* Return:
* * %1 if the update is not committed.
* * %0 if the update is committed.
* * %negative errno if some other failures happen.
*/
static int update_is_committed(const struct lu_env *env,
struct distribute_txn_replay_req *dtrq,
struct dt_object *dt_obj,
struct top_thandle *top_th,
struct sub_thandle *st)
{
struct seq_server_site *seq_site;
const struct lu_fid *fid = lu_object_fid(&dt_obj->do_lu);
struct distribute_txn_replay_req_sub *dtrqs;
__u32 mdt_index;
ENTRY;
if (st->st_sub_th != NULL)
RETURN(1);
if (st->st_committed)
RETURN(0);
seq_site = lu_site2seq(dt_obj->do_lu.lo_dev->ld_site);
if (fid_is_update_log(fid) || fid_is_update_log_dir(fid)) {
mdt_index = fid_oid(fid);
} else if (!fid_seq_in_fldb(fid_seq(fid))) {
mdt_index = seq_site->ss_node_id;
} else {
struct lu_server_fld *fld;
struct lu_seq_range range = {0};
int rc;
fld = seq_site->ss_server_fld;
fld_range_set_type(&range, LU_SEQ_RANGE_MDT);
LASSERT(fld->lsf_seq_lookup != NULL);
rc = fld->lsf_seq_lookup(env, fld, fid_seq(fid),
&range);
if (rc < 0)
RETURN(rc);
mdt_index = range.lsr_index;
}
dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
if (dtrqs != NULL || top_th->tt_multiple_thandle->tmt_committed) {
st->st_committed = 1;
if (dtrqs != NULL) {
struct sub_thandle_cookie *stc;
struct sub_thandle_cookie *tmp;
list_for_each_entry_safe(stc, tmp,
&dtrqs->dtrqs_cookie_list,
stc_list)
list_move(&stc->stc_list, &st->st_cookie_list);
}
RETURN(0);
}
CDEBUG(D_HA, "Update of "DFID "on MDT%u is not committed\n", PFID(fid),
mdt_index);
RETURN(1);
}
/**
* update_recovery_create() - Implementation of different update methods for
* update recovery.
* @env: execution environment
* @dt_obj: target object to be created
* @op: update operation to be replayed
* @params: common update parameters which holds all parameters of the operation
* @ta: pointer to the thandle_exec_args struct
* @th: transaction handle
*
* These following functions update_recovery_$(update_name) implement
* different updates recovery methods. They will extract the parameters
* from the common parameters area and call correspondent dt API to redo
* the update.
*
* Return:
* * %0 if it succeeds.
* * %negative errno if it fails.
*/
static int update_recovery_create(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
struct llog_update_record *lur = uti->uti_dtrq->dtrq_lur;
struct lu_attr *attr = &uti->uti_attr;
struct obdo *wobdo;
struct obdo *lobdo = &uti->uti_obdo;
struct dt_object_format dof;
__u16 size;
unsigned int param_count;
int rc;
ENTRY;
if (dt_object_exists(dt_obj))
RETURN(-EEXIST);
param_count = lur->lur_update_rec.ur_param_count;
wobdo = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (wobdo == NULL)
RETURN(-EIO);
if (size != sizeof(*wobdo))
RETURN(-EIO);
if (LLOG_REC_HDR_NEEDS_SWABBING(&lur->lur_hdr))
lustre_swab_obdo(wobdo);
lustre_get_wire_obdo(NULL, lobdo, wobdo);
la_from_obdo(attr, lobdo, lobdo->o_valid);
dof.dof_type = dt_mode_to_dft(attr->la_mode);
rc = out_tx_create(env, dt_obj, attr, NULL, &dof,
ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_destroy(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
int rc;
ENTRY;
rc = out_tx_destroy(env, dt_obj, ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_ref_add(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
int rc;
ENTRY;
rc = out_tx_ref_add(env, dt_obj, ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_ref_del(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
int rc;
ENTRY;
rc = out_tx_ref_del(env, dt_obj, ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_attr_set(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
struct llog_update_record *lur = uti->uti_dtrq->dtrq_lur;
struct obdo *wobdo;
struct obdo *lobdo = &uti->uti_obdo;
struct lu_attr *attr = &uti->uti_attr;
__u16 size;
unsigned int param_count;
int rc;
ENTRY;
param_count = lur->lur_update_rec.ur_param_count;
wobdo = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (wobdo == NULL)
RETURN(-EIO);
if (size != sizeof(*wobdo))
RETURN(-EIO);
if (LLOG_REC_HDR_NEEDS_SWABBING(&lur->lur_hdr))
lustre_swab_obdo(wobdo);
lustre_get_wire_obdo(NULL, lobdo, wobdo);
la_from_obdo(attr, lobdo, lobdo->o_valid);
rc = out_tx_attr_set(env, dt_obj, attr, ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_xattr_set(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
char *buf;
char *name;
int fl;
__u16 size;
__u32 param_count;
int rc;
ENTRY;
param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
name = update_params_get_param_buf(params,
op->uop_params_off[0],
param_count, &size);
if (name == NULL)
RETURN(-EIO);
buf = update_params_get_param_buf(params,
op->uop_params_off[1],
param_count, &size);
if (buf == NULL)
RETURN(-EIO);
uti->uti_buf.lb_buf = buf;
uti->uti_buf.lb_len = (size_t)size;
buf = update_params_get_param_buf(params, op->uop_params_off[2],
param_count, &size);
if (buf == NULL)
RETURN(-EIO);
if (size != sizeof(fl))
RETURN(-EIO);
fl = le32_to_cpu(*(int *)buf);
rc = out_tx_xattr_set(env, dt_obj, &uti->uti_buf, name, fl, ta, th,
NULL, 0);
RETURN(rc);
}
static int update_recovery_index_insert(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
struct lu_fid *fid;
char *name;
__u32 param_count;
__u32 *ptype;
__u32 type;
__u16 size;
int rc;
ENTRY;
param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
name = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (name == NULL)
RETURN(-EIO);
fid = update_params_get_param_buf(params, op->uop_params_off[1],
param_count, &size);
if (fid == NULL)
RETURN(-EIO);
if (size != sizeof(*fid))
RETURN(-EIO);
fid_le_to_cpu(fid, fid);
ptype = update_params_get_param_buf(params, op->uop_params_off[2],
param_count, &size);
if (ptype == NULL)
RETURN(-EIO);
if (size != sizeof(*ptype))
RETURN(-EIO);
type = le32_to_cpu(*ptype);
if (!dt_try_as_dir(env, dt_obj, false))
RETURN(-ENOTDIR);
uti->uti_rec.rec_fid = fid;
uti->uti_rec.rec_type = type;
rc = out_tx_index_insert(env, dt_obj,
(const struct dt_rec *)&uti->uti_rec,
(const struct dt_key *)name, ta, th,
NULL, 0);
RETURN(rc);
}
static int update_recovery_index_delete(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
__u32 param_count;
char *name;
__u16 size;
int rc;
ENTRY;
param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
name = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (name == NULL)
RETURN(-EIO);
if (!dt_try_as_dir(env, dt_obj, true))
RETURN(-ENOTDIR);
rc = out_tx_index_delete(env, dt_obj,
(const struct dt_key *)name, ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_write(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
char *buf;
__u32 param_count;
__u64 pos;
__u16 size;
int rc;
ENTRY;
param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
buf = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (buf == NULL)
RETURN(-EIO);
uti->uti_buf.lb_buf = buf;
uti->uti_buf.lb_len = size;
buf = update_params_get_param_buf(params, op->uop_params_off[1],
param_count, &size);
if (buf == NULL)
RETURN(-EIO);
pos = le64_to_cpu(*(__u64 *)buf);
rc = out_tx_write(env, dt_obj, &uti->uti_buf, pos,
ta, th, NULL, 0);
RETURN(rc);
}
static int update_recovery_xattr_del(const struct lu_env *env,
struct dt_object *dt_obj,
const struct update_op *op,
const struct update_params *params,
struct thandle_exec_args *ta,
struct thandle *th)
{
struct update_thread_info *uti = update_env_info(env);
__u32 param_count;
char *name;
__u16 size;
int rc;
ENTRY;
param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
name = update_params_get_param_buf(params, op->uop_params_off[0],
param_count, &size);
if (name == NULL)
RETURN(-EIO);
rc = out_tx_xattr_del(env, dt_obj, name, ta, th, NULL, 0);
RETURN(rc);
}
/**
* update_recovery_update_ses() - Update session information
* @env: execution environment.
* @tdtd: distribute data structure of the recovering tgt.
* @th: thandle of this update replay.
* @master_th: master sub thandle.
* @dtrq: pointer to distribute_txn_replay_req (replay request info)
* @ta_arg: the tx arg structure to hold the update for updating reply data.
*
* Update session information so tgt_txn_stop_cb()->tgt_last_rcvd_update()
* can be called correctly during update replay.
*/
static void update_recovery_update_ses(struct lu_env *env,
struct target_distribute_txn_data *tdtd,
struct thandle *th,
struct thandle *master_th,
struct distribute_txn_replay_req *dtrq,
struct tx_arg *ta_arg)
{
struct tgt_session_info *tsi;
struct lu_target *lut = tdtd->tdtd_lut;
struct lsd_reply_header *lrh = &lut->lut_reply_header;
struct lsd_reply_data *lrd;
struct top_thandle *top_th;
struct obd_export *export;
struct cfs_hash *hash;
size_t size;
tsi = tgt_ses_info(env);
if (tsi->tsi_exp != NULL)
return;
size = ta_arg->u.write.buf.lb_len;
lrd = ta_arg->u.write.buf.lb_buf;
if (size != lrh->lrh_reply_size || lrd == NULL)
return;
lrd->lrd_transno = le64_to_cpu(lrd->lrd_transno);
lrd->lrd_xid = le64_to_cpu(lrd->lrd_xid);
lrd->lrd_data = le64_to_cpu(lrd->lrd_data);
lrd->lrd_result = le32_to_cpu(lrd->lrd_result);
lrd->lrd_client_gen = le32_to_cpu(lrd->lrd_client_gen);
CDEBUG(D_HA, "xid=%llu transno=%llu\n", lrd->lrd_xid, lrd->lrd_transno);
if (lrd->lrd_transno != tgt_th_info(env)->tti_transno)
return;
hash = cfs_hash_getref(lut->lut_obd->obd_gen_hash);
if (hash == NULL)
return;
export = cfs_hash_lookup(hash, &lrd->lrd_client_gen);
if (export == NULL) {
cfs_hash_putref(hash);
return;
}
tsi->tsi_exp = export;
tsi->tsi_xid = lrd->lrd_xid;
tsi->tsi_opdata = lrd->lrd_data;
tsi->tsi_result = lrd->lrd_result;
tsi->tsi_client_gen = lrd->lrd_client_gen;
dtrq->dtrq_xid = lrd->lrd_xid;
top_th = container_of(th, struct top_thandle, tt_super);
top_th->tt_master_sub_thandle = master_th;
cfs_hash_putref(hash);
}
/**
* update_recovery_exec() - Execute updates in the update replay records
* @env: execution environment
* @tdtd: distribute txn replay data which hold all of replay reqs and all
* replay parameters.
* @dtrq: distribute transaction replay req.
* @ta: thandle execute args.
*
* Declare distribute txn replay by update records and add the updates
* to the execution list. Note: it will check if the update has been
* committed, and only execute the updates if it is not committed to
* disk.
*
* Return:
* * %0 if declare succeeds.
* * %negative errno if declare fails.
*/
static int update_recovery_exec(const struct lu_env *env,
struct target_distribute_txn_data *tdtd,
struct distribute_txn_replay_req *dtrq,
struct thandle_exec_args *ta)
{
struct llog_update_record *lur = dtrq->dtrq_lur;
struct update_records *records = &lur->lur_update_rec;
struct update_ops *ops = &records->ur_ops;
struct update_params *params = update_records_get_params(records);
struct top_thandle *top_th = container_of(ta->ta_handle,
struct top_thandle,
tt_super);
struct top_multiple_thandle *tmt = top_th->tt_multiple_thandle;
struct update_op *op;
unsigned int i;
int rc = 0;
ENTRY;
/* These records have been swabbed in llog_cat_process() */
for (i = 0, op = &ops->uops_op[0]; i < records->ur_update_count;
i++, op = update_op_next_op(op)) {
struct lu_fid *fid = &op->uop_fid;
struct dt_object *dt_obj;
struct dt_object *sub_dt_obj;
struct dt_device *sub_dt;
struct sub_thandle *st;
if (op->uop_type == OUT_NOOP)
continue;
dt_obj = dt_locate(env, tdtd->tdtd_dt, fid);
if (IS_ERR(dt_obj)) {
rc = PTR_ERR(dt_obj);
if (rc == -EREMCHG)
LCONSOLE_WARN("%.16s: hit invalid OI mapping for "
DFID" during recovering, "
"that may because auto scrub is disabled on related MDT, and "
"will cause recovery failure. Please enable auto scrub and "
"retry the recovery.\n",
tdtd->tdtd_lut->lut_obd->obd_name,
PFID(fid));
break;
}
sub_dt_obj = dt_object_child(dt_obj);
/* Create sub thandle if not */
sub_dt = lu2dt_dev(sub_dt_obj->do_lu.lo_dev);
st = lookup_sub_thandle(tmt, sub_dt);
if (st == NULL) {
st = create_sub_thandle(tmt, sub_dt);
if (IS_ERR(st))
GOTO(next, rc = PTR_ERR(st));
}
/* check if updates on the OSD/OSP are committed */
rc = update_is_committed(env, dtrq, dt_obj, top_th, st);
if (rc == 0)
/* If this is committed, goto next */
goto next;
if (rc < 0)
GOTO(next, rc);
/* Create thandle for sub thandle if needed */
if (st->st_sub_th == NULL) {
rc = sub_thandle_trans_create(env, top_th, st);
if (rc != 0)
GOTO(next, rc);
}
CDEBUG(D_HA, "replay %uth update\n", i);
switch (op->uop_type) {
case OUT_CREATE:
rc = update_recovery_create(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_DESTROY:
rc = update_recovery_destroy(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_REF_ADD:
rc = update_recovery_ref_add(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_REF_DEL:
rc = update_recovery_ref_del(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_ATTR_SET:
rc = update_recovery_attr_set(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_XATTR_SET:
rc = update_recovery_xattr_set(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_INDEX_INSERT:
rc = update_recovery_index_insert(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_INDEX_DELETE:
rc = update_recovery_index_delete(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_WRITE:
rc = update_recovery_write(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
case OUT_XATTR_DEL:
rc = update_recovery_xattr_del(env, sub_dt_obj,
op, params, ta,
st->st_sub_th);
break;
default:
CERROR("Unknown update type %u\n", (__u32)op->uop_type);
rc = -EINVAL;
break;
}
next:
dt_object_put(env, dt_obj);
if (rc < 0)
break;
}
ta->ta_handle->th_result = rc;
RETURN(rc);
}
/**
* distribute_txn_replay_handle() - redo updates on MDT if needed.
* @env: execution environment
* @tdtd: target distribute txn data, which holds the replay list and all
* parameters needed by replay process.
* @dtrq: distribute txn replay req.
*
* During DNE recovery, the recovery thread (target_recovery_thread) will call
* this function to replay distribute txn updates on all MDTs. It only replay
* updates on the MDT where the update record is missing.
*
* If the update already exists on the MDT, then it does not need replay the
* updates on that MDT, and only mark the sub transaction has been committed
* there.
*
* Return:
* * %0 if replay succeeds.
* * %negative errno if replay failes.
*/
int distribute_txn_replay_handle(struct lu_env *env,
struct target_distribute_txn_data *tdtd,
struct distribute_txn_replay_req *dtrq)
{
struct update_records *records = &dtrq->dtrq_lur->lur_update_rec;
struct thandle_exec_args *ta;
struct lu_context session_env;
struct thandle *th = NULL;
struct top_thandle *top_th;
struct top_multiple_thandle *tmt;
struct thandle_update_records *tur = NULL;
int i;
int rc = 0;
ENTRY;
/* initialize session, it is needed for the handler of target */
rc = lu_context_init(&session_env, LCT_SERVER_SESSION | LCT_NOREF);
if (rc) {
CERROR("%s: failure to initialize session: rc = %d\n",
tdtd->tdtd_lut->lut_obd->obd_name, rc);
RETURN(rc);
}
lu_context_enter(&session_env);
env->le_ses = &session_env;
lu_env_refill(env);
update_records_dump(records, D_HA, true);
th = top_trans_create(env, NULL);
if (IS_ERR(th))
GOTO(exit_session, rc = PTR_ERR(th));
ta = &update_env_info(env)->uti_tea;
ta->ta_argno = 0;
update_env_info(env)->uti_dtrq = dtrq;
/* Create distribute transaction structure for this top thandle */
top_th = container_of(th, struct top_thandle, tt_super);
rc = top_trans_create_tmt(env, top_th);
if (rc < 0)
GOTO(stop_trans, rc);
th->th_dev = tdtd->tdtd_dt;
ta->ta_handle = th;
/* check if the distribute transaction has been committed */
tmt = top_th->tt_multiple_thandle;
tmt->tmt_master_sub_dt = tdtd->tdtd_lut->lut_bottom;
tmt->tmt_batchid = dtrq->dtrq_batchid;
tgt_th_info(env)->tti_transno = dtrq->dtrq_master_transno;
if (tmt->tmt_batchid <= tdtd->tdtd_committed_batchid)
tmt->tmt_committed = 1;
rc = update_recovery_exec(env, tdtd, dtrq, ta);
if (rc < 0)
GOTO(stop_trans, rc);
/* If no updates are needed to be replayed, then mark this records as
* committed, so commit thread distribute_txn_commit_thread() will
* delete the record */
if (ta->ta_argno == 0)
tmt->tmt_committed = 1;
tur = &update_env_info(env)->uti_tur;
tur->tur_update_records = dtrq->dtrq_lur;
tur->tur_update_records_buf_size = dtrq->dtrq_lur_size;
tur->tur_update_params = NULL;
tur->tur_update_param_count = 0;
tmt->tmt_update_records = tur;
distribute_txn_insert_by_batchid(tmt);
rc = top_trans_start(env, NULL, th);
if (rc < 0)
GOTO(stop_trans, rc);
for (i = 0; i < ta->ta_argno; i++) {
struct tx_arg *ta_arg;
struct dt_object *dt_obj;
struct dt_device *sub_dt;
struct sub_thandle *st;
ta_arg = ta->ta_args[i];
dt_obj = ta_arg->object;
LASSERT(tmt->tmt_committed == 0);
sub_dt = lu2dt_dev(dt_obj->do_lu.lo_dev);
st = lookup_sub_thandle(tmt, sub_dt);
LASSERT(st != NULL);
LASSERT(st->st_sub_th != NULL);
rc = ta->ta_args[i]->exec_fn(env, st->st_sub_th,
ta->ta_args[i]);
/* If the update is to update the reply data, then
* we need set the session information, so
* tgt_last_rcvd_update() can be called correctly */
if (rc == 0 && dt_obj == tdtd->tdtd_lut->lut_reply_data)
update_recovery_update_ses(env, tdtd, th,
st->st_sub_th, dtrq, ta_arg);
if (unlikely(rc < 0)) {
CWARN("%s: error during execution of #%u from %s:%d: rc = %d\n",
dt_obd_name(ta->ta_handle->th_dev), i,
ta->ta_args[i]->file, ta->ta_args[i]->line, rc);
while (--i > 0) {
if (ta->ta_args[i]->undo_fn != NULL) {
dt_obj = ta->ta_args[i]->object;
sub_dt =
lu2dt_dev(dt_obj->do_lu.lo_dev);
st = lookup_sub_thandle(tmt, sub_dt);
LASSERT(st != NULL);
LASSERT(st->st_sub_th != NULL);
ta->ta_args[i]->undo_fn(env,
st->st_sub_th,
ta->ta_args[i]);
} else {
CERROR("%s: undo for %s:%d: rc = %d\n",
dt_obd_name(ta->ta_handle->th_dev),
ta->ta_args[i]->file,
ta->ta_args[i]->line, -ENOTSUPP);
}
}
break;
}
CDEBUG(D_HA, "%s: executed %u/%u: rc = %d\n",
dt_obd_name(sub_dt), i, ta->ta_argno, rc);
}
stop_trans:
if (rc < 0)
th->th_result = rc;
rc = top_trans_stop(env, tdtd->tdtd_dt, th);
for (i = 0; i < ta->ta_argno; i++) {
if (ta->ta_args[i]->object != NULL) {
dt_object_put(env, ta->ta_args[i]->object);
ta->ta_args[i]->object = NULL;
}
}
if (tur != NULL)
tur->tur_update_records = NULL;
if (tgt_ses_info(env)->tsi_exp != NULL) {
class_export_put(tgt_ses_info(env)->tsi_exp);
tgt_ses_info(env)->tsi_exp = NULL;
}
exit_session:
lu_context_exit(&session_env);
lu_context_fini(&session_env);
RETURN(rc);
}
EXPORT_SYMBOL(distribute_txn_replay_handle);
