* Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2011, 2012, Intel, Inc.
+ * Copyright (c) 2012, 2014, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*
* lustre/osp/osp_dev.c
*
- * Lustre OST Proxy Device
- *
* 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.
*/
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
#define DEBUG_SUBSYSTEM S_MDS
+#include <linux/kthread.h>
#include <obd_class.h>
+#include <lustre_ioctl.h>
#include <lustre_param.h>
+#include <lustre_log.h>
#include "osp_internal.h"
/* Slab for OSP object allocation */
-cfs_mem_cache_t *osp_object_kmem;
+struct kmem_cache *osp_object_kmem;
static struct lu_kmem_descr osp_caches[] = {
{
}
};
-struct lu_object *osp_object_alloc(const struct lu_env *env,
- const struct lu_object_header *hdr,
- struct lu_device *d)
+/**
+ * Implementation of lu_device_operations::ldo_object_alloc
+ *
+ * Allocates an OSP object in memory, whose FID is on the remote target.
+ *
+ * \param[in] env execution environment
+ * \param[in] hdr The header of the object stack. If it is NULL, it
+ * means the object is not built from top device, i.e.
+ * it is a sub-stripe object of striped directory or
+ * an OST object.
+ * \param[in] d OSP device
+ *
+ * \retval object object being created if the creation succeed.
+ * \retval NULL NULL if the creation failed.
+ */
+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;
+ struct lu_object_header *h = NULL;
struct osp_object *o;
struct lu_object *l;
- LASSERT(hdr == NULL);
-
- OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, CFS_ALLOC_IO);
+ OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
if (o != NULL) {
l = &o->opo_obj.do_lu;
- h = &o->opo_header;
- lu_object_header_init(h);
- dt_object_init(&o->opo_obj, h, d);
- lu_object_add_top(h, l);
+ /* If hdr is NULL, it means the object is not built
+ * from the top dev(MDT/OST), usually it happens when
+ * building striped object, like data object on MDT or
+ * striped object for directory */
+ if (hdr == NULL) {
+ h = &o->opo_header;
+ lu_object_header_init(h);
+ dt_object_init(&o->opo_obj, h, d);
+ lu_object_add_top(h, l);
+ } else {
+ dt_object_init(&o->opo_obj, h, d);
+ }
l->lo_ops = &osp_lu_obj_ops;
}
}
-/* Update opd_last_used_id along with checking for gap in objid sequence */
-void osp_update_last_id(struct osp_device *d, obd_id objid)
+/**
+ * Find or create the local object
+ *
+ * Finds or creates the local file referenced by \a reg_id and return the
+ * attributes of the local file.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ * \param[out] attr attributes of the object
+ * \param[in] reg_id the local object ID of the file. It will be used
+ * to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
+ * to identify the object.
+ *
+ * \retval object object(dt_object) found or created
+ * \retval ERR_PTR(errno) ERR_PTR(errno) if not get the object.
+ */
+static struct dt_object
+*osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
+ struct lu_attr *attr, __u32 reg_id)
{
- /*
- * we might have lost precreated objects due to VBR and precreate
- * orphans, the gap in objid can be calculated properly only here
- */
- if (objid > le64_to_cpu(d->opd_last_used_id)) {
- if (objid - le64_to_cpu(d->opd_last_used_id) > 1) {
- d->opd_gap_start = le64_to_cpu(d->opd_last_used_id) + 1;
- d->opd_gap_count = objid - d->opd_gap_start;
- CDEBUG(D_HA, "Gap in objids: %d, start = %llu\n",
- d->opd_gap_count, d->opd_gap_start);
- }
- d->opd_last_used_id = cpu_to_le64(objid);
+ struct osp_thread_info *osi = osp_env_info(env);
+ struct dt_object_format dof = { 0 };
+ struct dt_object *dto;
+ int rc;
+ ENTRY;
+
+ lu_local_obj_fid(&osi->osi_fid, reg_id);
+ attr->la_valid = LA_MODE;
+ attr->la_mode = S_IFREG | 0644;
+ dof.dof_type = DFT_REGULAR;
+ /* Find or create the local object by osi_fid. */
+ dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
+ &dof, attr);
+ if (IS_ERR(dto))
+ RETURN(dto);
+
+ /* Get attributes of the local object. */
+ 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);
+ RETURN(ERR_PTR(rc));
}
+ RETURN(dto);
}
-static int osp_last_used_init(const struct lu_env *env, struct osp_device *m)
+/**
+ * Write data buffer to a local file object.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ * \param[in] dt_obj object written to
+ * \param[in] buf buffer containing byte array and length
+ * \param[in] offset write offset in the object in bytes
+ *
+ * \retval 0 0 if write succeed
+ * \retval -EFAULT -EFAULT if only part of buffer is written.
+ * \retval negative other negative errno if write failed.
+ */
+static int osp_write_local_file(const struct lu_env *env,
+ struct osp_device *osp,
+ struct dt_object *dt_obj,
+ struct lu_buf *buf,
+ loff_t offset)
{
- struct osp_thread_info *osi = osp_env_info(env);
- struct dt_object_format dof = { 0 };
- struct dt_object *o;
- int rc;
-
- ENTRY;
+ struct thandle *th;
+ int rc;
- osi->osi_attr.la_valid = LA_MODE;
- osi->osi_attr.la_mode = S_IFREG | 0644;
- lu_local_obj_fid(&osi->osi_fid, MDD_LOV_OBJ_OID);
- dof.dof_type = DFT_REGULAR;
- o = dt_find_or_create(env, m->opd_storage, &osi->osi_fid, &dof,
- &osi->osi_attr);
- if (IS_ERR(o))
- RETURN(PTR_ERR(o));
+ th = dt_trans_create(env, osp->opd_storage);
+ if (IS_ERR(th))
+ RETURN(PTR_ERR(th));
- rc = dt_attr_get(env, o, &osi->osi_attr, NULL);
+ rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
if (rc)
GOTO(out, rc);
+ rc = dt_trans_start_local(env, osp->opd_storage, th);
+ if (rc)
+ GOTO(out, rc);
+
+ rc = dt_record_write(env, dt_obj, buf, &offset, th);
+out:
+ dt_trans_stop(env, osp->opd_storage, th);
+ RETURN(rc);
+}
+
+/**
+ * Initialize last ID object.
+ *
+ * This function initializes the LAST_ID file, which stores the current last
+ * used id of data objects. The MDT will use the last used id and the last_seq
+ * (\see osp_init_last_seq()) to synchronize the precreate object cache with
+ * OSTs.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ *
+ * \retval 0 0 if initialization succeed
+ * \retval negative negative errno if initialization failed
+ */
+static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
+{
+ struct osp_thread_info *osi = osp_env_info(env);
+ struct lu_fid *fid = &osp->opd_last_used_fid;
+ struct dt_object *dto;
+ int rc = -EFAULT;
+ ENTRY;
+
+ dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
+ MDD_LOV_OBJ_OID);
+ if (IS_ERR(dto))
+ RETURN(PTR_ERR(dto));
/* object will be released in device cleanup path */
- m->opd_last_used_file = o;
+ 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);
+ rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
+ if (rc != 0 && rc != -EFAULT)
+ GOTO(out, rc);
+ }
- if (osi->osi_attr.la_size >= sizeof(osi->osi_id) *
- (m->opd_index + 1)) {
- osp_objid_buf_prep(osi, m, m->opd_index);
- rc = dt_record_read(env, o, &osi->osi_lb, &osi->osi_off);
+ if (rc == -EFAULT) { /* fresh LAST_ID */
+ 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)
GOTO(out, rc);
- } else {
- /* reset value to 0, just to make sure and change file's size */
- struct thandle *th;
+ }
+ osp->opd_last_used_oid_file = dto;
+ RETURN(0);
+out:
+ /* 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);
+ osp->opd_last_used_oid_file = NULL;
+ RETURN(rc);
+}
- m->opd_last_used_id = 0;
- osp_objid_buf_prep(osi, m, m->opd_index);
+/**
+ * Initialize last sequence object.
+ *
+ * This function initializes the LAST_SEQ file in the local OSD, which stores
+ * the current last used sequence of data objects. The MDT will use the last
+ * sequence and last id (\see osp_init_last_objid()) to synchronize the
+ * precreate object cache with OSTs.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ *
+ * \retval 0 0 if initialization succeed
+ * \retval negative negative errno if initialization failed
+ */
+static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
+{
+ struct osp_thread_info *osi = osp_env_info(env);
+ struct lu_fid *fid = &osp->opd_last_used_fid;
+ struct dt_object *dto;
+ int rc = -EFAULT;
+ ENTRY;
- th = dt_trans_create(env, m->opd_storage);
- if (IS_ERR(th))
- GOTO(out, rc = PTR_ERR(th));
+ dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
+ MDD_LOV_OBJ_OSEQ);
+ if (IS_ERR(dto))
+ RETURN(PTR_ERR(dto));
- rc = dt_declare_record_write(env, m->opd_last_used_file,
- osi->osi_lb.lb_len, osi->osi_off,
- th);
- if (rc) {
- dt_trans_stop(env, m->opd_storage, th);
+ /* 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);
+ rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
+ if (rc != 0 && rc != -EFAULT)
GOTO(out, rc);
- }
+ }
+
+ 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);
+out:
+ /* 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);
+ osp->opd_last_used_seq_file = NULL;
+ RETURN(rc);
+}
- rc = dt_trans_start_local(env, m->opd_storage, th);
+/**
+ * Initialize last OID and sequence object.
+ *
+ * If the MDT is just upgraded to 2.4 from the lower version, where the
+ * LAST_SEQ file does not exist, the file will be created and IDIF sequence
+ * will be written into the file.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ *
+ * \retval 0 0 if initialization succeed
+ * \retval negative negative error if initialization failed
+ */
+static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
+{
+ struct osp_thread_info *osi = osp_env_info(env);
+ int rc;
+ ENTRY;
+
+ fid_zero(&osp->opd_last_used_fid);
+ rc = osp_init_last_objid(env, osp);
+ if (rc < 0) {
+ CERROR("%s: Can not get ids %d from old objid!\n",
+ osp->opd_obd->obd_name, rc);
+ RETURN(rc);
+ }
+
+ rc = osp_init_last_seq(env, osp);
+ if (rc < 0) {
+ CERROR("%s: Can not get ids %d from old objid!\n",
+ osp->opd_obd->obd_name, rc);
+ GOTO(out, rc);
+ }
+
+ if (fid_oid(&osp->opd_last_used_fid) != 0 &&
+ fid_seq(&osp->opd_last_used_fid) == 0) {
+ /* Just upgrade from the old version,
+ * set the seq to be IDIF */
+ osp->opd_last_used_fid.f_seq =
+ fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
+ osp->opd_index);
+ osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
+ &osp->opd_last_used_fid.f_seq,
+ osp->opd_index);
+ rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
+ &osi->osi_lb, osi->osi_off);
if (rc) {
- dt_trans_stop(env, m->opd_storage, th);
+ CERROR("%s : Can not write seq file: rc = %d\n",
+ osp->opd_obd->obd_name, rc);
GOTO(out, rc);
}
+ }
- rc = dt_record_write(env, m->opd_last_used_file, &osi->osi_lb,
- &osi->osi_off, th);
- dt_trans_stop(env, m->opd_storage, th);
- if (rc)
- GOTO(out, rc);
+ if (!fid_is_zero(&osp->opd_last_used_fid) &&
+ !fid_is_sane(&osp->opd_last_used_fid)) {
+ CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
+ PFID(&osp->opd_last_used_fid));
+ GOTO(out, rc = -EINVAL);
}
- RETURN(0);
+
+ CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
+ osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
out:
- CERROR("%s: can't initialize lov_objid: %d\n",
- m->opd_obd->obd_name, rc);
- lu_object_put(env, &o->do_lu);
- m->opd_last_used_file = NULL;
- return rc;
+ if (rc != 0) {
+ if (osp->opd_last_used_oid_file != NULL) {
+ lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
+ 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);
+ osp->opd_last_used_seq_file = NULL;
+ }
+ }
+
+ RETURN(rc);
}
-static void osp_last_used_fini(const struct lu_env *env, struct osp_device *d)
+/**
+ * Release the last sequence and OID file objects in OSP device.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ */
+static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
{
- if (d->opd_last_used_file != NULL) {
- lu_object_put(env, &d->opd_last_used_file->do_lu);
- d->opd_last_used_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);
+ 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);
+ osp->opd_last_used_seq_file = NULL;
}
}
-int osp_disconnect(struct osp_device *d)
+/**
+ * Disconnects the connection between OSP and its correspondent MDT or OST, and
+ * the import will be marked as inactive. It will only be called during OSP
+ * cleanup process.
+ *
+ * \param[in] d OSP device being disconnected
+ *
+ * \retval 0 0 if disconnection succeed
+ * \retval negative negative errno if disconnection failed
+ */
+static int osp_disconnect(struct osp_device *d)
{
struct obd_import *imp;
int rc = 0;
* of the cleanup RPCs fails (e.g. ldlm cancel, etc). We don't
* fully deactivate the import, or that would drop all requests. */
LASSERT(imp != NULL);
- cfs_spin_lock(&imp->imp_lock);
+ spin_lock(&imp->imp_lock);
imp->imp_deactive = 1;
- cfs_spin_unlock(&imp->imp_lock);
+ spin_unlock(&imp->imp_lock);
ptlrpc_deactivate_import(imp);
(void)ptlrpc_pinger_del_import(imp);
rc = ptlrpc_disconnect_import(imp, 0);
- if (rc && rc != -ETIMEDOUT)
+ if (rc != 0)
CERROR("%s: can't disconnect: rc = %d\n",
d->opd_obd->obd_name, rc);
RETURN(rc);
}
-static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
+/**
+ * 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)
{
- int rc = 0;
+ struct l_wait_info lwi = { 0 };
+ struct task_struct *task;
+
ENTRY;
- if (is_osp_on_ost(d->opd_obd->obd_name)) {
- rc = osp_disconnect(d);
+ LASSERT(osp->opd_connect_mdt);
+
+ 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;
+
+ /* 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(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.
+ *
+ * \param[in] env execution environment.
+ * \param[in] d OSP device being disconnected.
+ *
+ * \retval 0 0 if cleanup succeed
+ * \retval negative negative errno if cleanup failed
+ */
+static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
+{
+ int rc = 0;
+ ENTRY;
+
LASSERT(env);
- /* release last_used file */
- osp_last_used_fini(env, d);
rc = osp_disconnect(d);
- /* stop precreate thread */
- osp_precreate_fini(d);
+ if (!d->opd_connect_mdt) {
+ /* stop sync thread */
+ osp_sync_fini(d);
+
+ /* stop precreate thread */
+ osp_precreate_fini(d);
- /* stop sync thread */
- osp_sync_fini(d);
+ /* release last_used file */
+ osp_last_used_fini(env, d);
+ }
+
+ obd_fid_fini(d->opd_obd);
RETURN(rc);
}
+/**
+ * Implementation of osp_lu_ops::ldo_process_config
+ *
+ * This function processes config log records in OSP layer. It is usually
+ * called from the top layer of MDT stack, and goes through the stack by calling
+ * ldo_process_config of next layer.
+ *
+ * \param[in] env execution environment
+ * \param[in] dev lu_device of OSP
+ * \param[in] lcfg config log
+ *
+ * \retval 0 0 if the config log record is executed correctly.
+ * \retval negative negative errno if the record execution fails.
+ */
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 lprocfs_static_vars lvars = { 0 };
+ struct obd_device *obd = d->opd_obd;
int rc;
ENTRY;
switch (lcfg->lcfg_command) {
+ case LCFG_PRE_CLEANUP:
+ rc = osp_disconnect(d);
+ osp_update_fini(env, d);
+ break;
case LCFG_CLEANUP:
- if (!is_osp_on_ost(d->opd_obd->obd_name))
- lu_dev_del_linkage(dev->ld_site, dev);
+ lu_dev_del_linkage(dev->ld_site, dev);
rc = osp_shutdown(env, d);
break;
case LCFG_PARAM:
- lprocfs_osp_init_vars(&lvars);
-
- LASSERT(d->opd_obd);
- rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
- lcfg, d->opd_obd);
+ LASSERT(obd);
+ rc = class_process_proc_param(d->opd_connect_mdt ?
+ PARAM_OSP : PARAM_OSC,
+ obd->obd_vars, lcfg, obd);
if (rc > 0)
rc = 0;
if (rc == -ENOSYS) {
RETURN(rc);
}
+/**
+ * Implementation of osp_lu_ops::ldo_recovery_complete
+ *
+ * This function is called after recovery is finished, and OSP layer
+ * will wake up precreate thread here.
+ *
+ * \param[in] env execution environment
+ * \param[in] dev lu_device of OSP
+ *
+ * \retval 0 0 unconditionally
+ */
static int osp_recovery_complete(const struct lu_env *env,
struct lu_device *dev)
{
struct osp_device *osp = lu2osp_dev(dev);
- int rc = 0;
ENTRY;
osp->opd_recovery_completed = 1;
- cfs_waitq_signal(&osp->opd_pre_waitq);
- RETURN(rc);
+
+ if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
+ wake_up(&osp->opd_pre_waitq);
+
+ RETURN(0);
}
const struct lu_device_operations osp_lu_ops = {
};
/**
- * provides with statfs from corresponded OST
+ * Implementation of dt_device_operations::dt_statfs
+ *
+ * This function provides statfs status (for precreation) from
+ * corresponding OST. Note: this function only retrieves the status
+ * from the OSP device, and the real statfs RPC happens inside
+ * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
+ * not need to retrieve statfs data for now.
*
+ * \param[in] env execution environment.
+ * \param[in] dev dt_device of OSP.
+ * \param[out] sfs holds the retrieved statfs data.
+ *
+ * \retval 0 0 statfs data was retrieved successfully or
+ * retrieval was not needed
+ * \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)
if (unlikely(d->opd_imp_active == 0))
RETURN(-ENOTCONN);
+ if (d->opd_pre == NULL)
+ RETURN(0);
+
/* return recently updated data */
*sfs = d->opd_statfs;
* layer above osp (usually lod) can use ffree to estimate
* how many objects are available for immediate creation
*/
- cfs_spin_lock(&d->opd_pre_lock);
- sfs->os_fprecreated = d->opd_pre_last_created - d->opd_pre_used_id;
+ 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 -= d->opd_pre_reserved;
- cfs_spin_unlock(&d->opd_pre_lock);
+ spin_unlock(&d->opd_pre_lock);
- LASSERT(sfs->os_fprecreated <= OST_MAX_PRECREATE);
+ 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,
RETURN(0);
}
+static int osp_sync_timeout(void *data)
+{
+ return 1;
+}
+
+/**
+ * Implementation of dt_device_operations::dt_sync
+ *
+ * This function synchronizes the OSP cache to the remote target. It wakes
+ * up unlink log threads and sends out unlink records to the remote OST.
+ *
+ * \param[in] env execution environment
+ * \param[in] dev dt_device of OSP
+ *
+ * \retval 0 0 if synchronization succeeds
+ * \retval negative negative errno if synchronization fails
+ */
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();
ENTRY;
- /*
- * XXX: wake up sync thread, command it to start flushing asap?
- */
+ /* No Sync between MDTs yet. */
+ if (d->opd_connect_mdt)
+ RETURN(0);
- RETURN(0);
+ 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: async updates %d\n", d->opd_obd->obd_name,
+ atomic_read(&d->opd_async_updates_count));
+
+ /* make sure the connection is fine */
+ expire = cfs_time_shift(obd_timeout);
+ lwi = LWI_TIMEOUT(expire - cfs_time_current(), osp_sync_timeout, d);
+ rc = l_wait_event(d->opd_syn_barrier_waitq,
+ atomic_read(&d->opd_async_updates_count) == 0,
+ &lwi);
+ up_write(&d->opd_async_updates_rwsem);
+ if (rc != 0)
+ GOTO(out, rc);
+
+ CDEBUG(D_CACHE, "%s: id: used %lu, processed %lu\n",
+ d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
+
+ /* wait till all-in-line are processed */
+ while (d->opd_syn_last_processed_id < id) {
+
+ old = d->opd_syn_last_processed_id;
+
+ /* make sure the connection is fine */
+ expire = cfs_time_shift(obd_timeout);
+ lwi = LWI_TIMEOUT(expire - cfs_time_current(),
+ osp_sync_timeout, d);
+ l_wait_event(d->opd_syn_barrier_waitq,
+ d->opd_syn_last_processed_id >= id,
+ &lwi);
+
+ if (d->opd_syn_last_processed_id >= id)
+ break;
+
+ if (d->opd_syn_last_processed_id != old) {
+ /* some progress have been made,
+ * keep trying... */
+ continue;
+ }
+
+ /* no changes and expired, something is wrong */
+ GOTO(out, rc = -ETIMEDOUT);
+ }
+
+ /* block new processing (barrier>0 - few callers are possible */
+ atomic_inc(&d->opd_syn_barrier);
+
+ CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
+ d->opd_syn_rpc_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;
+
+ expire = cfs_time_shift(obd_timeout);
+ lwi = LWI_TIMEOUT(expire - cfs_time_current(),
+ osp_sync_timeout, d);
+ l_wait_event(d->opd_syn_barrier_waitq,
+ d->opd_syn_rpc_in_flight == 0, &lwi);
+
+ if (d->opd_syn_rpc_in_flight == 0)
+ break;
+
+ if (d->opd_syn_rpc_in_flight != old) {
+ /* some progress have been made */
+ continue;
+ }
+
+ /* no changes and expired, something is wrong */
+ GOTO(out, rc = -ETIMEDOUT);
+ }
+
+out:
+ /* resume normal processing (barrier=0) */
+ atomic_dec(&d->opd_syn_barrier);
+ __osp_sync_check_for_work(d);
+
+ CDEBUG(D_CACHE, "%s: done in %lu: rc = %d\n", d->opd_obd->obd_name,
+ cfs_time_current() - start, rc);
+
+ RETURN(rc);
}
const struct dt_device_operations osp_dt_ops = {
- .dt_statfs = osp_statfs,
- .dt_sync = osp_sync,
+ .dt_statfs = osp_statfs,
+ .dt_sync = osp_sync,
+ .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,
};
-static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *m,
+/**
+ * Connect OSP to local OSD.
+ *
+ * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
+ * OSP needs to access the local OSD to store some information. For example,
+ * during precreate, it needs to update last used OID and sequence file
+ * (LAST_SEQ) in local OSD.
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ * \param[in] nextdev the name of local OSD
+ *
+ * \retval 0 0 connection succeeded
+ * \retval negative negative errno connection failed
+ */
+static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
const char *nextdev)
{
struct obd_connect_data *data = NULL;
ENTRY;
- LASSERT(m->opd_storage_exp == NULL);
+ LASSERT(osp->opd_storage_exp == NULL);
OBD_ALLOC_PTR(data);
if (data == NULL)
obd = class_name2obd(nextdev);
if (obd == NULL) {
CERROR("%s: can't locate next device: %s\n",
- m->opd_obd->obd_name, nextdev);
+ osp->opd_obd->obd_name, nextdev);
GOTO(out, rc = -ENOTCONN);
}
- rc = obd_connect(env, &m->opd_storage_exp, obd, &obd->obd_uuid, data,
+ rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
NULL);
if (rc) {
CERROR("%s: cannot connect to next dev %s: rc = %d\n",
- m->opd_obd->obd_name, nextdev, rc);
+ osp->opd_obd->obd_name, nextdev, rc);
GOTO(out, rc);
}
- m->opd_dt_dev.dd_lu_dev.ld_site =
- m->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
- LASSERT(m->opd_dt_dev.dd_lu_dev.ld_site);
- m->opd_storage = lu2dt_dev(m->opd_storage_exp->exp_obd->obd_lu_dev);
+ osp->opd_dt_dev.dd_lu_dev.ld_site =
+ osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
+ LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
+ osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
out:
OBD_FREE_PTR(data);
RETURN(rc);
}
-static int osp_init0(const struct lu_env *env, struct osp_device *m,
+/**
+ * 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.
+ *
+ * Reconstruct the local device name from the configuration profile, and
+ * initialize necessary threads and structures according to the OSP type
+ * (MDT or OST).
+ *
+ * Since there is no record in the MDT configuration for the local disk
+ * device, we have to extract this from elsewhere in the profile.
+ * The only information we get at setup is from the OSC records:
+ * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
+ *
+ * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
+ * so, we need to reconstruct the name of the underlying OSD from this:
+ * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
+ *
+ * \param[in] env execution environment
+ * \param[in] osp OSP device
+ * \param[in] ldt lu device type of OSP
+ * \param[in] cfg configuration log
+ *
+ * \retval 0 0 if OSP initialization succeeded.
+ * \retval negative negative errno if OSP initialization failed.
+ */
+static int osp_init0(const struct lu_env *env, struct osp_device *osp,
struct lu_device_type *ldt, struct lustre_cfg *cfg)
{
- struct obd_device *obd;
- struct obd_import *imp;
- class_uuid_t uuid;
- char *src, *ost, *mdt, *osdname = NULL;
- int rc, idx;
+ 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) {
CERROR("Cannot find obd with name %s\n",
lustre_cfg_string(cfg, 0));
RETURN(-ENODEV);
}
- m->opd_obd = obd;
-
- /* There is no record in the MDT configuration for the local disk
- * device, so we have to extract this from elsewhere in the profile.
- * The only information we get at setup is from the OSC records:
- * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
- * Note that 1.8 generated configs are missing the -MDTxxxx part.
- * We need to reconstruct the name of the underlying OSD from this:
- * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd". We
- * also need to determine the OST index from this - will be used
- * to calculate the offset in shared lov_objids file later */
+ osp->opd_obd = obd;
src = lustre_cfg_string(cfg, 0);
if (src == NULL)
RETURN(-EINVAL);
- ost = strstr(src, "-OST");
- if (ost == NULL)
+ tgt = strrchr(src, '-');
+ if (tgt == NULL) {
+ CERROR("%s: invalid target name %s: rc = %d\n",
+ osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
+ -EINVAL);
RETURN(-EINVAL);
-
- idx = simple_strtol(ost + 4, &mdt, 16);
- if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
- CERROR("%s: invalid OST index in '%s'\n", obd->obd_name, src);
- GOTO(out_fini, rc = -EINVAL);
}
- m->opd_index = idx;
- idx = ost - src;
+ if (strncmp(tgt, "-osc", 4) == 0) {
+ /* Old OSC name fsname-OSTXXXX-osc */
+ for (tgt--; tgt > src && *tgt != '-'; tgt--)
+ ;
+ if (tgt == src) {
+ CERROR("%s: invalid target name %s: rc = %d\n",
+ osp->opd_obd->obd_name,
+ lustre_cfg_string(cfg, 0), -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ if (strncmp(tgt, "-OST", 4) != 0) {
+ CERROR("%s: invalid target name %s: rc = %d\n",
+ osp->opd_obd->obd_name,
+ lustre_cfg_string(cfg, 0), -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ idx = simple_strtol(tgt + 4, &mdt, 16);
+ if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
+ CERROR("%s: invalid OST index in '%s': rc = %d\n",
+ osp->opd_obd->obd_name, src, -EINVAL);
+ RETURN(-EINVAL);
+ }
+ osp->opd_index = idx;
+ osp->opd_group = 0;
+ idx = tgt - src;
+ } else {
+ /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
+ if (strncmp(tgt, "-MDT", 4) != 0 &&
+ strncmp(tgt, "-OST", 4) != 0) {
+ CERROR("%s: invalid target name %s: rc = %d\n",
+ osp->opd_obd->obd_name,
+ lustre_cfg_string(cfg, 0), -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ idx = simple_strtol(tgt + 4, &mdt, 16);
+ if (*mdt != '\0' || idx > INT_MAX || idx < 0) {
+ CERROR("%s: invalid OST index in '%s': rc = %d\n",
+ osp->opd_obd->obd_name, src, -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ /* Get MDT index from the name and set it to opd_group,
+ * which will be used by OSP to connect with OST */
+ osp->opd_group = idx;
+ if (tgt - src <= 12) {
+ CERROR("%s: invalid mdt index from %s: rc =%d\n",
+ osp->opd_obd->obd_name,
+ lustre_cfg_string(cfg, 0), -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ if (strncmp(tgt - 12, "-MDT", 4) == 0)
+ osp->opd_connect_mdt = 1;
+
+ idx = simple_strtol(tgt - 8, &mdt, 16);
+ if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
+ CERROR("%s: invalid OST index in '%s': rc =%d\n",
+ osp->opd_obd->obd_name, src, -EINVAL);
+ RETURN(-EINVAL);
+ }
+
+ osp->opd_index = idx;
+ idx = tgt - src - 12;
+ }
/* check the fsname length, and after this everything else will fit */
if (idx > MTI_NAME_MAXLEN) {
- CERROR("%s: fsname too long in '%s'\n", obd->obd_name, src);
- GOTO(out_fini, rc = -EINVAL);
+ CERROR("%s: fsname too long in '%s': rc = %d\n",
+ osp->opd_obd->obd_name, src, -EINVAL);
+ RETURN(-EINVAL);
}
OBD_ALLOC(osdname, MAX_OBD_NAME);
if (osdname == NULL)
- GOTO(out_fini, rc = -ENOMEM);
+ RETURN(-ENOMEM);
memcpy(osdname, src, idx); /* copy just the fsname part */
osdname[idx] = '\0';
strcat(osdname, "-osd");
CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
- m->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
- m->opd_dt_dev.dd_ops = &osp_dt_ops;
- obd->obd_lu_dev = &m->opd_dt_dev.dd_lu_dev;
+ 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->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
+ osp->opd_dt_dev.dd_ops = &osp_dt_ops;
- rc = osp_connect_to_osd(env, m, osdname);
+ obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
+
+ rc = osp_connect_to_osd(env, osp, osdname);
if (rc)
GOTO(out_fini, rc);
rc = client_obd_setup(obd, cfg);
if (rc) {
- CERROR("%s: can't setup obd: %d\n", m->opd_obd->obd_name, rc);
+ CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
+ rc);
GOTO(out_ref, rc);
}
- osp_lprocfs_init(m);
+ osp_lprocfs_init(osp);
- /*
- * Initialize last id from the storage - will be used in orphan cleanup
- */
- rc = osp_last_used_init(env, m);
- if (rc)
+ rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
+ LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
+ if (rc) {
+ CERROR("%s: fid init error: rc = %d\n",
+ osp->opd_obd->obd_name, rc);
GOTO(out_proc, rc);
+ }
- /*
- * Initialize precreation thread, it handles new connections as well
- */
- rc = osp_init_precreate(m);
- if (rc)
- GOTO(out_last_used, rc);
+ 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_fid, rc);
- /*
- * Initialize synhronization mechanism taking care of propogating
- * changes to OST in near transactional manner
- */
- rc = osp_sync_init(env, m);
- if (rc)
- GOTO(out_precreat, 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 {
+ rc = osp_update_init(osp);
+ if (rc != 0)
+ GOTO(out_fid, rc);
+ }
+
+ ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
/*
* Initiate connect to OST
*/
ll_generate_random_uuid(uuid);
- class_uuid_unparse(uuid, &m->opd_cluuid);
+ class_uuid_unparse(uuid, &osp->opd_cluuid);
imp = obd->u.cli.cl_import;
RETURN(0);
out:
- /* stop sync thread */
- osp_sync_fini(m);
+ if (!osp->opd_connect_mdt)
+ /* stop sync thread */
+ osp_sync_fini(osp);
out_precreat:
/* stop precreate thread */
- osp_precreate_fini(m);
+ if (!osp->opd_connect_mdt)
+ osp_precreate_fini(osp);
+ else
+ osp_update_fini(env, osp);
out_last_used:
- osp_last_used_fini(env, m);
+ 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);
- class_destroy_import(obd->u.cli.cl_import);
+ if (osp->opd_symlink)
+ lprocfs_remove(&osp->opd_symlink);
client_obd_cleanup(obd);
out_ref:
ptlrpcd_decref();
out_disconnect:
- obd_disconnect(m->opd_storage_exp);
+ 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)
OBD_FREE(osdname, MAX_OBD_NAME);
RETURN(rc);
}
+/**
+ * Implementation of lu_device_type_operations::ldto_device_free
+ *
+ * Free the OSP device in memory. No return value is needed for now,
+ * so always return NULL to comply with the interface.
+ *
+ * \param[in] env execution environment
+ * \param[in] lu lu_device of OSP
+ *
+ * \retval NULL NULL unconditionally
+ */
static struct lu_device *osp_device_free(const struct lu_env *env,
struct lu_device *lu)
{
- struct osp_device *m = lu2osp_dev(lu);
-
- ENTRY;
+ struct osp_device *osp = lu2osp_dev(lu);
- if (cfs_atomic_read(&lu->ld_ref) && lu->ld_site) {
+ if (atomic_read(&lu->ld_ref) && lu->ld_site) {
LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
}
- dt_device_fini(&m->opd_dt_dev);
- OBD_FREE_PTR(m);
- RETURN(NULL);
+ dt_device_fini(&osp->opd_dt_dev);
+ OBD_FREE_PTR(osp);
+
+ return NULL;
}
+/**
+ * Implementation of lu_device_type_operations::ldto_device_alloc
+ *
+ * This function allocates and initializes OSP device in memory according to
+ * the config log.
+ *
+ * \param[in] env execution environment
+ * \param[in] type device type of OSP
+ * \param[in] lcfg config log
+ *
+ * \retval pointer the pointer of allocated OSP if succeed.
+ * \retval ERR_PTR(errno) ERR_PTR(errno) if failed.
+ */
static struct lu_device *osp_device_alloc(const struct lu_env *env,
- struct lu_device_type *t,
+ struct lu_device_type *type,
struct lustre_cfg *lcfg)
{
- struct osp_device *m;
- struct lu_device *l;
+ struct osp_device *osp;
+ struct lu_device *ld;
- OBD_ALLOC_PTR(m);
- if (m == NULL) {
- l = ERR_PTR(-ENOMEM);
+ OBD_ALLOC_PTR(osp);
+ if (osp == NULL) {
+ ld = ERR_PTR(-ENOMEM);
} else {
int rc;
- l = osp2lu_dev(m);
- dt_device_init(&m->opd_dt_dev, t);
- if (is_osp_on_ost(lustre_cfg_string(lcfg, 0)))
- rc = osp_init_for_ost(env, m, t, lcfg);
- else
- rc = osp_init0(env, m, t, lcfg);
+ ld = osp2lu_dev(osp);
+ dt_device_init(&osp->opd_dt_dev, type);
+ rc = osp_init0(env, osp, type, lcfg);
if (rc != 0) {
- osp_device_free(env, l);
- l = ERR_PTR(rc);
+ osp_device_free(env, ld);
+ ld = ERR_PTR(rc);
}
}
- return l;
+ return ld;
}
+/**
+ * Implementation of lu_device_type_operations::ldto_device_fini
+ *
+ * This function cleans up the OSP device, i.e. release and free those
+ * attached items in osp_device.
+ *
+ * \param[in] env execution environment
+ * \param[in] ld lu_device of OSP
+ *
+ * \retval NULL NULL if cleanup succeeded.
+ * \retval ERR_PTR(errno) ERR_PTR(errno) if cleanup failed.
+ */
static struct lu_device *osp_device_fini(const struct lu_env *env,
- struct lu_device *d)
+ struct lu_device *ld)
{
- struct osp_device *m = lu2osp_dev(d);
+ struct osp_device *osp = lu2osp_dev(ld);
struct obd_import *imp;
int rc;
ENTRY;
- if (m->opd_storage_exp)
- obd_disconnect(m->opd_storage_exp);
-
- if (is_osp_on_ost(m->opd_obd->obd_name))
- osp_fini_for_ost(m);
+ if (osp->opd_async_requests != NULL) {
+ osp_update_request_destroy(osp->opd_async_requests);
+ osp->opd_async_requests = NULL;
+ }
- imp = m->opd_obd->u.cli.cl_import;
+ if (osp->opd_storage_exp)
+ obd_disconnect(osp->opd_storage_exp);
- if (imp->imp_rq_pool) {
- ptlrpc_free_rq_pool(imp->imp_rq_pool);
- imp->imp_rq_pool = NULL;
- }
+ imp = osp->opd_obd->u.cli.cl_import;
- obd_cleanup_client_import(m->opd_obd);
+ if (osp->opd_symlink)
+ lprocfs_remove(&osp->opd_symlink);
- if (m->opd_symlink)
- lprocfs_remove(&m->opd_symlink);
+ LASSERT(osp->opd_obd);
+ ptlrpc_lprocfs_unregister_obd(osp->opd_obd);
+ lprocfs_obd_cleanup(osp->opd_obd);
- LASSERT(m->opd_obd);
- ptlrpc_lprocfs_unregister_obd(m->opd_obd);
- lprocfs_obd_cleanup(m->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(m->opd_obd);
- LASSERTF(rc == 0, "error %d\n", rc);
+ rc = client_obd_cleanup(osp->opd_obd);
+ if (rc != 0) {
+ ptlrpcd_decref();
+ RETURN(ERR_PTR(rc));
+ }
ptlrpcd_decref();
RETURN(NULL);
}
+/**
+ * Implementation of obd_ops::o_reconnect
+ *
+ * This function is empty and does not need to do anything for now.
+ */
static int osp_reconnect(const struct lu_env *env,
struct obd_export *exp, struct obd_device *obd,
struct obd_uuid *cluuid,
}
/*
- * we use exports to track all LOD users
+ * Implementation of obd_ops::o_connect
+ *
+ * Connect OSP to the remote target (MDT or OST). Allocate the
+ * export and return it to the LOD, which calls this function
+ * for each OSP to connect it to the remote target. This function
+ * is currently only called once per OSP.
+ *
+ * \param[in] env execution environment
+ * \param[out] exp export connected to OSP
+ * \param[in] obd OSP device
+ * \param[in] cluuid OSP device client uuid
+ * \param[in] data connect_data to be used to connect to the remote
+ * target
+ * \param[in] localdata necessary for the API interface, but not used in
+ * this function
+ *
+ * \retval 0 0 if the connection succeeded.
+ * \retval negative negative errno if the connection failed.
*/
static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
struct obd_device *obd, struct obd_uuid *cluuid,
RETURN(rc);
*exp = class_conn2export(&conn);
- if (is_osp_on_ost(obd->obd_name))
- osp->opd_exp = *exp;
-
/* Why should there ever be more than 1 connect? */
osp->opd_connects++;
LASSERT(osp->opd_connects == 1);
+ osp->opd_exp = *exp;
+
imp = osp->opd_obd->u.cli.cl_import;
imp->imp_dlm_handle = conn;
+ LASSERT(data != NULL);
+ LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
ocd = &imp->imp_connect_data;
- ocd->ocd_connect_flags = OBD_CONNECT_AT |
- OBD_CONNECT_FULL20 |
- OBD_CONNECT_INDEX |
-#ifdef HAVE_LRU_RESIZE_SUPPORT
- OBD_CONNECT_LRU_RESIZE |
-#endif
- OBD_CONNECT_MDS |
- OBD_CONNECT_OSS_CAPA |
- OBD_CONNECT_REQPORTAL |
- OBD_CONNECT_SKIP_ORPHAN |
- OBD_CONNECT_VERSION |
- OBD_CONNECT_FID;
-
- if (is_osp_on_ost(osp->opd_obd->obd_name))
- ocd->ocd_connect_flags |= OBD_CONNECT_LIGHTWEIGHT;
+ *ocd = *data;
+
+ imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
ocd->ocd_version = LUSTRE_VERSION_CODE;
- LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
ocd->ocd_index = data->ocd_index;
imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
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);
-
out:
RETURN(rc);
}
-/*
- * once last export (we don't count self-export) disappeared
- * osp can be released
+/**
+ * Implementation of obd_ops::o_disconnect
+ *
+ * Disconnect the export for the OSP. This is called by LOD to release the
+ * OSP during cleanup (\see lod_del_device()). The OSP will be released after
+ * the export is released.
+ *
+ * \param[in] exp export to be disconnected.
+ *
+ * \retval 0 0 if disconnection succeed
+ * \retval negative negative errno if disconnection failed
*/
static int osp_obd_disconnect(struct obd_export *exp)
{
}
/* destroy the device */
- if (!is_osp_on_ost(obd->obd_name))
- class_manual_cleanup(obd);
+ class_manual_cleanup(obd);
RETURN(rc);
}
-/*
- * lprocfs helpers still use OBD API, let's keep obd_statfs() support
+/**
+ * Implementation of obd_ops::o_statfs
+ *
+ * Send a RPC to the remote target to get statfs status. This is only used
+ * in lprocfs helpers by obd_statfs.
+ *
+ * \param[in] env execution environment
+ * \param[in] exp connection state from this OSP to the parent (LOD)
+ * device
+ * \param[out] osfs hold the statfs result
+ * \param[in] unused Not used in this function for now
+ * \param[in] flags flags to indicate how OSP will issue the RPC
+ *
+ * \retval 0 0 if statfs succeeded.
+ * \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 max_age, __u32 flags)
+ struct obd_statfs *osfs, __u64 unused, __u32 flags)
{
struct obd_statfs *msfs;
struct ptlrpc_request *req;
/* Since the request might also come from lprocfs, so we need
* sync this with client_disconnect_export Bug15684 */
- cfs_down_read(&exp->exp_obd->u.cli.cl_sem);
+ down_read(&exp->exp_obd->u.cli.cl_sem);
if (exp->exp_obd->u.cli.cl_import)
imp = class_import_get(exp->exp_obd->u.cli.cl_import);
- cfs_up_read(&exp->exp_obd->u.cli.cl_sem);
+ up_read(&exp->exp_obd->u.cli.cl_sem);
if (!imp)
RETURN(-ENODEV);
- /* We could possibly pass max_age in the request (as an absolute
- * timestamp or a "seconds.usec ago") so the target can avoid doing
- * extra calls into the filesystem if that isn't necessary (e.g.
- * during mount that would help a bit). Having relative timestamps
- * is not so great if request processing is slow, while absolute
- * timestamps are not ideal because they need time synchronization. */
req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
class_import_put(imp);
return rc;
}
+/**
+ * Implementation of obd_ops::o_import_event
+ *
+ * This function is called when some related import event happens. It will
+ * mark the necessary flags according to the event and notify the necessary
+ * threads (mainly precreate thread).
+ *
+ * \param[in] obd OSP OBD device
+ * \param[in] imp import attached from OSP to remote (OST/MDT) service
+ * \param[in] event event related to remote service (IMP_EVENT_*)
+ *
+ * \retval 0 0 if the event handling succeeded.
+ * \retval negative negative errno if the event handling failed.
+ */
static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
enum obd_import_event event)
{
case IMP_EVENT_DISCON:
d->opd_got_disconnected = 1;
d->opd_imp_connected = 0;
- if (is_osp_on_ost(d->opd_obd->obd_name))
+ if (d->opd_connect_mdt)
break;
- osp_pre_update_status(d, -ENODEV);
- cfs_waitq_signal(&d->opd_pre_waitq);
+
+ if (d->opd_pre != NULL) {
+ osp_pre_update_status(d, -ENODEV);
+ wake_up(&d->opd_pre_waitq);
+ }
+
CDEBUG(D_HA, "got disconnected\n");
break;
case IMP_EVENT_INACTIVE:
d->opd_imp_active = 0;
- if (is_osp_on_ost(d->opd_obd->obd_name))
+ if (d->opd_connect_mdt)
break;
- osp_pre_update_status(d, -ENODEV);
- cfs_waitq_signal(&d->opd_pre_waitq);
+
+ if (d->opd_pre != NULL) {
+ osp_pre_update_status(d, -ENODEV);
+ wake_up(&d->opd_pre_waitq);
+ }
+
CDEBUG(D_HA, "got inactive\n");
break;
case IMP_EVENT_ACTIVE:
d->opd_imp_active = 1;
+
if (d->opd_got_disconnected)
d->opd_new_connection = 1;
d->opd_imp_connected = 1;
d->opd_imp_seen_connected = 1;
- if (is_osp_on_ost(d->opd_obd->obd_name))
+ if (d->opd_connect_mdt)
break;
- cfs_waitq_signal(&d->opd_pre_waitq);
+
+ if (d->opd_pre != NULL)
+ wake_up(&d->opd_pre_waitq);
+
__osp_sync_check_for_work(d);
CDEBUG(D_HA, "got connected\n");
break;
ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
break;
case IMP_EVENT_OCD:
+ case IMP_EVENT_DEACTIVATE:
+ case IMP_EVENT_ACTIVATE:
break;
default:
CERROR("%s: unsupported import event: %#x\n",
return 0;
}
+/**
+ * Implementation of obd_ops: o_iocontrol
+ *
+ * This function is the ioctl handler for OSP. Note: lctl will access the OSP
+ * directly by ioctl, instead of through the MDS stack.
+ *
+ * param[in] cmd ioctl command.
+ * param[in] exp export of this OSP.
+ * param[in] len data length of \a karg.
+ * param[in] karg input argument which is packed as
+ * obd_ioctl_data
+ * param[out] uarg pointer to userspace buffer (must access by
+ * copy_to_user()).
+ *
+ * \retval 0 0 if the ioctl handling succeeded.
+ * \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)
{
d = lu2osp_dev(obd->obd_lu_dev);
LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
- if (!cfs_try_module_get(THIS_MODULE)) {
- CERROR("%s: can't get module. Is it alive?", obd->obd_name);
+ if (!try_module_get(THIS_MODULE)) {
+ CERROR("%s: cannot get module '%s'\n", obd->obd_name,
+ module_name(THIS_MODULE));
return -EINVAL;
}
break;
default:
CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
- cmd, cfs_curproc_comm());
+ cmd, current_comm());
rc = -ENOTTY;
}
- cfs_module_put(THIS_MODULE);
+ module_put(THIS_MODULE);
return rc;
}
-static int osp_obd_health_check(const struct lu_env *env,
- struct obd_device *obd)
+/**
+ * Implementation of obd_ops::o_get_info
+ *
+ * Retrieve information by key. Retrieval starts from the top layer
+ * (MDT) of the MDS stack and traverses the stack by calling the
+ * obd_get_info() method of the next sub-layer.
+ *
+ * \param[in] env execution environment
+ * \param[in] exp export of this OSP
+ * \param[in] keylen length of \a key
+ * \param[in] key the key
+ * \param[out] vallen length of \a val
+ * \param[out] val holds the value returned by the key
+ *
+ * \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 osp_device *d = lu2osp_dev(obd->obd_lu_dev);
+ int rc = -EINVAL;
+
+ if (KEY_IS(KEY_OSP_CONNECTED)) {
+ struct obd_device *obd = exp->exp_obd;
+ struct osp_device *osp;
+
+ if (!obd->obd_set_up || obd->obd_stopping)
+ RETURN(-EAGAIN);
+
+ osp = lu2osp_dev(obd->obd_lu_dev);
+ LASSERT(osp);
+ /*
+ * 1.8/2.0 behaviour is that OST being connected once at least
+ * is considered "healthy". and one "healthy" OST is enough to
+ * allow lustre clients to connect to MDS
+ */
+ RETURN(!osp->opd_imp_seen_connected);
+ }
+
+ RETURN(rc);
+}
+/**
+ * Implementation of obd_ops: o_fid_alloc
+ *
+ * Allocate a FID. There are two cases in which OSP performs
+ * FID allocation.
+ *
+ * 1. FID precreation for data objects, which is done in
+ * osp_precreate_fids() instead of this function.
+ * 2. FID allocation for each sub-stripe of a striped directory.
+ * Similar to other FID clients, OSP requests the sequence
+ * from its corresponding remote MDT, which in turn requests
+ * sequences from the sequence controller (MDT0).
+ *
+ * \param[in] env execution environment
+ * \param[in] exp export of the OSP
+ * \param[out] fid FID being allocated
+ * \param[in] unused necessary for the interface but unused.
+ *
+ * \retval 0 0 FID allocated successfully.
+ * \retval 1 1 FID allocated successfully and new sequence
+ * requested from seq meta server
+ * \retval negative negative errno if FID allocation failed.
+ */
+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);
+ struct lu_client_seq *seq = cli->cl_seq;
ENTRY;
- /*
- * 1.8/2.0 behaviour is that OST being connected once at least
- * is considired "healthy". and one "healty" OST is enough to
- * allow lustre clients to connect to MDS
- */
- LASSERT(d);
- RETURN(!d->opd_imp_seen_connected);
+ LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
+ /* Sigh, fid client is not ready yet */
+ LASSERT(osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
+
+ RETURN(seq_client_alloc_fid(env, seq, fid));
}
/* context key constructor/destructor: mdt_key_init, mdt_key_fini */
LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
struct lu_context_key osp_txn_key = {
- .lct_tags = LCT_OSP_THREAD,
+ .lct_tags = LCT_OSP_THREAD | LCT_TX_HANDLE,
.lct_init = osp_txn_key_init,
.lct_fini = osp_txn_key_fini
};
.ldt_tags = LU_DEVICE_DT,
.ldt_name = LUSTRE_OSP_NAME,
.ldt_ops = &osp_device_type_ops,
- .ldt_ctx_tags = LCT_MD_THREAD
+ .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
};
static struct obd_ops osp_obd_device_ops = {
.o_reconnect = osp_reconnect,
.o_connect = osp_obd_connect,
.o_disconnect = osp_obd_disconnect,
- .o_health_check = osp_obd_health_check,
+ .o_get_info = osp_obd_get_info,
.o_import_event = osp_import_event,
.o_iocontrol = osp_iocontrol,
.o_statfs = osp_obd_statfs,
+ .o_fid_init = client_fid_init,
+ .o_fid_fini = client_fid_fini,
+ .o_fid_alloc = osp_fid_alloc,
};
+struct llog_operations osp_mds_ost_orig_logops;
+
+/**
+ * Initialize OSP module.
+ *
+ * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
+ * in obd_types (\see class_obd.c). Initialize procfs for the
+ * the OSP device. Note: OSP was called OSC before Lustre 2.4,
+ * so for compatibility it still uses the name "osc" in procfs.
+ * This is called at module load time.
+ *
+ * \retval 0 0 if initialization succeeds.
+ * \retval negative negative errno if initialization failed.
+ */
static int __init osp_mod_init(void)
{
- struct lprocfs_static_vars lvars;
- cfs_proc_dir_entry_t *osc_proc_dir;
- int rc;
+ struct obd_type *type;
+ int rc;
rc = lu_kmem_init(osp_caches);
if (rc)
return rc;
- lprocfs_osp_init_vars(&lvars);
- rc = class_register_type(&osp_obd_device_ops, NULL, lvars.module_vars,
+ rc = class_register_type(&osp_obd_device_ops, NULL, true, NULL,
LUSTRE_OSP_NAME, &osp_device_type);
+ if (rc != 0) {
+ lu_kmem_fini(osp_caches);
+ return rc;
+ }
- /* create "osc" entry in procfs for compatibility purposes */
+ rc = class_register_type(&lwp_obd_device_ops, NULL, true, NULL,
+ LUSTRE_LWP_NAME, &lwp_device_type);
if (rc != 0) {
+ class_unregister_type(LUSTRE_OSP_NAME);
lu_kmem_fini(osp_caches);
return rc;
}
- osc_proc_dir = lprocfs_srch(proc_lustre_root, "osc");
- if (osc_proc_dir == NULL) {
- osc_proc_dir = lprocfs_register("osc", proc_lustre_root, NULL,
- NULL);
- if (IS_ERR(osc_proc_dir))
- CERROR("osp: can't create compat entry \"osc\": %d\n",
- (int) PTR_ERR(osc_proc_dir));
+ /* 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_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;
}
return rc;
}
+/**
+ * Finalize OSP module.
+ *
+ * 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)
{
- lprocfs_try_remove_proc_entry("osc", proc_lustre_root);
-
+ 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_VERSION(LUSTRE_VERSION_STRING);
MODULE_LICENSE("GPL");
-cfs_module(osp, LUSTRE_VERSION_STRING, osp_mod_init, osp_mod_exit);
+module_init(osp_mod_init);
+module_exit(osp_mod_exit);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff