Viewing: ofd_fs.c
// SPDX-License-Identifier: GPL-2.0
/*
* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
* Copyright (c) 2012, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*
* This file provides helper functions to handle various data stored on disk.
* It uses OSD API and works with any OSD.
*
* Note: this file contains also functions for sequence handling, they are
* placed here improperly and will be moved to the ofd_dev.c and ofd_internal.h,
* this comment is to be removed after that.
*
* Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
* Author: Mikhail Pershin <mike.pershin@intel.com>
*/
#define DEBUG_SUBSYSTEM S_FILTER
#include "ofd_internal.h"
/**
* Restrict precreate batch count by its upper limit.
*
* The precreate batch count is a number of precreates to do in
* single transaction. It has upper limit - ofd_device::ofd_precreate_batch
* value which shouldn't be exceeded.
*
* \param[in] ofd OFD device
* \param[in] batch number of updates in the batch
*
* \retval \a batch limited by ofd_device::ofd_precreate_batch
*/
int ofd_precreate_batch(struct ofd_device *ofd, int batch)
{
int count;
spin_lock(&ofd->ofd_batch_lock);
count = min(ofd->ofd_precreate_batch, batch);
spin_unlock(&ofd->ofd_batch_lock);
return count;
}
/**
* Get ofd_seq for \a seq.
*
* Function finds appropriate structure by \a seq number and
* increases the reference counter of that structure.
*
* \param[in] ofd OFD device
* \param[in] seq sequence number, FID sequence number usually
*
* \retval pointer to the requested ofd_seq structure
* \retval NULL if ofd_seq is not found
*/
struct ofd_seq *ofd_seq_get(struct ofd_device *ofd, u64 seq)
{
struct ofd_seq *oseq;
read_lock(&ofd->ofd_seq_list_lock);
list_for_each_entry(oseq, &ofd->ofd_seq_list, os_list) {
if (ostid_seq(&oseq->os_oi) == seq) {
refcount_inc(&oseq->os_refc);
read_unlock(&ofd->ofd_seq_list_lock);
return oseq;
}
}
read_unlock(&ofd->ofd_seq_list_lock);
return NULL;
}
/**
* Drop a reference to ofd_seq.
*
* The paired function to the ofd_seq_get(). It decrease the reference counter
* of the ofd_seq structure and free it if that reference was last one.
*
* \param[in] env execution environment
* \param[in] oseq ofd_seq structure to put
*/
void ofd_seq_put(const struct lu_env *env, struct ofd_seq *oseq)
{
if (refcount_dec_and_test(&oseq->os_refc)) {
LASSERT(list_empty(&oseq->os_list));
LASSERT(oseq->os_lastid_obj != NULL);
dt_object_put(env, oseq->os_lastid_obj);
OBD_FREE_PTR(oseq);
}
}
/**
* Add a new ofd_seq to the given OFD device.
*
* First it checks if there is already existent ofd_seq with the same
* sequence number as used by \a new_seq.
* If such ofd_seq is not found then the \a new_seq is added to the list
* of all ofd_seq structures else the \a new_seq is dropped and the found
* ofd_seq is returned back.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
* \param[in] new_seq new ofd_seq to be added
*
* \retval ofd_seq structure
*/
static struct ofd_seq *ofd_seq_add(const struct lu_env *env,
struct ofd_device *ofd,
struct ofd_seq *new_seq)
{
struct ofd_seq *os = NULL;
write_lock(&ofd->ofd_seq_list_lock);
list_for_each_entry(os, &ofd->ofd_seq_list, os_list) {
if (ostid_seq(&os->os_oi) == ostid_seq(&new_seq->os_oi)) {
refcount_inc(&os->os_refc);
write_unlock(&ofd->ofd_seq_list_lock);
/* The seq has not been added to the list */
ofd_seq_put(env, new_seq);
return os;
}
}
refcount_inc(&new_seq->os_refc);
list_add_tail(&new_seq->os_list, &ofd->ofd_seq_list);
ofd->ofd_seq_count++;
write_unlock(&ofd->ofd_seq_list_lock);
return new_seq;
}
/**
* Get last object ID for the given sequence.
*
* \param[in] oseq OFD sequence structure
*
* \retval the last object ID for this sequence
*/
u64 ofd_seq_last_oid(struct ofd_seq *oseq)
{
u64 id;
spin_lock(&oseq->os_last_oid_lock);
id = ostid_id(&oseq->os_oi);
spin_unlock(&oseq->os_last_oid_lock);
return id;
}
/**
* Set new last object ID for the given sequence.
*
* \param[in] oseq OFD sequence
* \param[in] id the new OID to set
*/
void ofd_seq_last_oid_set(struct ofd_seq *oseq, u64 id)
{
spin_lock(&oseq->os_last_oid_lock);
if (likely(ostid_id(&oseq->os_oi) < id)) {
if (ostid_set_id(&oseq->os_oi, id)) {
CERROR("Bad %llu to set " DOSTID "\n",
(unsigned long long)id, POSTID(&oseq->os_oi));
}
}
spin_unlock(&oseq->os_last_oid_lock);
}
/**
* Update last used OID on disk for the given sequence.
*
* The last used object ID is stored persistently on disk and
* must be written when updated. This function writes the sequence data.
* The format is just an object ID of the latest used object FID.
* Each ID is stored in per-sequence file.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
* \param[in] oseq ofd_seq structure with data to write
*
* \retval 0 on successful write of data from \a oseq
* \retval negative value on error
*/
int ofd_seq_last_oid_write(const struct lu_env *env, struct ofd_device *ofd,
struct ofd_seq *oseq)
{
struct ofd_thread_info *info = ofd_info(env);
u64 tmp;
struct dt_object *obj = oseq->os_lastid_obj;
struct thandle *th;
int rc;
ENTRY;
if (ofd->ofd_osd->dd_rdonly)
RETURN(0);
tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
info->fti_buf.lb_buf = &tmp;
info->fti_buf.lb_len = sizeof(tmp);
info->fti_off = 0;
LASSERT(obj != NULL);
th = dt_trans_create(env, ofd->ofd_osd);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
rc = dt_declare_record_write(env, obj, &info->fti_buf,
info->fti_off, th);
if (rc < 0)
GOTO(out, rc);
rc = dt_trans_start_local(env, ofd->ofd_osd, th);
if (rc < 0)
GOTO(out, rc);
rc = dt_record_write(env, obj, &info->fti_buf, &info->fti_off,
th);
if (rc < 0)
GOTO(out, rc);
CDEBUG(D_INODE, "%s: write last_objid "DOSTID": rc = %d\n",
ofd_name(ofd), POSTID(&oseq->os_oi), rc);
EXIT;
out:
dt_trans_stop(env, ofd->ofd_osd, th);
return rc;
}
/**
* Deregister LWP items for FLDB and SEQ client on OFD.
*
* LWP is lightweight proxy - simplified connection between
* servers. It is used for FID Location Database (FLDB) and
* sequence (SEQ) client-server interactions.
*
* This function is used during server cleanup process to free
* LWP items that were previously set up upon OFD start.
*
* \param[in] ofd OFD device
*/
static void ofd_deregister_seq_exp(struct ofd_device *ofd)
{
struct seq_server_site *ss = &ofd->ofd_seq_site;
if (ss->ss_client_seq != NULL) {
lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
ss->ss_client_seq->lcs_exp = NULL;
}
if (ss->ss_server_fld != NULL) {
lustre_deregister_lwp_item(&ss->ss_server_fld->lsf_control_exp);
ss->ss_server_fld->lsf_control_exp = NULL;
}
}
/**
* Stop FLDB server on OFD.
*
* This function is part of OFD cleanup process.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
*
*/
static void ofd_fld_fini(const struct lu_env *env, struct ofd_device *ofd)
{
struct seq_server_site *ss = &ofd->ofd_seq_site;
if (ss != NULL && ss->ss_server_fld != NULL) {
fld_server_fini(env, ss->ss_server_fld);
OBD_FREE_PTR(ss->ss_server_fld);
ss->ss_server_fld = NULL;
}
}
/**
* Free sequence structures on OFD.
*
* This function is part of OFD cleanup process, it goes through
* the list of ofd_seq structures stored in ofd_device structure
* and frees them.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
*/
void ofd_seqs_free(const struct lu_env *env, struct ofd_device *ofd)
{
struct ofd_seq *oseq;
struct ofd_seq *tmp;
LIST_HEAD(dispose);
write_lock(&ofd->ofd_seq_list_lock);
list_for_each_entry_safe(oseq, tmp, &ofd->ofd_seq_list, os_list)
list_move(&oseq->os_list, &dispose);
write_unlock(&ofd->ofd_seq_list_lock);
while (!list_empty(&dispose)) {
oseq = container_of(dispose.next, struct ofd_seq, os_list);
list_del_init(&oseq->os_list);
ofd_seq_put(env, oseq);
}
}
/**
* Stop FLDB and SEQ services on OFD.
*
* This function is part of OFD cleanup process.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
*
*/
void ofd_seqs_fini(const struct lu_env *env, struct ofd_device *ofd)
{
int rc;
ofd_deregister_seq_exp(ofd);
rc = ofd_fid_fini(env, ofd);
if (rc != 0)
CERROR("%s: fid fini error: rc = %d\n", ofd_name(ofd), rc);
ofd_fld_fini(env, ofd);
ofd_seqs_free(env, ofd);
LASSERT(list_empty(&ofd->ofd_seq_list));
}
/**
* Return ofd_seq structure filled with valid data.
*
* This function gets the ofd_seq by sequence number and read
* corresponding data from disk.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
* \param[in] seq sequence number
*
* \retval ofd_seq structure filled with data
* \retval ERR_PTR pointer on error
*/
struct ofd_seq *ofd_seq_load(const struct lu_env *env, struct ofd_device *ofd,
u64 seq)
{
struct ofd_thread_info *info = ofd_info(env);
struct ofd_seq *oseq = NULL;
struct dt_object *dob;
u64 lastid;
int rc;
ENTRY;
/* if seq is already initialized */
oseq = ofd_seq_get(ofd, seq);
if (oseq != NULL) {
CDEBUG(D_TRACE, "%s: got sequence %#llx "DOSTID"\n",
ofd_name(ofd), seq, POSTID(&oseq->os_oi));
RETURN(oseq);
}
OBD_ALLOC_PTR(oseq);
if (oseq == NULL)
RETURN(ERR_PTR(-ENOMEM));
lu_last_id_fid(&info->fti_fid, seq, ofd->ofd_lut.lut_lsd.lsd_osd_index);
memset(&info->fti_attr, 0, sizeof(info->fti_attr));
info->fti_attr.la_valid = LA_MODE;
info->fti_attr.la_mode = S_IFREG | 0644;
info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
/* create object tracking per-seq last created
* id to be used by orphan recovery mechanism */
dob = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
&info->fti_dof, &info->fti_attr);
if (IS_ERR(dob)) {
OBD_FREE_PTR(oseq);
RETURN((void *)dob);
}
oseq->os_lastid_obj = dob;
INIT_LIST_HEAD(&oseq->os_list);
mutex_init(&oseq->os_create_lock);
spin_lock_init(&oseq->os_last_oid_lock);
ostid_set_seq(&oseq->os_oi, seq);
oseq->os_last_id_synced = 0;
refcount_set(&oseq->os_refc, 1);
atomic_set(&oseq->os_precreate_in_progress, 0);
rc = dt_attr_get(env, dob, &info->fti_attr);
if (rc)
GOTO(cleanup, rc);
if (info->fti_attr.la_size == 0) {
/* object is just created, initialize last id */
if (CFS_FAIL_CHECK(OBD_FAIL_OFD_SET_OID)) {
struct seq_server_site *ss = &ofd->ofd_seq_site;
struct lu_client_seq *client_seq = ss->ss_client_seq;
__u64 seq_width = fid_seq_is_norm(seq) ?
min(OBIF_MAX_OID, client_seq->lcs_width) :
min(IDIF_MAX_OID, client_seq->lcs_width);
ofd_seq_last_oid_set(oseq, seq_width > 255 ?
seq_width - 255 : seq_width);
} else {
ofd_seq_last_oid_set(oseq, OFD_INIT_OBJID);
}
ofd_seq_last_oid_write(env, ofd, oseq);
} else if (info->fti_attr.la_size == sizeof(lastid)) {
info->fti_off = 0;
info->fti_buf.lb_buf = &lastid;
info->fti_buf.lb_len = sizeof(lastid);
rc = dt_record_read(env, dob, &info->fti_buf, &info->fti_off);
if (rc) {
CERROR("%s: can't read last_id: rc = %d\n",
ofd_name(ofd), rc);
GOTO(cleanup, rc);
}
ofd_seq_last_oid_set(oseq, le64_to_cpu(lastid));
} else {
CERROR("%s: corrupted size %llu LAST_ID of seq %#llx\n",
ofd_name(ofd), (__u64)info->fti_attr.la_size, seq);
GOTO(cleanup, rc = -EINVAL);
}
CDEBUG(D_HA, "%s: adding sequence %#llx\n", ofd_name(ofd), seq);
oseq = ofd_seq_add(env, ofd, oseq);
RETURN((oseq != NULL) ? oseq : ERR_PTR(-ENOENT));
cleanup:
ofd_seq_put(env, oseq);
return ERR_PTR(rc);
}
/**
* initialize local FLDB server.
*
* \param[in] env execution environment
* \param[in] uuid unique name for this FLDS server
* \param[in] ofd OFD device
*
* \retval 0 on successful initialization
* \retval negative value on error
*/
static int ofd_fld_init(const struct lu_env *env, const char *uuid,
struct ofd_device *ofd)
{
struct seq_server_site *ss = &ofd->ofd_seq_site;
int rc;
ENTRY;
OBD_ALLOC_PTR(ss->ss_server_fld);
if (ss->ss_server_fld == NULL)
RETURN(rc = -ENOMEM);
rc = fld_server_init(env, ss->ss_server_fld, ofd->ofd_osd, uuid,
LU_SEQ_RANGE_OST);
if (rc < 0) {
OBD_FREE_PTR(ss->ss_server_fld);
ss->ss_server_fld = NULL;
RETURN(rc);
}
RETURN(0);
}
/**
* Update local FLDB copy from master server.
*
* This callback is called when LWP is connected to the server.
* It retrieves its FLDB entries from MDT0, and it only happens
* when upgrading the existing file system to 2.6.
*
* \param[in] data OFD device
*
* \retval 0 on successful FLDB update
* \retval negative value in case if failure
*/
static int ofd_register_lwp_callback(void *data)
{
struct lu_env *env;
struct ofd_device *ofd = data;
struct lu_server_fld *fld = ofd->ofd_seq_site.ss_server_fld;
int rc;
ENTRY;
if (!likely(fld->lsf_new))
RETURN(0);
OBD_ALLOC_PTR(env);
if (env == NULL)
RETURN(-ENOMEM);
rc = lu_env_init(env, LCT_DT_THREAD);
if (rc < 0)
GOTO(out, rc);
rc = fld_update_from_controller(env, fld);
if (rc < 0) {
CERROR("%s: cannot update controller: rc = %d\n",
ofd_name(ofd), rc);
GOTO(out, rc);
}
EXIT;
out:
lu_env_fini(env);
OBD_FREE_PTR(env);
return rc;
}
/**
* Get LWP exports from LWP connection for local FLDB server and SEQ client.
*
* This function is part of setup process and initialize FLDB server and SEQ
* client, so they may work with remote servers.
*
* \param[in] ofd OFD device
*
* \retval 0 on successful export get
* \retval negative value on error
*/
static int ofd_register_seq_exp(struct ofd_device *ofd)
{
struct seq_server_site *ss = &ofd->ofd_seq_site;
char *lwp_name = NULL;
int rc;
OBD_ALLOC(lwp_name, MAX_OBD_NAME);
if (lwp_name == NULL)
GOTO(out_free, rc = -ENOMEM);
rc = tgt_name2lwp_name(ofd_name(ofd), lwp_name, MAX_OBD_NAME, 0);
if (rc != 0)
GOTO(out_free, rc);
rc = lustre_register_lwp_item(lwp_name, &ss->ss_client_seq->lcs_exp,
NULL, NULL);
if (rc != 0)
GOTO(out_free, rc);
rc = lustre_register_lwp_item(lwp_name,
&ss->ss_server_fld->lsf_control_exp,
ofd_register_lwp_callback, ofd);
if (rc != 0) {
lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
ss->ss_client_seq->lcs_exp = NULL;
GOTO(out_free, rc);
}
out_free:
OBD_FREE(lwp_name, MAX_OBD_NAME);
return rc;
}
/**
* Initialize SEQ and FLD service on OFD.
*
* This is part of OFD setup process.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
*
* \retval 0 on successful services initialization
* \retval negative value on error
*/
int ofd_seqs_init(const struct lu_env *env, struct ofd_device *ofd)
{
int rc;
rwlock_init(&ofd->ofd_seq_list_lock);
INIT_LIST_HEAD(&ofd->ofd_seq_list);
ofd->ofd_seq_count = 0;
rc = ofd_fid_init(env, ofd);
if (rc != 0) {
CERROR("%s: fid init error: rc = %d\n", ofd_name(ofd), rc);
GOTO(out, rc);
}
rc = ofd_fld_init(env, ofd_name(ofd), ofd);
if (rc < 0) {
CERROR("%s: Can't init fld, rc %d\n", ofd_name(ofd), rc);
GOTO(out_fid, rc);
}
rc = ofd_register_seq_exp(ofd);
if (rc < 0) {
CERROR("%s: Can't init seq exp, rc %d\n", ofd_name(ofd), rc);
GOTO(out_fld, rc);
}
RETURN(0);
out_fld:
ofd_fld_fini(env, ofd);
out_fid:
ofd_fid_fini(env, ofd);
out:
return rc;
}
int ofd_failure_domain_write(const struct lu_env *env, struct ofd_device *ofd)
{
struct ofd_thread_info *info = ofd_info(env);
struct dt_object *fd;
struct thandle *th;
int rc;
memset(&info->fti_attr, 0, sizeof(info->fti_attr));
info->fti_buf.lb_buf = &ofd->ofd_failure_domain;
info->fti_buf.lb_len = sizeof(ofd->ofd_failure_domain);
info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
info->fti_off = 0;
lu_local_obj_fid(&info->fti_fid, OFD_FAILURE_DOMAIN_OID);
fd = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
&info->fti_dof, &info->fti_attr);
if (IS_ERR(fd))
GOTO(out, rc = PTR_ERR(fd));
th = dt_trans_create(env, ofd->ofd_osd);
if (IS_ERR(th))
GOTO(out_put, rc = PTR_ERR(th));
rc = dt_declare_record_write(env, fd, &info->fti_buf, info->fti_off,
th);
if (rc)
goto out_stop;
rc = dt_trans_start_local(env, ofd->ofd_osd, th);
if (rc)
goto out_stop;
rc = dt_record_write(env, fd, &info->fti_buf, &info->fti_off, th);
out_stop:
dt_trans_stop(env, ofd->ofd_osd, th);
out_put:
dt_object_put(env, fd);
out:
return rc;
}
static int ofd_failure_domain_read(const struct lu_env *env,
struct ofd_device *ofd)
{
struct ofd_thread_info *info = ofd_info(env);
struct dt_object *fd;
int rc;
memset(&info->fti_attr, 0, sizeof(info->fti_attr));
info->fti_buf.lb_buf = &ofd->ofd_failure_domain;
info->fti_buf.lb_len = sizeof(ofd->ofd_failure_domain);
info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
info->fti_off = 0;
lu_local_obj_fid(&info->fti_fid, OFD_FAILURE_DOMAIN_OID);
fd = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
&info->fti_dof, &info->fti_attr);
if (IS_ERR(fd))
GOTO(out, rc = PTR_ERR(fd));
rc = dt_record_read(env, fd, &info->fti_buf, &info->fti_off);
dt_object_put(env, fd);
out:
return rc;
}
/**
* Initialize storage for the OFD.
*
* This function sets up service files for OFD. Currently, the only
* service files are "health_check" and "failure_domain".
*
* \param[in] env execution environment
* \param[in] ofd OFD device
* \param[in] obd OBD device (unused now)
*
* \retval 0 on successful setup
* \retval negative value on error
*/
int ofd_fs_setup(const struct lu_env *env, struct ofd_device *ofd,
struct obd_device *obd)
{
struct ofd_thread_info *info = ofd_info(env);
struct dt_object *fo, *fd;
int rc = 0;
ENTRY;
rc = ofd_seqs_init(env, ofd);
if (rc)
GOTO(out, rc);
if (CFS_FAIL_CHECK(OBD_FAIL_MDS_FS_SETUP))
GOTO(out_seqs, rc = -ENOENT);
lu_local_obj_fid(&info->fti_fid, OFD_HEALTH_CHECK_OID);
memset(&info->fti_attr, 0, sizeof(info->fti_attr));
info->fti_attr.la_valid = LA_MODE;
info->fti_attr.la_mode = S_IFREG | 0644;
info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
fo = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
&info->fti_dof, &info->fti_attr);
if (IS_ERR(fo))
GOTO(out_seqs, rc = PTR_ERR(fo));
ofd->ofd_health_check_file = fo;
lu_local_obj_fid(&info->fti_fid, OFD_FAILURE_DOMAIN_OID);
fd = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
&info->fti_dof, &info->fti_attr);
if (IS_ERR(fd))
GOTO(out_seqs, rc = PTR_ERR(fd));
dt_object_put(env, fd);
ofd_failure_domain_read(env, ofd);
RETURN(0);
out_seqs:
ofd_seqs_fini(env, ofd);
out:
return rc;
}
/**
* Cleanup service files on OFD.
*
* This function syncs whole OFD device.
*
* \param[in] env execution environment
* \param[in] ofd OFD device
*/
void ofd_fs_cleanup(const struct lu_env *env, struct ofd_device *ofd)
{
int rc;
ENTRY;
ofd_seqs_fini(env, ofd);
rc = dt_sync(env, ofd->ofd_osd);
if (rc < 0)
CWARN("%s: can't sync OFD upon cleanup: %d\n",
ofd_name(ofd), rc);
if (ofd->ofd_health_check_file) {
dt_object_put(env, ofd->ofd_health_check_file);
ofd->ofd_health_check_file = NULL;
}
EXIT;
}
