4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License version 2 only,
8 * as published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful, but
11 * WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * General Public License version 2 for more details (a copy is included
14 * in the LICENSE file that accompanied this code).
16 * You should have received a copy of the GNU General Public License
17 * version 2 along with this program; If not, see
18 * http://www.gnu.org/licenses/gpl-2.0.html
23 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24 * Use is subject to license terms.
26 * Copyright (c) 2012, 2017, Intel Corporation.
29 * This file is part of Lustre, http://www.lustre.org/
31 * lustre/osp/osp_dev.c
33 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
34 * Author: Mikhail Pershin <mike.pershin@intel.com>
35 * Author: Di Wang <di.wang@intel.com>
38 * The Object Storage Proxy (OSP) module provides an implementation of
39 * the DT API for remote MDTs and OSTs. Every local OSP device (or
40 * object) is a proxy for a remote OSD device (or object). Thus OSP
41 * converts DT operations into RPCs, which are sent to the OUT service
42 * on a remote target, converted back to DT operations, and
43 * executed. Of course there are many ways in which this description
44 * is inaccurate but it's a good enough mental model. OSP is used by
45 * the MDT stack in several ways:
47 * - OSP devices allocate FIDs for the stripe sub-objects of a striped
50 * - OSP objects represent the remote MDT and OST objects that are
51 * the stripes of a striped object.
53 * - OSP devices log, send, and track synchronous operations (setattr
54 * and unlink) to remote targets.
56 * - OSP objects are the bottom slice of the compound LU object
57 * representing a remote MDT object: MDT/MDD/LOD/OSP.
59 * - OSP objects are used by LFSCK to represent remote OST objects
60 * during the verification of MDT-OST consistency.
62 * - OSP devices batch idempotent requests (declare_attr_get() and
63 * declare_xattr_get()) to the remote target and cache their results.
65 * In addition the OSP layer implements a subset of the OBD device API
66 * to support being a client of a remote target, connecting to other
67 * layers, and FID allocation.
70 #define DEBUG_SUBSYSTEM S_MDS
72 #include <linux/kthread.h>
74 #include <uapi/linux/lustre/lustre_ioctl.h>
75 #include <lustre_ioctl_old.h>
76 #include <lustre_log.h>
77 #include <lustre_obdo.h>
78 #include <uapi/linux/lustre/lustre_param.h>
79 #include <obd_class.h>
81 #include "osp_internal.h"
83 /* Slab for OSP object allocation */
84 struct kmem_cache *osp_object_kmem;
86 static struct lu_kmem_descr osp_caches[] = {
88 .ckd_cache = &osp_object_kmem,
89 .ckd_name = "osp_obj",
90 .ckd_size = sizeof(struct osp_object)
98 * Implementation of lu_device_operations::ldo_object_alloc
100 * Allocates an OSP object in memory, whose FID is on the remote target.
102 * \param[in] env execution environment
103 * \param[in] hdr The header of the object stack. If it is NULL, it
104 * means the object is not built from top device, i.e.
105 * it is a sub-stripe object of striped directory or
107 * \param[in] d OSP device
109 * \retval object object being created if the creation succeed.
110 * \retval NULL NULL if the creation failed.
112 static struct lu_object *osp_object_alloc(const struct lu_env *env,
113 const struct lu_object_header *hdr,
116 struct osp_object *o;
118 OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
120 struct lu_object *l = &o->opo_obj.do_lu;
122 /* If hdr is NULL, it means the object is not built
123 * from the top dev(MDT/OST), usually it happens when
124 * building striped object, like data object on MDT or
125 * striped object for directory */
127 struct lu_object_header *h = &o->opo_header;
129 lu_object_header_init(h);
130 dt_object_init(&o->opo_obj, h, d);
131 lu_object_add_top(h, l);
133 dt_object_init(&o->opo_obj, NULL, d);
136 l->lo_ops = &osp_lu_obj_ops;
138 init_rwsem(&o->opo_sem);
139 INIT_LIST_HEAD(&o->opo_xattr_list);
140 INIT_LIST_HEAD(&o->opo_invalidate_cb_list);
141 spin_lock_init(&o->opo_lock);
142 init_rwsem(&o->opo_invalidate_sem);
151 * Find or create the local object
153 * Finds or creates the local file referenced by \a reg_id and return the
154 * attributes of the local file.
156 * \param[in] env execution environment
157 * \param[in] osp OSP device
158 * \param[out] attr attributes of the object
159 * \param[in] reg_id the local object ID of the file. It will be used
160 * to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
161 * to identify the object.
163 * \retval object object(dt_object) found or created
164 * \retval ERR_PTR(errno) ERR_PTR(errno) if not get the object.
166 static struct dt_object
167 *osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
168 struct lu_attr *attr, __u32 reg_id)
170 struct osp_thread_info *osi = osp_env_info(env);
171 struct dt_object_format dof = { 0 };
172 struct dt_object *dto;
176 lu_local_obj_fid(&osi->osi_fid, reg_id);
177 attr->la_valid = LA_MODE;
178 attr->la_mode = S_IFREG | 0644;
179 dof.dof_type = DFT_REGULAR;
180 /* Find or create the local object by osi_fid. */
181 dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
186 /* Get attributes of the local object. */
187 rc = dt_attr_get(env, dto, attr);
189 CERROR("%s: can't be initialized: rc = %d\n",
190 osp->opd_obd->obd_name, rc);
191 dt_object_put(env, dto);
198 * Write data buffer to a local file object.
200 * \param[in] env execution environment
201 * \param[in] osp OSP device
202 * \param[in] dt_obj object written to
203 * \param[in] buf buffer containing byte array and length
204 * \param[in] offset write offset in the object in bytes
206 * \retval 0 0 if write succeed
207 * \retval -EFAULT -EFAULT if only part of buffer is written.
208 * \retval negative other negative errno if write failed.
210 static int osp_write_local_file(const struct lu_env *env,
211 struct osp_device *osp,
212 struct dt_object *dt_obj,
219 if (osp->opd_storage->dd_rdonly)
222 th = dt_trans_create(env, osp->opd_storage);
226 rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
229 rc = dt_trans_start_local(env, osp->opd_storage, th);
233 rc = dt_record_write(env, dt_obj, buf, &offset, th);
235 dt_trans_stop(env, osp->opd_storage, th);
240 * Initialize last ID object.
242 * This function initializes the LAST_ID file, which stores the current last
243 * used id of data objects. The MDT will use the last used id and the last_seq
244 * (\see osp_init_last_seq()) to synchronize the precreate object cache with
247 * \param[in] env execution environment
248 * \param[in] osp OSP device
250 * \retval 0 0 if initialization succeed
251 * \retval negative negative errno if initialization failed
253 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
255 struct osp_thread_info *osi = osp_env_info(env);
256 struct lu_fid *fid = &osp->opd_last_used_fid;
257 struct dt_object *dto;
261 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
264 RETURN(PTR_ERR(dto));
266 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &osp->opd_last_id,
269 /* object will be released in device cleanup path */
270 if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
271 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
272 if (rc != 0 && rc != -EFAULT)
274 /* In case of idif bits 32-48 go to f_seq
275 * (see osp_init_last_seq). So don't care
276 * about u64->u32 convertion. */
277 fid->f_oid = osp->opd_last_id;
280 if (rc == -EFAULT) { /* fresh LAST_ID */
281 osp->opd_last_id = 0;
283 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
288 osp->opd_last_used_oid_file = dto;
291 /* object will be released in device cleanup path */
292 CERROR("%s: can't initialize lov_objid: rc = %d\n",
293 osp->opd_obd->obd_name, rc);
294 dt_object_put(env, dto);
295 osp->opd_last_used_oid_file = NULL;
300 * Initialize last sequence object.
302 * This function initializes the LAST_SEQ file in the local OSD, which stores
303 * the current last used sequence of data objects. The MDT will use the last
304 * sequence and last id (\see osp_init_last_objid()) to synchronize the
305 * precreate object cache with OSTs.
307 * \param[in] env execution environment
308 * \param[in] osp OSP device
310 * \retval 0 0 if initialization succeed
311 * \retval negative negative errno if initialization failed
313 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
315 struct osp_thread_info *osi = osp_env_info(env);
316 struct lu_fid *fid = &osp->opd_last_used_fid;
317 struct dt_object *dto;
321 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
324 RETURN(PTR_ERR(dto));
326 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
329 /* object will be released in device cleanup path */
330 if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
331 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
332 if (rc != 0 && rc != -EFAULT)
334 if (fid_is_idif(fid))
335 fid->f_seq = fid_idif_seq(osp->opd_last_id,
339 if (rc == -EFAULT) { /* fresh OSP */
341 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
346 osp->opd_last_used_seq_file = dto;
349 /* object will be released in device cleanup path */
350 CERROR("%s: can't initialize lov_seq: rc = %d\n",
351 osp->opd_obd->obd_name, rc);
352 dt_object_put(env, dto);
353 osp->opd_last_used_seq_file = NULL;
358 * Initialize last OID and sequence object.
360 * If the MDT is just upgraded to 2.4 from the lower version, where the
361 * LAST_SEQ file does not exist, the file will be created and IDIF sequence
362 * will be written into the file.
364 * \param[in] env execution environment
365 * \param[in] osp OSP device
367 * \retval 0 0 if initialization succeed
368 * \retval negative negative error if initialization failed
370 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
372 struct osp_thread_info *osi = osp_env_info(env);
376 fid_zero(&osp->opd_last_used_fid);
377 rc = osp_init_last_objid(env, osp);
379 CERROR("%s: Can not get ids %d from old objid!\n",
380 osp->opd_obd->obd_name, rc);
384 rc = osp_init_last_seq(env, osp);
386 CERROR("%s: Can not get sequence %d from old objseq!\n",
387 osp->opd_obd->obd_name, rc);
391 if (fid_oid(&osp->opd_last_used_fid) != 0 &&
392 fid_seq(&osp->opd_last_used_fid) == 0) {
393 /* Just upgrade from the old version,
394 * set the seq to be IDIF */
395 osp->opd_last_used_fid.f_seq =
396 fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
398 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
399 &osp->opd_last_used_fid.f_seq,
401 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
402 &osi->osi_lb, osi->osi_off);
404 CERROR("%s : Can not write seq file: rc = %d\n",
405 osp->opd_obd->obd_name, rc);
410 if (!fid_is_zero(&osp->opd_last_used_fid) &&
411 !fid_is_sane(&osp->opd_last_used_fid)) {
412 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
413 PFID(&osp->opd_last_used_fid));
414 GOTO(out, rc = -EINVAL);
417 osp_fid_to_obdid(&osp->opd_last_used_fid, &osp->opd_last_id);
418 CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
419 osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
422 if (osp->opd_last_used_oid_file != NULL) {
423 dt_object_put(env, osp->opd_last_used_oid_file);
424 osp->opd_last_used_oid_file = NULL;
426 if (osp->opd_last_used_seq_file != NULL) {
427 dt_object_put(env, osp->opd_last_used_seq_file);
428 osp->opd_last_used_seq_file = NULL;
436 * Release the last sequence and OID file objects in OSP device.
438 * \param[in] env execution environment
439 * \param[in] osp OSP device
441 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
443 /* release last_used file */
444 if (osp->opd_last_used_oid_file != NULL) {
445 dt_object_put(env, osp->opd_last_used_oid_file);
446 osp->opd_last_used_oid_file = NULL;
449 if (osp->opd_last_used_seq_file != NULL) {
450 dt_object_put(env, osp->opd_last_used_seq_file);
451 osp->opd_last_used_seq_file = NULL;
456 * Disconnects the connection between OSP and its correspondent MDT or OST, and
457 * the import will be marked as inactive. It will only be called during OSP
460 * \param[in] d OSP device being disconnected
462 * \retval 0 0 if disconnection succeed
463 * \retval negative negative errno if disconnection failed
465 static int osp_disconnect(struct osp_device *d)
467 struct obd_device *obd = d->opd_obd;
468 struct obd_import *imp;
471 imp = obd->u.cli.cl_import;
473 /* Mark import deactivated now, so we don't try to reconnect if any
474 * of the cleanup RPCs fails (e.g. ldlm cancel, etc). We don't
475 * fully deactivate the import, or that would drop all requests. */
476 LASSERT(imp != NULL);
477 spin_lock(&imp->imp_lock);
478 imp->imp_deactive = 1;
479 spin_unlock(&imp->imp_lock);
481 ptlrpc_deactivate_import(imp);
483 /* Some non-replayable imports (MDS's OSCs) are pinged, so just
484 * delete it regardless. (It's safe to delete an import that was
486 (void)ptlrpc_pinger_del_import(imp);
488 /* Send disconnect on healthy import, do force disconnect otherwise */
489 spin_lock(&imp->imp_lock);
490 imp->imp_obd->obd_force = imp->imp_state != LUSTRE_IMP_FULL;
491 spin_unlock(&imp->imp_lock);
493 rc = ptlrpc_disconnect_import(imp, 0);
495 CERROR("%s: can't disconnect: rc = %d\n", obd->obd_name, rc);
497 ptlrpc_invalidate_import(imp);
503 * Initialize the osp_update structure in OSP device
505 * Allocate osp update structure and start update thread.
507 * \param[in] osp OSP device
509 * \retval 0 if initialization succeeds.
510 * \retval negative errno if initialization fails.
512 static int osp_update_init(struct osp_device *osp)
514 struct task_struct *task;
519 LASSERT(osp->opd_connect_mdt);
521 if (osp->opd_storage->dd_rdonly)
524 OBD_ALLOC_PTR(osp->opd_update);
525 if (osp->opd_update == NULL)
528 init_waitqueue_head(&osp->opd_update->ou_waitq);
529 spin_lock_init(&osp->opd_update->ou_lock);
530 INIT_LIST_HEAD(&osp->opd_update->ou_list);
531 osp->opd_update->ou_rpc_version = 1;
532 osp->opd_update->ou_version = 1;
533 osp->opd_update->ou_generation = 0;
535 rc = lu_env_init(&osp->opd_update->ou_env,
536 osp->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
538 CERROR("%s: init env error: rc = %d\n", osp->opd_obd->obd_name,
540 OBD_FREE_PTR(osp->opd_update);
541 osp->opd_update = NULL;
544 /* start thread handling sending updates to the remote MDT */
545 task = kthread_create(osp_send_update_thread, osp,
546 "osp_up%u-%u", osp->opd_index, osp->opd_group);
548 int rc = PTR_ERR(task);
550 lu_env_fini(&osp->opd_update->ou_env);
551 OBD_FREE_PTR(osp->opd_update);
552 osp->opd_update = NULL;
553 CERROR("%s: can't start precreate thread: rc = %d\n",
554 osp->opd_obd->obd_name, rc);
558 osp->opd_update->ou_update_task = task;
559 wake_up_process(task);
565 * Finialize osp_update structure in OSP device
567 * Stop the OSP update sending thread, then delete the left
568 * osp thandle in the sending list.
570 * \param [in] osp OSP device.
572 static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
574 struct obd_import *imp = osp->opd_obd->u.cli.cl_import;
575 struct osp_update_request *our;
576 struct osp_update_request *tmp;
577 struct osp_updates *ou = osp->opd_update;
582 kthread_stop(ou->ou_update_task);
583 lu_env_fini(&ou->ou_env);
586 * import invalidation can be running in a dedicated thread.
587 * we have to wait for the thread's completion as the thread
588 * invalidates this list as well.
590 wait_event_idle(imp->imp_recovery_waitq,
591 (atomic_read(&imp->imp_inval_count) == 0));
593 /* Remove the left osp thandle from the list */
594 list_for_each_entry_safe(our, tmp, &ou->ou_list, our_list) {
595 list_del_init(&our->our_list);
596 LASSERT(our->our_th != NULL);
597 osp_trans_callback(env, our->our_th, -EIO);
598 /* our will be destroyed in osp_thandle_put() */
599 osp_thandle_put(env, our->our_th);
603 osp->opd_update = NULL;
607 * Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
608 * precreate threads etc.
610 * \param[in] env execution environment.
611 * \param[in] d OSP device being disconnected.
613 * \retval 0 0 if cleanup succeed
614 * \retval negative negative errno if cleanup failed
616 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
623 rc = osp_disconnect(d);
627 if (!d->opd_connect_mdt) {
628 /* stop sync thread */
631 /* stop precreate thread */
632 osp_precreate_fini(d);
634 /* release last_used file */
635 osp_last_used_fini(env, d);
638 obd_fid_fini(d->opd_obd);
644 * Implementation of osp_lu_ops::ldo_process_config
646 * This function processes config log records in OSP layer. It is usually
647 * called from the top layer of MDT stack, and goes through the stack by calling
648 * ldo_process_config of next layer.
650 * \param[in] env execution environment
651 * \param[in] dev lu_device of OSP
652 * \param[in] lcfg config log
654 * \retval 0 0 if the config log record is executed correctly.
655 * \retval negative negative errno if the record execution fails.
657 static int osp_process_config(const struct lu_env *env,
658 struct lu_device *dev, struct lustre_cfg *lcfg)
660 struct osp_device *d = lu2osp_dev(dev);
661 struct dt_device *dt = lu2dt_dev(dev);
662 struct obd_device *obd = d->opd_obd;
668 switch (lcfg->lcfg_command) {
669 case LCFG_PRE_CLEANUP:
670 rc = osp_disconnect(d);
671 osp_update_fini(env, d);
675 * cleanup ldlm so that PRE_CLEANUP phase doesn't block
676 * awaiting for locks held by MDT threads awaiting for
677 * all OSPs to interrupt their in-flight RPCs
679 if (obd->obd_namespace != NULL)
680 ldlm_namespace_free_prior(obd->obd_namespace, NULL, 1);
681 lu_dev_del_linkage(dev->ld_site, dev);
682 rc = osp_shutdown(env, d);
685 count = class_modify_config(lcfg, d->opd_connect_mdt ?
686 PARAM_OSP : PARAM_OSC,
689 /* class_modify_config() haven't found matching
690 * parameter and returned an error so that layer(s)
691 * below could use that. But OSP is the bottom, so
694 CERROR("%s: unknown param %s\n",
695 (char *)lustre_cfg_string(lcfg, 0),
696 (char *)lustre_cfg_string(lcfg, 1));
701 CERROR("%s: unknown command %u\n",
702 (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
711 * Implementation of osp_lu_ops::ldo_recovery_complete
713 * This function is called after recovery is finished, and OSP layer
714 * will wake up precreate thread here.
716 * \param[in] env execution environment
717 * \param[in] dev lu_device of OSP
719 * \retval 0 0 unconditionally
721 static int osp_recovery_complete(const struct lu_env *env,
722 struct lu_device *dev)
724 struct osp_device *osp = lu2osp_dev(dev);
727 osp->opd_recovery_completed = 1;
729 if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
730 wake_up(&osp->opd_pre_waitq);
736 * Implementation of lu_device_operations::ldo_fid_alloc() for OSP
738 * Allocate FID from remote MDT.
740 * see include/lu_object.h for the details.
742 static int osp_fid_alloc(const struct lu_env *env, struct lu_device *d,
743 struct lu_fid *fid, struct lu_object *parent,
744 const struct lu_name *name)
746 struct osp_device *osp = lu2osp_dev(d);
747 struct client_obd *cli = &osp->opd_obd->u.cli;
748 struct lu_client_seq *seq = cli->cl_seq;
753 /* Sigh, fid client is not ready yet */
754 if (!osp->opd_obd->u.cli.cl_seq)
757 if (!osp->opd_obd->u.cli.cl_seq->lcs_exp)
760 rc = seq_client_alloc_fid(env, seq, fid);
765 const struct lu_device_operations osp_lu_ops = {
766 .ldo_object_alloc = osp_object_alloc,
767 .ldo_process_config = osp_process_config,
768 .ldo_recovery_complete = osp_recovery_complete,
769 .ldo_fid_alloc = osp_fid_alloc,
773 * Implementation of dt_device_operations::dt_statfs
775 * This function provides statfs status (for precreation) from
776 * corresponding OST. Note: this function only retrieves the status
777 * from the OSP device, and the real statfs RPC happens inside
778 * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
779 * not need to retrieve statfs data for now.
781 * \param[in] env execution environment.
782 * \param[in] dev dt_device of OSP.
783 * \param[out] sfs holds the retrieved statfs data.
785 * \retval 0 0 statfs data was retrieved successfully or
786 * retrieval was not needed
787 * \retval negative negative errno if get statfs failed.
789 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
790 struct obd_statfs *sfs, struct obd_statfs_info *info)
792 struct osp_device *d = dt2osp_dev(dev);
793 struct obd_import *imp = d->opd_obd->u.cli.cl_import;
797 if (imp->imp_state == LUSTRE_IMP_CLOSED)
800 if (unlikely(d->opd_imp_active == 0))
803 /* return recently updated data */
804 *sfs = d->opd_statfs;
806 info->os_reserved_mb_low = d->opd_reserved_mb_low;
807 info->os_reserved_mb_high = d->opd_reserved_mb_high;
811 "%s: blocks=%llu, bfree=%llu, bavail=%llu, bsize=%u, reserved_mb_low=%u, reserved_mb_high=%u, files=%llu, ffree=%llu, state=%x\n",
812 d->opd_obd->obd_name,
813 sfs->os_blocks, sfs->os_bfree, sfs->os_bavail, sfs->os_bsize,
814 d->opd_reserved_mb_low, d->opd_reserved_mb_high,
815 sfs->os_files, sfs->os_ffree, sfs->os_state);
817 if (d->opd_pre == NULL || (info && !info->os_enable_pre))
821 * The layer above osp (usually lod) can use f_precreated to
822 * estimate how many objects are available for immediate usage.
824 spin_lock(&d->opd_pre_lock);
825 sfs->os_fprecreated = osp_fid_diff(&d->opd_pre_last_created_fid,
826 &d->opd_pre_used_fid);
827 sfs->os_fprecreated -= d->opd_pre_reserved;
828 LASSERTF(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2,
829 "last_created "DFID", next_fid "DFID", reserved %llu\n",
830 PFID(&d->opd_pre_last_created_fid), PFID(&d->opd_pre_used_fid),
831 d->opd_pre_reserved);
832 spin_unlock(&d->opd_pre_lock);
837 * Implementation of dt_device_operations::dt_sync
839 * This function synchronizes the OSP cache to the remote target. It wakes
840 * up unlink log threads and sends out unlink records to the remote OST.
842 * \param[in] env execution environment
843 * \param[in] dev dt_device of OSP
845 * \retval 0 0 if synchronization succeeds
846 * \retval negative negative errno if synchronization fails
848 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
850 struct osp_device *d = dt2osp_dev(dev);
851 time64_t start = ktime_get_seconds();
857 /* No Sync between MDTs yet. */
858 if (d->opd_connect_mdt)
861 if (d->opd_storage->dd_rdonly)
864 recs = atomic_read(&d->opd_sync_changes);
865 old = atomic64_read(&d->opd_sync_processed_recs);
867 osp_sync_force(env, dt2osp_dev(dev));
869 if (unlikely(d->opd_imp_active == 0))
872 down_write(&d->opd_async_updates_rwsem);
874 CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
875 atomic_read(&d->opd_async_updates_count));
877 /* make sure the connection is fine */
878 rc = wait_event_idle_timeout(
879 d->opd_sync_barrier_waitq,
880 atomic_read(&d->opd_async_updates_count) == 0,
881 cfs_time_seconds(obd_timeout));
887 up_write(&d->opd_async_updates_rwsem);
891 CDEBUG(D_CACHE, "%s: processed %llu\n", d->opd_obd->obd_name,
892 (unsigned long long)atomic64_read(&d->opd_sync_processed_recs));
894 while (atomic64_read(&d->opd_sync_processed_recs) < old + recs) {
895 __u64 last = atomic64_read(&d->opd_sync_processed_recs);
896 /* make sure the connection is fine */
897 wait_event_idle_timeout(
898 d->opd_sync_barrier_waitq,
899 atomic64_read(&d->opd_sync_processed_recs)
901 cfs_time_seconds(obd_timeout));
903 if (atomic64_read(&d->opd_sync_processed_recs) >= old + recs)
906 if (atomic64_read(&d->opd_sync_processed_recs) != last) {
907 /* some progress have been made,
912 /* no changes and expired, something is wrong */
913 GOTO(out, rc = -ETIMEDOUT);
916 /* block new processing (barrier>0 - few callers are possible */
917 atomic_inc(&d->opd_sync_barrier);
919 CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
920 atomic_read(&d->opd_sync_rpcs_in_flight));
922 /* wait till all-in-flight are replied, so executed by the target */
923 /* XXX: this is used by LFSCK at the moment, which doesn't require
924 * all the changes to be committed, but in general it'd be
925 * better to wait till commit */
926 while (atomic_read(&d->opd_sync_rpcs_in_flight) > 0) {
927 old = atomic_read(&d->opd_sync_rpcs_in_flight);
929 wait_event_idle_timeout(
930 d->opd_sync_barrier_waitq,
931 atomic_read(&d->opd_sync_rpcs_in_flight) == 0,
932 cfs_time_seconds(obd_timeout));
934 if (atomic_read(&d->opd_sync_rpcs_in_flight) == 0)
937 if (atomic_read(&d->opd_sync_rpcs_in_flight) != old) {
938 /* some progress have been made */
942 /* no changes and expired, something is wrong */
943 GOTO(out, rc = -ETIMEDOUT);
947 /* resume normal processing (barrier=0) */
948 atomic_dec(&d->opd_sync_barrier);
949 osp_sync_check_for_work(d);
951 CDEBUG(D_CACHE, "%s: done in %lld: rc = %d\n", d->opd_obd->obd_name,
952 ktime_get_seconds() - start, rc);
957 static const struct dt_device_operations osp_dt_ops = {
958 .dt_statfs = osp_statfs,
960 .dt_trans_create = osp_trans_create,
961 .dt_trans_start = osp_trans_start,
962 .dt_trans_stop = osp_trans_stop,
963 .dt_trans_cb_add = osp_trans_cb_add,
967 * Connect OSP to local OSD.
969 * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
970 * OSP needs to access the local OSD to store some information. For example,
971 * during precreate, it needs to update last used OID and sequence file
972 * (LAST_SEQ) in local OSD.
974 * \param[in] env execution environment
975 * \param[in] osp OSP device
976 * \param[in] nextdev the name of local OSD
978 * \retval 0 0 connection succeeded
979 * \retval negative negative errno connection failed
981 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
984 struct obd_connect_data *data = NULL;
985 struct obd_device *obd;
990 LASSERT(osp->opd_storage_exp == NULL);
996 obd = class_name2obd(nextdev);
998 CERROR("%s: can't locate next device: %s\n",
999 osp->opd_obd->obd_name, nextdev);
1000 GOTO(out, rc = -ENOTCONN);
1003 rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
1006 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
1007 osp->opd_obd->obd_name, nextdev, rc);
1011 osp->opd_dt_dev.dd_lu_dev.ld_site =
1012 osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
1013 LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
1014 osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
1022 * Determine if the lock needs to be cancelled
1024 * Determine if the unused lock should be cancelled before replay, see
1025 * (ldlm_cancel_no_wait_policy()). Currently, only inode bits lock exists
1028 * \param[in] lock lock to be checked.
1030 * \retval 1 if the lock needs to be cancelled before replay.
1031 * \retval 0 if the lock does not need to be cancelled before
1034 static int osp_cancel_weight(struct ldlm_lock *lock)
1036 if (lock->l_resource->lr_type != LDLM_IBITS)
1043 * Initialize OSP device according to the parameters in the configuration
1046 * Reconstruct the local device name from the configuration profile, and
1047 * initialize necessary threads and structures according to the OSP type
1050 * Since there is no record in the MDT configuration for the local disk
1051 * device, we have to extract this from elsewhere in the profile.
1052 * The only information we get at setup is from the OSC records:
1053 * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
1055 * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
1056 * so, we need to reconstruct the name of the underlying OSD from this:
1057 * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
1059 * \param[in] env execution environment
1060 * \param[in] osp OSP device
1061 * \param[in] ldt lu device type of OSP
1062 * \param[in] cfg configuration log
1064 * \retval 0 0 if OSP initialization succeeded.
1065 * \retval negative negative errno if OSP initialization failed.
1067 static int osp_init0(const struct lu_env *env, struct osp_device *osp,
1068 struct lu_device_type *ldt, struct lustre_cfg *cfg)
1070 struct obd_device *obd;
1071 struct obd_import *imp;
1072 char *src, *tgt, *osdname = NULL;
1079 mutex_init(&osp->opd_async_requests_mutex);
1080 INIT_LIST_HEAD(&osp->opd_async_updates);
1081 init_rwsem(&osp->opd_async_updates_rwsem);
1082 atomic_set(&osp->opd_async_updates_count, 0);
1084 obd = class_name2obd(lustre_cfg_string(cfg, 0));
1086 CERROR("Cannot find obd with name %s\n",
1087 lustre_cfg_string(cfg, 0));
1092 src = lustre_cfg_string(cfg, 0);
1096 tgt = strrchr(src, '-');
1098 CERROR("%s: invalid target name %s: rc = %d\n",
1099 osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
1104 if (strncmp(tgt, "-osc", 4) == 0) {
1105 /* Old OSC name fsname-OSTXXXX-osc */
1106 for (tgt--; tgt > src && *tgt != '-'; tgt--)
1109 CERROR("%s: invalid target name %s: rc = %d\n",
1110 osp->opd_obd->obd_name,
1111 lustre_cfg_string(cfg, 0), -EINVAL);
1115 if (strncmp(tgt, "-OST", 4) != 0) {
1116 CERROR("%s: invalid target name %s: rc = %d\n",
1117 osp->opd_obd->obd_name,
1118 lustre_cfg_string(cfg, 0), -EINVAL);
1122 rc = target_name2index(tgt + 1, &idx, &mdt);
1123 if (rc < 0 || rc & LDD_F_SV_ALL || mdt[0] != '-') {
1124 CERROR("%s: invalid OST index in '%s': rc = %d\n",
1125 osp->opd_obd->obd_name, src, -EINVAL);
1128 osp->opd_index = idx;
1132 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
1133 if (strncmp(tgt, "-MDT", 4) != 0 &&
1134 strncmp(tgt, "-OST", 4) != 0) {
1135 CERROR("%s: invalid target name %s: rc = %d\n",
1136 osp->opd_obd->obd_name,
1137 lustre_cfg_string(cfg, 0), -EINVAL);
1141 rc = target_name2index(tgt + 1, &idx, &mdt);
1142 if (rc < 0 || rc & LDD_F_SV_ALL || *mdt != '\0') {
1143 CERROR("%s: invalid OST index in '%s': rc = %d\n",
1144 osp->opd_obd->obd_name, src, -EINVAL);
1148 /* Get MDT index from the name and set it to opd_group,
1149 * which will be used by OSP to connect with OST */
1150 osp->opd_group = idx;
1151 if (tgt - src <= 12) {
1152 CERROR("%s: invalid mdt index from %s: rc =%d\n",
1153 osp->opd_obd->obd_name,
1154 lustre_cfg_string(cfg, 0), -EINVAL);
1158 if (strncmp(tgt - 12, "-MDT", 4) == 0)
1159 osp->opd_connect_mdt = 1;
1161 rc = target_name2index(tgt - 11, &idx, &mdt);
1162 if (rc < 0 || rc & LDD_F_SV_ALL || mdt[0] != '-') {
1163 CERROR("%s: invalid OST index in '%s': rc =%d\n",
1164 osp->opd_obd->obd_name, src, -EINVAL);
1168 osp->opd_index = idx;
1169 idx = tgt - src - 12;
1171 /* check the fsname length, and after this everything else will fit */
1172 if (idx > MTI_NAME_MAXLEN) {
1173 CERROR("%s: fsname too long in '%s': rc = %d\n",
1174 osp->opd_obd->obd_name, src, -EINVAL);
1178 OBD_ALLOC(osdname, MAX_OBD_NAME);
1179 if (osdname == NULL)
1182 memcpy(osdname, src, idx); /* copy just the fsname part */
1183 osdname[idx] = '\0';
1185 mdt = strstr(mdt, "-MDT");
1186 if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
1187 strcat(osdname, "-MDT0000");
1189 strcat(osdname, mdt);
1190 strcat(osdname, "-osd");
1191 CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
1193 osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
1194 osp->opd_dt_dev.dd_ops = &osp_dt_ops;
1196 obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
1198 rc = osp_connect_to_osd(env, osp, osdname);
1202 rc = ptlrpcd_addref();
1204 GOTO(out_disconnect, rc);
1206 rc = client_obd_setup(obd, cfg);
1208 CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
1213 osp_tunables_init(osp);
1215 rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
1216 LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
1218 CERROR("%s: fid init error: rc = %d\n",
1219 osp->opd_obd->obd_name, rc);
1223 if (!osp->opd_connect_mdt) {
1224 /* Initialize last id from the storage - will be
1225 * used in orphan cleanup. */
1226 if (!osp->opd_storage->dd_rdonly) {
1227 rc = osp_last_used_init(env, osp);
1232 /* Initialize precreation thread, it handles new
1233 * connections as well. */
1234 rc = osp_init_precreate(osp);
1236 GOTO(out_last_used, rc);
1239 * Initialize synhronization mechanism taking
1240 * care of propogating changes to OST in near
1241 * transactional manner.
1243 rc = osp_sync_init(env, osp);
1245 GOTO(out_precreat, rc);
1247 osp->opd_got_disconnected = 1;
1248 rc = osp_update_init(osp);
1253 rc = osp_init_statfs(osp);
1255 GOTO(out_precreat, rc);
1257 ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
1260 * Initiate connect to OST
1262 imp = obd->u.cli.cl_import;
1264 rc = ptlrpc_init_import(imp);
1268 OBD_FREE(osdname, MAX_OBD_NAME);
1269 init_waitqueue_head(&osp->opd_out_waitq);
1273 if (!osp->opd_connect_mdt)
1274 /* stop sync thread */
1277 /* stop precreate thread */
1278 if (!osp->opd_connect_mdt)
1279 osp_precreate_fini(osp);
1281 osp_update_fini(env, osp);
1283 if (!osp->opd_connect_mdt)
1284 osp_last_used_fini(env, osp);
1286 obd_fid_fini(osp->opd_obd);
1288 osp_tunables_fini(osp);
1289 client_obd_cleanup(obd);
1293 obd_disconnect(osp->opd_storage_exp);
1296 OBD_FREE(osdname, MAX_OBD_NAME);
1301 * Implementation of lu_device_type_operations::ldto_device_free
1303 * Free the OSP device in memory. No return value is needed for now,
1304 * so always return NULL to comply with the interface.
1306 * \param[in] env execution environment
1307 * \param[in] lu lu_device of OSP
1309 * \retval NULL NULL unconditionally
1311 static struct lu_device *osp_device_free(const struct lu_env *env,
1312 struct lu_device *lu)
1314 struct osp_device *osp = lu2osp_dev(lu);
1316 lu_site_print(env, lu->ld_site, &lu->ld_ref, D_ERROR,
1318 dt_device_fini(&osp->opd_dt_dev);
1325 * Implementation of lu_device_type_operations::ldto_device_alloc
1327 * This function allocates and initializes OSP device in memory according to
1330 * \param[in] env execution environment
1331 * \param[in] type device type of OSP
1332 * \param[in] lcfg config log
1334 * \retval pointer the pointer of allocated OSP if succeed.
1335 * \retval ERR_PTR(errno) ERR_PTR(errno) if failed.
1337 static struct lu_device *osp_device_alloc(const struct lu_env *env,
1338 struct lu_device_type *type,
1339 struct lustre_cfg *lcfg)
1341 struct osp_device *osp;
1342 struct lu_device *ld;
1346 ld = ERR_PTR(-ENOMEM);
1350 ld = osp2lu_dev(osp);
1351 dt_device_init(&osp->opd_dt_dev, type);
1352 rc = osp_init0(env, osp, type, lcfg);
1354 osp_device_free(env, ld);
1362 * Implementation of lu_device_type_operations::ldto_device_fini
1364 * This function cleans up the OSP device, i.e. release and free those
1365 * attached items in osp_device.
1367 * \param[in] env execution environment
1368 * \param[in] ld lu_device of OSP
1370 * \retval NULL NULL if cleanup succeeded.
1371 * \retval ERR_PTR(errno) ERR_PTR(errno) if cleanup failed.
1373 static struct lu_device *osp_device_fini(const struct lu_env *env,
1374 struct lu_device *ld)
1376 struct osp_device *osp = lu2osp_dev(ld);
1381 if (osp->opd_async_requests != NULL) {
1382 osp_update_request_destroy(env, osp->opd_async_requests);
1383 osp->opd_async_requests = NULL;
1386 if (osp->opd_storage_exp) {
1387 /* wait for the commit callbacks to complete */
1388 wait_event(osp->opd_sync_waitq,
1389 atomic_read(&osp->opd_commits_registered) == 0);
1390 obd_disconnect(osp->opd_storage_exp);
1393 LASSERT(osp->opd_obd);
1395 rc = client_obd_cleanup(osp->opd_obd);
1398 RETURN(ERR_PTR(rc));
1401 osp_tunables_fini(osp);
1409 * Implementation of obd_ops::o_reconnect
1411 * This function is empty and does not need to do anything for now.
1413 static int osp_reconnect(const struct lu_env *env,
1414 struct obd_export *exp, struct obd_device *obd,
1415 struct obd_uuid *cluuid,
1416 struct obd_connect_data *data,
1423 * Implementation of obd_ops::o_connect
1425 * Connect OSP to the remote target (MDT or OST). Allocate the
1426 * export and return it to the LOD, which calls this function
1427 * for each OSP to connect it to the remote target. This function
1428 * is currently only called once per OSP.
1430 * \param[in] env execution environment
1431 * \param[out] exp export connected to OSP
1432 * \param[in] obd OSP device
1433 * \param[in] cluuid OSP device client uuid
1434 * \param[in] data connect_data to be used to connect to the remote
1436 * \param[in] localdata necessary for the API interface, but not used in
1439 * \retval 0 0 if the connection succeeded.
1440 * \retval negative negative errno if the connection failed.
1442 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
1443 struct obd_device *obd, struct obd_uuid *cluuid,
1444 struct obd_connect_data *data, void *localdata)
1446 struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1451 LASSERT(data != NULL);
1452 LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
1454 rc = client_connect_import(env, &osp->opd_exp, obd, cluuid, data,
1459 *exp = osp->opd_exp;
1461 osp->opd_obd->u.cli.cl_seq->lcs_exp = class_export_get(osp->opd_exp);
1462 /* precreate thread can be waiting for us to initialize fld client */
1463 wake_up(&osp->opd_pre_waitq);
1469 * Implementation of obd_ops::o_disconnect
1471 * Disconnect the export for the OSP. This is called by LOD to release the
1472 * OSP during cleanup (\see lod_del_device()). The OSP will be released after
1473 * the export is released.
1475 * \param[in] exp export to be disconnected.
1477 * \retval 0 0 if disconnection succeed
1478 * \retval negative negative errno if disconnection failed
1480 static int osp_obd_disconnect(struct obd_export *exp)
1482 struct obd_device *obd = exp->exp_obd;
1486 rc = class_disconnect(exp);
1488 CERROR("%s: class disconnect error: rc = %d\n",
1493 /* destroy the device */
1494 class_manual_cleanup(obd);
1500 * Implementation of obd_ops::o_statfs
1502 * Send a RPC to the remote target to get statfs status. This is only used
1503 * in lprocfs helpers by obd_statfs.
1505 * \param[in] env execution environment
1506 * \param[in] exp connection state from this OSP to the parent (LOD)
1508 * \param[out] osfs hold the statfs result
1509 * \param[in] unused Not used in this function for now
1510 * \param[in] flags flags to indicate how OSP will issue the RPC
1512 * \retval 0 0 if statfs succeeded.
1513 * \retval negative negative errno if statfs failed.
1515 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
1516 struct obd_statfs *osfs, time64_t unused, __u32 flags)
1518 struct obd_statfs *msfs;
1519 struct ptlrpc_request *req;
1520 struct obd_import *imp = NULL, *imp0;
1525 /* Since the request might also come from lprocfs, so we need
1526 * sync this with client_disconnect_export Bug15684
1528 with_imp_locked(exp->exp_obd, imp0, rc)
1529 imp = class_import_get(imp0);
1533 req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
1535 class_import_put(imp);
1540 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
1542 ptlrpc_request_free(req);
1545 ptlrpc_request_set_replen(req);
1546 req->rq_request_portal = OST_CREATE_PORTAL;
1547 ptlrpc_at_set_req_timeout(req);
1549 if (flags & OBD_STATFS_NODELAY) {
1550 /* procfs requests not want stat in wait for avoid deadlock */
1551 req->rq_no_resend = 1;
1552 req->rq_no_delay = 1;
1555 rc = ptlrpc_queue_wait(req);
1559 msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1561 GOTO(out, rc = -EPROTO);
1567 ptlrpc_req_finished(req);
1572 * Implementation of obd_ops::o_import_event
1574 * This function is called when some related import event happens. It will
1575 * mark the necessary flags according to the event and notify the necessary
1576 * threads (mainly precreate thread).
1578 * \param[in] obd OSP OBD device
1579 * \param[in] imp import attached from OSP to remote (OST/MDT) service
1580 * \param[in] event event related to remote service (IMP_EVENT_*)
1582 * \retval 0 0 if the event handling succeeded.
1583 * \retval negative negative errno if the event handling failed.
1585 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1586 enum obd_import_event event)
1588 struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1592 case IMP_EVENT_DISCON:
1593 d->opd_got_disconnected = 1;
1594 d->opd_imp_connected = 0;
1595 if (d->opd_connect_mdt)
1598 if (d->opd_pre != NULL) {
1599 osp_pre_update_status(d, -ENODEV);
1600 wake_up(&d->opd_pre_waitq);
1603 CDEBUG(D_HA, "got disconnected\n");
1605 case IMP_EVENT_INACTIVE:
1606 d->opd_imp_active = 0;
1607 d->opd_imp_connected = 0;
1608 d->opd_obd->obd_inactive = 1;
1609 if (d->opd_connect_mdt)
1611 if (d->opd_pre != NULL) {
1612 /* Import is invalid, we can`t get stripes so
1614 rc = imp->imp_deactive ? -ESHUTDOWN : -ENODEV;
1615 osp_pre_update_status(d, rc);
1616 wake_up(&d->opd_pre_waitq);
1619 CDEBUG(D_HA, "got inactive\n");
1621 case IMP_EVENT_ACTIVE:
1622 d->opd_imp_active = 1;
1624 d->opd_new_connection = 1;
1625 d->opd_imp_connected = 1;
1626 d->opd_imp_seen_connected = 1;
1627 d->opd_obd->obd_inactive = 0;
1628 wake_up(&d->opd_pre_waitq);
1629 if (d->opd_connect_mdt)
1632 osp_sync_check_for_work(d);
1633 CDEBUG(D_HA, "got connected\n");
1635 case IMP_EVENT_INVALIDATE:
1636 if (d->opd_connect_mdt)
1637 osp_invalidate_request(d);
1639 if (obd->obd_namespace == NULL)
1641 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1644 case IMP_EVENT_DEACTIVATE:
1645 case IMP_EVENT_ACTIVATE:
1648 CERROR("%s: unsupported import event: %#x\n",
1649 obd->obd_name, event);
1655 * Implementation of obd_ops: o_iocontrol
1657 * This function is the ioctl handler for OSP. Note: lctl will access the OSP
1658 * directly by ioctl, instead of through the MDS stack.
1660 * param[in] cmd ioctl command.
1661 * param[in] exp export of this OSP.
1662 * param[in] len data length of \a karg.
1663 * param[in] karg input argument which is packed as
1665 * param[out] uarg pointer to userspace buffer (must access by
1668 * \retval 0 0 if the ioctl handling succeeded.
1669 * \retval negative negative errno if the ioctl handling failed.
1671 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1672 void *karg, void __user *uarg)
1674 struct obd_device *obd = exp->exp_obd;
1675 struct osp_device *d;
1676 struct obd_ioctl_data *data;
1680 CDEBUG(D_IOCTL, "%s: cmd=%x len=%u karg=%pK uarg=%pK\n",
1681 exp->exp_obd->obd_name, cmd, len, karg, uarg);
1682 if (unlikely(karg == NULL)) {
1683 CERROR("%s: iocontrol from '%s' cmd=%x karg=NULL: rc = %d\n",
1684 obd->obd_name, current->comm, cmd, rc);
1689 LASSERT(obd->obd_lu_dev);
1690 d = lu2osp_dev(obd->obd_lu_dev);
1691 LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1693 if (!try_module_get(THIS_MODULE)) {
1694 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
1695 module_name(THIS_MODULE));
1700 case OBD_IOC_CLIENT_RECOVER:
1701 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1702 data->ioc_inlbuf1, 0);
1706 #ifdef IOC_OSC_SET_ACTIVE
1707 case_OBD_IOC_DEPRECATED_FT(IOC_OSC_SET_ACTIVE, obd->obd_name, 2, 17);
1709 case OBD_IOC_SET_ACTIVE:
1710 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1714 rc = OBD_IOC_ERROR(obd->obd_name, cmd, "unrecognized", -ENOTTY);
1717 module_put(THIS_MODULE);
1722 * Implementation of obd_ops::o_get_info
1724 * Retrieve information by key. Retrieval starts from the top layer
1725 * (MDT) of the MDS stack and traverses the stack by calling the
1726 * obd_get_info() method of the next sub-layer.
1728 * \param[in] env execution environment
1729 * \param[in] exp export of this OSP
1730 * \param[in] keylen length of \a key
1731 * \param[in] key the key
1732 * \param[out] vallen length of \a val
1733 * \param[out] val holds the value returned by the key
1735 * \retval 0 0 if getting information succeeded.
1736 * \retval negative negative errno if getting information failed.
1738 static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1739 __u32 keylen, void *key, __u32 *vallen, void *val)
1743 if (KEY_IS(KEY_OSP_CONNECTED)) {
1744 struct obd_device *obd = exp->exp_obd;
1745 struct osp_device *osp;
1747 if (!obd->obd_set_up || obd->obd_stopping)
1750 osp = lu2osp_dev(obd->obd_lu_dev);
1753 * 1.8/2.0 behaviour is that OST being connected once at least
1754 * is considered "healthy". and one "healthy" OST is enough to
1755 * allow lustre clients to connect to MDS
1757 RETURN(!osp->opd_imp_seen_connected);
1763 static int osp_obd_set_info_async(const struct lu_env *env,
1764 struct obd_export *exp,
1765 u32 keylen, void *key,
1766 u32 vallen, void *val,
1767 struct ptlrpc_request_set *set)
1769 struct obd_device *obd = exp->exp_obd;
1770 struct obd_import *imp = obd->u.cli.cl_import;
1771 struct osp_device *osp;
1772 struct ptlrpc_request *req;
1776 if (KEY_IS(KEY_SPTLRPC_CONF)) {
1777 sptlrpc_conf_client_adapt(exp->exp_obd);
1781 LASSERT(set != NULL);
1782 if (!obd->obd_set_up || obd->obd_stopping)
1784 osp = lu2osp_dev(obd->obd_lu_dev);
1786 req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
1790 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
1791 RCL_CLIENT, keylen);
1792 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
1793 RCL_CLIENT, vallen);
1794 if (osp->opd_connect_mdt)
1795 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SET_INFO);
1797 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
1799 ptlrpc_request_free(req);
1803 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
1804 memcpy(tmp, key, keylen);
1805 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
1806 memcpy(tmp, val, vallen);
1808 ptlrpc_request_set_replen(req);
1809 ptlrpc_set_add_req(set, req);
1810 ptlrpc_check_set(NULL, set);
1815 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1816 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1817 static void osp_key_exit(const struct lu_context *ctx,
1818 struct lu_context_key *key, void *data)
1820 struct osp_thread_info *info = data;
1822 info->osi_attr.la_valid = 0;
1825 struct lu_context_key osp_thread_key = {
1826 .lct_tags = LCT_MD_THREAD,
1827 .lct_init = osp_key_init,
1828 .lct_fini = osp_key_fini,
1829 .lct_exit = osp_key_exit
1832 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1833 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1835 struct lu_context_key osp_txn_key = {
1836 .lct_tags = LCT_OSP_THREAD,
1837 .lct_init = osp_txn_key_init,
1838 .lct_fini = osp_txn_key_fini
1840 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1842 static const struct lu_device_type_operations osp_device_type_ops = {
1843 .ldto_init = osp_type_init,
1844 .ldto_fini = osp_type_fini,
1846 .ldto_start = osp_type_start,
1847 .ldto_stop = osp_type_stop,
1849 .ldto_device_alloc = osp_device_alloc,
1850 .ldto_device_free = osp_device_free,
1852 .ldto_device_fini = osp_device_fini
1855 static struct lu_device_type osp_device_type = {
1856 .ldt_tags = LU_DEVICE_DT,
1857 .ldt_name = LUSTRE_OSP_NAME,
1858 .ldt_ops = &osp_device_type_ops,
1859 .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
1862 static const struct obd_ops osp_obd_device_ops = {
1863 .o_owner = THIS_MODULE,
1864 .o_add_conn = client_import_add_conn,
1865 .o_del_conn = client_import_del_conn,
1866 .o_reconnect = osp_reconnect,
1867 .o_connect = osp_obd_connect,
1868 .o_disconnect = osp_obd_disconnect,
1869 .o_get_info = osp_obd_get_info,
1870 .o_set_info_async = osp_obd_set_info_async,
1871 .o_import_event = osp_import_event,
1872 .o_iocontrol = osp_iocontrol,
1873 .o_statfs = osp_obd_statfs,
1874 .o_fid_init = client_fid_init,
1875 .o_fid_fini = client_fid_fini,
1879 * Initialize OSP module.
1881 * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
1882 * in obd_types (\see class_obd.c). Initialize procfs for the
1883 * the OSP device. Note: OSP was called OSC before Lustre 2.4,
1884 * so for compatibility it still uses the name "osc" in procfs.
1885 * This is called at module load time.
1887 * \retval 0 0 if initialization succeeds.
1888 * \retval negative negative errno if initialization failed.
1890 static int __init osp_init(void)
1892 struct obd_type *sym;
1895 rc = libcfs_setup();
1899 rc = lu_kmem_init(osp_caches);
1903 rc = class_register_type(&osp_obd_device_ops, NULL, false,
1904 LUSTRE_OSP_NAME, &osp_device_type);
1906 lu_kmem_fini(osp_caches);
1910 rc = class_register_type(&lwp_obd_device_ops, NULL, false,
1911 LUSTRE_LWP_NAME, &lwp_device_type);
1913 class_unregister_type(LUSTRE_OSP_NAME);
1914 lu_kmem_fini(osp_caches);
1918 /* create "osc" entry for compatibility purposes */
1919 sym = class_add_symlinks(LUSTRE_OSC_NAME, true);
1922 /* does real "osc" already exist ? */
1931 * Finalize OSP module.
1933 * This callback is called when kernel unloads OSP module from memory, and
1934 * it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
1936 static void __exit osp_exit(void)
1938 struct obd_type *sym = class_search_type(LUSTRE_OSC_NAME);
1940 /* if this was never fully initialized by the osc layer
1941 * then we are responsible for freeing this obd_type
1944 /* final put if we manage this obd type */
1945 if (sym->typ_sym_filter)
1946 kobject_put(&sym->typ_kobj);
1947 /* put reference taken by class_search_type */
1948 kobject_put(&sym->typ_kobj);
1951 class_unregister_type(LUSTRE_LWP_NAME);
1952 class_unregister_type(LUSTRE_OSP_NAME);
1953 lu_kmem_fini(osp_caches);
1956 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
1957 MODULE_DESCRIPTION("Lustre OSD Storage Proxy ("LUSTRE_OSP_NAME")");
1958 MODULE_VERSION(LUSTRE_VERSION_STRING);
1959 MODULE_LICENSE("GPL");
1961 module_init(osp_init);
1962 module_exit(osp_exit);