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) 1999, 2010, Oracle and/or its affiliates. All rights reserved.
24 * Use is subject to license terms.
26 * Copyright (c) 2011, 2017, Intel Corporation.
29 * This file is part of Lustre, http://www.lustre.org/
31 * lustre/obdclass/genops.c
33 * These are the only exported functions, they provide some generic
34 * infrastructure for managing object devices
37 #define DEBUG_SUBSYSTEM S_CLASS
39 #include <linux/pid_namespace.h>
40 #include <linux/workqueue.h>
41 #include <lustre_compat.h>
42 #include <obd_class.h>
43 #include <lustre_log.h>
44 #include <lprocfs_status.h>
45 #include <lustre_disk.h>
46 #include <lustre_kernelcomm.h>
48 DEFINE_RWLOCK(obd_dev_lock);
49 static struct obd_device *obd_devs[MAX_OBD_DEVICES];
51 static struct kmem_cache *obd_device_cachep;
52 static struct kobj_type class_ktype;
53 static struct workqueue_struct *zombie_wq;
55 static void obd_zombie_export_add(struct obd_export *exp);
56 static void obd_zombie_import_add(struct obd_import *imp);
57 static void print_export_data(struct obd_export *exp,
58 const char *status, int locks, int debug_level);
60 static LIST_HEAD(obd_stale_exports);
61 static DEFINE_SPINLOCK(obd_stale_export_lock);
62 static atomic_t obd_stale_export_num = ATOMIC_INIT(0);
65 * support functions: we could use inter-module communication, but this
66 * is more portable to other OS's
68 static struct obd_device *obd_device_alloc(void)
70 struct obd_device *obd;
72 OBD_SLAB_ALLOC_PTR_GFP(obd, obd_device_cachep, GFP_NOFS);
74 obd->obd_magic = OBD_DEVICE_MAGIC;
79 static void obd_device_free(struct obd_device *obd)
82 LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "obd %p obd_magic %08x != %08x\n",
83 obd, obd->obd_magic, OBD_DEVICE_MAGIC);
84 if (obd->obd_namespace != NULL) {
85 CERROR("obd %p: namespace %p was not properly cleaned up (obd_force=%d)!\n",
86 obd, obd->obd_namespace, obd->obd_force);
89 lu_ref_fini(&obd->obd_reference);
90 OBD_SLAB_FREE_PTR(obd, obd_device_cachep);
93 struct obd_type *class_search_type(const char *name)
95 struct kobject *kobj = kset_find_obj(lustre_kset, name);
97 if (kobj && kobj->ktype == &class_ktype)
98 return container_of(kobj, struct obd_type, typ_kobj);
103 EXPORT_SYMBOL(class_search_type);
105 struct obd_type *class_get_type(const char *name)
107 struct obd_type *type;
109 type = class_search_type(name);
110 #ifdef HAVE_MODULE_LOADING_SUPPORT
112 const char *modname = name;
114 #ifdef HAVE_SERVER_SUPPORT
115 if (strcmp(modname, "obdfilter") == 0)
118 if (strcmp(modname, LUSTRE_LWP_NAME) == 0)
119 modname = LUSTRE_OSP_NAME;
121 if (!strncmp(modname, LUSTRE_MDS_NAME, strlen(LUSTRE_MDS_NAME)))
122 modname = LUSTRE_MDT_NAME;
123 #endif /* HAVE_SERVER_SUPPORT */
125 if (!request_module("%s", modname)) {
126 CDEBUG(D_INFO, "Loaded module '%s'\n", modname);
127 type = class_search_type(name);
129 LCONSOLE_ERROR_MSG(0x158, "Can't load module '%s'\n",
135 if (try_module_get(type->typ_dt_ops->o_owner)) {
136 atomic_inc(&type->typ_refcnt);
137 /* class_search_type() returned a counted reference,
138 * but we don't need that count any more as
139 * we have one through typ_refcnt.
141 kobject_put(&type->typ_kobj);
143 kobject_put(&type->typ_kobj);
150 void class_put_type(struct obd_type *type)
153 module_put(type->typ_dt_ops->o_owner);
154 atomic_dec(&type->typ_refcnt);
157 static void class_sysfs_release(struct kobject *kobj)
159 struct obd_type *type = container_of(kobj, struct obd_type, typ_kobj);
161 debugfs_remove_recursive(type->typ_debugfs_entry);
162 type->typ_debugfs_entry = NULL;
165 lu_device_type_fini(type->typ_lu);
167 #ifdef CONFIG_PROC_FS
168 if (type->typ_name && type->typ_procroot)
169 remove_proc_subtree(type->typ_name, proc_lustre_root);
171 OBD_FREE(type, sizeof(*type));
174 static struct kobj_type class_ktype = {
175 .sysfs_ops = &lustre_sysfs_ops,
176 .release = class_sysfs_release,
179 #ifdef HAVE_SERVER_SUPPORT
180 struct obd_type *class_add_symlinks(const char *name, bool enable_proc)
182 struct dentry *symlink;
183 struct obd_type *type;
186 type = class_search_type(name);
188 kobject_put(&type->typ_kobj);
189 return ERR_PTR(-EEXIST);
192 OBD_ALLOC(type, sizeof(*type));
194 return ERR_PTR(-ENOMEM);
196 type->typ_kobj.kset = lustre_kset;
197 rc = kobject_init_and_add(&type->typ_kobj, &class_ktype,
198 &lustre_kset->kobj, "%s", name);
202 symlink = debugfs_create_dir(name, debugfs_lustre_root);
203 type->typ_debugfs_entry = symlink;
204 type->typ_sym_filter = true;
207 type->typ_procroot = lprocfs_register(name, proc_lustre_root,
209 if (IS_ERR(type->typ_procroot)) {
210 CERROR("%s: can't create compat proc entry: %d\n",
211 name, (int)PTR_ERR(type->typ_procroot));
212 type->typ_procroot = NULL;
218 EXPORT_SYMBOL(class_add_symlinks);
219 #endif /* HAVE_SERVER_SUPPORT */
221 #define CLASS_MAX_NAME 1024
223 int class_register_type(const struct obd_ops *dt_ops,
224 const struct md_ops *md_ops,
226 const char *name, struct lu_device_type *ldt)
228 struct obd_type *type;
233 LASSERT(strnlen(name, CLASS_MAX_NAME) < CLASS_MAX_NAME);
235 type = class_search_type(name);
237 #ifdef HAVE_SERVER_SUPPORT
238 if (type->typ_sym_filter)
240 #endif /* HAVE_SERVER_SUPPORT */
241 kobject_put(&type->typ_kobj);
242 CDEBUG(D_IOCTL, "Type %s already registered\n", name);
246 OBD_ALLOC(type, sizeof(*type));
250 type->typ_lu = ldt ? OBD_LU_TYPE_SETUP : NULL;
251 type->typ_kobj.kset = lustre_kset;
252 kobject_init(&type->typ_kobj, &class_ktype);
253 #ifdef HAVE_SERVER_SUPPORT
255 #endif /* HAVE_SERVER_SUPPORT */
257 type->typ_dt_ops = dt_ops;
258 type->typ_md_ops = md_ops;
260 #ifdef HAVE_SERVER_SUPPORT
261 if (type->typ_sym_filter) {
262 type->typ_sym_filter = false;
263 kobject_put(&type->typ_kobj);
267 #ifdef CONFIG_PROC_FS
268 if (enable_proc && !type->typ_procroot) {
269 type->typ_procroot = lprocfs_register(name,
272 if (IS_ERR(type->typ_procroot)) {
273 rc = PTR_ERR(type->typ_procroot);
274 type->typ_procroot = NULL;
279 type->typ_debugfs_entry = debugfs_create_dir(name, debugfs_lustre_root);
281 rc = kobject_add(&type->typ_kobj, &lustre_kset->kobj, "%s", name);
284 #ifdef HAVE_SERVER_SUPPORT
288 rc = lu_device_type_init(ldt);
289 smp_store_release(&type->typ_lu, rc ? NULL : ldt);
290 wake_up_var(&type->typ_lu);
298 kobject_put(&type->typ_kobj);
302 EXPORT_SYMBOL(class_register_type);
304 int class_unregister_type(const char *name)
306 struct obd_type *type = class_search_type(name);
311 CERROR("unknown obd type\n");
315 if (atomic_read(&type->typ_refcnt)) {
316 CERROR("type %s has refcount (%d)\n", name,
317 atomic_read(&type->typ_refcnt));
318 /* This is a bad situation, let's make the best of it */
319 /* Remove ops, but leave the name for debugging */
320 type->typ_dt_ops = NULL;
321 type->typ_md_ops = NULL;
322 GOTO(out_put, rc = -EBUSY);
325 /* Put the final ref */
326 kobject_put(&type->typ_kobj);
328 /* Put the ref returned by class_search_type() */
329 kobject_put(&type->typ_kobj);
332 } /* class_unregister_type */
333 EXPORT_SYMBOL(class_unregister_type);
336 * Create a new obd device.
338 * Allocate the new obd_device and initialize it.
340 * \param[in] type_name obd device type string.
341 * \param[in] name obd device name.
342 * \param[in] uuid obd device UUID
344 * \retval newdev pointer to created obd_device
345 * \retval ERR_PTR(errno) on error
347 struct obd_device *class_newdev(const char *type_name, const char *name,
350 struct obd_device *newdev;
351 struct obd_type *type = NULL;
354 if (strlen(name) >= MAX_OBD_NAME) {
355 CERROR("name/uuid must be < %u bytes long\n", MAX_OBD_NAME);
356 RETURN(ERR_PTR(-EINVAL));
359 type = class_get_type(type_name);
361 CERROR("OBD: unknown type: %s\n", type_name);
362 RETURN(ERR_PTR(-ENODEV));
365 newdev = obd_device_alloc();
366 if (newdev == NULL) {
367 class_put_type(type);
368 RETURN(ERR_PTR(-ENOMEM));
370 LASSERT(newdev->obd_magic == OBD_DEVICE_MAGIC);
371 strncpy(newdev->obd_name, name, sizeof(newdev->obd_name) - 1);
372 newdev->obd_type = type;
373 newdev->obd_minor = -1;
375 rwlock_init(&newdev->obd_pool_lock);
376 newdev->obd_pool_limit = 0;
377 newdev->obd_pool_slv = 0;
379 INIT_LIST_HEAD(&newdev->obd_exports);
380 newdev->obd_num_exports = 0;
381 newdev->obd_grant_check_threshold = 100;
382 INIT_LIST_HEAD(&newdev->obd_unlinked_exports);
383 INIT_LIST_HEAD(&newdev->obd_delayed_exports);
384 INIT_LIST_HEAD(&newdev->obd_exports_timed);
385 INIT_LIST_HEAD(&newdev->obd_nid_stats);
386 spin_lock_init(&newdev->obd_nid_lock);
387 spin_lock_init(&newdev->obd_dev_lock);
388 mutex_init(&newdev->obd_dev_mutex);
389 spin_lock_init(&newdev->obd_osfs_lock);
390 /* newdev->obd_osfs_age must be set to a value in the distant
391 * past to guarantee a fresh statfs is fetched on mount. */
392 newdev->obd_osfs_age = ktime_get_seconds() - 1000;
394 /* XXX belongs in setup not attach */
395 init_rwsem(&newdev->obd_observer_link_sem);
397 spin_lock_init(&newdev->obd_recovery_task_lock);
398 init_waitqueue_head(&newdev->obd_next_transno_waitq);
399 init_waitqueue_head(&newdev->obd_evict_inprogress_waitq);
400 INIT_LIST_HEAD(&newdev->obd_req_replay_queue);
401 INIT_LIST_HEAD(&newdev->obd_lock_replay_queue);
402 INIT_LIST_HEAD(&newdev->obd_final_req_queue);
403 INIT_LIST_HEAD(&newdev->obd_evict_list);
404 INIT_LIST_HEAD(&newdev->obd_lwp_list);
406 llog_group_init(&newdev->obd_olg);
407 /* Detach drops this */
408 atomic_set(&newdev->obd_refcount, 1);
409 lu_ref_init(&newdev->obd_reference);
410 lu_ref_add(&newdev->obd_reference, "newdev", newdev);
412 newdev->obd_conn_inprogress = 0;
414 strncpy(newdev->obd_uuid.uuid, uuid, UUID_MAX);
416 CDEBUG(D_IOCTL, "Allocate new device %s (%p)\n",
417 newdev->obd_name, newdev);
425 * \param[in] obd obd_device to be freed
429 void class_free_dev(struct obd_device *obd)
431 struct obd_type *obd_type = obd->obd_type;
433 LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "%p obd_magic %08x "
434 "!= %08x\n", obd, obd->obd_magic, OBD_DEVICE_MAGIC);
435 LASSERTF(obd->obd_minor == -1 || obd_devs[obd->obd_minor] == obd,
436 "obd %p != obd_devs[%d] %p\n",
437 obd, obd->obd_minor, obd_devs[obd->obd_minor]);
438 LASSERTF(atomic_read(&obd->obd_refcount) == 0,
439 "obd_refcount should be 0, not %d\n",
440 atomic_read(&obd->obd_refcount));
441 LASSERT(obd_type != NULL);
443 CDEBUG(D_INFO, "Release obd device %s obd_type name = %s\n",
444 obd->obd_name, obd->obd_type->typ_name);
446 CDEBUG(D_CONFIG, "finishing cleanup of obd %s (%s)\n",
447 obd->obd_name, obd->obd_uuid.uuid);
448 if (obd->obd_stopping) {
451 /* If we're not stopping, we were never set up */
452 err = obd_cleanup(obd);
454 CERROR("Cleanup %s returned %d\n",
458 obd_device_free(obd);
460 class_put_type(obd_type);
464 * Unregister obd device.
466 * Free slot in obd_dev[] used by \a obd.
468 * \param[in] new_obd obd_device to be unregistered
472 void class_unregister_device(struct obd_device *obd)
474 write_lock(&obd_dev_lock);
475 if (obd->obd_minor >= 0) {
476 LASSERT(obd_devs[obd->obd_minor] == obd);
477 obd_devs[obd->obd_minor] = NULL;
480 write_unlock(&obd_dev_lock);
484 * Register obd device.
486 * Find free slot in obd_devs[], fills it with \a new_obd.
488 * \param[in] new_obd obd_device to be registered
491 * \retval -EEXIST device with this name is registered
492 * \retval -EOVERFLOW obd_devs[] is full
494 int class_register_device(struct obd_device *new_obd)
498 int new_obd_minor = 0;
499 bool minor_assign = false;
500 bool retried = false;
503 write_lock(&obd_dev_lock);
504 for (i = 0; i < class_devno_max(); i++) {
505 struct obd_device *obd = class_num2obd(i);
508 (strcmp(new_obd->obd_name, obd->obd_name) == 0)) {
511 write_unlock(&obd_dev_lock);
513 /* the obd_device could be waited to be
514 * destroyed by the "obd_zombie_impexp_thread".
516 obd_zombie_barrier();
521 CERROR("%s: already exists, won't add\n",
523 /* in case we found a free slot before duplicate */
524 minor_assign = false;
528 if (!minor_assign && obd == NULL) {
535 new_obd->obd_minor = new_obd_minor;
536 LASSERTF(obd_devs[new_obd_minor] == NULL, "obd_devs[%d] "
537 "%p\n", new_obd_minor, obd_devs[new_obd_minor]);
538 obd_devs[new_obd_minor] = new_obd;
542 CERROR("%s: all %u/%u devices used, increase "
543 "MAX_OBD_DEVICES: rc = %d\n", new_obd->obd_name,
544 i, class_devno_max(), ret);
547 write_unlock(&obd_dev_lock);
552 static int class_name2dev_nolock(const char *name)
559 for (i = 0; i < class_devno_max(); i++) {
560 struct obd_device *obd = class_num2obd(i);
562 if (obd && strcmp(name, obd->obd_name) == 0) {
563 /* Make sure we finished attaching before we give
564 out any references */
565 LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
566 if (obd->obd_attached) {
576 int class_name2dev(const char *name)
583 read_lock(&obd_dev_lock);
584 i = class_name2dev_nolock(name);
585 read_unlock(&obd_dev_lock);
589 EXPORT_SYMBOL(class_name2dev);
591 struct obd_device *class_name2obd(const char *name)
593 int dev = class_name2dev(name);
595 if (dev < 0 || dev > class_devno_max())
597 return class_num2obd(dev);
599 EXPORT_SYMBOL(class_name2obd);
601 int class_uuid2dev_nolock(struct obd_uuid *uuid)
605 for (i = 0; i < class_devno_max(); i++) {
606 struct obd_device *obd = class_num2obd(i);
608 if (obd && obd_uuid_equals(uuid, &obd->obd_uuid)) {
609 LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
617 int class_uuid2dev(struct obd_uuid *uuid)
621 read_lock(&obd_dev_lock);
622 i = class_uuid2dev_nolock(uuid);
623 read_unlock(&obd_dev_lock);
627 EXPORT_SYMBOL(class_uuid2dev);
629 struct obd_device *class_uuid2obd(struct obd_uuid *uuid)
631 int dev = class_uuid2dev(uuid);
634 return class_num2obd(dev);
636 EXPORT_SYMBOL(class_uuid2obd);
639 * Get obd device from ::obd_devs[]
641 * \param num [in] array index
643 * \retval NULL if ::obd_devs[\a num] does not contains an obd device
644 * otherwise return the obd device there.
646 struct obd_device *class_num2obd(int num)
648 struct obd_device *obd = NULL;
650 if (num < class_devno_max()) {
655 LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC,
656 "%p obd_magic %08x != %08x\n",
657 obd, obd->obd_magic, OBD_DEVICE_MAGIC);
658 LASSERTF(obd->obd_minor == num,
659 "%p obd_minor %0d != %0d\n",
660 obd, obd->obd_minor, num);
665 EXPORT_SYMBOL(class_num2obd);
668 * Find obd in obd_dev[] by name or uuid.
670 * Increment obd's refcount if found.
672 * \param[in] str obd name or uuid
674 * \retval NULL if not found
675 * \retval target pointer to found obd_device
677 struct obd_device *class_dev_by_str(const char *str)
679 struct obd_device *target = NULL;
680 struct obd_uuid tgtuuid;
683 obd_str2uuid(&tgtuuid, str);
685 read_lock(&obd_dev_lock);
686 rc = class_uuid2dev_nolock(&tgtuuid);
688 rc = class_name2dev_nolock(str);
691 target = class_num2obd(rc);
694 class_incref(target, "find", current);
695 read_unlock(&obd_dev_lock);
699 EXPORT_SYMBOL(class_dev_by_str);
702 * Get obd devices count. Device in any
704 * \retval obd device count
706 int get_devices_count(void)
708 int index, max_index = class_devno_max(), dev_count = 0;
710 read_lock(&obd_dev_lock);
711 for (index = 0; index <= max_index; index++) {
712 struct obd_device *obd = class_num2obd(index);
716 read_unlock(&obd_dev_lock);
720 EXPORT_SYMBOL(get_devices_count);
722 void class_obd_list(void)
727 read_lock(&obd_dev_lock);
728 for (i = 0; i < class_devno_max(); i++) {
729 struct obd_device *obd = class_num2obd(i);
733 if (obd->obd_stopping)
735 else if (obd->obd_set_up)
737 else if (obd->obd_attached)
741 LCONSOLE(D_CONFIG, "%3d %s %s %s %s %d\n",
742 i, status, obd->obd_type->typ_name,
743 obd->obd_name, obd->obd_uuid.uuid,
744 atomic_read(&obd->obd_refcount));
746 read_unlock(&obd_dev_lock);
749 /* Search for a client OBD connected to tgt_uuid. If grp_uuid is
750 * specified, then only the client with that uuid is returned,
751 * otherwise any client connected to the tgt is returned.
753 struct obd_device *class_find_client_obd(struct obd_uuid *tgt_uuid,
754 const char *type_name,
755 struct obd_uuid *grp_uuid)
759 read_lock(&obd_dev_lock);
760 for (i = 0; i < class_devno_max(); i++) {
761 struct obd_device *obd = class_num2obd(i);
765 if ((strncmp(obd->obd_type->typ_name, type_name,
766 strlen(type_name)) == 0)) {
767 if (obd_uuid_equals(tgt_uuid,
768 &obd->u.cli.cl_target_uuid) &&
769 ((grp_uuid)? obd_uuid_equals(grp_uuid,
770 &obd->obd_uuid) : 1)) {
771 read_unlock(&obd_dev_lock);
776 read_unlock(&obd_dev_lock);
780 EXPORT_SYMBOL(class_find_client_obd);
782 /* Iterate the obd_device list looking devices have grp_uuid. Start
783 * searching at *next, and if a device is found, the next index to look
784 * at is saved in *next. If next is NULL, then the first matching device
785 * will always be returned.
787 struct obd_device *class_devices_in_group(struct obd_uuid *grp_uuid, int *next)
793 else if (*next >= 0 && *next < class_devno_max())
798 read_lock(&obd_dev_lock);
799 for (; i < class_devno_max(); i++) {
800 struct obd_device *obd = class_num2obd(i);
804 if (obd_uuid_equals(grp_uuid, &obd->obd_uuid)) {
807 read_unlock(&obd_dev_lock);
811 read_unlock(&obd_dev_lock);
815 EXPORT_SYMBOL(class_devices_in_group);
818 * to notify sptlrpc log for \a fsname has changed, let every relevant OBD
819 * adjust sptlrpc settings accordingly.
821 int class_notify_sptlrpc_conf(const char *fsname, int namelen)
823 struct obd_device *obd;
827 LASSERT(namelen > 0);
829 read_lock(&obd_dev_lock);
830 for (i = 0; i < class_devno_max(); i++) {
831 obd = class_num2obd(i);
833 if (obd == NULL || obd->obd_set_up == 0 || obd->obd_stopping)
836 /* only notify mdc, osc, osp, lwp, mdt, ost
837 * because only these have a -sptlrpc llog */
838 type = obd->obd_type->typ_name;
839 if (strcmp(type, LUSTRE_MDC_NAME) != 0 &&
840 strcmp(type, LUSTRE_OSC_NAME) != 0 &&
841 strcmp(type, LUSTRE_OSP_NAME) != 0 &&
842 strcmp(type, LUSTRE_LWP_NAME) != 0 &&
843 strcmp(type, LUSTRE_MDT_NAME) != 0 &&
844 strcmp(type, LUSTRE_OST_NAME) != 0)
847 if (strncmp(obd->obd_name, fsname, namelen))
850 class_incref(obd, __FUNCTION__, obd);
851 read_unlock(&obd_dev_lock);
852 rc2 = obd_set_info_async(NULL, obd->obd_self_export,
853 sizeof(KEY_SPTLRPC_CONF),
854 KEY_SPTLRPC_CONF, 0, NULL, NULL);
856 class_decref(obd, __FUNCTION__, obd);
857 read_lock(&obd_dev_lock);
859 read_unlock(&obd_dev_lock);
862 EXPORT_SYMBOL(class_notify_sptlrpc_conf);
864 void obd_cleanup_caches(void)
867 if (obd_device_cachep) {
868 kmem_cache_destroy(obd_device_cachep);
869 obd_device_cachep = NULL;
875 int obd_init_caches(void)
880 LASSERT(obd_device_cachep == NULL);
881 obd_device_cachep = kmem_cache_create_usercopy("ll_obd_dev_cache",
882 sizeof(struct obd_device),
883 0, 0, 0, sizeof(struct obd_device), NULL);
884 if (!obd_device_cachep)
885 GOTO(out, rc = -ENOMEM);
889 obd_cleanup_caches();
893 static const char export_handle_owner[] = "export";
895 /* map connection to client */
896 struct obd_export *class_conn2export(struct lustre_handle *conn)
898 struct obd_export *export;
902 CDEBUG(D_CACHE, "looking for null handle\n");
906 if (conn->cookie == -1) { /* this means assign a new connection */
907 CDEBUG(D_CACHE, "want a new connection\n");
911 CDEBUG(D_INFO, "looking for export cookie %#llx\n", conn->cookie);
912 export = class_handle2object(conn->cookie, export_handle_owner);
915 EXPORT_SYMBOL(class_conn2export);
917 struct obd_device *class_exp2obd(struct obd_export *exp)
923 EXPORT_SYMBOL(class_exp2obd);
925 struct obd_import *class_exp2cliimp(struct obd_export *exp)
927 struct obd_device *obd = exp->exp_obd;
930 return obd->u.cli.cl_import;
932 EXPORT_SYMBOL(class_exp2cliimp);
934 /* Export management functions */
935 static void class_export_destroy(struct obd_export *exp)
937 struct obd_device *obd = exp->exp_obd;
940 LASSERT(refcount_read(&exp->exp_handle.h_ref) == 0);
941 LASSERT(obd != NULL);
943 CDEBUG(D_IOCTL, "destroying export %p/%s for %s\n", exp,
944 exp->exp_client_uuid.uuid, obd->obd_name);
946 /* "Local" exports (lctl, LOV->{mdc,osc}) have no connection. */
947 ptlrpc_connection_put(exp->exp_connection);
949 LASSERT(list_empty(&exp->exp_outstanding_replies));
950 LASSERT(list_empty(&exp->exp_uncommitted_replies));
951 LASSERT(list_empty(&exp->exp_req_replay_queue));
952 LASSERT(list_empty(&exp->exp_hp_rpcs));
953 obd_destroy_export(exp);
954 /* self export doesn't hold a reference to an obd, although it
955 * exists until freeing of the obd */
956 if (exp != obd->obd_self_export)
957 class_decref(obd, "export", exp);
959 OBD_FREE_PRE(exp, sizeof(*exp), "rcu");
960 kfree_rcu(exp, exp_handle.h_rcu);
964 struct obd_export *class_export_get(struct obd_export *exp)
966 refcount_inc(&exp->exp_handle.h_ref);
967 CDEBUG(D_INFO, "GET export %p refcount=%d\n", exp,
968 refcount_read(&exp->exp_handle.h_ref));
971 EXPORT_SYMBOL(class_export_get);
973 void class_export_put(struct obd_export *exp)
975 LASSERT(exp != NULL);
976 LASSERT(refcount_read(&exp->exp_handle.h_ref) > 0);
977 LASSERT(refcount_read(&exp->exp_handle.h_ref) < LI_POISON);
978 CDEBUG(D_INFO, "PUTting export %p : new refcount %d\n", exp,
979 refcount_read(&exp->exp_handle.h_ref) - 1);
981 if (refcount_dec_and_test(&exp->exp_handle.h_ref)) {
982 struct obd_device *obd = exp->exp_obd;
984 CDEBUG(D_IOCTL, "final put %p/%s\n",
985 exp, exp->exp_client_uuid.uuid);
987 /* release nid stat refererence */
988 lprocfs_exp_cleanup(exp);
990 if (exp == obd->obd_self_export) {
991 /* self export should be destroyed without
992 * zombie thread as it doesn't hold a
993 * reference to obd and doesn't hold any
995 class_export_destroy(exp);
996 /* self export is destroyed, no class
997 * references exist and it is safe to free
1001 LASSERT(!list_empty(&exp->exp_obd_chain));
1002 obd_zombie_export_add(exp);
1007 EXPORT_SYMBOL(class_export_put);
1009 static void obd_zombie_exp_cull(struct work_struct *ws)
1011 struct obd_export *export;
1013 export = container_of(ws, struct obd_export, exp_zombie_work);
1014 class_export_destroy(export);
1017 /* Creates a new export, adds it to the hash table, and returns a
1018 * pointer to it. The refcount is 2: one for the hash reference, and
1019 * one for the pointer returned by this function. */
1020 struct obd_export *__class_new_export(struct obd_device *obd,
1021 struct obd_uuid *cluuid, bool is_self)
1023 struct obd_export *export;
1027 OBD_ALLOC_PTR(export);
1029 return ERR_PTR(-ENOMEM);
1031 export->exp_conn_cnt = 0;
1032 export->exp_lock_hash = NULL;
1033 export->exp_flock_hash = NULL;
1034 /* 2 = class_handle_hash + last */
1035 refcount_set(&export->exp_handle.h_ref, 2);
1036 atomic_set(&export->exp_rpc_count, 0);
1037 atomic_set(&export->exp_cb_count, 0);
1038 atomic_set(&export->exp_locks_count, 0);
1039 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1040 INIT_LIST_HEAD(&export->exp_locks_list);
1041 spin_lock_init(&export->exp_locks_list_guard);
1043 atomic_set(&export->exp_replay_count, 0);
1044 export->exp_obd = obd;
1045 INIT_LIST_HEAD(&export->exp_outstanding_replies);
1046 spin_lock_init(&export->exp_uncommitted_replies_lock);
1047 INIT_LIST_HEAD(&export->exp_uncommitted_replies);
1048 INIT_LIST_HEAD(&export->exp_req_replay_queue);
1049 INIT_HLIST_NODE(&export->exp_handle.h_link);
1050 INIT_LIST_HEAD(&export->exp_hp_rpcs);
1051 INIT_LIST_HEAD(&export->exp_reg_rpcs);
1052 class_handle_hash(&export->exp_handle, export_handle_owner);
1053 export->exp_last_request_time = ktime_get_real_seconds();
1054 spin_lock_init(&export->exp_lock);
1055 spin_lock_init(&export->exp_rpc_lock);
1056 INIT_HLIST_NODE(&export->exp_gen_hash);
1057 spin_lock_init(&export->exp_bl_list_lock);
1058 INIT_LIST_HEAD(&export->exp_bl_list);
1059 INIT_LIST_HEAD(&export->exp_stale_list);
1060 INIT_WORK(&export->exp_zombie_work, obd_zombie_exp_cull);
1062 export->exp_sp_peer = LUSTRE_SP_ANY;
1063 export->exp_flvr.sf_rpc = SPTLRPC_FLVR_INVALID;
1064 export->exp_client_uuid = *cluuid;
1065 obd_init_export(export);
1067 at_init(&export->exp_bl_lock_at, obd_timeout, 0);
1069 spin_lock(&obd->obd_dev_lock);
1070 if (!obd_uuid_equals(cluuid, &obd->obd_uuid)) {
1071 /* shouldn't happen, but might race */
1072 if (obd->obd_stopping)
1073 GOTO(exit_unlock, rc = -ENODEV);
1075 rc = obd_uuid_add(obd, export);
1077 LCONSOLE_WARN("%s: denying duplicate export for %s: rc = %d\n",
1078 obd->obd_name, cluuid->uuid, rc);
1079 GOTO(exit_unlock, rc = -EALREADY);
1084 class_incref(obd, "export", export);
1085 list_add_tail(&export->exp_obd_chain_timed,
1086 &obd->obd_exports_timed);
1087 list_add(&export->exp_obd_chain, &obd->obd_exports);
1088 obd->obd_num_exports++;
1090 INIT_LIST_HEAD(&export->exp_obd_chain_timed);
1091 INIT_LIST_HEAD(&export->exp_obd_chain);
1093 spin_unlock(&obd->obd_dev_lock);
1097 spin_unlock(&obd->obd_dev_lock);
1098 class_handle_unhash(&export->exp_handle);
1099 obd_destroy_export(export);
1100 OBD_FREE_PTR(export);
1104 struct obd_export *class_new_export(struct obd_device *obd,
1105 struct obd_uuid *uuid)
1107 return __class_new_export(obd, uuid, false);
1109 EXPORT_SYMBOL(class_new_export);
1111 struct obd_export *class_new_export_self(struct obd_device *obd,
1112 struct obd_uuid *uuid)
1114 return __class_new_export(obd, uuid, true);
1117 void class_unlink_export(struct obd_export *exp)
1119 class_handle_unhash(&exp->exp_handle);
1121 if (exp->exp_obd->obd_self_export == exp) {
1122 class_export_put(exp);
1126 spin_lock(&exp->exp_obd->obd_dev_lock);
1127 /* delete an uuid-export hashitem from hashtables */
1128 if (exp != exp->exp_obd->obd_self_export)
1129 obd_uuid_del(exp->exp_obd, exp);
1131 #ifdef HAVE_SERVER_SUPPORT
1132 if (!hlist_unhashed(&exp->exp_gen_hash)) {
1133 struct tg_export_data *ted = &exp->exp_target_data;
1134 struct cfs_hash *hash;
1136 /* Because obd_gen_hash will not be released until
1137 * class_cleanup(), so hash should never be NULL here */
1138 hash = cfs_hash_getref(exp->exp_obd->obd_gen_hash);
1139 LASSERT(hash != NULL);
1140 cfs_hash_del(hash, &ted->ted_lcd->lcd_generation,
1141 &exp->exp_gen_hash);
1142 cfs_hash_putref(hash);
1144 #endif /* HAVE_SERVER_SUPPORT */
1146 list_move(&exp->exp_obd_chain, &exp->exp_obd->obd_unlinked_exports);
1147 list_del_init(&exp->exp_obd_chain_timed);
1148 exp->exp_obd->obd_num_exports--;
1149 spin_unlock(&exp->exp_obd->obd_dev_lock);
1150 atomic_inc(&obd_stale_export_num);
1152 /* A reference is kept by obd_stale_exports list */
1153 obd_stale_export_put(exp);
1155 EXPORT_SYMBOL(class_unlink_export);
1157 /* Import management functions */
1158 static void obd_zombie_import_free(struct obd_import *imp)
1160 struct obd_import_conn *imp_conn;
1163 CDEBUG(D_IOCTL, "destroying import %p for %s\n", imp,
1164 imp->imp_obd->obd_name);
1166 LASSERT(refcount_read(&imp->imp_refcount) == 0);
1168 ptlrpc_connection_put(imp->imp_connection);
1170 while ((imp_conn = list_first_entry_or_null(&imp->imp_conn_list,
1171 struct obd_import_conn,
1172 oic_item)) != NULL) {
1173 list_del_init(&imp_conn->oic_item);
1174 ptlrpc_connection_put(imp_conn->oic_conn);
1175 OBD_FREE(imp_conn, sizeof(*imp_conn));
1178 LASSERT(imp->imp_sec == NULL);
1179 LASSERTF(atomic_read(&imp->imp_reqs) == 0, "%s: imp_reqs = %d\n",
1180 imp->imp_obd->obd_name, atomic_read(&imp->imp_reqs));
1181 class_decref(imp->imp_obd, "import", imp);
1186 struct obd_import *class_import_get(struct obd_import *import)
1188 refcount_inc(&import->imp_refcount);
1189 CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", import,
1190 refcount_read(&import->imp_refcount),
1191 import->imp_obd->obd_name);
1194 EXPORT_SYMBOL(class_import_get);
1196 void class_import_put(struct obd_import *imp)
1200 LASSERT(refcount_read(&imp->imp_refcount) > 0);
1202 CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", imp,
1203 refcount_read(&imp->imp_refcount) - 1,
1204 imp->imp_obd->obd_name);
1206 if (refcount_dec_and_test(&imp->imp_refcount)) {
1207 CDEBUG(D_INFO, "final put import %p\n", imp);
1208 obd_zombie_import_add(imp);
1213 EXPORT_SYMBOL(class_import_put);
1215 static void init_imp_at(struct imp_at *at) {
1217 at_init(&at->iat_net_latency, 0, 0);
1218 for (i = 0; i < IMP_AT_MAX_PORTALS; i++) {
1219 /* max service estimates are tracked on the server side, so
1220 don't use the AT history here, just use the last reported
1221 val. (But keep hist for proc histogram, worst_ever) */
1222 at_init(&at->iat_service_estimate[i], INITIAL_CONNECT_TIMEOUT,
1227 static void obd_zombie_imp_cull(struct work_struct *ws)
1229 struct obd_import *import;
1231 import = container_of(ws, struct obd_import, imp_zombie_work);
1232 obd_zombie_import_free(import);
1235 struct obd_import *class_new_import(struct obd_device *obd)
1237 struct obd_import *imp;
1238 struct pid_namespace *curr_pid_ns = ll_task_pid_ns(current);
1240 OBD_ALLOC(imp, sizeof(*imp));
1244 INIT_LIST_HEAD(&imp->imp_pinger_chain);
1245 INIT_LIST_HEAD(&imp->imp_replay_list);
1246 INIT_LIST_HEAD(&imp->imp_sending_list);
1247 INIT_LIST_HEAD(&imp->imp_delayed_list);
1248 INIT_LIST_HEAD(&imp->imp_committed_list);
1249 INIT_LIST_HEAD(&imp->imp_unreplied_list);
1250 imp->imp_known_replied_xid = 0;
1251 imp->imp_replay_cursor = &imp->imp_committed_list;
1252 spin_lock_init(&imp->imp_lock);
1253 imp->imp_last_success_conn = 0;
1254 imp->imp_state = LUSTRE_IMP_NEW;
1255 imp->imp_obd = class_incref(obd, "import", imp);
1256 rwlock_init(&imp->imp_sec_lock);
1257 init_waitqueue_head(&imp->imp_recovery_waitq);
1258 INIT_WORK(&imp->imp_zombie_work, obd_zombie_imp_cull);
1260 if (curr_pid_ns && curr_pid_ns->child_reaper)
1261 imp->imp_sec_refpid = curr_pid_ns->child_reaper->pid;
1263 imp->imp_sec_refpid = 1;
1265 refcount_set(&imp->imp_refcount, 2);
1266 atomic_set(&imp->imp_unregistering, 0);
1267 atomic_set(&imp->imp_reqs, 0);
1268 atomic_set(&imp->imp_inflight, 0);
1269 atomic_set(&imp->imp_replay_inflight, 0);
1270 init_waitqueue_head(&imp->imp_replay_waitq);
1271 atomic_set(&imp->imp_inval_count, 0);
1272 INIT_LIST_HEAD(&imp->imp_conn_list);
1273 init_imp_at(&imp->imp_at);
1275 /* the default magic is V2, will be used in connect RPC, and
1276 * then adjusted according to the flags in request/reply. */
1277 imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
1281 EXPORT_SYMBOL(class_new_import);
1283 void class_destroy_import(struct obd_import *import)
1285 LASSERT(import != NULL);
1286 LASSERT(import != LP_POISON);
1288 spin_lock(&import->imp_lock);
1289 import->imp_generation++;
1290 spin_unlock(&import->imp_lock);
1291 class_import_put(import);
1293 EXPORT_SYMBOL(class_destroy_import);
1295 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1297 void __class_export_add_lock_ref(struct obd_export *exp, struct ldlm_lock *lock)
1299 spin_lock(&exp->exp_locks_list_guard);
1301 LASSERT(lock->l_exp_refs_nr >= 0);
1303 if (lock->l_exp_refs_target != NULL &&
1304 lock->l_exp_refs_target != exp) {
1305 LCONSOLE_WARN("setting export %p for lock %p which already has export %p\n",
1306 exp, lock, lock->l_exp_refs_target);
1308 if ((lock->l_exp_refs_nr ++) == 0) {
1309 list_add(&lock->l_exp_refs_link, &exp->exp_locks_list);
1310 lock->l_exp_refs_target = exp;
1312 CDEBUG(D_INFO, "lock = %p, export = %p, refs = %u\n",
1313 lock, exp, lock->l_exp_refs_nr);
1314 spin_unlock(&exp->exp_locks_list_guard);
1316 EXPORT_SYMBOL(__class_export_add_lock_ref);
1318 void __class_export_del_lock_ref(struct obd_export *exp, struct ldlm_lock *lock)
1320 spin_lock(&exp->exp_locks_list_guard);
1321 LASSERT(lock->l_exp_refs_nr > 0);
1322 if (lock->l_exp_refs_target != exp) {
1323 LCONSOLE_WARN("lock %p, "
1324 "mismatching export pointers: %p, %p\n",
1325 lock, lock->l_exp_refs_target, exp);
1327 if (-- lock->l_exp_refs_nr == 0) {
1328 list_del_init(&lock->l_exp_refs_link);
1329 lock->l_exp_refs_target = NULL;
1331 CDEBUG(D_INFO, "lock = %p, export = %p, refs = %u\n",
1332 lock, exp, lock->l_exp_refs_nr);
1333 spin_unlock(&exp->exp_locks_list_guard);
1335 EXPORT_SYMBOL(__class_export_del_lock_ref);
1338 /* A connection defines an export context in which preallocation can
1339 be managed. This releases the export pointer reference, and returns
1340 the export handle, so the export refcount is 1 when this function
1342 int class_connect(struct lustre_handle *conn, struct obd_device *obd,
1343 struct obd_uuid *cluuid)
1345 struct obd_export *export;
1346 LASSERT(conn != NULL);
1347 LASSERT(obd != NULL);
1348 LASSERT(cluuid != NULL);
1351 export = class_new_export(obd, cluuid);
1353 RETURN(PTR_ERR(export));
1355 conn->cookie = export->exp_handle.h_cookie;
1356 class_export_put(export);
1358 CDEBUG(D_IOCTL, "connect: client %s, cookie %#llx\n",
1359 cluuid->uuid, conn->cookie);
1362 EXPORT_SYMBOL(class_connect);
1364 /* if export is involved in recovery then clean up related things */
1365 static void class_export_recovery_cleanup(struct obd_export *exp)
1367 struct obd_device *obd = exp->exp_obd;
1369 spin_lock(&obd->obd_recovery_task_lock);
1370 if (obd->obd_recovering) {
1371 if (exp->exp_in_recovery) {
1372 spin_lock(&exp->exp_lock);
1373 exp->exp_in_recovery = 0;
1374 spin_unlock(&exp->exp_lock);
1375 LASSERT_ATOMIC_POS(&obd->obd_connected_clients);
1376 atomic_dec(&obd->obd_connected_clients);
1379 /* if called during recovery then should update
1380 * obd_stale_clients counter,
1381 * lightweight exports are not counted */
1382 if ((exp_connect_flags(exp) & OBD_CONNECT_LIGHTWEIGHT) == 0)
1383 exp->exp_obd->obd_stale_clients++;
1385 spin_unlock(&obd->obd_recovery_task_lock);
1387 spin_lock(&exp->exp_lock);
1388 /** Cleanup req replay fields */
1389 if (exp->exp_req_replay_needed) {
1390 exp->exp_req_replay_needed = 0;
1392 LASSERT(atomic_read(&obd->obd_req_replay_clients));
1393 atomic_dec(&obd->obd_req_replay_clients);
1396 /** Cleanup lock replay data */
1397 if (exp->exp_lock_replay_needed) {
1398 exp->exp_lock_replay_needed = 0;
1400 LASSERT(atomic_read(&obd->obd_lock_replay_clients));
1401 atomic_dec(&obd->obd_lock_replay_clients);
1403 spin_unlock(&exp->exp_lock);
1406 /* This function removes 1-3 references from the export:
1407 * 1 - for export pointer passed
1408 * and if disconnect really need
1409 * 2 - removing from hash
1410 * 3 - in client_unlink_export
1411 * The export pointer passed to this function can destroyed */
1412 int class_disconnect(struct obd_export *export)
1414 int already_disconnected;
1417 if (export == NULL) {
1418 CWARN("attempting to free NULL export %p\n", export);
1422 spin_lock(&export->exp_lock);
1423 already_disconnected = export->exp_disconnected;
1424 export->exp_disconnected = 1;
1425 #ifdef HAVE_SERVER_SUPPORT
1426 /* We hold references of export for uuid hash
1427 * and nid_hash and export link at least. So
1428 * it is safe to call rh*table_remove_fast in
1431 obd_nid_del(export->exp_obd, export);
1432 #endif /* HAVE_SERVER_SUPPORT */
1433 spin_unlock(&export->exp_lock);
1435 /* class_cleanup(), abort_recovery(), and class_fail_export()
1436 * all end up in here, and if any of them race we shouldn't
1437 * call extra class_export_puts(). */
1438 if (already_disconnected)
1439 GOTO(no_disconn, already_disconnected);
1441 CDEBUG(D_IOCTL, "disconnect: cookie %#llx\n",
1442 export->exp_handle.h_cookie);
1444 class_export_recovery_cleanup(export);
1445 class_unlink_export(export);
1447 class_export_put(export);
1450 EXPORT_SYMBOL(class_disconnect);
1452 /* Return non-zero for a fully connected export */
1453 int class_connected_export(struct obd_export *exp)
1458 spin_lock(&exp->exp_lock);
1459 connected = (exp->exp_conn_cnt > 0) && !exp->exp_failed;
1460 spin_unlock(&exp->exp_lock);
1464 EXPORT_SYMBOL(class_connected_export);
1466 static void class_disconnect_export_list(struct list_head *list,
1467 enum obd_option flags)
1470 struct obd_export *exp;
1473 /* It's possible that an export may disconnect itself, but
1474 * nothing else will be added to this list.
1476 while ((exp = list_first_entry_or_null(list, struct obd_export,
1477 exp_obd_chain)) != NULL) {
1478 /* need for safe call CDEBUG after obd_disconnect */
1479 class_export_get(exp);
1481 spin_lock(&exp->exp_lock);
1482 exp->exp_flags = flags;
1483 spin_unlock(&exp->exp_lock);
1485 if (obd_uuid_equals(&exp->exp_client_uuid,
1486 &exp->exp_obd->obd_uuid)) {
1488 "exp %p export uuid == obd uuid, don't discon\n",
1490 /* Need to delete this now so we don't end up pointing
1491 * to work_list later when this export is cleaned up. */
1492 list_del_init(&exp->exp_obd_chain);
1493 class_export_put(exp);
1497 class_export_get(exp);
1498 CDEBUG(D_HA, "%s: disconnecting export at %s (%p), "
1499 "last request at %lld\n",
1500 exp->exp_obd->obd_name, obd_export_nid2str(exp),
1501 exp, exp->exp_last_request_time);
1502 /* release one export reference anyway */
1503 rc = obd_disconnect(exp);
1505 CDEBUG(D_HA, "disconnected export at %s (%p): rc %d\n",
1506 obd_export_nid2str(exp), exp, rc);
1507 class_export_put(exp);
1512 void class_disconnect_exports(struct obd_device *obd)
1514 LIST_HEAD(work_list);
1517 /* Move all of the exports from obd_exports to a work list, en masse. */
1518 spin_lock(&obd->obd_dev_lock);
1519 list_splice_init(&obd->obd_exports, &work_list);
1520 list_splice_init(&obd->obd_delayed_exports, &work_list);
1521 spin_unlock(&obd->obd_dev_lock);
1523 if (!list_empty(&work_list)) {
1524 CDEBUG(D_HA, "OBD device %d (%p) has exports, "
1525 "disconnecting them\n", obd->obd_minor, obd);
1526 class_disconnect_export_list(&work_list,
1527 exp_flags_from_obd(obd));
1529 CDEBUG(D_HA, "OBD device %d (%p) has no exports\n",
1530 obd->obd_minor, obd);
1533 EXPORT_SYMBOL(class_disconnect_exports);
1535 /* Remove exports that have not completed recovery.
1537 void class_disconnect_stale_exports(struct obd_device *obd,
1538 int (*test_export)(struct obd_export *))
1540 LIST_HEAD(work_list);
1541 struct obd_export *exp, *n;
1545 spin_lock(&obd->obd_dev_lock);
1546 list_for_each_entry_safe(exp, n, &obd->obd_exports,
1548 /* don't count self-export as client */
1549 if (obd_uuid_equals(&exp->exp_client_uuid,
1550 &exp->exp_obd->obd_uuid))
1553 /* don't evict clients which have no slot in last_rcvd
1554 * (e.g. lightweight connection) */
1555 if (exp->exp_target_data.ted_lr_idx == -1)
1558 spin_lock(&exp->exp_lock);
1559 if (exp->exp_failed || test_export(exp)) {
1560 spin_unlock(&exp->exp_lock);
1563 exp->exp_failed = 1;
1564 spin_unlock(&exp->exp_lock);
1566 list_move(&exp->exp_obd_chain, &work_list);
1568 CDEBUG(D_HA, "%s: disconnect stale client %s@%s\n",
1569 obd->obd_name, exp->exp_client_uuid.uuid,
1570 obd_export_nid2str(exp));
1571 print_export_data(exp, "EVICTING", 0, D_HA);
1573 spin_unlock(&obd->obd_dev_lock);
1576 LCONSOLE_WARN("%s: disconnecting %d stale clients\n",
1577 obd->obd_name, evicted);
1579 class_disconnect_export_list(&work_list, exp_flags_from_obd(obd) |
1580 OBD_OPT_ABORT_RECOV);
1583 EXPORT_SYMBOL(class_disconnect_stale_exports);
1585 void class_fail_export(struct obd_export *exp)
1587 int rc, already_failed;
1589 spin_lock(&exp->exp_lock);
1590 already_failed = exp->exp_failed;
1591 exp->exp_failed = 1;
1592 spin_unlock(&exp->exp_lock);
1594 if (already_failed) {
1595 CDEBUG(D_HA, "disconnecting dead export %p/%s; skipping\n",
1596 exp, exp->exp_client_uuid.uuid);
1600 CDEBUG(D_HA, "disconnecting export %p/%s\n",
1601 exp, exp->exp_client_uuid.uuid);
1603 if (obd_dump_on_timeout)
1604 libcfs_debug_dumplog();
1606 /* need for safe call CDEBUG after obd_disconnect */
1607 class_export_get(exp);
1609 /* Most callers into obd_disconnect are removing their own reference
1610 * (request, for example) in addition to the one from the hash table.
1611 * We don't have such a reference here, so make one. */
1612 class_export_get(exp);
1613 rc = obd_disconnect(exp);
1615 CERROR("disconnecting export %p failed: %d\n", exp, rc);
1617 CDEBUG(D_HA, "disconnected export %p/%s\n",
1618 exp, exp->exp_client_uuid.uuid);
1619 class_export_put(exp);
1621 EXPORT_SYMBOL(class_fail_export);
1623 #ifdef HAVE_SERVER_SUPPORT
1625 static int take_first(struct obd_export *exp, void *data)
1627 struct obd_export **expp = data;
1630 /* already have one */
1632 if (exp->exp_failed)
1633 /* Don't want this one */
1635 if (!refcount_inc_not_zero(&exp->exp_handle.h_ref))
1636 /* Cannot get a ref on this one */
1642 int obd_export_evict_by_nid(struct obd_device *obd, const char *nid)
1644 struct lnet_nid nid_key;
1645 struct obd_export *doomed_exp;
1646 int exports_evicted = 0;
1648 libcfs_strnid(&nid_key, nid);
1650 spin_lock(&obd->obd_dev_lock);
1651 /* umount has run already, so evict thread should leave
1652 * its task to umount thread now */
1653 if (obd->obd_stopping) {
1654 spin_unlock(&obd->obd_dev_lock);
1655 return exports_evicted;
1657 spin_unlock(&obd->obd_dev_lock);
1660 while (obd_nid_export_for_each(obd, &nid_key,
1661 take_first, &doomed_exp) > 0) {
1663 LASSERTF(doomed_exp != obd->obd_self_export,
1664 "self-export is hashed by NID?\n");
1666 LCONSOLE_WARN("%s: evicting %s (at %s) by administrative request\n",
1668 obd_uuid2str(&doomed_exp->exp_client_uuid),
1669 obd_export_nid2str(doomed_exp));
1671 class_fail_export(doomed_exp);
1672 class_export_put(doomed_exp);
1677 if (!exports_evicted)
1679 "%s: can't disconnect NID '%s': no exports found\n",
1680 obd->obd_name, nid);
1681 return exports_evicted;
1683 EXPORT_SYMBOL(obd_export_evict_by_nid);
1685 int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid)
1687 struct obd_export *doomed_exp = NULL;
1688 struct obd_uuid doomed_uuid;
1689 int exports_evicted = 0;
1691 spin_lock(&obd->obd_dev_lock);
1692 if (obd->obd_stopping) {
1693 spin_unlock(&obd->obd_dev_lock);
1694 return exports_evicted;
1696 spin_unlock(&obd->obd_dev_lock);
1698 obd_str2uuid(&doomed_uuid, uuid);
1699 if (obd_uuid_equals(&doomed_uuid, &obd->obd_uuid)) {
1700 CERROR("%s: can't evict myself\n", obd->obd_name);
1701 return exports_evicted;
1704 doomed_exp = obd_uuid_lookup(obd, &doomed_uuid);
1705 if (doomed_exp == NULL) {
1706 CERROR("%s: can't disconnect %s: no exports found\n",
1707 obd->obd_name, uuid);
1709 CWARN("%s: evicting %s at adminstrative request\n",
1710 obd->obd_name, doomed_exp->exp_client_uuid.uuid);
1711 class_fail_export(doomed_exp);
1712 class_export_put(doomed_exp);
1713 obd_uuid_del(obd, doomed_exp);
1717 return exports_evicted;
1719 #endif /* HAVE_SERVER_SUPPORT */
1721 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1722 void (*class_export_dump_hook)(struct obd_export*) = NULL;
1723 EXPORT_SYMBOL(class_export_dump_hook);
1726 static void print_export_data(struct obd_export *exp, const char *status,
1727 int locks, int debug_level)
1729 struct ptlrpc_reply_state *rs;
1730 struct ptlrpc_reply_state *first_reply = NULL;
1733 spin_lock(&exp->exp_lock);
1734 list_for_each_entry(rs, &exp->exp_outstanding_replies,
1740 spin_unlock(&exp->exp_lock);
1742 CDEBUG(debug_level, "%s: %s %p %s %s %d (%d %d %d) %d %d %d %d: "
1743 "%p %s %llu stale:%d\n",
1744 exp->exp_obd->obd_name, status, exp, exp->exp_client_uuid.uuid,
1745 obd_export_nid2str(exp),
1746 refcount_read(&exp->exp_handle.h_ref),
1747 atomic_read(&exp->exp_rpc_count),
1748 atomic_read(&exp->exp_cb_count),
1749 atomic_read(&exp->exp_locks_count),
1750 exp->exp_disconnected, exp->exp_delayed, exp->exp_failed,
1751 nreplies, first_reply, nreplies > 3 ? "..." : "",
1752 exp->exp_last_committed, !list_empty(&exp->exp_stale_list));
1753 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1754 if (locks && class_export_dump_hook != NULL)
1755 class_export_dump_hook(exp);
1759 void dump_exports(struct obd_device *obd, int locks, int debug_level)
1761 struct obd_export *exp;
1763 spin_lock(&obd->obd_dev_lock);
1764 list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain)
1765 print_export_data(exp, "ACTIVE", locks, debug_level);
1766 list_for_each_entry(exp, &obd->obd_unlinked_exports, exp_obd_chain)
1767 print_export_data(exp, "UNLINKED", locks, debug_level);
1768 list_for_each_entry(exp, &obd->obd_delayed_exports, exp_obd_chain)
1769 print_export_data(exp, "DELAYED", locks, debug_level);
1770 spin_unlock(&obd->obd_dev_lock);
1773 void obd_exports_barrier(struct obd_device *obd)
1776 LASSERT(list_empty(&obd->obd_exports));
1777 spin_lock(&obd->obd_dev_lock);
1778 while (!list_empty(&obd->obd_unlinked_exports)) {
1779 spin_unlock(&obd->obd_dev_lock);
1780 schedule_timeout_uninterruptible(cfs_time_seconds(waited));
1781 if (waited > 5 && is_power_of_2(waited)) {
1782 LCONSOLE_WARN("%s is waiting for obd_unlinked_exports "
1783 "more than %d seconds. "
1784 "The obd refcount = %d. Is it stuck?\n",
1785 obd->obd_name, waited,
1786 atomic_read(&obd->obd_refcount));
1787 dump_exports(obd, 1, D_CONSOLE | D_WARNING);
1790 spin_lock(&obd->obd_dev_lock);
1792 spin_unlock(&obd->obd_dev_lock);
1794 EXPORT_SYMBOL(obd_exports_barrier);
1797 * Add export to the obd_zombe thread and notify it.
1799 static void obd_zombie_export_add(struct obd_export *exp) {
1800 atomic_dec(&obd_stale_export_num);
1801 spin_lock(&exp->exp_obd->obd_dev_lock);
1802 LASSERT(!list_empty(&exp->exp_obd_chain));
1803 list_del_init(&exp->exp_obd_chain);
1804 spin_unlock(&exp->exp_obd->obd_dev_lock);
1806 queue_work(zombie_wq, &exp->exp_zombie_work);
1810 * Add import to the obd_zombe thread and notify it.
1812 static void obd_zombie_import_add(struct obd_import *imp) {
1813 LASSERT(imp->imp_sec == NULL);
1815 queue_work(zombie_wq, &imp->imp_zombie_work);
1819 * wait when obd_zombie import/export queues become empty
1821 void obd_zombie_barrier(void)
1823 flush_workqueue(zombie_wq);
1825 EXPORT_SYMBOL(obd_zombie_barrier);
1828 struct obd_export *obd_stale_export_get(void)
1830 struct obd_export *exp = NULL;
1833 spin_lock(&obd_stale_export_lock);
1834 if (!list_empty(&obd_stale_exports)) {
1835 exp = list_first_entry(&obd_stale_exports,
1836 struct obd_export, exp_stale_list);
1837 list_del_init(&exp->exp_stale_list);
1839 spin_unlock(&obd_stale_export_lock);
1842 CDEBUG(D_DLMTRACE, "Get export %p: total %d\n", exp,
1843 atomic_read(&obd_stale_export_num));
1847 EXPORT_SYMBOL(obd_stale_export_get);
1849 void obd_stale_export_put(struct obd_export *exp)
1853 LASSERT(list_empty(&exp->exp_stale_list));
1854 if (exp->exp_lock_hash &&
1855 atomic_read(&exp->exp_lock_hash->hs_count)) {
1856 CDEBUG(D_DLMTRACE, "Put export %p: total %d\n", exp,
1857 atomic_read(&obd_stale_export_num));
1859 spin_lock_bh(&exp->exp_bl_list_lock);
1860 spin_lock(&obd_stale_export_lock);
1861 /* Add to the tail if there is no blocked locks,
1862 * to the head otherwise. */
1863 if (list_empty(&exp->exp_bl_list))
1864 list_add_tail(&exp->exp_stale_list,
1865 &obd_stale_exports);
1867 list_add(&exp->exp_stale_list,
1868 &obd_stale_exports);
1870 spin_unlock(&obd_stale_export_lock);
1871 spin_unlock_bh(&exp->exp_bl_list_lock);
1873 class_export_put(exp);
1877 EXPORT_SYMBOL(obd_stale_export_put);
1880 * Adjust the position of the export in the stale list,
1881 * i.e. move to the head of the list if is needed.
1883 void obd_stale_export_adjust(struct obd_export *exp)
1885 LASSERT(exp != NULL);
1886 spin_lock_bh(&exp->exp_bl_list_lock);
1887 spin_lock(&obd_stale_export_lock);
1889 if (!list_empty(&exp->exp_stale_list) &&
1890 !list_empty(&exp->exp_bl_list))
1891 list_move(&exp->exp_stale_list, &obd_stale_exports);
1893 spin_unlock(&obd_stale_export_lock);
1894 spin_unlock_bh(&exp->exp_bl_list_lock);
1896 EXPORT_SYMBOL(obd_stale_export_adjust);
1899 * start destroy zombie import/export thread
1901 int obd_zombie_impexp_init(void)
1903 zombie_wq = cfs_cpt_bind_workqueue("obd_zombid", cfs_cpt_tab,
1905 cfs_cpt_number(cfs_cpt_tab));
1907 return IS_ERR(zombie_wq) ? PTR_ERR(zombie_wq) : 0;
1911 * stop destroy zombie import/export thread
1913 void obd_zombie_impexp_stop(void)
1915 destroy_workqueue(zombie_wq);
1916 LASSERT(list_empty(&obd_stale_exports));
1919 /***** Kernel-userspace comm helpers *******/
1921 /* Get length of entire message, including header */
1922 int kuc_len(int payload_len)
1924 return sizeof(struct kuc_hdr) + payload_len;
1926 EXPORT_SYMBOL(kuc_len);
1928 /* Get a pointer to kuc header, given a ptr to the payload
1929 * @param p Pointer to payload area
1930 * @returns Pointer to kuc header
1932 struct kuc_hdr * kuc_ptr(void *p)
1934 struct kuc_hdr *lh = ((struct kuc_hdr *)p) - 1;
1935 LASSERT(lh->kuc_magic == KUC_MAGIC);
1938 EXPORT_SYMBOL(kuc_ptr);
1940 /* Alloc space for a message, and fill in header
1941 * @return Pointer to payload area
1943 void *kuc_alloc(int payload_len, int transport, int type)
1946 int len = kuc_len(payload_len);
1950 return ERR_PTR(-ENOMEM);
1952 lh->kuc_magic = KUC_MAGIC;
1953 lh->kuc_transport = transport;
1954 lh->kuc_msgtype = type;
1955 lh->kuc_msglen = len;
1957 return (void *)(lh + 1);
1959 EXPORT_SYMBOL(kuc_alloc);
1961 /* Takes pointer to payload area */
1962 void kuc_free(void *p, int payload_len)
1964 struct kuc_hdr *lh = kuc_ptr(p);
1965 OBD_FREE(lh, kuc_len(payload_len));
1967 EXPORT_SYMBOL(kuc_free);
1969 struct obd_request_slot_waiter {
1970 struct list_head orsw_entry;
1971 wait_queue_head_t orsw_waitq;
1975 static bool obd_request_slot_avail(struct client_obd *cli,
1976 struct obd_request_slot_waiter *orsw)
1980 spin_lock(&cli->cl_loi_list_lock);
1981 avail = !!list_empty(&orsw->orsw_entry);
1982 spin_unlock(&cli->cl_loi_list_lock);
1988 * For network flow control, the RPC sponsor needs to acquire a credit
1989 * before sending the RPC. The credits count for a connection is defined
1990 * by the "cl_max_rpcs_in_flight". If all the credits are occpuied, then
1991 * the subsequent RPC sponsors need to wait until others released their
1992 * credits, or the administrator increased the "cl_max_rpcs_in_flight".
1994 int obd_get_request_slot(struct client_obd *cli)
1996 struct obd_request_slot_waiter orsw;
1999 spin_lock(&cli->cl_loi_list_lock);
2000 if (cli->cl_rpcs_in_flight < cli->cl_max_rpcs_in_flight) {
2001 cli->cl_rpcs_in_flight++;
2002 spin_unlock(&cli->cl_loi_list_lock);
2006 init_waitqueue_head(&orsw.orsw_waitq);
2007 list_add_tail(&orsw.orsw_entry, &cli->cl_flight_waiters);
2008 orsw.orsw_signaled = false;
2009 spin_unlock(&cli->cl_loi_list_lock);
2011 rc = l_wait_event_abortable(orsw.orsw_waitq,
2012 obd_request_slot_avail(cli, &orsw) ||
2013 orsw.orsw_signaled);
2015 /* Here, we must take the lock to avoid the on-stack 'orsw' to be
2016 * freed but other (such as obd_put_request_slot) is using it. */
2017 spin_lock(&cli->cl_loi_list_lock);
2019 if (!orsw.orsw_signaled) {
2020 if (list_empty(&orsw.orsw_entry))
2021 cli->cl_rpcs_in_flight--;
2023 list_del(&orsw.orsw_entry);
2028 if (orsw.orsw_signaled) {
2029 LASSERT(list_empty(&orsw.orsw_entry));
2033 spin_unlock(&cli->cl_loi_list_lock);
2037 EXPORT_SYMBOL(obd_get_request_slot);
2039 void obd_put_request_slot(struct client_obd *cli)
2041 struct obd_request_slot_waiter *orsw;
2043 spin_lock(&cli->cl_loi_list_lock);
2044 cli->cl_rpcs_in_flight--;
2046 /* If there is free slot, wakeup the first waiter. */
2047 if (!list_empty(&cli->cl_flight_waiters) &&
2048 likely(cli->cl_rpcs_in_flight < cli->cl_max_rpcs_in_flight)) {
2049 orsw = list_first_entry(&cli->cl_flight_waiters,
2050 struct obd_request_slot_waiter,
2052 list_del_init(&orsw->orsw_entry);
2053 cli->cl_rpcs_in_flight++;
2054 wake_up(&orsw->orsw_waitq);
2056 spin_unlock(&cli->cl_loi_list_lock);
2058 EXPORT_SYMBOL(obd_put_request_slot);
2060 __u32 obd_get_max_rpcs_in_flight(struct client_obd *cli)
2062 return cli->cl_max_rpcs_in_flight;
2064 EXPORT_SYMBOL(obd_get_max_rpcs_in_flight);
2066 int obd_set_max_rpcs_in_flight(struct client_obd *cli, __u32 max)
2068 struct obd_request_slot_waiter *orsw;
2074 if (max > OBD_MAX_RIF_MAX || max < 1)
2077 CDEBUG(D_INFO, "%s: max = %hu max_mod = %u rif = %u\n",
2078 cli->cl_import->imp_obd->obd_name, max,
2079 cli->cl_max_mod_rpcs_in_flight, cli->cl_max_rpcs_in_flight);
2081 if (strcmp(cli->cl_import->imp_obd->obd_type->typ_name,
2082 LUSTRE_MDC_NAME) == 0) {
2083 /* adjust max_mod_rpcs_in_flight to ensure it is always
2084 * strictly lower that max_rpcs_in_flight */
2086 CERROR("%s: cannot set mdc.*.max_rpcs_in_flight=1\n",
2087 cli->cl_import->imp_obd->obd_name);
2090 if (max <= cli->cl_max_mod_rpcs_in_flight) {
2091 rc = obd_set_max_mod_rpcs_in_flight(cli, max - 1);
2097 spin_lock(&cli->cl_loi_list_lock);
2098 old = cli->cl_max_rpcs_in_flight;
2099 cli->cl_max_rpcs_in_flight = max;
2100 client_adjust_max_dirty(cli);
2104 /* We increase the max_rpcs_in_flight, then wakeup some waiters. */
2105 for (i = 0; i < diff; i++) {
2106 orsw = list_first_entry_or_null(&cli->cl_loi_read_list,
2107 struct obd_request_slot_waiter,
2112 list_del_init(&orsw->orsw_entry);
2113 cli->cl_rpcs_in_flight++;
2114 wake_up(&orsw->orsw_waitq);
2116 spin_unlock(&cli->cl_loi_list_lock);
2120 EXPORT_SYMBOL(obd_set_max_rpcs_in_flight);
2122 __u16 obd_get_max_mod_rpcs_in_flight(struct client_obd *cli)
2124 return cli->cl_max_mod_rpcs_in_flight;
2126 EXPORT_SYMBOL(obd_get_max_mod_rpcs_in_flight);
2128 int obd_set_max_mod_rpcs_in_flight(struct client_obd *cli, __u16 max)
2130 struct obd_connect_data *ocd;
2134 if (max > OBD_MAX_RIF_MAX || max < 1)
2137 ocd = &cli->cl_import->imp_connect_data;
2138 CDEBUG(D_INFO, "%s: max = %hu flags = %llx, max_mod = %u rif = %u\n",
2139 cli->cl_import->imp_obd->obd_name, max, ocd->ocd_connect_flags,
2140 ocd->ocd_maxmodrpcs, cli->cl_max_rpcs_in_flight);
2142 if (max == OBD_MAX_RIF_MAX)
2143 max = OBD_MAX_RIF_MAX - 1;
2145 /* Cannot exceed or equal max_rpcs_in_flight. If we are asked to
2146 * increase this value, also bump up max_rpcs_in_flight to match.
2148 if (max >= cli->cl_max_rpcs_in_flight) {
2150 "%s: increasing max_rpcs_in_flight=%hu to allow larger max_mod_rpcs_in_flight=%u\n",
2151 cli->cl_import->imp_obd->obd_name, max + 1, max);
2152 obd_set_max_rpcs_in_flight(cli, max + 1);
2155 /* cannot exceed max modify RPCs in flight supported by the server,
2156 * but verify ocd_connect_flags is at least initialized first. If
2157 * not, allow it and fix value later in ptlrpc_connect_set_flags().
2159 if (!ocd->ocd_connect_flags) {
2160 maxmodrpcs = cli->cl_max_rpcs_in_flight - 1;
2161 } else if (ocd->ocd_connect_flags & OBD_CONNECT_MULTIMODRPCS) {
2162 maxmodrpcs = ocd->ocd_maxmodrpcs;
2163 if (maxmodrpcs == 0) { /* connection not finished yet */
2164 maxmodrpcs = cli->cl_max_rpcs_in_flight - 1;
2166 "%s: partial connect, assume maxmodrpcs=%hu\n",
2167 cli->cl_import->imp_obd->obd_name, maxmodrpcs);
2172 if (max > maxmodrpcs) {
2173 CERROR("%s: can't set max_mod_rpcs_in_flight=%hu higher than ocd_maxmodrpcs=%hu returned by the server at connection\n",
2174 cli->cl_import->imp_obd->obd_name,
2179 spin_lock(&cli->cl_mod_rpcs_lock);
2181 prev = cli->cl_max_mod_rpcs_in_flight;
2182 cli->cl_max_mod_rpcs_in_flight = max;
2184 /* wakeup waiters if limit has been increased */
2185 if (cli->cl_max_mod_rpcs_in_flight > prev)
2186 wake_up(&cli->cl_mod_rpcs_waitq);
2188 spin_unlock(&cli->cl_mod_rpcs_lock);
2192 EXPORT_SYMBOL(obd_set_max_mod_rpcs_in_flight);
2194 int obd_mod_rpc_stats_seq_show(struct client_obd *cli,
2195 struct seq_file *seq)
2197 unsigned long mod_tot = 0, mod_cum;
2200 spin_lock(&cli->cl_mod_rpcs_lock);
2201 lprocfs_stats_header(seq, ktime_get_real(), cli->cl_mod_rpcs_init, 25,
2203 seq_printf(seq, "modify_RPCs_in_flight: %hu\n",
2204 cli->cl_mod_rpcs_in_flight);
2206 seq_printf(seq, "\n\t\t\tmodify\n");
2207 seq_printf(seq, "rpcs in flight rpcs %% cum %%\n");
2209 mod_tot = lprocfs_oh_sum(&cli->cl_mod_rpcs_hist);
2212 for (i = 0; i < OBD_HIST_MAX; i++) {
2213 unsigned long mod = cli->cl_mod_rpcs_hist.oh_buckets[i];
2216 seq_printf(seq, "%d:\t\t%10lu %3u %3u\n",
2217 i, mod, pct(mod, mod_tot),
2218 pct(mod_cum, mod_tot));
2219 if (mod_cum == mod_tot)
2223 spin_unlock(&cli->cl_mod_rpcs_lock);
2227 EXPORT_SYMBOL(obd_mod_rpc_stats_seq_show);
2229 /* The number of modify RPCs sent in parallel is limited
2230 * because the server has a finite number of slots per client to
2231 * store request result and ensure reply reconstruction when needed.
2232 * On the client, this limit is stored in cl_max_mod_rpcs_in_flight
2233 * that takes into account server limit and cl_max_rpcs_in_flight
2235 * On the MDC client, to avoid a potential deadlock (see Bugzilla 3462),
2236 * one close request is allowed above the maximum.
2238 static inline bool obd_mod_rpc_slot_avail_locked(struct client_obd *cli,
2243 /* A slot is available if
2244 * - number of modify RPCs in flight is less than the max
2245 * - it's a close RPC and no other close request is in flight
2247 avail = cli->cl_mod_rpcs_in_flight < cli->cl_max_mod_rpcs_in_flight ||
2248 (close_req && cli->cl_close_rpcs_in_flight == 0);
2253 static inline bool obd_mod_rpc_slot_avail(struct client_obd *cli,
2258 spin_lock(&cli->cl_mod_rpcs_lock);
2259 avail = obd_mod_rpc_slot_avail_locked(cli, close_req);
2260 spin_unlock(&cli->cl_mod_rpcs_lock);
2265 /* Get a modify RPC slot from the obd client @cli according
2266 * to the kind of operation @opc that is going to be sent
2267 * and the intent @it of the operation if it applies.
2268 * If the maximum number of modify RPCs in flight is reached
2269 * the thread is put to sleep.
2270 * Returns the tag to be set in the request message. Tag 0
2271 * is reserved for non-modifying requests.
2273 __u16 obd_get_mod_rpc_slot(struct client_obd *cli, __u32 opc)
2275 bool close_req = false;
2278 if (opc == MDS_CLOSE)
2282 spin_lock(&cli->cl_mod_rpcs_lock);
2283 max = cli->cl_max_mod_rpcs_in_flight;
2284 if (obd_mod_rpc_slot_avail_locked(cli, close_req)) {
2285 /* there is a slot available */
2286 cli->cl_mod_rpcs_in_flight++;
2288 cli->cl_close_rpcs_in_flight++;
2289 lprocfs_oh_tally(&cli->cl_mod_rpcs_hist,
2290 cli->cl_mod_rpcs_in_flight);
2291 /* find a free tag */
2292 i = find_first_zero_bit(cli->cl_mod_tag_bitmap,
2294 LASSERT(i < OBD_MAX_RIF_MAX);
2295 LASSERT(!test_and_set_bit(i, cli->cl_mod_tag_bitmap));
2296 spin_unlock(&cli->cl_mod_rpcs_lock);
2297 /* tag 0 is reserved for non-modify RPCs */
2300 "%s: modify RPC slot %u is allocated opc %u, max %hu\n",
2301 cli->cl_import->imp_obd->obd_name,
2306 spin_unlock(&cli->cl_mod_rpcs_lock);
2308 CDEBUG(D_RPCTRACE, "%s: sleeping for a modify RPC slot "
2309 "opc %u, max %hu\n",
2310 cli->cl_import->imp_obd->obd_name, opc, max);
2312 wait_event_idle_exclusive(cli->cl_mod_rpcs_waitq,
2313 obd_mod_rpc_slot_avail(cli,
2317 EXPORT_SYMBOL(obd_get_mod_rpc_slot);
2319 /* Put a modify RPC slot from the obd client @cli according
2320 * to the kind of operation @opc that has been sent.
2322 void obd_put_mod_rpc_slot(struct client_obd *cli, __u32 opc, __u16 tag)
2324 bool close_req = false;
2329 if (opc == MDS_CLOSE)
2332 spin_lock(&cli->cl_mod_rpcs_lock);
2333 cli->cl_mod_rpcs_in_flight--;
2335 cli->cl_close_rpcs_in_flight--;
2336 /* release the tag in the bitmap */
2337 LASSERT(tag - 1 < OBD_MAX_RIF_MAX);
2338 LASSERT(test_and_clear_bit(tag - 1, cli->cl_mod_tag_bitmap) != 0);
2339 spin_unlock(&cli->cl_mod_rpcs_lock);
2340 /* LU-14741 - to prevent close RPCs stuck behind normal ones */
2342 wake_up_all(&cli->cl_mod_rpcs_waitq);
2344 wake_up(&cli->cl_mod_rpcs_waitq);
2346 EXPORT_SYMBOL(obd_put_mod_rpc_slot);