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/
30 * Lustre is a trademark of Sun Microsystems, Inc.
32 * lustre/osp/osp_dev.c
34 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
35 * Author: Mikhail Pershin <mike.pershin@intel.com>
36 * Author: Di Wang <di.wang@intel.com>
39 * The Object Storage Proxy (OSP) module provides an implementation of
40 * the DT API for remote MDTs and OSTs. Every local OSP device (or
41 * object) is a proxy for a remote OSD device (or object). Thus OSP
42 * converts DT operations into RPCs, which are sent to the OUT service
43 * on a remote target, converted back to DT operations, and
44 * executed. Of course there are many ways in which this description
45 * is inaccurate but it's a good enough mental model. OSP is used by
46 * the MDT stack in several ways:
48 * - OSP devices allocate FIDs for the stripe sub-objects of a striped
51 * - OSP objects represent the remote MDT and OST objects that are
52 * the stripes of a striped object.
54 * - OSP devices log, send, and track synchronous operations (setattr
55 * and unlink) to remote targets.
57 * - OSP objects are the bottom slice of the compound LU object
58 * representing a remote MDT object: MDT/MDD/LOD/OSP.
60 * - OSP objects are used by LFSCK to represent remote OST objects
61 * during the verification of MDT-OST consistency.
63 * - OSP devices batch idempotent requests (declare_attr_get() and
64 * declare_xattr_get()) to the remote target and cache their results.
66 * In addition the OSP layer implements a subset of the OBD device API
67 * to support being a client of a remote target, connecting to other
68 * layers, and FID allocation.
71 #define DEBUG_SUBSYSTEM S_MDS
73 #include <linux/kthread.h>
75 #include <uapi/linux/lustre/lustre_ioctl.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 lu_object_header *h = NULL;
117 struct osp_object *o;
120 OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
122 l = &o->opo_obj.do_lu;
124 /* If hdr is NULL, it means the object is not built
125 * from the top dev(MDT/OST), usually it happens when
126 * building striped object, like data object on MDT or
127 * striped object for directory */
130 lu_object_header_init(h);
131 dt_object_init(&o->opo_obj, h, d);
132 lu_object_add_top(h, l);
134 dt_object_init(&o->opo_obj, h, d);
137 l->lo_ops = &osp_lu_obj_ops;
146 * Find or create the local object
148 * Finds or creates the local file referenced by \a reg_id and return the
149 * attributes of the local file.
151 * \param[in] env execution environment
152 * \param[in] osp OSP device
153 * \param[out] attr attributes of the object
154 * \param[in] reg_id the local object ID of the file. It will be used
155 * to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
156 * to identify the object.
158 * \retval object object(dt_object) found or created
159 * \retval ERR_PTR(errno) ERR_PTR(errno) if not get the object.
161 static struct dt_object
162 *osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
163 struct lu_attr *attr, __u32 reg_id)
165 struct osp_thread_info *osi = osp_env_info(env);
166 struct dt_object_format dof = { 0 };
167 struct dt_object *dto;
171 lu_local_obj_fid(&osi->osi_fid, reg_id);
172 attr->la_valid = LA_MODE;
173 attr->la_mode = S_IFREG | 0644;
174 dof.dof_type = DFT_REGULAR;
175 /* Find or create the local object by osi_fid. */
176 dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
181 /* Get attributes of the local object. */
182 rc = dt_attr_get(env, dto, attr);
184 CERROR("%s: can't be initialized: rc = %d\n",
185 osp->opd_obd->obd_name, rc);
186 dt_object_put(env, dto);
193 * Write data buffer to a local file object.
195 * \param[in] env execution environment
196 * \param[in] osp OSP device
197 * \param[in] dt_obj object written to
198 * \param[in] buf buffer containing byte array and length
199 * \param[in] offset write offset in the object in bytes
201 * \retval 0 0 if write succeed
202 * \retval -EFAULT -EFAULT if only part of buffer is written.
203 * \retval negative other negative errno if write failed.
205 static int osp_write_local_file(const struct lu_env *env,
206 struct osp_device *osp,
207 struct dt_object *dt_obj,
214 if (osp->opd_storage->dd_rdonly)
217 th = dt_trans_create(env, osp->opd_storage);
221 rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
224 rc = dt_trans_start_local(env, osp->opd_storage, th);
228 rc = dt_record_write(env, dt_obj, buf, &offset, th);
230 dt_trans_stop(env, osp->opd_storage, th);
235 * Initialize last ID object.
237 * This function initializes the LAST_ID file, which stores the current last
238 * used id of data objects. The MDT will use the last used id and the last_seq
239 * (\see osp_init_last_seq()) to synchronize the precreate object cache with
242 * \param[in] env execution environment
243 * \param[in] osp OSP device
245 * \retval 0 0 if initialization succeed
246 * \retval negative negative errno if initialization failed
248 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
250 struct osp_thread_info *osi = osp_env_info(env);
251 struct lu_fid *fid = &osp->opd_last_used_fid;
252 struct dt_object *dto;
256 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
259 RETURN(PTR_ERR(dto));
261 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &osp->opd_last_id,
264 /* object will be released in device cleanup path */
265 if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
266 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
267 if (rc != 0 && rc != -EFAULT)
269 /* In case of idif bits 32-48 go to f_seq
270 * (see osp_init_last_seq). So don't care
271 * about u64->u32 convertion. */
272 fid->f_oid = osp->opd_last_id;
275 if (rc == -EFAULT) { /* fresh LAST_ID */
276 osp->opd_last_id = 0;
278 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
283 osp->opd_last_used_oid_file = dto;
286 /* object will be released in device cleanup path */
287 CERROR("%s: can't initialize lov_objid: rc = %d\n",
288 osp->opd_obd->obd_name, rc);
289 dt_object_put(env, dto);
290 osp->opd_last_used_oid_file = NULL;
295 * Initialize last sequence object.
297 * This function initializes the LAST_SEQ file in the local OSD, which stores
298 * the current last used sequence of data objects. The MDT will use the last
299 * sequence and last id (\see osp_init_last_objid()) to synchronize the
300 * precreate object cache with OSTs.
302 * \param[in] env execution environment
303 * \param[in] osp OSP device
305 * \retval 0 0 if initialization succeed
306 * \retval negative negative errno if initialization failed
308 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
310 struct osp_thread_info *osi = osp_env_info(env);
311 struct lu_fid *fid = &osp->opd_last_used_fid;
312 struct dt_object *dto;
316 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
319 RETURN(PTR_ERR(dto));
321 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
324 /* object will be released in device cleanup path */
325 if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
326 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
327 if (rc != 0 && rc != -EFAULT)
329 if (fid_is_idif(fid))
330 fid->f_seq = fid_idif_seq(osp->opd_last_id,
334 if (rc == -EFAULT) { /* fresh OSP */
336 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
341 osp->opd_last_used_seq_file = dto;
344 /* object will be released in device cleanup path */
345 CERROR("%s: can't initialize lov_seq: rc = %d\n",
346 osp->opd_obd->obd_name, rc);
347 dt_object_put(env, dto);
348 osp->opd_last_used_seq_file = NULL;
353 * Initialize last OID and sequence object.
355 * If the MDT is just upgraded to 2.4 from the lower version, where the
356 * LAST_SEQ file does not exist, the file will be created and IDIF sequence
357 * will be written into the file.
359 * \param[in] env execution environment
360 * \param[in] osp OSP device
362 * \retval 0 0 if initialization succeed
363 * \retval negative negative error if initialization failed
365 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
367 struct osp_thread_info *osi = osp_env_info(env);
371 fid_zero(&osp->opd_last_used_fid);
372 rc = osp_init_last_objid(env, osp);
374 CERROR("%s: Can not get ids %d from old objid!\n",
375 osp->opd_obd->obd_name, rc);
379 rc = osp_init_last_seq(env, osp);
381 CERROR("%s: Can not get sequence %d from old objseq!\n",
382 osp->opd_obd->obd_name, rc);
386 if (fid_oid(&osp->opd_last_used_fid) != 0 &&
387 fid_seq(&osp->opd_last_used_fid) == 0) {
388 /* Just upgrade from the old version,
389 * set the seq to be IDIF */
390 osp->opd_last_used_fid.f_seq =
391 fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
393 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
394 &osp->opd_last_used_fid.f_seq,
396 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
397 &osi->osi_lb, osi->osi_off);
399 CERROR("%s : Can not write seq file: rc = %d\n",
400 osp->opd_obd->obd_name, rc);
405 if (!fid_is_zero(&osp->opd_last_used_fid) &&
406 !fid_is_sane(&osp->opd_last_used_fid)) {
407 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
408 PFID(&osp->opd_last_used_fid));
409 GOTO(out, rc = -EINVAL);
412 osp_fid_to_obdid(&osp->opd_last_used_fid, &osp->opd_last_id);
413 CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
414 osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
417 if (osp->opd_last_used_oid_file != NULL) {
418 dt_object_put(env, osp->opd_last_used_oid_file);
419 osp->opd_last_used_oid_file = NULL;
421 if (osp->opd_last_used_seq_file != NULL) {
422 dt_object_put(env, osp->opd_last_used_seq_file);
423 osp->opd_last_used_seq_file = NULL;
431 * Release the last sequence and OID file objects in OSP device.
433 * \param[in] env execution environment
434 * \param[in] osp OSP device
436 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
438 /* release last_used file */
439 if (osp->opd_last_used_oid_file != NULL) {
440 dt_object_put(env, osp->opd_last_used_oid_file);
441 osp->opd_last_used_oid_file = NULL;
444 if (osp->opd_last_used_seq_file != NULL) {
445 dt_object_put(env, osp->opd_last_used_seq_file);
446 osp->opd_last_used_seq_file = NULL;
451 * Disconnects the connection between OSP and its correspondent MDT or OST, and
452 * the import will be marked as inactive. It will only be called during OSP
455 * \param[in] d OSP device being disconnected
457 * \retval 0 0 if disconnection succeed
458 * \retval negative negative errno if disconnection failed
460 static int osp_disconnect(struct osp_device *d)
462 struct obd_device *obd = d->opd_obd;
463 struct obd_import *imp;
466 imp = obd->u.cli.cl_import;
468 /* Mark import deactivated now, so we don't try to reconnect if any
469 * of the cleanup RPCs fails (e.g. ldlm cancel, etc). We don't
470 * fully deactivate the import, or that would drop all requests. */
471 LASSERT(imp != NULL);
472 spin_lock(&imp->imp_lock);
473 imp->imp_deactive = 1;
474 spin_unlock(&imp->imp_lock);
476 ptlrpc_deactivate_import(imp);
478 /* Some non-replayable imports (MDS's OSCs) are pinged, so just
479 * delete it regardless. (It's safe to delete an import that was
481 (void)ptlrpc_pinger_del_import(imp);
483 rc = ptlrpc_disconnect_import(imp, 0);
485 CERROR("%s: can't disconnect: rc = %d\n", obd->obd_name, rc);
487 ptlrpc_invalidate_import(imp);
493 * Initialize the osp_update structure in OSP device
495 * Allocate osp update structure and start update thread.
497 * \param[in] osp OSP device
499 * \retval 0 if initialization succeeds.
500 * \retval negative errno if initialization fails.
502 static int osp_update_init(struct osp_device *osp)
504 struct l_wait_info lwi = { 0 };
505 struct task_struct *task;
509 LASSERT(osp->opd_connect_mdt);
511 if (osp->opd_storage->dd_rdonly)
514 OBD_ALLOC_PTR(osp->opd_update);
515 if (osp->opd_update == NULL)
518 init_waitqueue_head(&osp->opd_update_thread.t_ctl_waitq);
519 init_waitqueue_head(&osp->opd_update->ou_waitq);
520 spin_lock_init(&osp->opd_update->ou_lock);
521 INIT_LIST_HEAD(&osp->opd_update->ou_list);
522 osp->opd_update->ou_rpc_version = 1;
523 osp->opd_update->ou_version = 1;
524 osp->opd_update->ou_generation = 0;
526 /* start thread handling sending updates to the remote MDT */
527 task = kthread_run(osp_send_update_thread, osp,
528 "osp_up%u-%u", osp->opd_index, osp->opd_group);
530 int rc = PTR_ERR(task);
532 OBD_FREE_PTR(osp->opd_update);
533 osp->opd_update = NULL;
534 CERROR("%s: can't start precreate thread: rc = %d\n",
535 osp->opd_obd->obd_name, rc);
539 l_wait_event(osp->opd_update_thread.t_ctl_waitq,
540 osp_send_update_thread_running(osp) ||
541 osp_send_update_thread_stopped(osp), &lwi);
547 * Finialize osp_update structure in OSP device
549 * Stop the OSP update sending thread, then delete the left
550 * osp thandle in the sending list.
552 * \param [in] osp OSP device.
554 static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
556 struct osp_update_request *our;
557 struct osp_update_request *tmp;
558 struct osp_updates *ou = osp->opd_update;
563 osp->opd_update_thread.t_flags = SVC_STOPPING;
564 wake_up(&ou->ou_waitq);
566 wait_event(osp->opd_update_thread.t_ctl_waitq,
567 osp->opd_update_thread.t_flags & SVC_STOPPED);
569 /* Remove the left osp thandle from the list */
570 spin_lock(&ou->ou_lock);
571 list_for_each_entry_safe(our, tmp, &ou->ou_list,
573 list_del_init(&our->our_list);
574 LASSERT(our->our_th != NULL);
575 osp_trans_callback(env, our->our_th, -EIO);
576 /* our will be destroyed in osp_thandle_put() */
577 osp_thandle_put(env, our->our_th);
579 spin_unlock(&ou->ou_lock);
582 osp->opd_update = NULL;
586 * Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
587 * precreate threads etc.
589 * \param[in] env execution environment.
590 * \param[in] d OSP device being disconnected.
592 * \retval 0 0 if cleanup succeed
593 * \retval negative negative errno if cleanup failed
595 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
602 rc = osp_disconnect(d);
606 if (!d->opd_connect_mdt) {
607 /* stop sync thread */
610 /* stop precreate thread */
611 osp_precreate_fini(d);
613 /* release last_used file */
614 osp_last_used_fini(env, d);
617 obd_fid_fini(d->opd_obd);
623 * Implementation of osp_lu_ops::ldo_process_config
625 * This function processes config log records in OSP layer. It is usually
626 * called from the top layer of MDT stack, and goes through the stack by calling
627 * ldo_process_config of next layer.
629 * \param[in] env execution environment
630 * \param[in] dev lu_device of OSP
631 * \param[in] lcfg config log
633 * \retval 0 0 if the config log record is executed correctly.
634 * \retval negative negative errno if the record execution fails.
636 static int osp_process_config(const struct lu_env *env,
637 struct lu_device *dev, struct lustre_cfg *lcfg)
639 struct osp_device *d = lu2osp_dev(dev);
640 struct dt_device *dt = lu2dt_dev(dev);
641 struct obd_device *obd = d->opd_obd;
647 switch (lcfg->lcfg_command) {
648 case LCFG_PRE_CLEANUP:
649 rc = osp_disconnect(d);
650 osp_update_fini(env, d);
651 if (obd->obd_namespace != NULL)
652 ldlm_namespace_free_prior(obd->obd_namespace, NULL, 1);
655 lu_dev_del_linkage(dev->ld_site, dev);
656 rc = osp_shutdown(env, d);
659 count = class_modify_config(lcfg, d->opd_connect_mdt ?
660 PARAM_OSP : PARAM_OSC,
663 /* class_modify_config() haven't found matching
664 * parameter and returned an error so that layer(s)
665 * below could use that. But OSP is the bottom, so
668 CERROR("%s: unknown param %s\n",
669 (char *)lustre_cfg_string(lcfg, 0),
670 (char *)lustre_cfg_string(lcfg, 1));
675 CERROR("%s: unknown command %u\n",
676 (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
685 * Implementation of osp_lu_ops::ldo_recovery_complete
687 * This function is called after recovery is finished, and OSP layer
688 * will wake up precreate thread here.
690 * \param[in] env execution environment
691 * \param[in] dev lu_device of OSP
693 * \retval 0 0 unconditionally
695 static int osp_recovery_complete(const struct lu_env *env,
696 struct lu_device *dev)
698 struct osp_device *osp = lu2osp_dev(dev);
701 osp->opd_recovery_completed = 1;
703 if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
704 wake_up(&osp->opd_pre_waitq);
709 const struct lu_device_operations osp_lu_ops = {
710 .ldo_object_alloc = osp_object_alloc,
711 .ldo_process_config = osp_process_config,
712 .ldo_recovery_complete = osp_recovery_complete,
716 * Implementation of dt_device_operations::dt_statfs
718 * This function provides statfs status (for precreation) from
719 * corresponding OST. Note: this function only retrieves the status
720 * from the OSP device, and the real statfs RPC happens inside
721 * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
722 * not need to retrieve statfs data for now.
724 * \param[in] env execution environment.
725 * \param[in] dev dt_device of OSP.
726 * \param[out] sfs holds the retrieved statfs data.
728 * \retval 0 0 statfs data was retrieved successfully or
729 * retrieval was not needed
730 * \retval negative negative errno if get statfs failed.
732 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
733 struct obd_statfs *sfs)
735 struct osp_device *d = dt2osp_dev(dev);
736 struct obd_import *imp = d->opd_obd->u.cli.cl_import;
740 if (imp->imp_state == LUSTRE_IMP_CLOSED)
743 if (unlikely(d->opd_imp_active == 0))
746 /* return recently updated data */
747 *sfs = d->opd_statfs;
749 if (d->opd_pre == NULL)
753 * layer above osp (usually lod) can use ffree to estimate
754 * how many objects are available for immediate creation
756 spin_lock(&d->opd_pre_lock);
757 LASSERTF(fid_seq(&d->opd_pre_last_created_fid) ==
758 fid_seq(&d->opd_pre_used_fid),
759 "last_created "DFID", next_fid "DFID"\n",
760 PFID(&d->opd_pre_last_created_fid),
761 PFID(&d->opd_pre_used_fid));
762 sfs->os_fprecreated = fid_oid(&d->opd_pre_last_created_fid) -
763 fid_oid(&d->opd_pre_used_fid);
764 sfs->os_fprecreated -= d->opd_pre_reserved;
765 LASSERTF(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2,
766 "last_created "DFID", next_fid "DFID", reserved %llu\n",
767 PFID(&d->opd_pre_last_created_fid), PFID(&d->opd_pre_used_fid),
768 d->opd_pre_reserved);
769 spin_unlock(&d->opd_pre_lock);
771 CDEBUG(D_OTHER, "%s: %llu blocks, %llu free, %llu avail, "
772 "%llu files, %llu free files\n", d->opd_obd->obd_name,
773 sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
774 sfs->os_files, sfs->os_ffree);
778 static int osp_sync_timeout(void *data)
784 * Implementation of dt_device_operations::dt_sync
786 * This function synchronizes the OSP cache to the remote target. It wakes
787 * up unlink log threads and sends out unlink records to the remote OST.
789 * \param[in] env execution environment
790 * \param[in] dev dt_device of OSP
792 * \retval 0 0 if synchronization succeeds
793 * \retval negative negative errno if synchronization fails
795 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
797 struct osp_device *d = dt2osp_dev(dev);
798 struct l_wait_info lwi = { 0 };
799 time64_t start = ktime_get_seconds();
805 /* No Sync between MDTs yet. */
806 if (d->opd_connect_mdt)
809 recs = atomic_read(&d->opd_sync_changes);
810 old = atomic64_read(&d->opd_sync_processed_recs);
812 osp_sync_force(env, dt2osp_dev(dev));
814 if (unlikely(d->opd_imp_active == 0))
817 down_write(&d->opd_async_updates_rwsem);
819 CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
820 atomic_read(&d->opd_async_updates_count));
822 /* make sure the connection is fine */
823 lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout), osp_sync_timeout, d);
824 rc = l_wait_event(d->opd_sync_barrier_waitq,
825 atomic_read(&d->opd_async_updates_count) == 0,
827 up_write(&d->opd_async_updates_rwsem);
831 CDEBUG(D_CACHE, "%s: processed %lu\n", d->opd_obd->obd_name,
832 atomic64_read(&d->opd_sync_processed_recs));
834 while (atomic64_read(&d->opd_sync_processed_recs) < old + recs) {
835 __u64 last = atomic64_read(&d->opd_sync_processed_recs);
836 /* make sure the connection is fine */
837 lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
838 osp_sync_timeout, d);
839 l_wait_event(d->opd_sync_barrier_waitq,
840 atomic64_read(&d->opd_sync_processed_recs)
841 >= old + recs, &lwi);
843 if (atomic64_read(&d->opd_sync_processed_recs) >= old + recs)
846 if (atomic64_read(&d->opd_sync_processed_recs) != last) {
847 /* some progress have been made,
852 /* no changes and expired, something is wrong */
853 GOTO(out, rc = -ETIMEDOUT);
856 /* block new processing (barrier>0 - few callers are possible */
857 atomic_inc(&d->opd_sync_barrier);
859 CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
860 atomic_read(&d->opd_sync_rpcs_in_flight));
862 /* wait till all-in-flight are replied, so executed by the target */
863 /* XXX: this is used by LFSCK at the moment, which doesn't require
864 * all the changes to be committed, but in general it'd be
865 * better to wait till commit */
866 while (atomic_read(&d->opd_sync_rpcs_in_flight) > 0) {
867 old = atomic_read(&d->opd_sync_rpcs_in_flight);
869 lwi = LWI_TIMEOUT(cfs_time_seconds(obd_timeout),
870 osp_sync_timeout, d);
871 l_wait_event(d->opd_sync_barrier_waitq,
872 atomic_read(&d->opd_sync_rpcs_in_flight) == 0,
875 if (atomic_read(&d->opd_sync_rpcs_in_flight) == 0)
878 if (atomic_read(&d->opd_sync_rpcs_in_flight) != old) {
879 /* some progress have been made */
883 /* no changes and expired, something is wrong */
884 GOTO(out, rc = -ETIMEDOUT);
888 /* resume normal processing (barrier=0) */
889 atomic_dec(&d->opd_sync_barrier);
890 osp_sync_check_for_work(d);
892 CDEBUG(D_CACHE, "%s: done in %lld: rc = %d\n", d->opd_obd->obd_name,
893 ktime_get_seconds() - start, rc);
898 const struct dt_device_operations osp_dt_ops = {
899 .dt_statfs = osp_statfs,
901 .dt_trans_create = osp_trans_create,
902 .dt_trans_start = osp_trans_start,
903 .dt_trans_stop = osp_trans_stop,
904 .dt_trans_cb_add = osp_trans_cb_add,
908 * Connect OSP to local OSD.
910 * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
911 * OSP needs to access the local OSD to store some information. For example,
912 * during precreate, it needs to update last used OID and sequence file
913 * (LAST_SEQ) in local OSD.
915 * \param[in] env execution environment
916 * \param[in] osp OSP device
917 * \param[in] nextdev the name of local OSD
919 * \retval 0 0 connection succeeded
920 * \retval negative negative errno connection failed
922 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
925 struct obd_connect_data *data = NULL;
926 struct obd_device *obd;
931 LASSERT(osp->opd_storage_exp == NULL);
937 obd = class_name2obd(nextdev);
939 CERROR("%s: can't locate next device: %s\n",
940 osp->opd_obd->obd_name, nextdev);
941 GOTO(out, rc = -ENOTCONN);
944 rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
947 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
948 osp->opd_obd->obd_name, nextdev, rc);
952 osp->opd_dt_dev.dd_lu_dev.ld_site =
953 osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
954 LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
955 osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
963 * Determine if the lock needs to be cancelled
965 * Determine if the unused lock should be cancelled before replay, see
966 * (ldlm_cancel_no_wait_policy()). Currently, only inode bits lock exists
969 * \param[in] lock lock to be checked.
971 * \retval 1 if the lock needs to be cancelled before replay.
972 * \retval 0 if the lock does not need to be cancelled before
975 static int osp_cancel_weight(struct ldlm_lock *lock)
977 if (lock->l_resource->lr_type != LDLM_IBITS)
984 * Initialize OSP device according to the parameters in the configuration
987 * Reconstruct the local device name from the configuration profile, and
988 * initialize necessary threads and structures according to the OSP type
991 * Since there is no record in the MDT configuration for the local disk
992 * device, we have to extract this from elsewhere in the profile.
993 * The only information we get at setup is from the OSC records:
994 * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
996 * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
997 * so, we need to reconstruct the name of the underlying OSD from this:
998 * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
1000 * \param[in] env execution environment
1001 * \param[in] osp OSP device
1002 * \param[in] ldt lu device type of OSP
1003 * \param[in] cfg configuration log
1005 * \retval 0 0 if OSP initialization succeeded.
1006 * \retval negative negative errno if OSP initialization failed.
1008 static int osp_init0(const struct lu_env *env, struct osp_device *osp,
1009 struct lu_device_type *ldt, struct lustre_cfg *cfg)
1011 struct obd_device *obd;
1012 struct obd_import *imp;
1014 char *src, *tgt, *mdt, *osdname = NULL;
1020 mutex_init(&osp->opd_async_requests_mutex);
1021 INIT_LIST_HEAD(&osp->opd_async_updates);
1022 init_rwsem(&osp->opd_async_updates_rwsem);
1023 atomic_set(&osp->opd_async_updates_count, 0);
1025 obd = class_name2obd(lustre_cfg_string(cfg, 0));
1027 CERROR("Cannot find obd with name %s\n",
1028 lustre_cfg_string(cfg, 0));
1033 src = lustre_cfg_string(cfg, 0);
1037 tgt = strrchr(src, '-');
1039 CERROR("%s: invalid target name %s: rc = %d\n",
1040 osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
1045 if (strncmp(tgt, "-osc", 4) == 0) {
1046 /* Old OSC name fsname-OSTXXXX-osc */
1047 for (tgt--; tgt > src && *tgt != '-'; tgt--)
1050 CERROR("%s: invalid target name %s: rc = %d\n",
1051 osp->opd_obd->obd_name,
1052 lustre_cfg_string(cfg, 0), -EINVAL);
1056 if (strncmp(tgt, "-OST", 4) != 0) {
1057 CERROR("%s: invalid target name %s: rc = %d\n",
1058 osp->opd_obd->obd_name,
1059 lustre_cfg_string(cfg, 0), -EINVAL);
1063 idx = simple_strtol(tgt + 4, &mdt, 16);
1064 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
1065 CERROR("%s: invalid OST index in '%s': rc = %d\n",
1066 osp->opd_obd->obd_name, src, -EINVAL);
1069 osp->opd_index = idx;
1073 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
1074 if (strncmp(tgt, "-MDT", 4) != 0 &&
1075 strncmp(tgt, "-OST", 4) != 0) {
1076 CERROR("%s: invalid target name %s: rc = %d\n",
1077 osp->opd_obd->obd_name,
1078 lustre_cfg_string(cfg, 0), -EINVAL);
1082 idx = simple_strtol(tgt + 4, &mdt, 16);
1083 if (*mdt != '\0' || idx > INT_MAX || idx < 0) {
1084 CERROR("%s: invalid OST index in '%s': rc = %d\n",
1085 osp->opd_obd->obd_name, src, -EINVAL);
1089 /* Get MDT index from the name and set it to opd_group,
1090 * which will be used by OSP to connect with OST */
1091 osp->opd_group = idx;
1092 if (tgt - src <= 12) {
1093 CERROR("%s: invalid mdt index from %s: rc =%d\n",
1094 osp->opd_obd->obd_name,
1095 lustre_cfg_string(cfg, 0), -EINVAL);
1099 if (strncmp(tgt - 12, "-MDT", 4) == 0)
1100 osp->opd_connect_mdt = 1;
1102 idx = simple_strtol(tgt - 8, &mdt, 16);
1103 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
1104 CERROR("%s: invalid OST index in '%s': rc =%d\n",
1105 osp->opd_obd->obd_name, src, -EINVAL);
1109 osp->opd_index = idx;
1110 idx = tgt - src - 12;
1112 /* check the fsname length, and after this everything else will fit */
1113 if (idx > MTI_NAME_MAXLEN) {
1114 CERROR("%s: fsname too long in '%s': rc = %d\n",
1115 osp->opd_obd->obd_name, src, -EINVAL);
1119 OBD_ALLOC(osdname, MAX_OBD_NAME);
1120 if (osdname == NULL)
1123 memcpy(osdname, src, idx); /* copy just the fsname part */
1124 osdname[idx] = '\0';
1126 mdt = strstr(mdt, "-MDT");
1127 if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
1128 strcat(osdname, "-MDT0000");
1130 strcat(osdname, mdt);
1131 strcat(osdname, "-osd");
1132 CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
1134 osp_init_rpc_lock(osp);
1136 osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
1137 osp->opd_dt_dev.dd_ops = &osp_dt_ops;
1139 obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
1141 rc = osp_connect_to_osd(env, osp, osdname);
1145 rc = ptlrpcd_addref();
1147 GOTO(out_disconnect, rc);
1149 rc = client_obd_setup(obd, cfg);
1151 CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
1156 osp_tunables_init(osp);
1158 rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
1159 LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
1161 CERROR("%s: fid init error: rc = %d\n",
1162 osp->opd_obd->obd_name, rc);
1166 if (!osp->opd_connect_mdt) {
1167 /* Initialize last id from the storage - will be
1168 * used in orphan cleanup. */
1169 if (!osp->opd_storage->dd_rdonly) {
1170 rc = osp_last_used_init(env, osp);
1175 /* Initialize precreation thread, it handles new
1176 * connections as well. */
1177 rc = osp_init_precreate(osp);
1179 GOTO(out_last_used, rc);
1182 * Initialize synhronization mechanism taking
1183 * care of propogating changes to OST in near
1184 * transactional manner.
1186 rc = osp_sync_init(env, osp);
1188 GOTO(out_precreat, rc);
1190 osp->opd_got_disconnected = 1;
1191 rc = osp_update_init(osp);
1196 rc = osp_init_statfs(osp);
1198 GOTO(out_precreat, rc);
1200 ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
1203 * Initiate connect to OST
1205 ll_generate_random_uuid(uuid);
1206 class_uuid_unparse(uuid, &osp->opd_cluuid);
1208 imp = obd->u.cli.cl_import;
1210 rc = ptlrpc_init_import(imp);
1214 OBD_FREE(osdname, MAX_OBD_NAME);
1218 if (!osp->opd_connect_mdt)
1219 /* stop sync thread */
1222 /* stop precreate thread */
1223 if (!osp->opd_connect_mdt)
1224 osp_precreate_fini(osp);
1226 osp_update_fini(env, osp);
1228 if (!osp->opd_connect_mdt)
1229 osp_last_used_fini(env, osp);
1231 obd_fid_fini(osp->opd_obd);
1233 osp_tunables_fini(osp);
1234 client_obd_cleanup(obd);
1238 obd_disconnect(osp->opd_storage_exp);
1241 OBD_FREE(osdname, MAX_OBD_NAME);
1246 * Implementation of lu_device_type_operations::ldto_device_free
1248 * Free the OSP device in memory. No return value is needed for now,
1249 * so always return NULL to comply with the interface.
1251 * \param[in] env execution environment
1252 * \param[in] lu lu_device of OSP
1254 * \retval NULL NULL unconditionally
1256 static struct lu_device *osp_device_free(const struct lu_env *env,
1257 struct lu_device *lu)
1259 struct osp_device *osp = lu2osp_dev(lu);
1261 if (atomic_read(&lu->ld_ref) && lu->ld_site) {
1262 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1263 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
1265 dt_device_fini(&osp->opd_dt_dev);
1272 * Implementation of lu_device_type_operations::ldto_device_alloc
1274 * This function allocates and initializes OSP device in memory according to
1277 * \param[in] env execution environment
1278 * \param[in] type device type of OSP
1279 * \param[in] lcfg config log
1281 * \retval pointer the pointer of allocated OSP if succeed.
1282 * \retval ERR_PTR(errno) ERR_PTR(errno) if failed.
1284 static struct lu_device *osp_device_alloc(const struct lu_env *env,
1285 struct lu_device_type *type,
1286 struct lustre_cfg *lcfg)
1288 struct osp_device *osp;
1289 struct lu_device *ld;
1293 ld = ERR_PTR(-ENOMEM);
1297 ld = osp2lu_dev(osp);
1298 dt_device_init(&osp->opd_dt_dev, type);
1299 rc = osp_init0(env, osp, type, lcfg);
1301 osp_device_free(env, ld);
1309 * Implementation of lu_device_type_operations::ldto_device_fini
1311 * This function cleans up the OSP device, i.e. release and free those
1312 * attached items in osp_device.
1314 * \param[in] env execution environment
1315 * \param[in] ld lu_device of OSP
1317 * \retval NULL NULL if cleanup succeeded.
1318 * \retval ERR_PTR(errno) ERR_PTR(errno) if cleanup failed.
1320 static struct lu_device *osp_device_fini(const struct lu_env *env,
1321 struct lu_device *ld)
1323 struct osp_device *osp = lu2osp_dev(ld);
1328 if (osp->opd_async_requests != NULL) {
1329 osp_update_request_destroy(env, osp->opd_async_requests);
1330 osp->opd_async_requests = NULL;
1333 if (osp->opd_storage_exp) {
1334 /* wait for the commit callbacks to complete */
1335 wait_event(osp->opd_sync_waitq,
1336 atomic_read(&osp->opd_commits_registered) == 0);
1337 obd_disconnect(osp->opd_storage_exp);
1340 LASSERT(osp->opd_obd);
1341 osp_tunables_fini(osp);
1343 rc = client_obd_cleanup(osp->opd_obd);
1346 RETURN(ERR_PTR(rc));
1355 * Implementation of obd_ops::o_reconnect
1357 * This function is empty and does not need to do anything for now.
1359 static int osp_reconnect(const struct lu_env *env,
1360 struct obd_export *exp, struct obd_device *obd,
1361 struct obd_uuid *cluuid,
1362 struct obd_connect_data *data,
1369 * Implementation of obd_ops::o_connect
1371 * Connect OSP to the remote target (MDT or OST). Allocate the
1372 * export and return it to the LOD, which calls this function
1373 * for each OSP to connect it to the remote target. This function
1374 * is currently only called once per OSP.
1376 * \param[in] env execution environment
1377 * \param[out] exp export connected to OSP
1378 * \param[in] obd OSP device
1379 * \param[in] cluuid OSP device client uuid
1380 * \param[in] data connect_data to be used to connect to the remote
1382 * \param[in] localdata necessary for the API interface, but not used in
1385 * \retval 0 0 if the connection succeeded.
1386 * \retval negative negative errno if the connection failed.
1388 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
1389 struct obd_device *obd, struct obd_uuid *cluuid,
1390 struct obd_connect_data *data, void *localdata)
1392 struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1393 struct obd_connect_data *ocd;
1394 struct obd_import *imp;
1395 struct lustre_handle conn;
1400 CDEBUG(D_CONFIG, "connect #%d\n", osp->opd_connects);
1402 rc = class_connect(&conn, obd, cluuid);
1406 *exp = class_conn2export(&conn);
1407 /* Why should there ever be more than 1 connect? */
1408 osp->opd_connects++;
1409 LASSERT(osp->opd_connects == 1);
1411 osp->opd_exp = *exp;
1413 imp = osp->opd_obd->u.cli.cl_import;
1414 imp->imp_dlm_handle = conn;
1416 LASSERT(data != NULL);
1417 LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
1418 ocd = &imp->imp_connect_data;
1421 imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1422 imp->imp_connect_flags2_orig = ocd->ocd_connect_flags2;
1424 ocd->ocd_version = LUSTRE_VERSION_CODE;
1425 ocd->ocd_index = data->ocd_index;
1427 rc = ptlrpc_connect_import(imp);
1429 CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
1432 osp->opd_obd->u.cli.cl_seq->lcs_exp =
1433 class_export_get(osp->opd_exp);
1436 ptlrpc_pinger_add_import(imp);
1442 * Implementation of obd_ops::o_disconnect
1444 * Disconnect the export for the OSP. This is called by LOD to release the
1445 * OSP during cleanup (\see lod_del_device()). The OSP will be released after
1446 * the export is released.
1448 * \param[in] exp export to be disconnected.
1450 * \retval 0 0 if disconnection succeed
1451 * \retval negative negative errno if disconnection failed
1453 static int osp_obd_disconnect(struct obd_export *exp)
1455 struct obd_device *obd = exp->exp_obd;
1456 struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1460 /* Only disconnect the underlying layers on the final disconnect. */
1461 LASSERT(osp->opd_connects == 1);
1462 osp->opd_connects--;
1464 rc = class_disconnect(exp);
1466 CERROR("%s: class disconnect error: rc = %d\n",
1471 /* destroy the device */
1472 class_manual_cleanup(obd);
1478 * Implementation of obd_ops::o_statfs
1480 * Send a RPC to the remote target to get statfs status. This is only used
1481 * in lprocfs helpers by obd_statfs.
1483 * \param[in] env execution environment
1484 * \param[in] exp connection state from this OSP to the parent (LOD)
1486 * \param[out] osfs hold the statfs result
1487 * \param[in] unused Not used in this function for now
1488 * \param[in] flags flags to indicate how OSP will issue the RPC
1490 * \retval 0 0 if statfs succeeded.
1491 * \retval negative negative errno if statfs failed.
1493 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
1494 struct obd_statfs *osfs, time64_t unused, __u32 flags)
1496 struct obd_statfs *msfs;
1497 struct ptlrpc_request *req;
1498 struct obd_import *imp = NULL;
1503 /* Since the request might also come from lprocfs, so we need
1504 * sync this with client_disconnect_export Bug15684 */
1505 down_read(&exp->exp_obd->u.cli.cl_sem);
1506 if (exp->exp_obd->u.cli.cl_import)
1507 imp = class_import_get(exp->exp_obd->u.cli.cl_import);
1508 up_read(&exp->exp_obd->u.cli.cl_sem);
1512 req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
1514 class_import_put(imp);
1519 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
1521 ptlrpc_request_free(req);
1524 ptlrpc_request_set_replen(req);
1525 req->rq_request_portal = OST_CREATE_PORTAL;
1526 ptlrpc_at_set_req_timeout(req);
1528 if (flags & OBD_STATFS_NODELAY) {
1529 /* procfs requests not want stat in wait for avoid deadlock */
1530 req->rq_no_resend = 1;
1531 req->rq_no_delay = 1;
1534 rc = ptlrpc_queue_wait(req);
1538 msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1540 GOTO(out, rc = -EPROTO);
1546 ptlrpc_req_finished(req);
1551 * Implementation of obd_ops::o_import_event
1553 * This function is called when some related import event happens. It will
1554 * mark the necessary flags according to the event and notify the necessary
1555 * threads (mainly precreate thread).
1557 * \param[in] obd OSP OBD device
1558 * \param[in] imp import attached from OSP to remote (OST/MDT) service
1559 * \param[in] event event related to remote service (IMP_EVENT_*)
1561 * \retval 0 0 if the event handling succeeded.
1562 * \retval negative negative errno if the event handling failed.
1564 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1565 enum obd_import_event event)
1567 struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1571 case IMP_EVENT_DISCON:
1572 d->opd_got_disconnected = 1;
1573 d->opd_imp_connected = 0;
1574 if (d->opd_connect_mdt)
1577 if (d->opd_pre != NULL) {
1578 osp_pre_update_status(d, -ENODEV);
1579 wake_up(&d->opd_pre_waitq);
1582 CDEBUG(D_HA, "got disconnected\n");
1584 case IMP_EVENT_INACTIVE:
1585 d->opd_imp_active = 0;
1586 d->opd_imp_connected = 0;
1587 d->opd_obd->obd_inactive = 1;
1588 if (d->opd_connect_mdt)
1590 if (d->opd_pre != NULL) {
1591 /* Import is invalid, we can`t get stripes so
1593 rc = imp->imp_deactive ? -ESHUTDOWN : -ENODEV;
1594 osp_pre_update_status(d, rc);
1595 wake_up(&d->opd_pre_waitq);
1598 CDEBUG(D_HA, "got inactive\n");
1600 case IMP_EVENT_ACTIVE:
1601 d->opd_imp_active = 1;
1603 if (d->opd_got_disconnected)
1604 d->opd_new_connection = 1;
1605 d->opd_imp_connected = 1;
1606 d->opd_imp_seen_connected = 1;
1607 d->opd_obd->obd_inactive = 0;
1608 wake_up(&d->opd_pre_waitq);
1609 if (d->opd_connect_mdt)
1612 osp_sync_check_for_work(d);
1613 CDEBUG(D_HA, "got connected\n");
1615 case IMP_EVENT_INVALIDATE:
1616 if (d->opd_connect_mdt)
1617 osp_invalidate_request(d);
1619 if (obd->obd_namespace == NULL)
1621 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1624 case IMP_EVENT_DEACTIVATE:
1625 case IMP_EVENT_ACTIVATE:
1628 CERROR("%s: unsupported import event: %#x\n",
1629 obd->obd_name, event);
1635 * Implementation of obd_ops: o_iocontrol
1637 * This function is the ioctl handler for OSP. Note: lctl will access the OSP
1638 * directly by ioctl, instead of through the MDS stack.
1640 * param[in] cmd ioctl command.
1641 * param[in] exp export of this OSP.
1642 * param[in] len data length of \a karg.
1643 * param[in] karg input argument which is packed as
1645 * param[out] uarg pointer to userspace buffer (must access by
1648 * \retval 0 0 if the ioctl handling succeeded.
1649 * \retval negative negative errno if the ioctl handling failed.
1651 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1652 void *karg, void __user *uarg)
1654 struct obd_device *obd = exp->exp_obd;
1655 struct osp_device *d;
1656 struct obd_ioctl_data *data = karg;
1661 LASSERT(obd->obd_lu_dev);
1662 d = lu2osp_dev(obd->obd_lu_dev);
1663 LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1665 if (!try_module_get(THIS_MODULE)) {
1666 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
1667 module_name(THIS_MODULE));
1672 case OBD_IOC_CLIENT_RECOVER:
1673 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1674 data->ioc_inlbuf1, 0);
1678 case IOC_OSC_SET_ACTIVE:
1679 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1683 CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
1684 cmd, current_comm());
1687 module_put(THIS_MODULE);
1693 * Implementation of obd_ops::o_get_info
1695 * Retrieve information by key. Retrieval starts from the top layer
1696 * (MDT) of the MDS stack and traverses the stack by calling the
1697 * obd_get_info() method of the next sub-layer.
1699 * \param[in] env execution environment
1700 * \param[in] exp export of this OSP
1701 * \param[in] keylen length of \a key
1702 * \param[in] key the key
1703 * \param[out] vallen length of \a val
1704 * \param[out] val holds the value returned by the key
1706 * \retval 0 0 if getting information succeeded.
1707 * \retval negative negative errno if getting information failed.
1709 static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1710 __u32 keylen, void *key, __u32 *vallen, void *val)
1714 if (KEY_IS(KEY_OSP_CONNECTED)) {
1715 struct obd_device *obd = exp->exp_obd;
1716 struct osp_device *osp;
1718 if (!obd->obd_set_up || obd->obd_stopping)
1721 osp = lu2osp_dev(obd->obd_lu_dev);
1724 * 1.8/2.0 behaviour is that OST being connected once at least
1725 * is considered "healthy". and one "healthy" OST is enough to
1726 * allow lustre clients to connect to MDS
1728 RETURN(!osp->opd_imp_seen_connected);
1734 static int osp_obd_set_info_async(const struct lu_env *env,
1735 struct obd_export *exp,
1736 u32 keylen, void *key,
1737 u32 vallen, void *val,
1738 struct ptlrpc_request_set *set)
1740 struct obd_device *obd = exp->exp_obd;
1741 struct obd_import *imp = obd->u.cli.cl_import;
1742 struct osp_device *osp;
1743 struct ptlrpc_request *req;
1747 if (KEY_IS(KEY_SPTLRPC_CONF)) {
1748 sptlrpc_conf_client_adapt(exp->exp_obd);
1752 LASSERT(set != NULL);
1753 if (!obd->obd_set_up || obd->obd_stopping)
1755 osp = lu2osp_dev(obd->obd_lu_dev);
1757 req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
1761 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
1762 RCL_CLIENT, keylen);
1763 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
1764 RCL_CLIENT, vallen);
1765 if (osp->opd_connect_mdt)
1766 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SET_INFO);
1768 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
1770 ptlrpc_request_free(req);
1774 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
1775 memcpy(tmp, key, keylen);
1776 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
1777 memcpy(tmp, val, vallen);
1779 ptlrpc_request_set_replen(req);
1780 ptlrpc_set_add_req(set, req);
1781 ptlrpc_check_set(NULL, set);
1787 * Implementation of obd_ops: o_fid_alloc
1789 * Allocate a FID. There are two cases in which OSP performs
1792 * 1. FID precreation for data objects, which is done in
1793 * osp_precreate_fids() instead of this function.
1794 * 2. FID allocation for each sub-stripe of a striped directory.
1795 * Similar to other FID clients, OSP requests the sequence
1796 * from its corresponding remote MDT, which in turn requests
1797 * sequences from the sequence controller (MDT0).
1799 * \param[in] env execution environment
1800 * \param[in] exp export of the OSP
1801 * \param[out] fid FID being allocated
1802 * \param[in] unused necessary for the interface but unused.
1804 * \retval 0 0 FID allocated successfully.
1805 * \retval 1 1 FID allocated successfully and new sequence
1806 * requested from seq meta server
1807 * \retval negative negative errno if FID allocation failed.
1809 static int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
1810 struct lu_fid *fid, struct md_op_data *unused)
1812 struct client_obd *cli = &exp->exp_obd->u.cli;
1813 struct osp_device *osp = lu2osp_dev(exp->exp_obd->obd_lu_dev);
1814 struct lu_client_seq *seq = cli->cl_seq;
1817 LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
1818 /* Sigh, fid client is not ready yet */
1819 LASSERT(osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
1821 RETURN(seq_client_alloc_fid(env, seq, fid));
1824 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1825 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1826 static void osp_key_exit(const struct lu_context *ctx,
1827 struct lu_context_key *key, void *data)
1829 struct osp_thread_info *info = data;
1831 info->osi_attr.la_valid = 0;
1834 struct lu_context_key osp_thread_key = {
1835 .lct_tags = LCT_MD_THREAD,
1836 .lct_init = osp_key_init,
1837 .lct_fini = osp_key_fini,
1838 .lct_exit = osp_key_exit
1841 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1842 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1844 struct lu_context_key osp_txn_key = {
1845 .lct_tags = LCT_OSP_THREAD,
1846 .lct_init = osp_txn_key_init,
1847 .lct_fini = osp_txn_key_fini
1849 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1851 static struct lu_device_type_operations osp_device_type_ops = {
1852 .ldto_init = osp_type_init,
1853 .ldto_fini = osp_type_fini,
1855 .ldto_start = osp_type_start,
1856 .ldto_stop = osp_type_stop,
1858 .ldto_device_alloc = osp_device_alloc,
1859 .ldto_device_free = osp_device_free,
1861 .ldto_device_fini = osp_device_fini
1864 static struct lu_device_type osp_device_type = {
1865 .ldt_tags = LU_DEVICE_DT,
1866 .ldt_name = LUSTRE_OSP_NAME,
1867 .ldt_ops = &osp_device_type_ops,
1868 .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
1871 static struct obd_ops osp_obd_device_ops = {
1872 .o_owner = THIS_MODULE,
1873 .o_add_conn = client_import_add_conn,
1874 .o_del_conn = client_import_del_conn,
1875 .o_reconnect = osp_reconnect,
1876 .o_connect = osp_obd_connect,
1877 .o_disconnect = osp_obd_disconnect,
1878 .o_get_info = osp_obd_get_info,
1879 .o_set_info_async = osp_obd_set_info_async,
1880 .o_import_event = osp_import_event,
1881 .o_iocontrol = osp_iocontrol,
1882 .o_statfs = osp_obd_statfs,
1883 .o_fid_init = client_fid_init,
1884 .o_fid_fini = client_fid_fini,
1885 .o_fid_alloc = osp_fid_alloc,
1888 static struct obd_type sym;
1891 * Initialize OSP module.
1893 * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
1894 * in obd_types (\see class_obd.c). Initialize procfs for the
1895 * the OSP device. Note: OSP was called OSC before Lustre 2.4,
1896 * so for compatibility it still uses the name "osc" in procfs.
1897 * This is called at module load time.
1899 * \retval 0 0 if initialization succeeds.
1900 * \retval negative negative errno if initialization failed.
1902 static int __init osp_init(void)
1904 struct dentry *symlink;
1905 struct obd_type *type;
1906 struct kobject *kobj;
1910 rc = lu_kmem_init(osp_caches);
1914 rc = class_register_type(&osp_obd_device_ops, NULL, true, NULL,
1915 LUSTRE_OSP_NAME, &osp_device_type);
1917 lu_kmem_fini(osp_caches);
1921 rc = class_register_type(&lwp_obd_device_ops, NULL, false, NULL,
1922 LUSTRE_LWP_NAME, &lwp_device_type);
1924 class_unregister_type(LUSTRE_OSP_NAME);
1925 lu_kmem_fini(osp_caches);
1929 /* create "osc" entry for compatibility purposes */
1931 dname.len = strlen(dname.name);
1932 dname.hash = ll_full_name_hash(debugfs_lustre_root, dname.name,
1934 symlink = d_lookup(debugfs_lustre_root, &dname);
1936 symlink = debugfs_create_dir(dname.name, debugfs_lustre_root);
1937 if (IS_ERR_OR_NULL(symlink)) {
1938 rc = symlink ? PTR_ERR(symlink) : -ENOMEM;
1941 sym.typ_debugfs_entry = symlink;
1946 kobj = kset_find_obj(lustre_kset, dname.name);
1952 kobj = class_setup_tunables(dname.name);
1955 if (sym.typ_debugfs_entry)
1956 ldebugfs_remove(&sym.typ_debugfs_entry);
1959 sym.typ_kobj = kobj;
1962 type = class_search_type(LUSTRE_OSC_NAME);
1963 if (type != NULL && type->typ_procroot != NULL)
1966 type = class_search_type(LUSTRE_OSP_NAME);
1967 type->typ_procsym = lprocfs_register("osc", proc_lustre_root,
1969 if (IS_ERR(type->typ_procsym)) {
1970 CERROR("osp: can't create compat entry \"osc\": %d\n",
1971 (int) PTR_ERR(type->typ_procsym));
1972 type->typ_procsym = NULL;
1979 * Finalize OSP module.
1981 * This callback is called when kernel unloads OSP module from memory, and
1982 * it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
1984 static void __exit osp_exit(void)
1986 ldebugfs_remove(&sym.typ_debugfs_entry);
1987 kobject_put(sym.typ_kobj);
1988 class_unregister_type(LUSTRE_LWP_NAME);
1989 class_unregister_type(LUSTRE_OSP_NAME);
1990 lu_kmem_fini(osp_caches);
1993 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
1994 MODULE_DESCRIPTION("Lustre OSD Storage Proxy ("LUSTRE_OSP_NAME")");
1995 MODULE_VERSION(LUSTRE_VERSION_STRING);
1996 MODULE_LICENSE("GPL");
1998 module_init(osp_init);
1999 module_exit(osp_exit);