Viewing: lod_sub_object.c
// SPDX-License-Identifier: GPL-2.0
/*
* Copyright (c) 2015, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*
* LOD sub object methods
*
* This file implements sub-object methods for LOD.
*
* LOD is Logic volume layer in the MDS stack, which will handle striping
* and distribute the update to different OSP/OSD. After directing the updates
* to one specific OSD/OSP, it also needs to do some thing before calling
* OSD/OSP API, for example recording updates for cross-MDT operation, get
* the next level transaction etc.
*
* Author: Di Wang <di.wang@intel.com>
*/
#define DEBUG_SUBSYSTEM S_MDS
#include <obd.h>
#include <obd_class.h>
#include <uapi/linux/lustre/lustre_ver.h>
#include <obd_support.h>
#include <lprocfs_status.h>
#include <lustre_fid.h>
#include <uapi/linux/lustre/lustre_param.h>
#include <md_object.h>
#include <lustre_linkea.h>
#include <lustre_log.h>
#include "lod_internal.h"
struct thandle *lod_sub_get_thandle(const struct lu_env *env,
struct thandle *th,
const struct dt_object *sub_obj,
bool *record_update)
{
struct lod_device *lod = dt2lod_dev(th->th_dev);
struct top_thandle *tth;
struct thandle *sub_th;
int type = LU_SEQ_RANGE_ANY;
__u32 mdt_index;
int rc;
ENTRY;
if (record_update != NULL)
*record_update = false;
if (th->th_top == NULL)
RETURN(th);
tth = container_of(th, struct top_thandle, tt_super);
tth->tt_master_sub_thandle->th_ignore_quota = th->th_ignore_quota;
tth->tt_master_sub_thandle->th_ignore_root_proj_quota =
th->th_ignore_root_proj_quota;
/* local object must be mdt object, Note: during ost object
* creation, FID is not assigned until osp_create(),
* so if the FID of sub_obj is zero, it means OST object. */
if (!dt_object_remote(sub_obj) ||
fid_is_zero(lu_object_fid(&sub_obj->do_lu))) {
/* local MDT object */
if (fid_is_sane(lu_object_fid(&sub_obj->do_lu)) &&
tth->tt_multiple_thandle != NULL &&
record_update != NULL &&
th->th_result == 0)
*record_update = true;
RETURN(tth->tt_master_sub_thandle);
}
rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
&mdt_index, &type);
if (rc < 0)
RETURN(ERR_PTR(rc));
/* th_complex means we need track all of updates for this
* transaction, include changes on OST */
if (type == LU_SEQ_RANGE_OST && !th->th_complex)
RETURN(tth->tt_master_sub_thandle);
sub_th = thandle_get_sub(env, th, sub_obj);
if (IS_ERR(sub_th))
RETURN(sub_th);
sub_th->th_ignore_quota = th->th_ignore_quota;
sub_th->th_ignore_root_proj_quota = th->th_ignore_root_proj_quota;
if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
th->th_result == 0)
*record_update = true;
RETURN(sub_th);
}
/**
* lod_sub_declare_create() - Declare sub-object creation.
* @env: execution environment
* @dt: the object being created
* @attr: the attributes of the object being created
* @hint: the hint of the creation
* @dof: the object format of the creation
* @th: the transaction handle
*
* Get transaction of next layer and declare the creation of the object.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_create(const struct lu_env *env, struct dt_object *dt,
struct lu_attr *attr,
struct dt_allocation_hint *hint,
struct dt_object_format *dof, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
return PTR_ERR(sub_th);
if (record_update)
update_record_size(env, create, th, lu_object_fid(&dt->do_lu),
attr, hint, dof);
return dt_declare_create(env, dt, attr, hint, dof, sub_th);
}
/**
* lod_sub_create() - Create sub-object.
* @env: execution environment
* @dt: the object being created
* @attr: the attributes of the object being created
* @hint: the hint of the creation
* @dof: the object format of the creation
* @th: the transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation, and create the object.
*
* Return:
* * %0 if the creation succeeds.
* * %negative errno if the creation fails.
*/
int lod_sub_create(const struct lu_env *env, struct dt_object *dt,
struct lu_attr *attr, struct dt_allocation_hint *hint,
struct dt_object_format *dof, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(create, th,
lu_object_fid(&dt->do_lu),
attr, hint, dof);
if (rc < 0)
RETURN(rc);
}
rc = dt_create(env, dt, attr, hint, dof, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_ref_add() - Declare adding reference for the sub-object
* @env: execution environment
* @dt: dt object to add reference
* @th: transaction handle
*
* Get transaction of next layer and declare the reference adding.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_ref_add(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
rc = dt_declare_ref_add(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_ref_add() - Add reference for the sub-object
* @env: execution environment
* @dt: dt object to add reference
* @th: transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation and add reference of the object.
*
* Return:
* * %0 if it succeeds.
* * %negative errno if the addition fails.
*/
int lod_sub_ref_add(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(ref_add, th,
lu_object_fid(&dt->do_lu));
if (rc < 0)
RETURN(rc);
}
rc = dt_ref_add(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_ref_del() - Declare deleting reference for the sub-object
* @env: execution environment
* @dt: dt object to delete reference
* @th: transaction handle
*
* Get transaction of next layer and declare the reference deleting.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_ref_del(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
rc = dt_declare_ref_del(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_ref_del() - Delete reference for the sub-object
* @env: execution environment
* @dt: dt object to delete reference
* @th: transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation and delete reference of the object.
*
* Return:
* * %0 0 if it succeeds.
* * %negative errno if it fails.
*/
int lod_sub_ref_del(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(ref_del, th,
lu_object_fid(&dt->do_lu));
if (rc < 0)
RETURN(rc);
}
rc = dt_ref_del(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_destroy() - Declare destroying sub-object
* @env: execution environment
* @dt: dt object to be destroyed
* @th: transaction handle
*
* Get transaction of next layer and declare the sub-object destroy.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_destroy(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, destroy, th, lu_object_fid(&dt->do_lu));
rc = dt_declare_destroy(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_destroy() - Destroy sub-object
* @env: execution environment
* @dt: dt object to be destroyed
* @th: transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation and destroy the object.
*
* Return:
* * %0 if the destroy succeeds.
* * %negative errno if the destroy fails.
*/
int lod_sub_destroy(const struct lu_env *env, struct dt_object *dt,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(destroy, th, lu_object_fid(&dt->do_lu));
if (rc < 0)
RETURN(rc);
}
rc = dt_destroy(env, dt, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_insert() - Declare sub-object index insert
* @env: execution environment
* @dt: object for which to insert index
* @rec: record of the index which will be inserted
* @key: key of the index which will be inserted
* @th: the transaction handle
*
* Get transaction of next layer and declare index insert.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_insert(const struct lu_env *env, struct dt_object *dt,
const struct dt_rec *rec,
const struct dt_key *key, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
return PTR_ERR(sub_th);
if (record_update)
update_record_size(env, index_insert, th,
lu_object_fid(&dt->do_lu), rec, key);
return dt_declare_insert(env, dt, rec, key, sub_th);
}
/**
* lod_sub_insert() - Insert index of sub object
* @env: execution environment
* @dt: object for which to insert index
* @rec: record of the index to be inserted
* @key: key of the index to be inserted
* @th: the transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation, and insert the index.
*
* Return:
* * %0 if the insertion succeeds.
* * %negative errno if the insertion fails.
*/
int lod_sub_insert(const struct lu_env *env, struct dt_object *dt,
const struct dt_rec *rec, const struct dt_key *key,
struct thandle *th)
{
struct thandle *sub_th;
int rc;
bool record_update;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
return PTR_ERR(sub_th);
if (record_update) {
rc = update_record_pack(index_insert, th,
lu_object_fid(&dt->do_lu), rec, key);
if (rc < 0)
return rc;
}
return dt_insert(env, dt, rec, key, sub_th);
}
/**
* lod_sub_declare_delete() - Declare sub-object index delete
* @env: execution environment
* @dt: object for which to delete index
* @key: key of the index which will be deleted
* @th: the transaction handle
*
* Get transaction of next layer and declare index deletion.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_delete(const struct lu_env *env, struct dt_object *dt,
const struct dt_key *key, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
return PTR_ERR(sub_th);
if (record_update)
update_record_size(env, index_delete, th,
lu_object_fid(&dt->do_lu), key);
return dt_declare_delete(env, dt, key, sub_th);
}
/**
* lod_sub_delete() - Delete index of sub object
* @env: execution environment
* @dt: object for which to delete index
* @name: key name of the sub-object to be deleted
* @th: the transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation, and delete the index.
*
* Return:
* * %0 if the deletion succeeds.
* * %negative errno if the deletion fails.
*/
int lod_sub_delete(const struct lu_env *env, struct dt_object *dt,
const struct dt_key *name, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(index_delete, th,
lu_object_fid(&dt->do_lu), name);
if (rc < 0)
RETURN(rc);
}
rc = dt_delete(env, dt, name, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_xattr_set() - Declare xattr_set
* @env: execution environment
* @dt: object on which to set xattr
* @buf: xattr to be set
* @name: name of the xattr
* @fl: flag for setting xattr
* @th: transaction handle
*
* Get transaction of next layer, and declare xattr set.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
const struct lu_buf *buf, const char *name,
int fl, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, xattr_set, th,
lu_object_fid(&dt->do_lu),
buf, name, fl);
rc = dt_declare_xattr_set(env, dt, NULL, buf, name, fl, sub_th);
RETURN(rc);
}
/**
* lod_sub_xattr_set() - Set xattr
* @env: execution environment
* @dt: object on which to set xattr
* @buf: xattr to be set
* @name: name of the xattr
* @fl: flag for setting xattr
* @th: transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation, and set xattr to the object.
*
* Return:
* * %0 if the xattr setting succeeds.
* * %negative errno if xattr setting fails.
*/
int lod_sub_xattr_set(const struct lu_env *env, struct dt_object *dt,
const struct lu_buf *buf, const char *name, int fl,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(xattr_set, th,
lu_object_fid(&dt->do_lu),
buf, name, fl);
if (rc < 0)
RETURN(rc);
}
rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_attr_set() - Declare attr_set
* @env: execution environment
* @dt: object on which to set attr
* @attr: attributes to be set
* @th: transaction handle
*
* Get transaction of next layer, and declare attr set.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
const struct lu_attr *attr, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, attr_set, th,
lu_object_fid(&dt->do_lu), attr);
rc = dt_declare_attr_set(env, dt, attr, sub_th);
RETURN(rc);
}
/**
* lod_sub_attr_set() - attributes set
* @env: execution environment
* @dt: object on which to set attr
* @attr: attrbutes to be set
* @th: transaction handle
*
* Get transaction of next layer, record updates if it belongs to cross-MDT
* operation, and set attributes to the object.
*
* Return:
* * %0 if attributes setting succeeds.
* * %negative errno if the attributes setting fails.
*/
int lod_sub_attr_set(const struct lu_env *env, struct dt_object *dt,
const struct lu_attr *attr, struct thandle *th)
{
bool record_update;
struct thandle *sub_th;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
attr);
if (rc < 0)
RETURN(rc);
}
rc = dt_attr_set(env, dt, attr, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_xattr_del() - Declare xattr_del
* @env: execution environment
* @dt: object on which to delete xattr
* @name: name of the xattr to be deleted
* @th: transaction handle
*
* Get transaction of next layer, and declare xattr deletion.
*
* Return:
* * %0 if the declaration succeeds.
* * %negative errno if the declaration fails.
*/
int lod_sub_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
const char *name, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, xattr_del, th,
lu_object_fid(&dt->do_lu),
name);
rc = dt_declare_xattr_del(env, dt, name, sub_th);
RETURN(rc);
}
/**
* lod_sub_xattr_del() - xattribute deletion
* @env: execution environment
* @dt: object on which to delete xattr
* @name: name of the xattr to be deleted
* @th: transaction handle
*
* Get transaction of next layer, record update if it belongs to cross-MDT
* operation and delete xattr.
*
* Return:
* * %0 if the deletion succeeds.
* * %negative errno if the deletion fails.
*/
int lod_sub_xattr_del(const struct lu_env *env, struct dt_object *dt,
const char *name, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(xattr_del, th,
lu_object_fid(&dt->do_lu), name);
if (rc < 0)
RETURN(rc);
}
rc = dt_xattr_del(env, dt, name, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_write() - Declare buffer write
* @env: execution environment
* @dt: object to be written
* @buf: buffer to write which includes an embedded size field
* @pos: offet in the object to start writing at
* @th: transaction handle
*
* Get transaction of next layer and declare buffer write.
*
* Return:
* * %0 if the insertion succeeds.
* * %negative errno if the insertion fails.
*/
int lod_sub_declare_write(const struct lu_env *env, struct dt_object *dt,
const struct lu_buf *buf, loff_t pos,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, write, th,
lu_object_fid(&dt->do_lu),
buf, pos);
rc = dt_declare_write(env, dt, buf, pos, sub_th);
RETURN(rc);
}
/**
* lod_sub_write() - Write buffer to sub object
* @env: execution environment
* @dt: object to be written
* @buf: buffer to write which includes an embedded size field
* @pos: offet in the object to start writing at
* @th: transaction handle
*
* Get transaction of next layer, records buffer write if it belongs to
* Cross-MDT operation, and write buffer.
*
* Return:
* * %size in bytes(buffer) if it succeeds.
* * %negative errno if the insertion fails.
*/
ssize_t lod_sub_write(const struct lu_env *env, struct dt_object *dt,
const struct lu_buf *buf, loff_t *pos,
struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
ssize_t rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
buf, *pos);
if (rc < 0)
RETURN(rc);
}
rc = dt_write(env, dt, buf, pos, sub_th);
RETURN(rc);
}
/**
* lod_sub_declare_punch() - Declare punch
* @env: execution environment
* @dt: object to be written
* @start: start offset of punch
* @end: end offet of punch
* @th: transaction handle
*
* Get transaction of next layer and declare punch.
*
* Return:
* * %0 if the insertion succeeds.
* * %negative errno if the insertion fails.
*/
int lod_sub_declare_punch(const struct lu_env *env, struct dt_object *dt,
__u64 start, __u64 end, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update)
update_record_size(env, punch, th,
lu_object_fid(&dt->do_lu),
start, end);
rc = dt_declare_punch(env, dt, start, end, sub_th);
RETURN(rc);
}
/**
* lod_sub_punch() - Punch to sub object
* @env: execution environment
* @dt: object to be written
* @start: start offset of punch
* @end: end offset of punch
* @th: transaction handle
*
* Get transaction of next layer, records buffer write if it belongs to
* Cross-MDT operation, and punch object.
*
* Return:
* * %size in bytes(buffer) if it succeeds.
* * %negative errno if it fails.
*/
int lod_sub_punch(const struct lu_env *env, struct dt_object *dt,
__u64 start, __u64 end, struct thandle *th)
{
struct thandle *sub_th;
bool record_update;
int rc;
ENTRY;
sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
if (IS_ERR(sub_th))
RETURN(PTR_ERR(sub_th));
if (record_update) {
rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
start, end);
if (rc < 0)
RETURN(rc);
}
rc = dt_punch(env, dt, start, end, sub_th);
RETURN(rc);
}
int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
struct dt_device *dt, int index)
{
struct lod_thread_info *lti = lod_env_info(env);
struct llog_ctxt *ctxt;
struct llog_handle *lgh;
struct llog_catid *cid = <i->lti_cid;
struct lu_fid *fid = <i->lti_fid;
struct obd_device *obd;
int rc;
bool need_put = false;
ENTRY;
lu_update_log_fid(fid, index);
rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
if (rc < 0)
RETURN(rc);
rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
if (rc != 0) {
CERROR("%s: can't get id from catalogs: rc = %d\n",
lod2obd(lod)->obd_name, rc);
RETURN(rc);
}
obd = dt->dd_lu_dev.ld_obd;
ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
LASSERT(ctxt != NULL);
/* concurrent config processing (e.g. setting MDT active)
* can try to initialize llog again before causing double
* initialization. check for this */
if (ctxt->loc_handle)
GOTO(out_put, rc = 0);
ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
if (likely(logid_id(&cid->lci_logid) != 0)) {
rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
LLOG_OPEN_EXISTS);
/* re-create llog if it is missing */
if (rc == -ENOENT || rc == -EREMCHG) {
logid_set_id(&cid->lci_logid, 0);
} else if (rc < 0) {
CWARN("%s: can't open llog "DFID": rc = %d\n",
lod2obd(lod)->obd_name,
PLOGID(&cid->lci_logid), rc);
GOTO(out_put, rc);
}
}
if (unlikely(logid_id(&cid->lci_logid) == 0)) {
renew:
rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
if (rc < 0) {
CWARN("%s: can't create new llog: rc = %d\n",
lod2obd(lod)->obd_name, rc);
GOTO(out_put, rc);
}
cid->lci_logid = lgh->lgh_id;
need_put = true;
}
LASSERT(lgh != NULL);
rc = llog_init_handle(env, lgh, LLOG_F_IS_CAT, NULL);
if (rc) {
/* Update llog is incorrect, renew it */
if (rc == -EINVAL && need_put == false) {
CWARN("%s: renew invalid update log "DFID": rc = %d\n",
lod2obd(lod)->obd_name, PLOGID(&cid->lci_logid),
rc);
llog_cat_close(env, lgh);
GOTO(renew, 0);
}
GOTO(out_close, rc);
}
if (need_put) {
rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
if (rc) {
CERROR("%s: can't update id in catalogs: rc = %d\n",
lod2obd(lod)->obd_name, rc);
GOTO(out_close, rc);
}
}
LASSERT(!ctxt->loc_handle);
ctxt->loc_handle = lgh;
CDEBUG(D_INFO, "%s: init llog for index %d - catid "DFID"\n",
obd->obd_name, index, PLOGID(&cid->lci_logid));
out_close:
if (rc != 0)
llog_cat_close(env, lgh);
out_put:
llog_ctxt_put(ctxt);
RETURN(rc);
}
