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.sun.com/software/products/lustre/docs/GPLv2.pdf
20 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21 * CA 95054 USA or visit www.sun.com if you need additional information or
27 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28 * Use is subject to license terms.
30 * Copyright (c) 2012, 2014, Intel Corporation.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lustre/osp/osp_dev.c
38 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
39 * Author: Mikhail Pershin <mike.pershin@intel.com>
40 * Author: Di Wang <di.wang@intel.com>
43 * The Object Storage Proxy (OSP) module provides an implementation of
44 * the DT API for remote MDTs and OSTs. Every local OSP device (or
45 * object) is a proxy for a remote OSD device (or object). Thus OSP
46 * converts DT operations into RPCs, which are sent to the OUT service
47 * on a remote target, converted back to DT operations, and
48 * executed. Of course there are many ways in which this description
49 * is inaccurate but it's a good enough mental model. OSP is used by
50 * the MDT stack in several ways:
52 * - OSP devices allocate FIDs for the stripe sub-objects of a striped
55 * - OSP objects represent the remote MDT and OST objects that are
56 * the stripes of a striped object.
58 * - OSP devices log, send, and track synchronous operations (setattr
59 * and unlink) to remote targets.
61 * - OSP objects are the bottom slice of the compound LU object
62 * representing a remote MDT object: MDT/MDD/LOD/OSP.
64 * - OSP objects are used by LFSCK to represent remote OST objects
65 * during the verification of MDT-OST consistency.
67 * - OSP devices batch idempotent requests (declare_attr_get() and
68 * declare_xattr_get()) to the remote target and cache their results.
70 * In addition the OSP layer implements a subset of the OBD device API
71 * to support being a client of a remote target, connecting to other
72 * layers, and FID allocation.
75 #define DEBUG_SUBSYSTEM S_MDS
77 #include <linux/kthread.h>
78 #include <obd_class.h>
79 #include <lustre_ioctl.h>
80 #include <lustre_param.h>
81 #include <lustre_log.h>
83 #include "osp_internal.h"
85 /* Slab for OSP object allocation */
86 struct kmem_cache *osp_object_kmem;
88 static struct lu_kmem_descr osp_caches[] = {
90 .ckd_cache = &osp_object_kmem,
91 .ckd_name = "osp_obj",
92 .ckd_size = sizeof(struct osp_object)
100 * Implementation of lu_device_operations::ldo_object_alloc
102 * Allocates an OSP object in memory, whose FID is on the remote target.
104 * \param[in] env execution environment
105 * \param[in] hdr The header of the object stack. If it is NULL, it
106 * means the object is not built from top device, i.e.
107 * it is a sub-stripe object of striped directory or
109 * \param[in] d OSP device
111 * \retval object object being created if the creation succeed.
112 * \retval NULL NULL if the creation failed.
114 static struct lu_object *osp_object_alloc(const struct lu_env *env,
115 const struct lu_object_header *hdr,
118 struct lu_object_header *h = NULL;
119 struct osp_object *o;
122 OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
124 l = &o->opo_obj.do_lu;
126 /* If hdr is NULL, it means the object is not built
127 * from the top dev(MDT/OST), usually it happens when
128 * building striped object, like data object on MDT or
129 * striped object for directory */
132 lu_object_header_init(h);
133 dt_object_init(&o->opo_obj, h, d);
134 lu_object_add_top(h, l);
136 dt_object_init(&o->opo_obj, h, d);
139 l->lo_ops = &osp_lu_obj_ops;
148 * Find or create the local object
150 * Finds or creates the local file referenced by \a reg_id and return the
151 * attributes of the local file.
153 * \param[in] env execution environment
154 * \param[in] osp OSP device
155 * \param[out] attr attributes of the object
156 * \param[in] reg_id the local object ID of the file. It will be used
157 * to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
158 * to identify the object.
160 * \retval object object(dt_object) found or created
161 * \retval ERR_PTR(errno) ERR_PTR(errno) if not get the object.
163 static struct dt_object
164 *osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
165 struct lu_attr *attr, __u32 reg_id)
167 struct osp_thread_info *osi = osp_env_info(env);
168 struct dt_object_format dof = { 0 };
169 struct dt_object *dto;
173 lu_local_obj_fid(&osi->osi_fid, reg_id);
174 attr->la_valid = LA_MODE;
175 attr->la_mode = S_IFREG | 0644;
176 dof.dof_type = DFT_REGULAR;
177 /* Find or create the local object by osi_fid. */
178 dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
183 /* Get attributes of the local object. */
184 rc = dt_attr_get(env, dto, attr);
186 CERROR("%s: can't be initialized: rc = %d\n",
187 osp->opd_obd->obd_name, rc);
188 lu_object_put(env, &dto->do_lu);
195 * Write data buffer to a local file object.
197 * \param[in] env execution environment
198 * \param[in] osp OSP device
199 * \param[in] dt_obj object written to
200 * \param[in] buf buffer containing byte array and length
201 * \param[in] offset write offset in the object in bytes
203 * \retval 0 0 if write succeed
204 * \retval -EFAULT -EFAULT if only part of buffer is written.
205 * \retval negative other negative errno if write failed.
207 static int osp_write_local_file(const struct lu_env *env,
208 struct osp_device *osp,
209 struct dt_object *dt_obj,
216 th = dt_trans_create(env, osp->opd_storage);
220 rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
223 rc = dt_trans_start_local(env, osp->opd_storage, th);
227 rc = dt_record_write(env, dt_obj, buf, &offset, th);
229 dt_trans_stop(env, osp->opd_storage, th);
234 * Initialize last ID object.
236 * This function initializes the LAST_ID file, which stores the current last
237 * used id of data objects. The MDT will use the last used id and the last_seq
238 * (\see osp_init_last_seq()) to synchronize the precreate object cache with
241 * \param[in] env execution environment
242 * \param[in] osp OSP device
244 * \retval 0 0 if initialization succeed
245 * \retval negative negative errno if initialization failed
247 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
249 struct osp_thread_info *osi = osp_env_info(env);
250 struct lu_fid *fid = &osp->opd_last_used_fid;
251 struct dt_object *dto;
255 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
258 RETURN(PTR_ERR(dto));
260 /* object will be released in device cleanup path */
261 if (osi->osi_attr.la_size >=
262 sizeof(osi->osi_id) * (osp->opd_index + 1)) {
263 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
265 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
266 if (rc != 0 && rc != -EFAULT)
270 if (rc == -EFAULT) { /* fresh LAST_ID */
272 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
274 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
279 osp->opd_last_used_oid_file = dto;
282 /* object will be released in device cleanup path */
283 CERROR("%s: can't initialize lov_objid: rc = %d\n",
284 osp->opd_obd->obd_name, rc);
285 lu_object_put(env, &dto->do_lu);
286 osp->opd_last_used_oid_file = NULL;
291 * Initialize last sequence object.
293 * This function initializes the LAST_SEQ file in the local OSD, which stores
294 * the current last used sequence of data objects. The MDT will use the last
295 * sequence and last id (\see osp_init_last_objid()) to synchronize the
296 * precreate object cache with OSTs.
298 * \param[in] env execution environment
299 * \param[in] osp OSP device
301 * \retval 0 0 if initialization succeed
302 * \retval negative negative errno if initialization failed
304 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
306 struct osp_thread_info *osi = osp_env_info(env);
307 struct lu_fid *fid = &osp->opd_last_used_fid;
308 struct dt_object *dto;
312 dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
315 RETURN(PTR_ERR(dto));
317 /* object will be released in device cleanup path */
318 if (osi->osi_attr.la_size >=
319 sizeof(osi->osi_id) * (osp->opd_index + 1)) {
320 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
322 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
323 if (rc != 0 && rc != -EFAULT)
327 if (rc == -EFAULT) { /* fresh OSP */
329 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
331 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
336 osp->opd_last_used_seq_file = dto;
339 /* object will be released in device cleanup path */
340 CERROR("%s: can't initialize lov_seq: rc = %d\n",
341 osp->opd_obd->obd_name, rc);
342 lu_object_put(env, &dto->do_lu);
343 osp->opd_last_used_seq_file = NULL;
348 * Initialize last OID and sequence object.
350 * If the MDT is just upgraded to 2.4 from the lower version, where the
351 * LAST_SEQ file does not exist, the file will be created and IDIF sequence
352 * will be written into the file.
354 * \param[in] env execution environment
355 * \param[in] osp OSP device
357 * \retval 0 0 if initialization succeed
358 * \retval negative negative error if initialization failed
360 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
362 struct osp_thread_info *osi = osp_env_info(env);
366 fid_zero(&osp->opd_last_used_fid);
367 rc = osp_init_last_objid(env, osp);
369 CERROR("%s: Can not get ids %d from old objid!\n",
370 osp->opd_obd->obd_name, rc);
374 rc = osp_init_last_seq(env, osp);
376 CERROR("%s: Can not get ids %d from old objid!\n",
377 osp->opd_obd->obd_name, rc);
381 if (fid_oid(&osp->opd_last_used_fid) != 0 &&
382 fid_seq(&osp->opd_last_used_fid) == 0) {
383 /* Just upgrade from the old version,
384 * set the seq to be IDIF */
385 osp->opd_last_used_fid.f_seq =
386 fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
388 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
389 &osp->opd_last_used_fid.f_seq,
391 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
392 &osi->osi_lb, osi->osi_off);
394 CERROR("%s : Can not write seq file: rc = %d\n",
395 osp->opd_obd->obd_name, rc);
400 if (!fid_is_zero(&osp->opd_last_used_fid) &&
401 !fid_is_sane(&osp->opd_last_used_fid)) {
402 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
403 PFID(&osp->opd_last_used_fid));
404 GOTO(out, rc = -EINVAL);
407 CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
408 osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
411 if (osp->opd_last_used_oid_file != NULL) {
412 lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
413 osp->opd_last_used_oid_file = NULL;
415 if (osp->opd_last_used_seq_file != NULL) {
416 lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
417 osp->opd_last_used_seq_file = NULL;
425 * Release the last sequence and OID file objects in OSP device.
427 * \param[in] env execution environment
428 * \param[in] osp OSP device
430 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
432 /* release last_used file */
433 if (osp->opd_last_used_oid_file != NULL) {
434 lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
435 osp->opd_last_used_oid_file = NULL;
438 if (osp->opd_last_used_seq_file != NULL) {
439 lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
440 osp->opd_last_used_seq_file = NULL;
445 * Disconnects the connection between OSP and its correspondent MDT or OST, and
446 * the import will be marked as inactive. It will only be called during OSP
449 * \param[in] d OSP device being disconnected
451 * \retval 0 0 if disconnection succeed
452 * \retval negative negative errno if disconnection failed
454 static int osp_disconnect(struct osp_device *d)
456 struct obd_import *imp;
459 imp = d->opd_obd->u.cli.cl_import;
461 /* Mark import deactivated now, so we don't try to reconnect if any
462 * of the cleanup RPCs fails (e.g. ldlm cancel, etc). We don't
463 * fully deactivate the import, or that would drop all requests. */
464 LASSERT(imp != NULL);
465 spin_lock(&imp->imp_lock);
466 imp->imp_deactive = 1;
467 spin_unlock(&imp->imp_lock);
469 ptlrpc_deactivate_import(imp);
471 /* Some non-replayable imports (MDS's OSCs) are pinged, so just
472 * delete it regardless. (It's safe to delete an import that was
474 (void)ptlrpc_pinger_del_import(imp);
476 rc = ptlrpc_disconnect_import(imp, 0);
478 CERROR("%s: can't disconnect: rc = %d\n",
479 d->opd_obd->obd_name, rc);
481 ptlrpc_invalidate_import(imp);
487 * Initialize the osp_update structure in OSP device
489 * Allocate osp update structure and start update thread.
491 * \param[in] osp OSP device
493 * \retval 0 if initialization succeeds.
494 * \retval negative errno if initialization fails.
496 static int osp_update_init(struct osp_device *osp)
498 struct l_wait_info lwi = { 0 };
499 struct task_struct *task;
503 LASSERT(osp->opd_connect_mdt);
505 OBD_ALLOC_PTR(osp->opd_update);
506 if (osp->opd_update == NULL)
509 init_waitqueue_head(&osp->opd_update_thread.t_ctl_waitq);
510 init_waitqueue_head(&osp->opd_update->ou_waitq);
511 spin_lock_init(&osp->opd_update->ou_lock);
512 INIT_LIST_HEAD(&osp->opd_update->ou_list);
513 osp->opd_update->ou_rpc_version = 1;
514 osp->opd_update->ou_version = 1;
516 /* start thread handling sending updates to the remote MDT */
517 task = kthread_run(osp_send_update_thread, osp,
518 "osp_up%u-%u", osp->opd_index, osp->opd_group);
520 int rc = PTR_ERR(task);
522 OBD_FREE_PTR(osp->opd_update);
523 osp->opd_update = NULL;
524 CERROR("%s: can't start precreate thread: rc = %d\n",
525 osp->opd_obd->obd_name, rc);
529 l_wait_event(osp->opd_update_thread.t_ctl_waitq,
530 osp_send_update_thread_running(osp) ||
531 osp_send_update_thread_stopped(osp), &lwi);
537 * Finialize osp_update structure in OSP device
539 * Stop the OSP update sending thread, then delete the left
540 * osp thandle in the sending list.
542 * \param [in] osp OSP device.
544 static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
546 struct osp_update_request *our;
547 struct osp_update_request *tmp;
548 struct osp_updates *ou = osp->opd_update;
553 osp->opd_update_thread.t_flags = SVC_STOPPING;
554 wake_up(&ou->ou_waitq);
556 wait_event(osp->opd_update_thread.t_ctl_waitq,
557 osp->opd_update_thread.t_flags & SVC_STOPPED);
559 /* Remove the left osp thandle from the list */
560 spin_lock(&ou->ou_lock);
561 list_for_each_entry_safe(our, tmp, &ou->ou_list,
563 list_del_init(&our->our_list);
564 LASSERT(our->our_th != NULL);
565 osp_trans_callback(env, our->our_th, -EIO);
566 /* our will be destroyed in osp_thandle_put() */
567 osp_thandle_put(our->our_th);
569 spin_unlock(&ou->ou_lock);
572 osp->opd_update = NULL;
576 * Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
577 * precreate threads etc.
579 * \param[in] env execution environment.
580 * \param[in] d OSP device being disconnected.
582 * \retval 0 0 if cleanup succeed
583 * \retval negative negative errno if cleanup failed
585 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
592 rc = osp_disconnect(d);
594 if (!d->opd_connect_mdt) {
595 /* stop sync thread */
598 /* stop precreate thread */
599 osp_precreate_fini(d);
601 /* release last_used file */
602 osp_last_used_fini(env, d);
605 obd_fid_fini(d->opd_obd);
611 * Implementation of osp_lu_ops::ldo_process_config
613 * This function processes config log records in OSP layer. It is usually
614 * called from the top layer of MDT stack, and goes through the stack by calling
615 * ldo_process_config of next layer.
617 * \param[in] env execution environment
618 * \param[in] dev lu_device of OSP
619 * \param[in] lcfg config log
621 * \retval 0 0 if the config log record is executed correctly.
622 * \retval negative negative errno if the record execution fails.
624 static int osp_process_config(const struct lu_env *env,
625 struct lu_device *dev, struct lustre_cfg *lcfg)
627 struct osp_device *d = lu2osp_dev(dev);
628 struct obd_device *obd = d->opd_obd;
633 switch (lcfg->lcfg_command) {
634 case LCFG_PRE_CLEANUP:
635 rc = osp_disconnect(d);
636 osp_update_fini(env, d);
639 lu_dev_del_linkage(dev->ld_site, dev);
640 rc = osp_shutdown(env, d);
644 rc = class_process_proc_param(PARAM_OSC, obd->obd_vars,
649 /* class_process_proc_param() haven't found matching
650 * parameter and returned ENOSYS so that layer(s)
651 * below could use that. But OSP is the bottom, so
653 CERROR("%s: unknown param %s\n",
654 (char *)lustre_cfg_string(lcfg, 0),
655 (char *)lustre_cfg_string(lcfg, 1));
660 CERROR("%s: unknown command %u\n",
661 (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
670 * Implementation of osp_lu_ops::ldo_recovery_complete
672 * This function is called after recovery is finished, and OSP layer
673 * will wake up precreate thread here.
675 * \param[in] env execution environment
676 * \param[in] dev lu_device of OSP
678 * \retval 0 0 unconditionally
680 static int osp_recovery_complete(const struct lu_env *env,
681 struct lu_device *dev)
683 struct osp_device *osp = lu2osp_dev(dev);
686 osp->opd_recovery_completed = 1;
688 if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
689 wake_up(&osp->opd_pre_waitq);
694 const struct lu_device_operations osp_lu_ops = {
695 .ldo_object_alloc = osp_object_alloc,
696 .ldo_process_config = osp_process_config,
697 .ldo_recovery_complete = osp_recovery_complete,
701 * Implementation of dt_device_operations::dt_statfs
703 * This function provides statfs status (for precreation) from
704 * corresponding OST. Note: this function only retrieves the status
705 * from the OSP device, and the real statfs RPC happens inside
706 * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
707 * not need to retrieve statfs data for now.
709 * \param[in] env execution environment.
710 * \param[in] dev dt_device of OSP.
711 * \param[out] sfs holds the retrieved statfs data.
713 * \retval 0 0 statfs data was retrieved successfully or
714 * retrieval was not needed
715 * \retval negative negative errno if get statfs failed.
717 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
718 struct obd_statfs *sfs)
720 struct osp_device *d = dt2osp_dev(dev);
724 if (unlikely(d->opd_imp_active == 0))
727 if (d->opd_pre == NULL)
730 /* return recently updated data */
731 *sfs = d->opd_statfs;
734 * layer above osp (usually lod) can use ffree to estimate
735 * how many objects are available for immediate creation
737 spin_lock(&d->opd_pre_lock);
738 LASSERTF(fid_seq(&d->opd_pre_last_created_fid) ==
739 fid_seq(&d->opd_pre_used_fid),
740 "last_created "DFID", next_fid "DFID"\n",
741 PFID(&d->opd_pre_last_created_fid),
742 PFID(&d->opd_pre_used_fid));
743 sfs->os_fprecreated = fid_oid(&d->opd_pre_last_created_fid) -
744 fid_oid(&d->opd_pre_used_fid);
745 sfs->os_fprecreated -= d->opd_pre_reserved;
746 spin_unlock(&d->opd_pre_lock);
748 LASSERT(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2);
750 CDEBUG(D_OTHER, "%s: "LPU64" blocks, "LPU64" free, "LPU64" avail, "
751 LPU64" files, "LPU64" free files\n", d->opd_obd->obd_name,
752 sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
753 sfs->os_files, sfs->os_ffree);
757 static int osp_sync_timeout(void *data)
763 * Implementation of dt_device_operations::dt_sync
765 * This function synchronizes the OSP cache to the remote target. It wakes
766 * up unlink log threads and sends out unlink records to the remote OST.
768 * \param[in] env execution environment
769 * \param[in] dev dt_device of OSP
771 * \retval 0 0 if synchronization succeeds
772 * \retval negative negative errno if synchronization fails
774 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
776 struct osp_device *d = dt2osp_dev(dev);
778 struct l_wait_info lwi = { 0 };
779 unsigned long id, old;
781 unsigned long start = cfs_time_current();
784 /* No Sync between MDTs yet. */
785 if (d->opd_connect_mdt)
788 if (unlikely(d->opd_imp_active == 0))
791 id = d->opd_syn_last_used_id;
792 down_write(&d->opd_async_updates_rwsem);
794 CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
795 atomic_read(&d->opd_async_updates_count));
797 /* make sure the connection is fine */
798 expire = cfs_time_shift(obd_timeout);
799 lwi = LWI_TIMEOUT(expire - cfs_time_current(), osp_sync_timeout, d);
800 rc = l_wait_event(d->opd_syn_barrier_waitq,
801 atomic_read(&d->opd_async_updates_count) == 0,
803 up_write(&d->opd_async_updates_rwsem);
807 CDEBUG(D_CACHE, "%s: id: used %lu, processed %lu\n",
808 d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
810 /* wait till all-in-line are processed */
811 while (d->opd_syn_last_processed_id < id) {
813 old = d->opd_syn_last_processed_id;
815 /* make sure the connection is fine */
816 expire = cfs_time_shift(obd_timeout);
817 lwi = LWI_TIMEOUT(expire - cfs_time_current(),
818 osp_sync_timeout, d);
819 l_wait_event(d->opd_syn_barrier_waitq,
820 d->opd_syn_last_processed_id >= id,
823 if (d->opd_syn_last_processed_id >= id)
826 if (d->opd_syn_last_processed_id != old) {
827 /* some progress have been made,
832 /* no changes and expired, something is wrong */
833 GOTO(out, rc = -ETIMEDOUT);
836 /* block new processing (barrier>0 - few callers are possible */
837 atomic_inc(&d->opd_syn_barrier);
839 CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
840 d->opd_syn_rpc_in_flight);
842 /* wait till all-in-flight are replied, so executed by the target */
843 /* XXX: this is used by LFSCK at the moment, which doesn't require
844 * all the changes to be committed, but in general it'd be
845 * better to wait till commit */
846 while (d->opd_syn_rpc_in_flight > 0) {
848 old = d->opd_syn_rpc_in_flight;
850 expire = cfs_time_shift(obd_timeout);
851 lwi = LWI_TIMEOUT(expire - cfs_time_current(),
852 osp_sync_timeout, d);
853 l_wait_event(d->opd_syn_barrier_waitq,
854 d->opd_syn_rpc_in_flight == 0, &lwi);
856 if (d->opd_syn_rpc_in_flight == 0)
859 if (d->opd_syn_rpc_in_flight != old) {
860 /* some progress have been made */
864 /* no changes and expired, something is wrong */
865 GOTO(out, rc = -ETIMEDOUT);
869 /* resume normal processing (barrier=0) */
870 atomic_dec(&d->opd_syn_barrier);
871 __osp_sync_check_for_work(d);
873 CDEBUG(D_CACHE, "%s: done in %lu: rc = %d\n", d->opd_obd->obd_name,
874 cfs_time_current() - start, rc);
879 const struct dt_device_operations osp_dt_ops = {
880 .dt_statfs = osp_statfs,
882 .dt_trans_create = osp_trans_create,
883 .dt_trans_start = osp_trans_start,
884 .dt_trans_stop = osp_trans_stop,
885 .dt_trans_cb_add = osp_trans_cb_add,
889 * Connect OSP to local OSD.
891 * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
892 * OSP needs to access the local OSD to store some information. For example,
893 * during precreate, it needs to update last used OID and sequence file
894 * (LAST_SEQ) in local OSD.
896 * \param[in] env execution environment
897 * \param[in] osp OSP device
898 * \param[in] nextdev the name of local OSD
900 * \retval 0 0 connection succeeded
901 * \retval negative negative errno connection failed
903 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
906 struct obd_connect_data *data = NULL;
907 struct obd_device *obd;
912 LASSERT(osp->opd_storage_exp == NULL);
918 obd = class_name2obd(nextdev);
920 CERROR("%s: can't locate next device: %s\n",
921 osp->opd_obd->obd_name, nextdev);
922 GOTO(out, rc = -ENOTCONN);
925 rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
928 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
929 osp->opd_obd->obd_name, nextdev, rc);
933 osp->opd_dt_dev.dd_lu_dev.ld_site =
934 osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
935 LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
936 osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
944 * Determine if the lock needs to be cancelled
946 * Determine if the unused lock should be cancelled before replay, see
947 * (ldlm_cancel_no_wait_policy()). Currently, only inode bits lock exists
950 * \param[in] lock lock to be checked.
952 * \retval 1 if the lock needs to be cancelled before replay.
953 * \retval 0 if the lock does not need to be cancelled before
956 static int osp_cancel_weight(struct ldlm_lock *lock)
958 if (lock->l_resource->lr_type != LDLM_IBITS)
965 * Initialize OSP device according to the parameters in the configuration
968 * Reconstruct the local device name from the configuration profile, and
969 * initialize necessary threads and structures according to the OSP type
972 * Since there is no record in the MDT configuration for the local disk
973 * device, we have to extract this from elsewhere in the profile.
974 * The only information we get at setup is from the OSC records:
975 * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
977 * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
978 * so, we need to reconstruct the name of the underlying OSD from this:
979 * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
981 * \param[in] env execution environment
982 * \param[in] osp OSP device
983 * \param[in] ldt lu device type of OSP
984 * \param[in] cfg configuration log
986 * \retval 0 0 if OSP initialization succeeded.
987 * \retval negative negative errno if OSP initialization failed.
989 static int osp_init0(const struct lu_env *env, struct osp_device *osp,
990 struct lu_device_type *ldt, struct lustre_cfg *cfg)
992 struct obd_device *obd;
993 struct obd_import *imp;
995 char *src, *tgt, *mdt, *osdname = NULL;
1001 mutex_init(&osp->opd_async_requests_mutex);
1002 INIT_LIST_HEAD(&osp->opd_async_updates);
1003 init_rwsem(&osp->opd_async_updates_rwsem);
1004 atomic_set(&osp->opd_async_updates_count, 0);
1006 obd = class_name2obd(lustre_cfg_string(cfg, 0));
1008 CERROR("Cannot find obd with name %s\n",
1009 lustre_cfg_string(cfg, 0));
1014 src = lustre_cfg_string(cfg, 0);
1018 tgt = strrchr(src, '-');
1020 CERROR("%s: invalid target name %s: rc = %d\n",
1021 osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
1026 if (strncmp(tgt, "-osc", 4) == 0) {
1027 /* Old OSC name fsname-OSTXXXX-osc */
1028 for (tgt--; tgt > src && *tgt != '-'; tgt--)
1031 CERROR("%s: invalid target name %s: rc = %d\n",
1032 osp->opd_obd->obd_name,
1033 lustre_cfg_string(cfg, 0), -EINVAL);
1037 if (strncmp(tgt, "-OST", 4) != 0) {
1038 CERROR("%s: invalid target name %s: rc = %d\n",
1039 osp->opd_obd->obd_name,
1040 lustre_cfg_string(cfg, 0), -EINVAL);
1044 idx = simple_strtol(tgt + 4, &mdt, 16);
1045 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
1046 CERROR("%s: invalid OST index in '%s': rc = %d\n",
1047 osp->opd_obd->obd_name, src, -EINVAL);
1050 osp->opd_index = idx;
1054 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
1055 if (strncmp(tgt, "-MDT", 4) != 0 &&
1056 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);
1070 /* Get MDT index from the name and set it to opd_group,
1071 * which will be used by OSP to connect with OST */
1072 osp->opd_group = idx;
1073 if (tgt - src <= 12) {
1074 CERROR("%s: invalid mdt index from %s: rc =%d\n",
1075 osp->opd_obd->obd_name,
1076 lustre_cfg_string(cfg, 0), -EINVAL);
1080 if (strncmp(tgt - 12, "-MDT", 4) == 0)
1081 osp->opd_connect_mdt = 1;
1083 idx = simple_strtol(tgt - 8, &mdt, 16);
1084 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
1085 CERROR("%s: invalid OST index in '%s': rc =%d\n",
1086 osp->opd_obd->obd_name, src, -EINVAL);
1090 osp->opd_index = idx;
1091 idx = tgt - src - 12;
1093 /* check the fsname length, and after this everything else will fit */
1094 if (idx > MTI_NAME_MAXLEN) {
1095 CERROR("%s: fsname too long in '%s': rc = %d\n",
1096 osp->opd_obd->obd_name, src, -EINVAL);
1100 OBD_ALLOC(osdname, MAX_OBD_NAME);
1101 if (osdname == NULL)
1104 memcpy(osdname, src, idx); /* copy just the fsname part */
1105 osdname[idx] = '\0';
1107 mdt = strstr(mdt, "-MDT");
1108 if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
1109 strcat(osdname, "-MDT0000");
1111 strcat(osdname, mdt);
1112 strcat(osdname, "-osd");
1113 CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
1115 if (osp->opd_connect_mdt) {
1116 struct client_obd *cli = &osp->opd_obd->u.cli;
1118 OBD_ALLOC(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
1119 if (!cli->cl_rpc_lock)
1120 GOTO(out_fini, rc = -ENOMEM);
1121 osp_init_rpc_lock(cli->cl_rpc_lock);
1124 osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
1125 osp->opd_dt_dev.dd_ops = &osp_dt_ops;
1127 obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
1129 rc = osp_connect_to_osd(env, osp, osdname);
1133 rc = ptlrpcd_addref();
1135 GOTO(out_disconnect, rc);
1137 rc = client_obd_setup(obd, cfg);
1139 CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
1144 osp_lprocfs_init(osp);
1146 rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
1147 LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
1149 CERROR("%s: fid init error: rc = %d\n",
1150 osp->opd_obd->obd_name, rc);
1154 if (!osp->opd_connect_mdt) {
1155 /* Initialize last id from the storage - will be
1156 * used in orphan cleanup. */
1157 rc = osp_last_used_init(env, osp);
1162 /* Initialize precreation thread, it handles new
1163 * connections as well. */
1164 rc = osp_init_precreate(osp);
1166 GOTO(out_last_used, rc);
1169 * Initialize synhronization mechanism taking
1170 * care of propogating changes to OST in near
1171 * transactional manner.
1173 rc = osp_sync_init(env, osp);
1175 GOTO(out_precreat, rc);
1177 rc = osp_update_init(osp);
1182 ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
1185 * Initiate connect to OST
1187 ll_generate_random_uuid(uuid);
1188 class_uuid_unparse(uuid, &osp->opd_cluuid);
1190 imp = obd->u.cli.cl_import;
1192 rc = ptlrpc_init_import(imp);
1196 OBD_FREE(osdname, MAX_OBD_NAME);
1200 if (!osp->opd_connect_mdt)
1201 /* stop sync thread */
1204 /* stop precreate thread */
1205 if (!osp->opd_connect_mdt)
1206 osp_precreate_fini(osp);
1208 osp_update_fini(env, osp);
1210 if (!osp->opd_connect_mdt)
1211 osp_last_used_fini(env, osp);
1213 obd_fid_fini(osp->opd_obd);
1215 ptlrpc_lprocfs_unregister_obd(obd);
1216 lprocfs_obd_cleanup(obd);
1217 if (osp->opd_symlink)
1218 lprocfs_remove(&osp->opd_symlink);
1219 client_obd_cleanup(obd);
1223 if (osp->opd_connect_mdt) {
1224 struct client_obd *cli = &osp->opd_obd->u.cli;
1225 if (cli->cl_rpc_lock != NULL) {
1226 OBD_FREE_PTR(cli->cl_rpc_lock);
1227 cli->cl_rpc_lock = NULL;
1230 obd_disconnect(osp->opd_storage_exp);
1233 OBD_FREE(osdname, MAX_OBD_NAME);
1238 * Implementation of lu_device_type_operations::ldto_device_free
1240 * Free the OSP device in memory. No return value is needed for now,
1241 * so always return NULL to comply with the interface.
1243 * \param[in] env execution environment
1244 * \param[in] lu lu_device of OSP
1246 * \retval NULL NULL unconditionally
1248 static struct lu_device *osp_device_free(const struct lu_env *env,
1249 struct lu_device *lu)
1251 struct osp_device *osp = lu2osp_dev(lu);
1253 if (atomic_read(&lu->ld_ref) && lu->ld_site) {
1254 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1255 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
1257 dt_device_fini(&osp->opd_dt_dev);
1264 * Implementation of lu_device_type_operations::ldto_device_alloc
1266 * This function allocates and initializes OSP device in memory according to
1269 * \param[in] env execution environment
1270 * \param[in] type device type of OSP
1271 * \param[in] lcfg config log
1273 * \retval pointer the pointer of allocated OSP if succeed.
1274 * \retval ERR_PTR(errno) ERR_PTR(errno) if failed.
1276 static struct lu_device *osp_device_alloc(const struct lu_env *env,
1277 struct lu_device_type *type,
1278 struct lustre_cfg *lcfg)
1280 struct osp_device *osp;
1281 struct lu_device *ld;
1285 ld = ERR_PTR(-ENOMEM);
1289 ld = osp2lu_dev(osp);
1290 dt_device_init(&osp->opd_dt_dev, type);
1291 rc = osp_init0(env, osp, type, lcfg);
1293 osp_device_free(env, ld);
1301 * Implementation of lu_device_type_operations::ldto_device_fini
1303 * This function cleans up the OSP device, i.e. release and free those
1304 * attached items in osp_device.
1306 * \param[in] env execution environment
1307 * \param[in] ld lu_device of OSP
1309 * \retval NULL NULL if cleanup succeeded.
1310 * \retval ERR_PTR(errno) ERR_PTR(errno) if cleanup failed.
1312 static struct lu_device *osp_device_fini(const struct lu_env *env,
1313 struct lu_device *ld)
1315 struct osp_device *osp = lu2osp_dev(ld);
1316 struct obd_import *imp;
1321 if (osp->opd_async_requests != NULL) {
1322 osp_update_request_destroy(osp->opd_async_requests);
1323 osp->opd_async_requests = NULL;
1326 if (osp->opd_storage_exp)
1327 obd_disconnect(osp->opd_storage_exp);
1329 imp = osp->opd_obd->u.cli.cl_import;
1331 if (imp->imp_rq_pool) {
1332 ptlrpc_free_rq_pool(imp->imp_rq_pool);
1333 imp->imp_rq_pool = NULL;
1336 if (osp->opd_symlink)
1337 lprocfs_remove(&osp->opd_symlink);
1339 LASSERT(osp->opd_obd);
1340 ptlrpc_lprocfs_unregister_obd(osp->opd_obd);
1341 lprocfs_obd_cleanup(osp->opd_obd);
1343 if (osp->opd_connect_mdt) {
1344 struct client_obd *cli = &osp->opd_obd->u.cli;
1345 if (cli->cl_rpc_lock != NULL) {
1346 OBD_FREE_PTR(cli->cl_rpc_lock);
1347 cli->cl_rpc_lock = NULL;
1351 rc = client_obd_cleanup(osp->opd_obd);
1354 RETURN(ERR_PTR(rc));
1363 * Implementation of obd_ops::o_reconnect
1365 * This function is empty and does not need to do anything for now.
1367 static int osp_reconnect(const struct lu_env *env,
1368 struct obd_export *exp, struct obd_device *obd,
1369 struct obd_uuid *cluuid,
1370 struct obd_connect_data *data,
1377 * Implementation of obd_ops::o_connect
1379 * Connect OSP to the remote target (MDT or OST). Allocate the
1380 * export and return it to the LOD, which calls this function
1381 * for each OSP to connect it to the remote target. This function
1382 * is currently only called once per OSP.
1384 * \param[in] env execution environment
1385 * \param[out] exp export connected to OSP
1386 * \param[in] obd OSP device
1387 * \param[in] cluuid OSP device client uuid
1388 * \param[in] data connect_data to be used to connect to the remote
1390 * \param[in] localdata necessary for the API interface, but not used in
1393 * \retval 0 0 if the connection succeeded.
1394 * \retval negative negative errno if the connection failed.
1396 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
1397 struct obd_device *obd, struct obd_uuid *cluuid,
1398 struct obd_connect_data *data, void *localdata)
1400 struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1401 struct obd_connect_data *ocd;
1402 struct obd_import *imp;
1403 struct lustre_handle conn;
1408 CDEBUG(D_CONFIG, "connect #%d\n", osp->opd_connects);
1410 rc = class_connect(&conn, obd, cluuid);
1414 *exp = class_conn2export(&conn);
1415 /* Why should there ever be more than 1 connect? */
1416 osp->opd_connects++;
1417 LASSERT(osp->opd_connects == 1);
1419 osp->opd_exp = *exp;
1421 imp = osp->opd_obd->u.cli.cl_import;
1422 imp->imp_dlm_handle = conn;
1424 LASSERT(data != NULL);
1425 LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
1426 ocd = &imp->imp_connect_data;
1429 imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1431 ocd->ocd_version = LUSTRE_VERSION_CODE;
1432 ocd->ocd_index = data->ocd_index;
1433 imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1435 rc = ptlrpc_connect_import(imp);
1437 CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
1440 osp->opd_obd->u.cli.cl_seq->lcs_exp =
1441 class_export_get(osp->opd_exp);
1444 ptlrpc_pinger_add_import(imp);
1450 * Implementation of obd_ops::o_disconnect
1452 * Disconnect the export for the OSP. This is called by LOD to release the
1453 * OSP during cleanup (\see lod_del_device()). The OSP will be released after
1454 * the export is released.
1456 * \param[in] exp export to be disconnected.
1458 * \retval 0 0 if disconnection succeed
1459 * \retval negative negative errno if disconnection failed
1461 static int osp_obd_disconnect(struct obd_export *exp)
1463 struct obd_device *obd = exp->exp_obd;
1464 struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1468 /* Only disconnect the underlying layers on the final disconnect. */
1469 LASSERT(osp->opd_connects == 1);
1470 osp->opd_connects--;
1472 rc = class_disconnect(exp);
1474 CERROR("%s: class disconnect error: rc = %d\n",
1479 /* destroy the device */
1480 class_manual_cleanup(obd);
1486 * Implementation of obd_ops::o_statfs
1488 * Send a RPC to the remote target to get statfs status. This is only used
1489 * in lprocfs helpers by obd_statfs.
1491 * \param[in] env execution environment
1492 * \param[in] exp connection state from this OSP to the parent (LOD)
1494 * \param[out] osfs hold the statfs result
1495 * \param[in] unused Not used in this function for now
1496 * \param[in] flags flags to indicate how OSP will issue the RPC
1498 * \retval 0 0 if statfs succeeded.
1499 * \retval negative negative errno if statfs failed.
1501 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
1502 struct obd_statfs *osfs, __u64 unused, __u32 flags)
1504 struct obd_statfs *msfs;
1505 struct ptlrpc_request *req;
1506 struct obd_import *imp = NULL;
1511 /* Since the request might also come from lprocfs, so we need
1512 * sync this with client_disconnect_export Bug15684 */
1513 down_read(&exp->exp_obd->u.cli.cl_sem);
1514 if (exp->exp_obd->u.cli.cl_import)
1515 imp = class_import_get(exp->exp_obd->u.cli.cl_import);
1516 up_read(&exp->exp_obd->u.cli.cl_sem);
1520 req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
1522 class_import_put(imp);
1527 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
1529 ptlrpc_request_free(req);
1532 ptlrpc_request_set_replen(req);
1533 req->rq_request_portal = OST_CREATE_PORTAL;
1534 ptlrpc_at_set_req_timeout(req);
1536 if (flags & OBD_STATFS_NODELAY) {
1537 /* procfs requests not want stat in wait for avoid deadlock */
1538 req->rq_no_resend = 1;
1539 req->rq_no_delay = 1;
1542 rc = ptlrpc_queue_wait(req);
1546 msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1548 GOTO(out, rc = -EPROTO);
1554 ptlrpc_req_finished(req);
1559 * Implementation of obd_ops::o_import_event
1561 * This function is called when some related import event happens. It will
1562 * mark the necessary flags according to the event and notify the necessary
1563 * threads (mainly precreate thread).
1565 * \param[in] obd OSP OBD device
1566 * \param[in] imp import attached from OSP to remote (OST/MDT) service
1567 * \param[in] event event related to remote service (IMP_EVENT_*)
1569 * \retval 0 0 if the event handling succeeded.
1570 * \retval negative negative errno if the event handling failed.
1572 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1573 enum obd_import_event event)
1575 struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1578 case IMP_EVENT_DISCON:
1579 d->opd_got_disconnected = 1;
1580 d->opd_imp_connected = 0;
1581 if (d->opd_connect_mdt)
1584 if (d->opd_pre != NULL) {
1585 osp_pre_update_status(d, -ENODEV);
1586 wake_up(&d->opd_pre_waitq);
1589 CDEBUG(D_HA, "got disconnected\n");
1591 case IMP_EVENT_INACTIVE:
1592 d->opd_imp_active = 0;
1593 if (d->opd_connect_mdt)
1596 if (d->opd_pre != NULL) {
1597 osp_pre_update_status(d, -ENODEV);
1598 wake_up(&d->opd_pre_waitq);
1601 CDEBUG(D_HA, "got inactive\n");
1603 case IMP_EVENT_ACTIVE:
1604 d->opd_imp_active = 1;
1606 if (d->opd_got_disconnected)
1607 d->opd_new_connection = 1;
1608 d->opd_imp_connected = 1;
1609 d->opd_imp_seen_connected = 1;
1610 if (d->opd_connect_mdt)
1613 if (d->opd_pre != NULL)
1614 wake_up(&d->opd_pre_waitq);
1616 __osp_sync_check_for_work(d);
1617 CDEBUG(D_HA, "got connected\n");
1619 case IMP_EVENT_INVALIDATE:
1620 if (obd->obd_namespace == NULL)
1622 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1625 case IMP_EVENT_DEACTIVATE:
1626 case IMP_EVENT_ACTIVATE:
1629 CERROR("%s: unsupported import event: %#x\n",
1630 obd->obd_name, event);
1636 * Implementation of obd_ops: o_iocontrol
1638 * This function is the ioctl handler for OSP. Note: lctl will access the OSP
1639 * directly by ioctl, instead of through the MDS stack.
1641 * param[in] cmd ioctl command.
1642 * param[in] exp export of this OSP.
1643 * param[in] len data length of \a karg.
1644 * param[in] karg input argument which is packed as
1646 * param[out] uarg pointer to userspace buffer (must access by
1649 * \retval 0 0 if the ioctl handling succeeded.
1650 * \retval negative negative errno if the ioctl handling failed.
1652 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1653 void *karg, void *uarg)
1655 struct obd_device *obd = exp->exp_obd;
1656 struct osp_device *d;
1657 struct obd_ioctl_data *data = karg;
1662 LASSERT(obd->obd_lu_dev);
1663 d = lu2osp_dev(obd->obd_lu_dev);
1664 LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1666 if (!try_module_get(THIS_MODULE)) {
1667 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
1668 module_name(THIS_MODULE));
1673 case OBD_IOC_CLIENT_RECOVER:
1674 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1675 data->ioc_inlbuf1, 0);
1679 case IOC_OSC_SET_ACTIVE:
1680 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1683 case OBD_IOC_PING_TARGET:
1684 rc = ptlrpc_obd_ping(obd);
1687 CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
1688 cmd, current_comm());
1691 module_put(THIS_MODULE);
1696 * Implementation of obd_ops::o_get_info
1698 * Retrieve information by key. Retrieval starts from the top layer
1699 * (MDT) of the MDS stack and traverses the stack by calling the
1700 * obd_get_info() method of the next sub-layer.
1702 * \param[in] env execution environment
1703 * \param[in] exp export of this OSP
1704 * \param[in] keylen length of \a key
1705 * \param[in] key the key
1706 * \param[out] vallen length of \a val
1707 * \param[out] val holds the value returned by the key
1709 * \retval 0 0 if getting information succeeded.
1710 * \retval negative negative errno if getting information failed.
1712 static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1713 __u32 keylen, void *key, __u32 *vallen, void *val)
1717 if (KEY_IS(KEY_OSP_CONNECTED)) {
1718 struct obd_device *obd = exp->exp_obd;
1719 struct osp_device *osp;
1721 if (!obd->obd_set_up || obd->obd_stopping)
1724 osp = lu2osp_dev(obd->obd_lu_dev);
1727 * 1.8/2.0 behaviour is that OST being connected once at least
1728 * is considered "healthy". and one "healthy" OST is enough to
1729 * allow lustre clients to connect to MDS
1731 RETURN(!osp->opd_imp_seen_connected);
1738 * Implementation of obd_ops: o_fid_alloc
1740 * Allocate a FID. There are two cases in which OSP performs
1743 * 1. FID precreation for data objects, which is done in
1744 * osp_precreate_fids() instead of this function.
1745 * 2. FID allocation for each sub-stripe of a striped directory.
1746 * Similar to other FID clients, OSP requests the sequence
1747 * from its corresponding remote MDT, which in turn requests
1748 * sequences from the sequence controller (MDT0).
1750 * \param[in] env execution environment
1751 * \param[in] exp export of the OSP
1752 * \param[out] fid FID being allocated
1753 * \param[in] unused necessary for the interface but unused.
1755 * \retval 0 0 FID allocated successfully.
1756 * \retval 1 1 FID allocated successfully and new sequence
1757 * requested from seq meta server
1758 * \retval negative negative errno if FID allocation failed.
1760 static int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
1761 struct lu_fid *fid, struct md_op_data *unused)
1763 struct client_obd *cli = &exp->exp_obd->u.cli;
1764 struct osp_device *osp = lu2osp_dev(exp->exp_obd->obd_lu_dev);
1765 struct lu_client_seq *seq = cli->cl_seq;
1768 LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
1769 /* Sigh, fid client is not ready yet */
1770 LASSERT(osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL);
1772 RETURN(seq_client_alloc_fid(env, seq, fid));
1775 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1776 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1777 static void osp_key_exit(const struct lu_context *ctx,
1778 struct lu_context_key *key, void *data)
1780 struct osp_thread_info *info = data;
1782 info->osi_attr.la_valid = 0;
1785 struct lu_context_key osp_thread_key = {
1786 .lct_tags = LCT_MD_THREAD,
1787 .lct_init = osp_key_init,
1788 .lct_fini = osp_key_fini,
1789 .lct_exit = osp_key_exit
1792 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1793 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1795 struct lu_context_key osp_txn_key = {
1796 .lct_tags = LCT_OSP_THREAD | LCT_TX_HANDLE,
1797 .lct_init = osp_txn_key_init,
1798 .lct_fini = osp_txn_key_fini
1800 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1802 static struct lu_device_type_operations osp_device_type_ops = {
1803 .ldto_init = osp_type_init,
1804 .ldto_fini = osp_type_fini,
1806 .ldto_start = osp_type_start,
1807 .ldto_stop = osp_type_stop,
1809 .ldto_device_alloc = osp_device_alloc,
1810 .ldto_device_free = osp_device_free,
1812 .ldto_device_fini = osp_device_fini
1815 static struct lu_device_type osp_device_type = {
1816 .ldt_tags = LU_DEVICE_DT,
1817 .ldt_name = LUSTRE_OSP_NAME,
1818 .ldt_ops = &osp_device_type_ops,
1819 .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
1822 static struct obd_ops osp_obd_device_ops = {
1823 .o_owner = THIS_MODULE,
1824 .o_add_conn = client_import_add_conn,
1825 .o_del_conn = client_import_del_conn,
1826 .o_reconnect = osp_reconnect,
1827 .o_connect = osp_obd_connect,
1828 .o_disconnect = osp_obd_disconnect,
1829 .o_get_info = osp_obd_get_info,
1830 .o_import_event = osp_import_event,
1831 .o_iocontrol = osp_iocontrol,
1832 .o_statfs = osp_obd_statfs,
1833 .o_fid_init = client_fid_init,
1834 .o_fid_fini = client_fid_fini,
1835 .o_fid_alloc = osp_fid_alloc,
1838 struct llog_operations osp_mds_ost_orig_logops;
1841 * Initialize OSP module.
1843 * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
1844 * in obd_types (\see class_obd.c). Initialize procfs for the
1845 * the OSP device. Note: OSP was called OSC before Lustre 2.4,
1846 * so for compatibility it still uses the name "osc" in procfs.
1847 * This is called at module load time.
1849 * \retval 0 0 if initialization succeeds.
1850 * \retval negative negative errno if initialization failed.
1852 static int __init osp_mod_init(void)
1854 struct obd_type *type;
1857 rc = lu_kmem_init(osp_caches);
1862 rc = class_register_type(&osp_obd_device_ops, NULL, true, NULL,
1863 LUSTRE_OSP_NAME, &osp_device_type);
1865 lu_kmem_fini(osp_caches);
1869 rc = class_register_type(&lwp_obd_device_ops, NULL, true, NULL,
1870 LUSTRE_LWP_NAME, &lwp_device_type);
1872 class_unregister_type(LUSTRE_OSP_NAME);
1873 lu_kmem_fini(osp_caches);
1877 /* Note: add_rec/delcare_add_rec will be only used by catalogs */
1878 osp_mds_ost_orig_logops = llog_osd_ops;
1879 osp_mds_ost_orig_logops.lop_add = llog_cat_add_rec;
1880 osp_mds_ost_orig_logops.lop_declare_add = llog_cat_declare_add_rec;
1882 /* create "osc" entry in procfs for compatibility purposes */
1883 type = class_search_type(LUSTRE_OSC_NAME);
1884 if (type != NULL && type->typ_procroot != NULL)
1887 type = class_search_type(LUSTRE_OSP_NAME);
1888 type->typ_procsym = lprocfs_register("osc", proc_lustre_root,
1890 if (IS_ERR(type->typ_procsym)) {
1891 CERROR("osp: can't create compat entry \"osc\": %d\n",
1892 (int) PTR_ERR(type->typ_procsym));
1893 type->typ_procsym = NULL;
1899 * Finalize OSP module.
1901 * This callback is called when kernel unloads OSP module from memory, and
1902 * it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
1904 static void __exit osp_mod_exit(void)
1906 class_unregister_type(LUSTRE_LWP_NAME);
1907 class_unregister_type(LUSTRE_OSP_NAME);
1908 lu_kmem_fini(osp_caches);
1911 MODULE_AUTHOR("Intel, Inc. <http://www.intel.com/>");
1912 MODULE_DESCRIPTION("Lustre OST Proxy Device ("LUSTRE_OSP_NAME")");
1913 MODULE_VERSION(LUSTRE_VERSION_STRING);
1914 MODULE_LICENSE("GPL");
1916 module_init(osp_mod_init);
1917 module_exit(osp_mod_exit);