*
* You should have received a copy of the GNU General Public License
* version 2 along with this program; If not, see
- * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
- *
- * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
- * CA 95054 USA or visit www.sun.com if you need additional information or
- * have any questions.
+ * http://www.gnu.org/licenses/gpl-2.0.html
*
* GPL HEADER END
*/
* Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2012, 2013, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*
* lustre/osp/osp_dev.c
*
- * Lustre OST/MDT Proxy Device (OSP) is in the MDS stack, and used as
- * a proxy device to communicate with other MDTs and OSTs.
- *
- * The purpose is to export the OSD API from the remote MDT or OST's underlying
- * OSD for access and modification by the local service (typically MDD/LOD, or
- * LFSCK) as if it were a local OSD. The goal is that the OSP provides a
- * transparent interface for access to the remote OSD.
- *
* Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
* Author: Mikhail Pershin <mike.pershin@intel.com>
* Author: Di Wang <di.wang@intel.com>
*/
+/*
+ * The Object Storage Proxy (OSP) module provides an implementation of
+ * the DT API for remote MDTs and OSTs. Every local OSP device (or
+ * object) is a proxy for a remote OSD device (or object). Thus OSP
+ * converts DT operations into RPCs, which are sent to the OUT service
+ * on a remote target, converted back to DT operations, and
+ * executed. Of course there are many ways in which this description
+ * is inaccurate but it's a good enough mental model. OSP is used by
+ * the MDT stack in several ways:
+ *
+ * - OSP devices allocate FIDs for the stripe sub-objects of a striped
+ * file or directory.
+ *
+ * - OSP objects represent the remote MDT and OST objects that are
+ * the stripes of a striped object.
+ *
+ * - OSP devices log, send, and track synchronous operations (setattr
+ * and unlink) to remote targets.
+ *
+ * - OSP objects are the bottom slice of the compound LU object
+ * representing a remote MDT object: MDT/MDD/LOD/OSP.
+ *
+ * - OSP objects are used by LFSCK to represent remote OST objects
+ * during the verification of MDT-OST consistency.
+ *
+ * - OSP devices batch idempotent requests (declare_attr_get() and
+ * declare_xattr_get()) to the remote target and cache their results.
+ *
+ * In addition the OSP layer implements a subset of the OBD device API
+ * to support being a client of a remote target, connecting to other
+ * layers, and FID allocation.
+ */
#define DEBUG_SUBSYSTEM S_MDS
-#include <obd_class.h>
-#include <lustre_ioctl.h>
-#include <lustre_param.h>
+#include <linux/kthread.h>
+
+#include <uapi/linux/lustre/lustre_ioctl.h>
#include <lustre_log.h>
-#include <lustre_mdc.h>
+#include <lustre_obdo.h>
+#include <uapi/linux/lustre/lustre_param.h>
+#include <obd_class.h>
#include "osp_internal.h"
* \retval object object being created if the creation succeed.
* \retval NULL NULL if the creation failed.
*/
-struct lu_object *osp_object_alloc(const struct lu_env *env,
- const struct lu_object_header *hdr,
- struct lu_device *d)
+static struct lu_object *osp_object_alloc(const struct lu_env *env,
+ const struct lu_object_header *hdr,
+ struct lu_device *d)
{
struct lu_object_header *h = NULL;
struct osp_object *o;
RETURN(dto);
/* Get attributes of the local object. */
- rc = dt_attr_get(env, dto, attr, NULL);
+ rc = dt_attr_get(env, dto, attr);
if (rc) {
CERROR("%s: can't be initialized: rc = %d\n",
osp->opd_obd->obd_name, rc);
- lu_object_put(env, &dto->do_lu);
+ dt_object_put(env, dto);
RETURN(ERR_PTR(rc));
}
RETURN(dto);
struct thandle *th;
int rc;
+ if (osp->opd_storage->dd_rdonly)
+ RETURN(0);
+
th = dt_trans_create(env, osp->opd_storage);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
struct osp_thread_info *osi = osp_env_info(env);
struct lu_fid *fid = &osp->opd_last_used_fid;
struct dt_object *dto;
- int rc;
+ int rc = -EFAULT;
ENTRY;
dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
if (IS_ERR(dto))
RETURN(PTR_ERR(dto));
+ osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &osp->opd_last_id,
+ osp->opd_index);
+
/* object will be released in device cleanup path */
- if (osi->osi_attr.la_size >=
- sizeof(osi->osi_id) * (osp->opd_index + 1)) {
- osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
- osp->opd_index);
+ if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
- if (rc != 0)
+ if (rc != 0 && rc != -EFAULT)
GOTO(out, rc);
- } else {
+ /* In case of idif bits 32-48 go to f_seq
+ * (see osp_init_last_seq). So don't care
+ * about u64->u32 convertion. */
+ fid->f_oid = osp->opd_last_id;
+ }
+
+ if (rc == -EFAULT) { /* fresh LAST_ID */
+ osp->opd_last_id = 0;
fid->f_oid = 0;
- osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
- osp->opd_index);
rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
osi->osi_off);
if (rc != 0)
/* object will be released in device cleanup path */
CERROR("%s: can't initialize lov_objid: rc = %d\n",
osp->opd_obd->obd_name, rc);
- lu_object_put(env, &dto->do_lu);
+ dt_object_put(env, dto);
osp->opd_last_used_oid_file = NULL;
RETURN(rc);
}
struct osp_thread_info *osi = osp_env_info(env);
struct lu_fid *fid = &osp->opd_last_used_fid;
struct dt_object *dto;
- int rc;
+ int rc = -EFAULT;
ENTRY;
dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
if (IS_ERR(dto))
RETURN(PTR_ERR(dto));
+ osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
+ osp->opd_index);
+
/* object will be released in device cleanup path */
- if (osi->osi_attr.la_size >=
- sizeof(osi->osi_id) * (osp->opd_index + 1)) {
- osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
- osp->opd_index);
+ if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
- if (rc != 0)
+ if (rc != 0 && rc != -EFAULT)
GOTO(out, rc);
- } else {
+ if (fid_is_idif(fid))
+ fid->f_seq = fid_idif_seq(osp->opd_last_id,
+ osp->opd_index);
+ }
+
+ if (rc == -EFAULT) { /* fresh OSP */
fid->f_seq = 0;
- osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
- osp->opd_index);
rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
osi->osi_off);
+ if (rc != 0)
+ GOTO(out, rc);
}
osp->opd_last_used_seq_file = dto;
RETURN(0);
/* object will be released in device cleanup path */
CERROR("%s: can't initialize lov_seq: rc = %d\n",
osp->opd_obd->obd_name, rc);
- lu_object_put(env, &dto->do_lu);
+ dt_object_put(env, dto);
osp->opd_last_used_seq_file = NULL;
RETURN(rc);
}
rc = osp_init_last_seq(env, osp);
if (rc < 0) {
- CERROR("%s: Can not get ids %d from old objid!\n",
+ CERROR("%s: Can not get sequence %d from old objseq!\n",
osp->opd_obd->obd_name, rc);
GOTO(out, rc);
}
GOTO(out, rc = -EINVAL);
}
+ osp_fid_to_obdid(&osp->opd_last_used_fid, &osp->opd_last_id);
CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
out:
if (rc != 0) {
if (osp->opd_last_used_oid_file != NULL) {
- lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
+ dt_object_put(env, osp->opd_last_used_oid_file);
osp->opd_last_used_oid_file = NULL;
}
if (osp->opd_last_used_seq_file != NULL) {
- lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
+ dt_object_put(env, osp->opd_last_used_seq_file);
osp->opd_last_used_seq_file = NULL;
}
}
{
/* release last_used file */
if (osp->opd_last_used_oid_file != NULL) {
- lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
+ dt_object_put(env, osp->opd_last_used_oid_file);
osp->opd_last_used_oid_file = NULL;
}
if (osp->opd_last_used_seq_file != NULL) {
- lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
+ dt_object_put(env, osp->opd_last_used_seq_file);
osp->opd_last_used_seq_file = NULL;
}
}
*/
static int osp_disconnect(struct osp_device *d)
{
+ struct obd_device *obd = d->opd_obd;
struct obd_import *imp;
int rc = 0;
- imp = d->opd_obd->u.cli.cl_import;
+ imp = obd->u.cli.cl_import;
/* Mark import deactivated now, so we don't try to reconnect if any
* of the cleanup RPCs fails (e.g. ldlm cancel, etc). We don't
rc = ptlrpc_disconnect_import(imp, 0);
if (rc != 0)
- CERROR("%s: can't disconnect: rc = %d\n",
- d->opd_obd->obd_name, rc);
+ CERROR("%s: can't disconnect: rc = %d\n", obd->obd_name, rc);
ptlrpc_invalidate_import(imp);
}
/**
+ * Initialize the osp_update structure in OSP device
+ *
+ * Allocate osp update structure and start update thread.
+ *
+ * \param[in] osp OSP device
+ *
+ * \retval 0 if initialization succeeds.
+ * \retval negative errno if initialization fails.
+ */
+static int osp_update_init(struct osp_device *osp)
+{
+ struct l_wait_info lwi = { 0 };
+ struct task_struct *task;
+
+ ENTRY;
+
+ LASSERT(osp->opd_connect_mdt);
+
+ if (osp->opd_storage->dd_rdonly)
+ RETURN(0);
+
+ OBD_ALLOC_PTR(osp->opd_update);
+ if (osp->opd_update == NULL)
+ RETURN(-ENOMEM);
+
+ init_waitqueue_head(&osp->opd_update_thread.t_ctl_waitq);
+ init_waitqueue_head(&osp->opd_update->ou_waitq);
+ spin_lock_init(&osp->opd_update->ou_lock);
+ INIT_LIST_HEAD(&osp->opd_update->ou_list);
+ osp->opd_update->ou_rpc_version = 1;
+ osp->opd_update->ou_version = 1;
+ osp->opd_update->ou_generation = 0;
+
+ /* start thread handling sending updates to the remote MDT */
+ task = kthread_run(osp_send_update_thread, osp,
+ "osp_up%u-%u", osp->opd_index, osp->opd_group);
+ if (IS_ERR(task)) {
+ int rc = PTR_ERR(task);
+
+ OBD_FREE_PTR(osp->opd_update);
+ osp->opd_update = NULL;
+ CERROR("%s: can't start precreate thread: rc = %d\n",
+ osp->opd_obd->obd_name, rc);
+ RETURN(rc);
+ }
+
+ l_wait_event(osp->opd_update_thread.t_ctl_waitq,
+ osp_send_update_thread_running(osp) ||
+ osp_send_update_thread_stopped(osp), &lwi);
+
+ RETURN(0);
+}
+
+/**
+ * Finialize osp_update structure in OSP device
+ *
+ * Stop the OSP update sending thread, then delete the left
+ * osp thandle in the sending list.
+ *
+ * \param [in] osp OSP device.
+ */
+static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
+{
+ struct osp_update_request *our;
+ struct osp_update_request *tmp;
+ struct osp_updates *ou = osp->opd_update;
+
+ if (ou == NULL)
+ return;
+
+ osp->opd_update_thread.t_flags = SVC_STOPPING;
+ wake_up(&ou->ou_waitq);
+
+ wait_event(osp->opd_update_thread.t_ctl_waitq,
+ osp->opd_update_thread.t_flags & SVC_STOPPED);
+
+ /* Remove the left osp thandle from the list */
+ spin_lock(&ou->ou_lock);
+ list_for_each_entry_safe(our, tmp, &ou->ou_list,
+ our_list) {
+ list_del_init(&our->our_list);
+ LASSERT(our->our_th != NULL);
+ osp_trans_callback(env, our->our_th, -EIO);
+ /* our will be destroyed in osp_thandle_put() */
+ osp_thandle_put(env, our->our_th);
+ }
+ spin_unlock(&ou->ou_lock);
+
+ OBD_FREE_PTR(ou);
+ osp->opd_update = NULL;
+}
+
+/**
* Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
* precreate threads etc.
*
rc = osp_disconnect(d);
- osp_sync_fini(d);
+ osp_statfs_fini(d);
if (!d->opd_connect_mdt) {
+ /* stop sync thread */
+ osp_sync_fini(d);
+
/* stop precreate thread */
osp_precreate_fini(d);
static int osp_process_config(const struct lu_env *env,
struct lu_device *dev, struct lustre_cfg *lcfg)
{
- struct osp_device *d = lu2osp_dev(dev);
- struct obd_device *obd = d->opd_obd;
- int rc;
+ struct osp_device *d = lu2osp_dev(dev);
+ struct dt_device *dt = lu2dt_dev(dev);
+ struct obd_device *obd = d->opd_obd;
+ ssize_t count;
+ int rc;
ENTRY;
switch (lcfg->lcfg_command) {
case LCFG_PRE_CLEANUP:
rc = osp_disconnect(d);
+ osp_update_fini(env, d);
+ if (obd->obd_namespace != NULL)
+ ldlm_namespace_free_prior(obd->obd_namespace, NULL, 1);
break;
case LCFG_CLEANUP:
lu_dev_del_linkage(dev->ld_site, dev);
rc = osp_shutdown(env, d);
break;
case LCFG_PARAM:
- LASSERT(obd);
- rc = class_process_proc_seq_param(PARAM_OSC, obd->obd_vars,
- lcfg, obd);
- if (rc > 0)
- rc = 0;
- if (rc == -ENOSYS) {
- /* class_process_proc_param() haven't found matching
- * parameter and returned ENOSYS so that layer(s)
+ count = class_modify_config(lcfg, d->opd_connect_mdt ?
+ PARAM_OSP : PARAM_OSC,
+ &dt->dd_kobj);
+ if (count < 0) {
+ /* class_modify_config() haven't found matching
+ * parameter and returned an error so that layer(s)
* below could use that. But OSP is the bottom, so
- * just ignore it */
+ * just ignore it
+ */
CERROR("%s: unknown param %s\n",
(char *)lustre_cfg_string(lcfg, 0),
(char *)lustre_cfg_string(lcfg, 1));
- rc = 0;
}
+ rc = 0;
break;
default:
CERROR("%s: unknown command %u\n",
* \retval negative negative errno if get statfs failed.
*/
static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
- struct obd_statfs *sfs)
+ struct obd_statfs *sfs, struct obd_statfs_info *info)
{
struct osp_device *d = dt2osp_dev(dev);
+ struct obd_import *imp = d->opd_obd->u.cli.cl_import;
ENTRY;
+ if (imp->imp_state == LUSTRE_IMP_CLOSED)
+ RETURN(-ESHUTDOWN);
+
if (unlikely(d->opd_imp_active == 0))
RETURN(-ENOTCONN);
+ /* return recently updated data */
+ *sfs = d->opd_statfs;
+ if (info) {
+ info->os_reserved_mb_low = d->opd_reserved_mb_low;
+ info->os_reserved_mb_high = d->opd_reserved_mb_high;
+ }
+
if (d->opd_pre == NULL)
RETURN(0);
- /* return recently updated data */
- *sfs = d->opd_statfs;
+ CDEBUG(D_OTHER, "%s: %llu blocks, %llu free, %llu avail, "
+ "%u reserved mb low, %u reserved mb high,"
+ "%llu files, %llu free files\n", d->opd_obd->obd_name,
+ sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
+ d->opd_reserved_mb_low, d->opd_reserved_mb_high,
+ sfs->os_files, sfs->os_ffree);
+
+
+ if (info && !info->os_enable_pre)
+ RETURN(0);
/*
* layer above osp (usually lod) can use ffree to estimate
* how many objects are available for immediate creation
*/
spin_lock(&d->opd_pre_lock);
- LASSERTF(fid_seq(&d->opd_pre_last_created_fid) ==
- fid_seq(&d->opd_pre_used_fid),
- "last_created "DFID", next_fid "DFID"\n",
- PFID(&d->opd_pre_last_created_fid),
- PFID(&d->opd_pre_used_fid));
- sfs->os_fprecreated = fid_oid(&d->opd_pre_last_created_fid) -
- fid_oid(&d->opd_pre_used_fid);
+ sfs->os_fprecreated = osp_fid_diff(&d->opd_pre_last_created_fid,
+ &d->opd_pre_used_fid);
sfs->os_fprecreated -= d->opd_pre_reserved;
+ LASSERTF(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2,
+ "last_created "DFID", next_fid "DFID", reserved %llu\n",
+ PFID(&d->opd_pre_last_created_fid), PFID(&d->opd_pre_used_fid),
+ d->opd_pre_reserved);
spin_unlock(&d->opd_pre_lock);
-
- LASSERT(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2);
-
- CDEBUG(D_OTHER, "%s: "LPU64" blocks, "LPU64" free, "LPU64" avail, "
- LPU64" files, "LPU64" free files\n", d->opd_obd->obd_name,
- sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
- sfs->os_files, sfs->os_ffree);
RETURN(0);
}
static int osp_sync(const struct lu_env *env, struct dt_device *dev)
{
struct osp_device *d = dt2osp_dev(dev);
- cfs_time_t expire;
struct l_wait_info lwi = { 0 };
- unsigned long id, old;
- int rc = 0;
- unsigned long start = cfs_time_current();
+ time64_t start = ktime_get_seconds();
+ int recs, rc = 0;
+ u64 old;
+
ENTRY;
+ /* No Sync between MDTs yet. */
+ if (d->opd_connect_mdt)
+ RETURN(0);
+
+ recs = atomic_read(&d->opd_sync_changes);
+ old = atomic64_read(&d->opd_sync_processed_recs);
+
+ osp_sync_force(env, dt2osp_dev(dev));
+
if (unlikely(d->opd_imp_active == 0))
RETURN(-ENOTCONN);
- id = d->opd_syn_last_used_id;
+ down_write(&d->opd_async_updates_rwsem);
- CDEBUG(D_OTHER, "%s: id: used %lu, processed %lu\n",
- d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
+ CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
+ atomic_read(&d->opd_async_updates_count));
- /* wait till all-in-line are processed */
- while (d->opd_syn_last_processed_id < id) {
+ /* make sure the connection is fine */
+ lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout), osp_sync_timeout, d);
+ rc = l_wait_event(d->opd_sync_barrier_waitq,
+ atomic_read(&d->opd_async_updates_count) == 0,
+ &lwi);
+ up_write(&d->opd_async_updates_rwsem);
+ if (rc != 0)
+ GOTO(out, rc);
- old = d->opd_syn_last_processed_id;
+ CDEBUG(D_CACHE, "%s: processed %llu\n", d->opd_obd->obd_name,
+ (unsigned long long)atomic64_read(&d->opd_sync_processed_recs));
+ while (atomic64_read(&d->opd_sync_processed_recs) < old + recs) {
+ __u64 last = atomic64_read(&d->opd_sync_processed_recs);
/* make sure the connection is fine */
- expire = cfs_time_shift(obd_timeout);
- lwi = LWI_TIMEOUT(expire - cfs_time_current(),
+ lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
osp_sync_timeout, d);
- l_wait_event(d->opd_syn_barrier_waitq,
- d->opd_syn_last_processed_id >= id,
- &lwi);
+ l_wait_event(d->opd_sync_barrier_waitq,
+ atomic64_read(&d->opd_sync_processed_recs)
+ >= old + recs, &lwi);
- if (d->opd_syn_last_processed_id >= id)
+ if (atomic64_read(&d->opd_sync_processed_recs) >= old + recs)
break;
- if (d->opd_syn_last_processed_id != old) {
+ if (atomic64_read(&d->opd_sync_processed_recs) != last) {
/* some progress have been made,
* keep trying... */
continue;
}
/* block new processing (barrier>0 - few callers are possible */
- atomic_inc(&d->opd_syn_barrier);
+ atomic_inc(&d->opd_sync_barrier);
- CDEBUG(D_OTHER, "%s: %u in flight\n", d->opd_obd->obd_name,
- d->opd_syn_rpc_in_flight);
+ CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
+ atomic_read(&d->opd_sync_rpcs_in_flight));
/* wait till all-in-flight are replied, so executed by the target */
/* XXX: this is used by LFSCK at the moment, which doesn't require
* all the changes to be committed, but in general it'd be
* better to wait till commit */
- while (d->opd_syn_rpc_in_flight > 0) {
-
- old = d->opd_syn_rpc_in_flight;
+ while (atomic_read(&d->opd_sync_rpcs_in_flight) > 0) {
+ old = atomic_read(&d->opd_sync_rpcs_in_flight);
- expire = cfs_time_shift(obd_timeout);
- lwi = LWI_TIMEOUT(expire - cfs_time_current(),
+ lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
osp_sync_timeout, d);
- l_wait_event(d->opd_syn_barrier_waitq,
- d->opd_syn_rpc_in_flight == 0, &lwi);
+ l_wait_event(d->opd_sync_barrier_waitq,
+ atomic_read(&d->opd_sync_rpcs_in_flight) == 0,
+ &lwi);
- if (d->opd_syn_rpc_in_flight == 0)
+ if (atomic_read(&d->opd_sync_rpcs_in_flight) == 0)
break;
- if (d->opd_syn_rpc_in_flight != old) {
+ if (atomic_read(&d->opd_sync_rpcs_in_flight) != old) {
/* some progress have been made */
continue;
}
GOTO(out, rc = -ETIMEDOUT);
}
- CDEBUG(D_OTHER, "%s: done in %lu\n", d->opd_obd->obd_name,
- cfs_time_current() - start);
out:
/* resume normal processing (barrier=0) */
- atomic_dec(&d->opd_syn_barrier);
- __osp_sync_check_for_work(d);
+ atomic_dec(&d->opd_sync_barrier);
+ osp_sync_check_for_work(d);
+
+ CDEBUG(D_CACHE, "%s: done in %lld: rc = %d\n", d->opd_obd->obd_name,
+ ktime_get_seconds() - start, rc);
RETURN(rc);
}
.dt_trans_create = osp_trans_create,
.dt_trans_start = osp_trans_start,
.dt_trans_stop = osp_trans_stop,
+ .dt_trans_cb_add = osp_trans_cb_add,
};
/**
}
/**
+ * Determine if the lock needs to be cancelled
+ *
+ * Determine if the unused lock should be cancelled before replay, see
+ * (ldlm_cancel_no_wait_policy()). Currently, only inode bits lock exists
+ * between MDTs.
+ *
+ * \param[in] lock lock to be checked.
+ *
+ * \retval 1 if the lock needs to be cancelled before replay.
+ * \retval 0 if the lock does not need to be cancelled before
+ * replay.
+ */
+static int osp_cancel_weight(struct ldlm_lock *lock)
+{
+ if (lock->l_resource->lr_type != LDLM_IBITS)
+ RETURN(0);
+
+ RETURN(1);
+}
+
+/**
* Initialize OSP device according to the parameters in the configuration
* log \a cfg.
*
{
struct obd_device *obd;
struct obd_import *imp;
- class_uuid_t uuid;
char *src, *tgt, *mdt, *osdname = NULL;
int rc;
long idx;
ENTRY;
mutex_init(&osp->opd_async_requests_mutex);
+ INIT_LIST_HEAD(&osp->opd_async_updates);
+ init_rwsem(&osp->opd_async_updates_rwsem);
+ atomic_set(&osp->opd_async_updates_count, 0);
obd = class_name2obd(lustre_cfg_string(cfg, 0));
if (obd == NULL) {
strcat(osdname, "-osd");
CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
- if (osp->opd_connect_mdt) {
- struct client_obd *cli = &osp->opd_obd->u.cli;
-
- OBD_ALLOC(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
- if (!cli->cl_rpc_lock)
- GOTO(out_fini, rc = -ENOMEM);
- osp_init_rpc_lock(cli->cl_rpc_lock);
- }
+ osp_init_rpc_lock(osp);
osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
osp->opd_dt_dev.dd_ops = &osp_dt_ops;
GOTO(out_ref, rc);
}
- osp_lprocfs_init(osp);
+ osp_tunables_init(osp);
rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
if (!osp->opd_connect_mdt) {
/* Initialize last id from the storage - will be
* used in orphan cleanup. */
- rc = osp_last_used_init(env, osp);
- if (rc)
- GOTO(out_proc, rc);
-
+ if (!osp->opd_storage->dd_rdonly) {
+ rc = osp_last_used_init(env, osp);
+ if (rc)
+ GOTO(out_fid, rc);
+ }
/* Initialize precreation thread, it handles new
* connections as well. */
rc = osp_init_precreate(osp);
if (rc)
GOTO(out_last_used, rc);
+
+ /*
+ * Initialize synhronization mechanism taking
+ * care of propogating changes to OST in near
+ * transactional manner.
+ */
+ rc = osp_sync_init(env, osp);
+ if (rc < 0)
+ GOTO(out_precreat, rc);
+ } else {
+ osp->opd_got_disconnected = 1;
+ rc = osp_update_init(osp);
+ if (rc != 0)
+ GOTO(out_fid, rc);
}
- /*
- * Initialize synhronization mechanism taking
- * care of propogating changes to OST in near
- * transactional manner.
- */
- rc = osp_sync_init(env, osp);
+ rc = osp_init_statfs(osp);
if (rc)
GOTO(out_precreat, rc);
+ ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
+
/*
* Initiate connect to OST
*/
- ll_generate_random_uuid(uuid);
- class_uuid_unparse(uuid, &osp->opd_cluuid);
-
imp = obd->u.cli.cl_import;
rc = ptlrpc_init_import(imp);
RETURN(0);
out:
- /* stop sync thread */
- osp_sync_fini(osp);
+ if (!osp->opd_connect_mdt)
+ /* stop sync thread */
+ osp_sync_fini(osp);
out_precreat:
/* stop precreate thread */
if (!osp->opd_connect_mdt)
osp_precreate_fini(osp);
+ else
+ osp_update_fini(env, osp);
out_last_used:
if (!osp->opd_connect_mdt)
osp_last_used_fini(env, osp);
+out_fid:
+ obd_fid_fini(osp->opd_obd);
out_proc:
- ptlrpc_lprocfs_unregister_obd(obd);
- lprocfs_obd_cleanup(obd);
- if (osp->opd_symlink)
- lprocfs_remove(&osp->opd_symlink);
+ osp_tunables_fini(osp);
client_obd_cleanup(obd);
out_ref:
ptlrpcd_decref();
out_disconnect:
- if (osp->opd_connect_mdt) {
- struct client_obd *cli = &osp->opd_obd->u.cli;
- if (cli->cl_rpc_lock != NULL) {
- OBD_FREE_PTR(cli->cl_rpc_lock);
- cli->cl_rpc_lock = NULL;
- }
- }
obd_disconnect(osp->opd_storage_exp);
out_fini:
if (osdname)
struct lu_device *ld)
{
struct osp_device *osp = lu2osp_dev(ld);
- struct obd_import *imp;
int rc;
ENTRY;
if (osp->opd_async_requests != NULL) {
- out_destroy_update_req(osp->opd_async_requests);
+ osp_update_request_destroy(env, osp->opd_async_requests);
osp->opd_async_requests = NULL;
}
- if (osp->opd_storage_exp)
+ if (osp->opd_storage_exp) {
+ /* wait for the commit callbacks to complete */
+ wait_event(osp->opd_sync_waitq,
+ atomic_read(&osp->opd_commits_registered) == 0);
obd_disconnect(osp->opd_storage_exp);
-
- imp = osp->opd_obd->u.cli.cl_import;
-
- if (imp->imp_rq_pool) {
- ptlrpc_free_rq_pool(imp->imp_rq_pool);
- imp->imp_rq_pool = NULL;
}
- if (osp->opd_symlink)
- lprocfs_remove(&osp->opd_symlink);
-
LASSERT(osp->opd_obd);
- ptlrpc_lprocfs_unregister_obd(osp->opd_obd);
- lprocfs_obd_cleanup(osp->opd_obd);
-
- if (osp->opd_connect_mdt) {
- struct client_obd *cli = &osp->opd_obd->u.cli;
- if (cli->cl_rpc_lock != NULL) {
- OBD_FREE_PTR(cli->cl_rpc_lock);
- cli->cl_rpc_lock = NULL;
- }
- }
rc = client_obd_cleanup(osp->opd_obd);
if (rc != 0) {
RETURN(ERR_PTR(rc));
}
+ osp_tunables_fini(osp);
+
ptlrpcd_decref();
RETURN(NULL);
*ocd = *data;
imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
+ imp->imp_connect_flags2_orig = ocd->ocd_connect_flags2;
ocd->ocd_version = LUSTRE_VERSION_CODE;
ocd->ocd_index = data->ocd_index;
- imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
rc = ptlrpc_connect_import(imp);
if (rc) {
CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
GOTO(out, rc);
+ } else {
+ osp->opd_obd->u.cli.cl_seq->lcs_exp =
+ class_export_get(osp->opd_exp);
}
ptlrpc_pinger_add_import(imp);
* \retval negative negative errno if statfs failed.
*/
static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
- struct obd_statfs *osfs, __u64 unused, __u32 flags)
+ struct obd_statfs *osfs, time64_t unused, __u32 flags)
{
struct obd_statfs *msfs;
struct ptlrpc_request *req;
}
/**
- * Prepare fid client.
- *
- * This function prepares the FID client for the OSP. It will check and assign
- * the export (to MDT0) for its FID client, so OSP can allocate super sequence
- * or lookup sequence in FLDB of MDT0.
- *
- * \param[in] osp OSP device
- */
-static void osp_prepare_fid_client(struct osp_device *osp)
-{
- LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
- if (osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL)
- return;
-
- LASSERT(osp->opd_exp != NULL);
- osp->opd_obd->u.cli.cl_seq->lcs_exp =
- class_export_get(osp->opd_exp);
-}
-
-/**
* Implementation of obd_ops::o_import_event
*
* This function is called when some related import event happens. It will
enum obd_import_event event)
{
struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
+ int rc;
switch (event) {
case IMP_EVENT_DISCON:
break;
case IMP_EVENT_INACTIVE:
d->opd_imp_active = 0;
+ d->opd_imp_connected = 0;
+ d->opd_obd->obd_inactive = 1;
if (d->opd_connect_mdt)
break;
-
if (d->opd_pre != NULL) {
- osp_pre_update_status(d, -ENODEV);
+ /* Import is invalid, we can`t get stripes so
+ * wakeup waiters */
+ rc = imp->imp_deactive ? -ESHUTDOWN : -ENODEV;
+ osp_pre_update_status(d, rc);
wake_up(&d->opd_pre_waitq);
}
case IMP_EVENT_ACTIVE:
d->opd_imp_active = 1;
- osp_prepare_fid_client(d);
if (d->opd_got_disconnected)
d->opd_new_connection = 1;
d->opd_imp_connected = 1;
d->opd_imp_seen_connected = 1;
+ d->opd_obd->obd_inactive = 0;
+ wake_up(&d->opd_pre_waitq);
if (d->opd_connect_mdt)
break;
- if (d->opd_pre != NULL)
- wake_up(&d->opd_pre_waitq);
-
- __osp_sync_check_for_work(d);
+ osp_sync_check_for_work(d);
CDEBUG(D_HA, "got connected\n");
break;
case IMP_EVENT_INVALIDATE:
+ if (d->opd_connect_mdt)
+ osp_invalidate_request(d);
+
if (obd->obd_namespace == NULL)
break;
ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
* \retval negative negative errno if the ioctl handling failed.
*/
static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
- void *karg, void *uarg)
+ void *karg, void __user *uarg)
{
struct obd_device *obd = exp->exp_obd;
struct osp_device *d;
rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
data->ioc_offset);
break;
- case OBD_IOC_PING_TARGET:
- rc = ptlrpc_obd_ping(obd);
- break;
default:
CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
cmd, current_comm());
return rc;
}
+
/**
* Implementation of obd_ops::o_get_info
*
* \param[in] key the key
* \param[out] vallen length of \a val
* \param[out] val holds the value returned by the key
- * \param[in] unused necessary for the interface but unused
*
* \retval 0 0 if getting information succeeded.
* \retval negative negative errno if getting information failed.
*/
static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
- __u32 keylen, void *key, __u32 *vallen, void *val,
- struct lov_stripe_md *unused)
+ __u32 keylen, void *key, __u32 *vallen, void *val)
{
int rc = -EINVAL;
RETURN(rc);
}
+static int osp_obd_set_info_async(const struct lu_env *env,
+ struct obd_export *exp,
+ u32 keylen, void *key,
+ u32 vallen, void *val,
+ struct ptlrpc_request_set *set)
+{
+ struct obd_device *obd = exp->exp_obd;
+ struct obd_import *imp = obd->u.cli.cl_import;
+ struct osp_device *osp;
+ struct ptlrpc_request *req;
+ char *tmp;
+ int rc;
+
+ if (KEY_IS(KEY_SPTLRPC_CONF)) {
+ sptlrpc_conf_client_adapt(exp->exp_obd);
+ RETURN(0);
+ }
+
+ LASSERT(set != NULL);
+ if (!obd->obd_set_up || obd->obd_stopping)
+ RETURN(-EAGAIN);
+ osp = lu2osp_dev(obd->obd_lu_dev);
+
+ req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
+ if (req == NULL)
+ RETURN(-ENOMEM);
+
+ req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
+ RCL_CLIENT, keylen);
+ req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
+ RCL_CLIENT, vallen);
+ if (osp->opd_connect_mdt)
+ rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SET_INFO);
+ else
+ rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
+ if (rc) {
+ ptlrpc_request_free(req);
+ RETURN(rc);
+ }
+
+ tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
+ memcpy(tmp, key, keylen);
+ tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
+ memcpy(tmp, val, vallen);
+
+ ptlrpc_request_set_replen(req);
+ ptlrpc_set_add_req(set, req);
+ ptlrpc_check_set(NULL, set);
+
+ RETURN(0);
+}
+
/**
* Implementation of obd_ops: o_fid_alloc
*
* requested from seq meta server
* \retval negative negative errno if FID allocation failed.
*/
-int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
- struct lu_fid *fid, struct md_op_data *unused)
+static int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
+ struct lu_fid *fid, struct md_op_data *unused)
{
struct client_obd *cli = &exp->exp_obd->u.cli;
struct osp_device *osp = lu2osp_dev(exp->exp_obd->obd_lu_dev);
LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
/* Sigh, fid client is not ready yet */
- if (osp->opd_obd->u.cli.cl_seq->lcs_exp == NULL)
- RETURN(-ENOTCONN);
+ LASSERT(osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
RETURN(seq_client_alloc_fid(env, seq, fid));
}
LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
struct lu_context_key osp_txn_key = {
- .lct_tags = LCT_OSP_THREAD | LCT_TX_HANDLE,
+ .lct_tags = LCT_OSP_THREAD,
.lct_init = osp_txn_key_init,
.lct_fini = osp_txn_key_fini
};
.o_connect = osp_obd_connect,
.o_disconnect = osp_obd_disconnect,
.o_get_info = osp_obd_get_info,
+ .o_set_info_async = osp_obd_set_info_async,
.o_import_event = osp_import_event,
.o_iocontrol = osp_iocontrol,
.o_statfs = osp_obd_statfs,
.o_fid_alloc = osp_fid_alloc,
};
-struct llog_operations osp_mds_ost_orig_logops;
+static struct obd_type *sym;
/**
* Initialize OSP module.
* \retval 0 0 if initialization succeeds.
* \retval negative negative errno if initialization failed.
*/
-static int __init osp_mod_init(void)
+static int __init osp_init(void)
{
- struct obd_type *type;
int rc;
rc = lu_kmem_init(osp_caches);
if (rc)
return rc;
-
- rc = class_register_type(&osp_obd_device_ops, NULL, true, NULL,
-#ifndef HAVE_ONLY_PROCFS_SEQ
- NULL,
-#endif
+ rc = class_register_type(&osp_obd_device_ops, NULL, false, NULL,
LUSTRE_OSP_NAME, &osp_device_type);
if (rc != 0) {
lu_kmem_fini(osp_caches);
return rc;
}
- rc = class_register_type(&lwp_obd_device_ops, NULL, true, NULL,
-#ifndef HAVE_ONLY_PROCFS_SEQ
- NULL,
-#endif
+ rc = class_register_type(&lwp_obd_device_ops, NULL, false, NULL,
LUSTRE_LWP_NAME, &lwp_device_type);
if (rc != 0) {
class_unregister_type(LUSTRE_OSP_NAME);
return rc;
}
- /* Note: add_rec/delcare_add_rec will be only used by catalogs */
- osp_mds_ost_orig_logops = llog_osd_ops;
- osp_mds_ost_orig_logops.lop_add = llog_cat_add_rec;
- osp_mds_ost_orig_logops.lop_declare_add = llog_cat_declare_add_rec;
-
- /* create "osc" entry in procfs for compatibility purposes */
- type = class_search_type(LUSTRE_OSC_NAME);
- if (type != NULL && type->typ_procroot != NULL)
- return rc;
-
- type = class_search_type(LUSTRE_OSP_NAME);
- type->typ_procsym = lprocfs_seq_register("osc", proc_lustre_root,
- NULL, NULL);
- if (IS_ERR(type->typ_procsym)) {
- CERROR("osp: can't create compat entry \"osc\": %d\n",
- (int) PTR_ERR(type->typ_procsym));
- type->typ_procsym = NULL;
+ /* create "osc" entry for compatibility purposes */
+ sym = class_add_symlinks(LUSTRE_OSC_NAME, false);
+ if (IS_ERR(sym)) {
+ rc = PTR_ERR(sym);
+ /* does real "osc" already exist ? */
+ if (rc == -EEXIST)
+ rc = 0;
}
+
return rc;
}
* This callback is called when kernel unloads OSP module from memory, and
* it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
*/
-static void __exit osp_mod_exit(void)
+static void __exit osp_exit(void)
{
+ if (!IS_ERR_OR_NULL(sym))
+ kobject_put(&sym->typ_kobj);
+
class_unregister_type(LUSTRE_LWP_NAME);
class_unregister_type(LUSTRE_OSP_NAME);
lu_kmem_fini(osp_caches);
}
-MODULE_AUTHOR("Intel, Inc. <http://www.intel.com/>");
-MODULE_DESCRIPTION("Lustre OST Proxy Device ("LUSTRE_OSP_NAME")");
+MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
+MODULE_DESCRIPTION("Lustre OSD Storage Proxy ("LUSTRE_OSP_NAME")");
+MODULE_VERSION(LUSTRE_VERSION_STRING);
MODULE_LICENSE("GPL");
-cfs_module(osp, LUSTRE_VERSION_STRING, osp_mod_init, osp_mod_exit);
+module_init(osp_init);
+module_exit(osp_exit);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff