X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fobdclass%2Fgenops.c;h=a74ee1844b542b608693d5e956f12b60d9dad1f2;hp=a630808642c9149fcd4e8449dcc1dfdb5d379314;hb=41ab260642f4923106da3567d7c75e1ca3c61ea4;hpb=de3c3fdf2597ba110a5247e271470b5dd626df75 diff --git a/lustre/obdclass/genops.c b/lustre/obdclass/genops.c index a630808..a74ee18 100644 --- a/lustre/obdclass/genops.c +++ b/lustre/obdclass/genops.c @@ -1,25 +1,39 @@ /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- * vim:expandtab:shiftwidth=8:tabstop=8: * - * Copyright (c) 2001-2003 Cluster File Systems, Inc. + * GPL HEADER START * - * This file is part of the Lustre file system, http://www.lustre.org - * Lustre is a trademark of Cluster File Systems, Inc. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * You may have signed or agreed to another license before downloading - * this software. If so, you are bound by the terms and conditions - * of that agreement, and the following does not apply to you. See the - * LICENSE file included with this distribution for more information. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 only, + * as published by the Free Software Foundation. * - * If you did not agree to a different license, then this copy of Lustre - * is open source software; you can redistribute it and/or modify it - * under the terms of version 2 of the GNU General Public License as - * published by the Free Software Foundation. + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License version 2 for more details (a copy is included + * in the LICENSE file that accompanied this code). * - * In either case, Lustre is distributed in the hope that it will be - * useful, but WITHOUT ANY WARRANTY; without even the implied warranty - * of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * license text for more details. + * You should have received a copy of the GNU General Public License + * version 2 along with this program; If not, see + * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf + * + * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, + * CA 95054 USA or visit www.sun.com if you need additional information or + * have any questions. + * + * GPL HEADER END + */ +/* + * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Use is subject to license terms. + */ +/* + * This file is part of Lustre, http://www.lustre.org/ + * Lustre is a trademark of Sun Microsystems, Inc. + * + * lustre/obdclass/genops.c * * These are the only exported functions, they provide some generic * infrastructure for managing object devices @@ -46,6 +60,8 @@ struct list_head obd_zombie_imports; struct list_head obd_zombie_exports; spinlock_t obd_zombie_impexp_lock; static void obd_zombie_impexp_notify(void); +static void obd_zombie_export_add(struct obd_export *exp); +static void obd_zombie_import_add(struct obd_import *imp); int (*ptlrpc_put_connection_superhack)(struct ptlrpc_connection *c); @@ -57,7 +73,7 @@ static struct obd_device *obd_device_alloc(void) { struct obd_device *obd; - OBD_SLAB_ALLOC_PTR(obd, obd_device_cachep); + OBD_SLAB_ALLOC_PTR_GFP(obd, obd_device_cachep, CFS_ALLOC_IO); if (obd != NULL) { obd->obd_magic = OBD_DEVICE_MAGIC; } @@ -68,16 +84,16 @@ EXPORT_SYMBOL(obd_device_alloc); static void obd_device_free(struct obd_device *obd) { LASSERT(obd != NULL); - LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "obd %p obd_magic %08x != %08x\n", + LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "obd %p obd_magic %08x != %08x\n", obd, obd->obd_magic, OBD_DEVICE_MAGIC); if (obd->obd_namespace != NULL) { - CERROR("obd %p: namespace %p was not properly cleaned up (obd_force=%d)!\n", + CERROR("obd %p: namespace %p was not properly cleaned up (obd_force=%d)!\n", obd, obd->obd_namespace, obd->obd_force); LBUG(); } + lu_ref_fini(&obd->obd_reference); OBD_SLAB_FREE_PTR(obd, obd_device_cachep); } -EXPORT_SYMBOL(obd_device_free); struct obd_type *class_search_type(const char *name) { @@ -132,8 +148,8 @@ void class_put_type(struct obd_type *type) #define CLASS_MAX_NAME 1024 -int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops, - struct lprocfs_vars *vars, const char *name, +int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops, + struct lprocfs_vars *vars, const char *name, struct lu_device_type *ldt) { struct obd_type *type; @@ -156,9 +172,9 @@ int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops, OBD_ALLOC_PTR(type->typ_dt_ops); OBD_ALLOC_PTR(type->typ_md_ops); OBD_ALLOC(type->typ_name, strlen(name) + 1); - - if (type->typ_dt_ops == NULL || - type->typ_md_ops == NULL || + + if (type->typ_dt_ops == NULL || + type->typ_md_ops == NULL || type->typ_name == NULL) GOTO (failed, rc); @@ -180,7 +196,7 @@ int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops, #endif if (ldt != NULL) { type->typ_lu = ldt; - rc = ldt->ldt_ops->ldto_init(ldt); + rc = lu_device_type_init(ldt); if (rc != 0) GOTO (failed, rc); } @@ -226,7 +242,7 @@ int class_unregister_type(const char *name) } if (type->typ_lu) - type->typ_lu->ldt_ops->ldto_fini(type->typ_lu); + lu_device_type_fini(type->typ_lu); spin_lock(&obd_types_lock); list_del(&type->typ_chain); @@ -240,6 +256,17 @@ int class_unregister_type(const char *name) RETURN(0); } /* class_unregister_type */ +/** + * Create a new obd device. + * + * Find an empty slot in ::obd_devs[], create a new obd device in it. + * + * \param typename [in] obd device type string. + * \param name [in] obd device name. + * + * \retval NULL if create fails, otherwise return the obd device + * pointer created. + */ struct obd_device *class_newdev(const char *type_name, const char *name) { struct obd_device *result = NULL; @@ -248,19 +275,19 @@ struct obd_device *class_newdev(const char *type_name, const char *name) int i; int new_obd_minor = 0; - if (strlen(name) > MAX_OBD_NAME) { + if (strlen(name) >= MAX_OBD_NAME) { CERROR("name/uuid must be < %u bytes long\n", MAX_OBD_NAME); RETURN(ERR_PTR(-EINVAL)); } - type = class_get_type(type_name); + type = class_get_type(type_name); if (type == NULL){ CERROR("OBD: unknown type: %s\n", type_name); RETURN(ERR_PTR(-ENODEV)); } newdev = obd_device_alloc(); - if (newdev == NULL) { + if (newdev == NULL) { class_put_type(type); RETURN(ERR_PTR(-ENOMEM)); } @@ -291,18 +318,19 @@ struct obd_device *class_newdev(const char *type_name, const char *name) result->obd_minor = i; new_obd_minor = i; result->obd_type = type; - memcpy(result->obd_name, name, strlen(name)); + strncpy(result->obd_name, name, + sizeof(result->obd_name) - 1); obd_devs[i] = result; } } spin_unlock(&obd_dev_lock); - + if (result == NULL && i >= class_devno_max()) { CERROR("all %u OBD devices used, increase MAX_OBD_DEVICES\n", class_devno_max()); result = ERR_PTR(-EOVERFLOW); } - + if (IS_ERR(result)) { obd_device_free(newdev); class_put_type(type); @@ -395,15 +423,22 @@ struct obd_device *class_uuid2obd(struct obd_uuid *uuid) return class_num2obd(dev); } +/** + * Get obd device from ::obd_devs[] + * + * \param num [in] array index + * + * \retval NULL if ::obd_devs[\a num] does not contains an obd device + * otherwise return the obd device there. + */ struct obd_device *class_num2obd(int num) { struct obd_device *obd = NULL; if (num < class_devno_max()) { obd = obd_devs[num]; - if (obd == NULL) { + if (obd == NULL) return NULL; - } LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "%p obd_magic %08x != %08x\n", @@ -473,18 +508,6 @@ struct obd_device * class_find_client_obd(struct obd_uuid *tgt_uuid, return NULL; } -struct obd_device *class_find_client_notype(struct obd_uuid *tgt_uuid, - struct obd_uuid *grp_uuid) -{ - struct obd_device *obd; - - obd = class_find_client_obd(tgt_uuid, LUSTRE_MDC_NAME, NULL); - if (!obd) - obd = class_find_client_obd(tgt_uuid, LUSTRE_OSC_NAME, - grp_uuid); - return obd; -} - /* Iterate the obd_device list looking devices have grp_uuid. Start searching at *next, and if a device is found, the next index to look at is saved in *next. If next is NULL, then the first matching device @@ -517,6 +540,49 @@ struct obd_device * class_devices_in_group(struct obd_uuid *grp_uuid, int *next) return NULL; } +/** + * to notify sptlrpc log for @fsname has changed, let every relevant OBD + * adjust sptlrpc settings accordingly. + */ +int class_notify_sptlrpc_conf(const char *fsname, int namelen) +{ + struct obd_device *obd; + const char *type; + int i, rc = 0, rc2; + + LASSERT(namelen > 0); + + spin_lock(&obd_dev_lock); + for (i = 0; i < class_devno_max(); i++) { + obd = class_num2obd(i); + + if (obd == NULL || obd->obd_set_up == 0 || obd->obd_stopping) + continue; + + /* only notify mdc, osc, mdt, ost */ + type = obd->obd_type->typ_name; + if (strcmp(type, LUSTRE_MDC_NAME) != 0 && + strcmp(type, LUSTRE_OSC_NAME) != 0 && + strcmp(type, LUSTRE_MDT_NAME) != 0 && + strcmp(type, LUSTRE_OST_NAME) != 0) + continue; + + if (strncmp(obd->obd_name, fsname, namelen)) + continue; + + class_incref(obd, __FUNCTION__, obd); + spin_unlock(&obd_dev_lock); + rc2 = obd_set_info_async(obd->obd_self_export, + sizeof(KEY_SPTLRPC_CONF), + KEY_SPTLRPC_CONF, 0, NULL, NULL); + rc = rc ? rc : rc2; + class_decref(obd, __FUNCTION__, obd); + spin_lock(&obd_dev_lock); + } + spin_unlock(&obd_dev_lock); + return rc; +} +EXPORT_SYMBOL(class_notify_sptlrpc_conf); void obd_cleanup_caches(void) { @@ -552,7 +618,7 @@ int obd_init_caches(void) LASSERT(obd_device_cachep == NULL); obd_device_cachep = cfs_mem_cache_create("ll_obd_dev_cache", - sizeof(struct obd_device), + sizeof(struct obd_device), 0, 0); if (!obd_device_cachep) GOTO(out, -ENOMEM); @@ -640,30 +706,7 @@ struct obd_import *class_conn2cliimp(struct lustre_handle *conn) } /* Export management functions */ -static void export_handle_addref(void *export) -{ - class_export_get(export); -} - -void __class_export_put(struct obd_export *exp) -{ - if (atomic_dec_and_test(&exp->exp_refcount)) { - LASSERT (list_empty(&exp->exp_obd_chain)); - - CDEBUG(D_IOCTL, "final put %p/%s\n", - exp, exp->exp_client_uuid.uuid); - - spin_lock(&obd_zombie_impexp_lock); - list_add(&exp->exp_obd_chain, &obd_zombie_exports); - spin_unlock(&obd_zombie_impexp_lock); - - if (obd_zombie_impexp_notify != NULL) - obd_zombie_impexp_notify(); - } -} -EXPORT_SYMBOL(__class_export_put); - -void class_export_destroy(struct obd_export *exp) +static void class_export_destroy(struct obd_export *exp) { struct obd_device *obd = exp->exp_obd; ENTRY; @@ -680,14 +723,46 @@ void class_export_destroy(struct obd_export *exp) ptlrpc_put_connection_superhack(exp->exp_connection); LASSERT(list_empty(&exp->exp_outstanding_replies)); + LASSERT(list_empty(&exp->exp_uncommitted_replies)); LASSERT(list_empty(&exp->exp_req_replay_queue)); + LASSERT(list_empty(&exp->exp_queued_rpc)); obd_destroy_export(exp); - + class_decref(obd, "export", exp); + OBD_FREE_RCU(exp, sizeof(*exp), &exp->exp_handle); - class_decref(obd); EXIT; } +static void export_handle_addref(void *export) +{ + class_export_get(export); +} + +struct obd_export *class_export_get(struct obd_export *exp) +{ + atomic_inc(&exp->exp_refcount); + CDEBUG(D_INFO, "GETting export %p : new refcount %d\n", exp, + atomic_read(&exp->exp_refcount)); + return exp; +} +EXPORT_SYMBOL(class_export_get); + +void class_export_put(struct obd_export *exp) +{ + LASSERT(exp != NULL); + CDEBUG(D_INFO, "PUTting export %p : new refcount %d\n", exp, + atomic_read(&exp->exp_refcount) - 1); + LASSERT(atomic_read(&exp->exp_refcount) > 0); + LASSERT(atomic_read(&exp->exp_refcount) < 0x5a5a5a); + + if (atomic_dec_and_test(&exp->exp_refcount)) { + CDEBUG(D_IOCTL, "final put %p/%s\n", + exp, exp->exp_client_uuid.uuid); + obd_zombie_export_add(exp); + } +} +EXPORT_SYMBOL(class_export_put); + /* Creates a new export, adds it to the hash table, and returns a * pointer to it. The refcount is 2: one for the hash reference, and * one for the pointer returned by this function. */ @@ -702,16 +777,16 @@ struct obd_export *class_new_export(struct obd_device *obd, return ERR_PTR(-ENOMEM); export->exp_conn_cnt = 0; + export->exp_lock_hash = NULL; atomic_set(&export->exp_refcount, 2); atomic_set(&export->exp_rpc_count, 0); export->exp_obd = obd; CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies); + spin_lock_init(&export->exp_uncommitted_replies_lock); + CFS_INIT_LIST_HEAD(&export->exp_uncommitted_replies); CFS_INIT_LIST_HEAD(&export->exp_req_replay_queue); - /* XXX this should be in LDLM init */ - CFS_INIT_LIST_HEAD(&export->exp_ldlm_data.led_held_locks); - spin_lock_init(&export->exp_ldlm_data.led_lock); - CFS_INIT_LIST_HEAD(&export->exp_handle.h_link); + CFS_INIT_LIST_HEAD(&export->exp_queued_rpc); class_handle_hash(&export->exp_handle, export_handle_addref); export->exp_last_request_time = cfs_time_current_sec(); spin_lock_init(&export->exp_lock); @@ -725,20 +800,20 @@ struct obd_export *class_new_export(struct obd_device *obd, spin_lock(&obd->obd_dev_lock); if (!obd_uuid_equals(cluuid, &obd->obd_uuid)) { - rc = lustre_hash_additem_unique(obd->obd_uuid_hash_body, cluuid, - &export->exp_uuid_hash); - if (rc != 0) { - CWARN("%s: denying duplicate export for %s\n", - obd->obd_name, cluuid->uuid); - spin_unlock(&obd->obd_dev_lock); - class_handle_unhash(&export->exp_handle); - OBD_FREE_PTR(export); - return ERR_PTR(-EALREADY); - } + rc = lustre_hash_add_unique(obd->obd_uuid_hash, cluuid, + &export->exp_uuid_hash); + if (rc != 0) { + LCONSOLE_WARN("%s: denying duplicate export for %s, %d\n", + obd->obd_name, cluuid->uuid, rc); + spin_unlock(&obd->obd_dev_lock); + class_handle_unhash(&export->exp_handle); + OBD_FREE_PTR(export); + return ERR_PTR(-EALREADY); + } } LASSERT(!obd->obd_stopping); /* shouldn't happen, but might race */ - class_incref(obd); + class_incref(obd, "export", export); list_add(&export->exp_obd_chain, &export->exp_obd->obd_exports); list_add_tail(&export->exp_obd_chain_timed, &export->exp_obd->obd_exports_timed); @@ -755,20 +830,57 @@ void class_unlink_export(struct obd_export *exp) spin_lock(&exp->exp_obd->obd_dev_lock); /* delete an uuid-export hashitem from hashtables */ - if (!hlist_unhashed(&exp->exp_uuid_hash)) { - lustre_hash_delitem(exp->exp_obd->obd_uuid_hash_body, - &exp->exp_client_uuid, &exp->exp_uuid_hash); - } + if (!hlist_unhashed(&exp->exp_uuid_hash)) + lustre_hash_del(exp->exp_obd->obd_uuid_hash, + &exp->exp_client_uuid, + &exp->exp_uuid_hash); + list_del_init(&exp->exp_obd_chain); list_del_init(&exp->exp_obd_chain_timed); exp->exp_obd->obd_num_exports--; spin_unlock(&exp->exp_obd->obd_dev_lock); + /* Keep these counter valid always */ + spin_lock_bh(&exp->exp_obd->obd_processing_task_lock); + if (exp->exp_delayed) + exp->exp_obd->obd_delayed_clients--; + else if (exp->exp_in_recovery) + exp->exp_obd->obd_recoverable_clients--; + else if (exp->exp_obd->obd_recovering) + exp->exp_obd->obd_max_recoverable_clients--; + spin_unlock_bh(&exp->exp_obd->obd_processing_task_lock); class_export_put(exp); } EXPORT_SYMBOL(class_unlink_export); /* Import management functions */ +void class_import_destroy(struct obd_import *imp) +{ + ENTRY; + + CDEBUG(D_IOCTL, "destroying import %p for %s\n", imp, + imp->imp_obd->obd_name); + + LASSERT(atomic_read(&imp->imp_refcount) == 0); + + ptlrpc_put_connection_superhack(imp->imp_connection); + + while (!list_empty(&imp->imp_conn_list)) { + struct obd_import_conn *imp_conn; + + imp_conn = list_entry(imp->imp_conn_list.next, + struct obd_import_conn, oic_item); + list_del_init(&imp_conn->oic_item); + ptlrpc_put_connection_superhack(imp_conn->oic_conn); + OBD_FREE(imp_conn, sizeof(*imp_conn)); + } + + LASSERT(imp->imp_sec == NULL); + class_decref(imp->imp_obd, "import", imp); + OBD_FREE_RCU(imp, sizeof(*imp), &imp->imp_handle); + EXIT; +} + static void import_handle_addref(void *import) { class_import_get(import); @@ -779,66 +891,34 @@ struct obd_import *class_import_get(struct obd_import *import) LASSERT(atomic_read(&import->imp_refcount) >= 0); LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a); atomic_inc(&import->imp_refcount); - CDEBUG(D_INFO, "import %p refcount=%d\n", import, - atomic_read(&import->imp_refcount)); + CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", import, + atomic_read(&import->imp_refcount), + import->imp_obd->obd_name); return import; } EXPORT_SYMBOL(class_import_get); -void class_import_put(struct obd_import *import) +void class_import_put(struct obd_import *imp) { ENTRY; - CDEBUG(D_INFO, "import %p refcount=%d\n", import, - atomic_read(&import->imp_refcount) - 1); + LASSERT(atomic_read(&imp->imp_refcount) > 0); + LASSERT(atomic_read(&imp->imp_refcount) < 0x5a5a5a); + LASSERT(list_empty(&imp->imp_zombie_chain)); - LASSERT(atomic_read(&import->imp_refcount) > 0); - LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a); - LASSERT(list_empty(&import->imp_zombie_chain)); - - if (atomic_dec_and_test(&import->imp_refcount)) { + CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", imp, + atomic_read(&imp->imp_refcount) - 1, + imp->imp_obd->obd_name); - CDEBUG(D_INFO, "final put import %p\n", import); - - spin_lock(&obd_zombie_impexp_lock); - list_add(&import->imp_zombie_chain, &obd_zombie_imports); - spin_unlock(&obd_zombie_impexp_lock); - - if (obd_zombie_impexp_notify != NULL) - obd_zombie_impexp_notify(); + if (atomic_dec_and_test(&imp->imp_refcount)) { + CDEBUG(D_INFO, "final put import %p\n", imp); + obd_zombie_import_add(imp); } EXIT; } EXPORT_SYMBOL(class_import_put); -void class_import_destroy(struct obd_import *import) -{ - ENTRY; - - CDEBUG(D_IOCTL, "destroying import %p for %s\n", import, - import->imp_obd->obd_name); - - LASSERT(atomic_read(&import->imp_refcount) == 0); - - ptlrpc_put_connection_superhack(import->imp_connection); - - while (!list_empty(&import->imp_conn_list)) { - struct obd_import_conn *imp_conn; - - imp_conn = list_entry(import->imp_conn_list.next, - struct obd_import_conn, oic_item); - list_del(&imp_conn->oic_item); - ptlrpc_put_connection_superhack(imp_conn->oic_conn); - OBD_FREE(imp_conn, sizeof(*imp_conn)); - } - - LASSERT(import->imp_sec == NULL); - class_decref(import->imp_obd); - OBD_FREE_RCU(import, sizeof(*import), &import->imp_handle); - EXIT; -} - static void init_imp_at(struct imp_at *at) { int i; at_init(&at->iat_net_latency, 0, 0); @@ -866,11 +946,12 @@ struct obd_import *class_new_import(struct obd_device *obd) spin_lock_init(&imp->imp_lock); imp->imp_last_success_conn = 0; imp->imp_state = LUSTRE_IMP_NEW; - imp->imp_obd = class_incref(obd); + imp->imp_obd = class_incref(obd, "import", imp); sema_init(&imp->imp_sec_mutex, 1); cfs_waitq_init(&imp->imp_recovery_waitq); atomic_set(&imp->imp_refcount, 2); + atomic_set(&imp->imp_unregistering, 0); atomic_set(&imp->imp_inflight, 0); atomic_set(&imp->imp_replay_inflight, 0); atomic_set(&imp->imp_inval_count, 0); @@ -952,16 +1033,18 @@ void class_export_recovery_cleanup(struct obd_export *exp) exp->exp_lock_replay_needed = 0; spin_unlock(&exp->exp_lock); LASSERT(atomic_read(&obd->obd_lock_replay_clients)); - atomic_dec(&obd->obd_lock_replay_clients); + atomic_dec(&obd->obd_lock_replay_clients); } } spin_unlock_bh(&obd->obd_processing_task_lock); } -/* This function removes two references from the export: one for the - * hash entry and one for the export pointer passed in. The export - * pointer passed to this function is destroyed should not be used - * again. */ +/* This function removes 1-3 references from the export: + * 1 - for export pointer passed + * and if disconnect really need + * 2 - removing from hash + * 3 - in client_unlink_export + * The export pointer passed to this function can destroyed */ int class_disconnect(struct obd_export *export) { int already_disconnected; @@ -976,39 +1059,43 @@ int class_disconnect(struct obd_export *export) spin_lock(&export->exp_lock); already_disconnected = export->exp_disconnected; export->exp_disconnected = 1; - - if (!hlist_unhashed(&export->exp_nid_hash)) { - lustre_hash_delitem(export->exp_obd->obd_nid_hash_body, - &export->exp_connection->c_peer.nid, &export->exp_nid_hash); - } spin_unlock(&export->exp_lock); /* class_cleanup(), abort_recovery(), and class_fail_export() * all end up in here, and if any of them race we shouldn't * call extra class_export_puts(). */ - if (already_disconnected) - RETURN(0); + if (already_disconnected) { + LASSERT(hlist_unhashed(&export->exp_nid_hash)); + GOTO(no_disconn, already_disconnected); + } CDEBUG(D_IOCTL, "disconnect: cookie "LPX64"\n", export->exp_handle.h_cookie); + if (!hlist_unhashed(&export->exp_nid_hash)) + lustre_hash_del(export->exp_obd->obd_nid_hash, + &export->exp_connection->c_peer.nid, + &export->exp_nid_hash); + class_export_recovery_cleanup(export); class_unlink_export(export); +no_disconn: class_export_put(export); RETURN(0); } -static void class_disconnect_export_list(struct list_head *list, int flags) +static void class_disconnect_export_list(struct list_head *list, + enum obd_option flags) { int rc; - struct lustre_handle fake_conn; - struct obd_export *fake_exp, *exp; + struct obd_export *exp; ENTRY; /* It's possible that an export may disconnect itself, but * nothing else will be added to this list. */ while (!list_empty(list)) { exp = list_entry(list->next, struct obd_export, exp_obd_chain); + /* need for safe call CDEBUG after obd_disconnect */ class_export_get(exp); spin_lock(&exp->exp_lock); @@ -1027,49 +1114,38 @@ static void class_disconnect_export_list(struct list_head *list, int flags) continue; } - fake_conn.cookie = exp->exp_handle.h_cookie; - fake_exp = class_conn2export(&fake_conn); - if (!fake_exp) { - class_export_put(exp); - continue; - } - - spin_lock(&fake_exp->exp_lock); - fake_exp->exp_flags = flags; - spin_unlock(&fake_exp->exp_lock); - + class_export_get(exp); CDEBUG(D_HA, "%s: disconnecting export at %s (%p), " - "last request at %ld\n", + "last request at "CFS_TIME_T"\n", exp->exp_obd->obd_name, obd_export_nid2str(exp), exp, exp->exp_last_request_time); - rc = obd_disconnect(fake_exp); + /* release one export reference anyway */ + rc = obd_disconnect(exp); + + CDEBUG(D_HA, "disconnected export at %s (%p): rc %d\n", + obd_export_nid2str(exp), exp, rc); class_export_put(exp); } EXIT; } -static inline int get_exp_flags_from_obd(struct obd_device *obd) -{ - return ((obd->obd_fail ? OBD_OPT_FAILOVER : 0) | - (obd->obd_force ? OBD_OPT_FORCE : 0)); -} - void class_disconnect_exports(struct obd_device *obd) { struct list_head work_list; ENTRY; /* Move all of the exports from obd_exports to a work list, en masse. */ + CFS_INIT_LIST_HEAD(&work_list); spin_lock(&obd->obd_dev_lock); - list_add(&work_list, &obd->obd_exports); - list_del_init(&obd->obd_exports); + list_splice_init(&obd->obd_exports, &work_list); + list_splice_init(&obd->obd_delayed_exports, &work_list); spin_unlock(&obd->obd_dev_lock); - + if (!list_empty(&work_list)) { CDEBUG(D_HA, "OBD device %d (%p) has exports, " "disconnecting them\n", obd->obd_minor, obd); - class_disconnect_export_list(&work_list, - get_exp_flags_from_obd(obd)); + class_disconnect_export_list(&work_list, + exp_flags_from_obd(obd)); } else CDEBUG(D_HA, "OBD device %d (%p) has no exports\n", obd->obd_minor, obd); @@ -1079,30 +1155,30 @@ EXPORT_SYMBOL(class_disconnect_exports); /* Remove exports that have not completed recovery. */ -int class_disconnect_stale_exports(struct obd_device *obd, - int (*test_export)(struct obd_export *)) +void class_disconnect_stale_exports(struct obd_device *obd, + int (*test_export)(struct obd_export *), + enum obd_option flags) { struct list_head work_list; struct list_head *pos, *n; struct obd_export *exp; - int cnt = 0; ENTRY; CFS_INIT_LIST_HEAD(&work_list); spin_lock(&obd->obd_dev_lock); + obd->obd_stale_clients = 0; list_for_each_safe(pos, n, &obd->obd_exports) { exp = list_entry(pos, struct obd_export, exp_obd_chain); if (test_export(exp)) continue; - - list_del(&exp->exp_obd_chain); - list_add(&exp->exp_obd_chain, &work_list); + + list_move(&exp->exp_obd_chain, &work_list); /* don't count self-export as client */ if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid)) continue; - cnt++; + obd->obd_stale_clients++; CDEBUG(D_ERROR, "%s: disconnect stale client %s@%s\n", obd->obd_name, exp->exp_client_uuid.uuid, exp->exp_connection == NULL ? "" : @@ -1110,152 +1186,13 @@ int class_disconnect_stale_exports(struct obd_device *obd, } spin_unlock(&obd->obd_dev_lock); - CDEBUG(D_ERROR, "%s: disconnecting %d stale clients\n", - obd->obd_name, cnt); - class_disconnect_export_list(&work_list, get_exp_flags_from_obd(obd)); - RETURN(cnt); -} -EXPORT_SYMBOL(class_disconnect_stale_exports); - -int oig_init(struct obd_io_group **oig_out) -{ - struct obd_io_group *oig; - ENTRY; - - OBD_ALLOC(oig, sizeof(*oig)); - if (oig == NULL) - RETURN(-ENOMEM); - - spin_lock_init(&oig->oig_lock); - oig->oig_rc = 0; - oig->oig_pending = 0; - atomic_set(&oig->oig_refcount, 1); - cfs_waitq_init(&oig->oig_waitq); - CFS_INIT_LIST_HEAD(&oig->oig_occ_list); - - *oig_out = oig; - RETURN(0); -}; -EXPORT_SYMBOL(oig_init); - -static inline void oig_grab(struct obd_io_group *oig) -{ - atomic_inc(&oig->oig_refcount); -} - -void oig_release(struct obd_io_group *oig) -{ - if (atomic_dec_and_test(&oig->oig_refcount)) - OBD_FREE(oig, sizeof(*oig)); -} -EXPORT_SYMBOL(oig_release); - -int oig_add_one(struct obd_io_group *oig, struct oig_callback_context *occ) -{ - int rc = 0; - CDEBUG(D_CACHE, "oig %p ready to roll\n", oig); - spin_lock(&oig->oig_lock); - if (oig->oig_rc) { - rc = oig->oig_rc; - } else { - oig->oig_pending++; - if (occ != NULL) - list_add_tail(&occ->occ_oig_item, &oig->oig_occ_list); - } - spin_unlock(&oig->oig_lock); - oig_grab(oig); - - return rc; -} -EXPORT_SYMBOL(oig_add_one); - -void oig_complete_one(struct obd_io_group *oig, - struct oig_callback_context *occ, int rc) -{ - cfs_waitq_t *wake = NULL; - int old_rc; - - spin_lock(&oig->oig_lock); - - if (occ != NULL) - list_del_init(&occ->occ_oig_item); - - old_rc = oig->oig_rc; - if (oig->oig_rc == 0 && rc != 0) - oig->oig_rc = rc; - - if (--oig->oig_pending <= 0) - wake = &oig->oig_waitq; - - spin_unlock(&oig->oig_lock); + CDEBUG(D_HA, "%s: disconnecting %d stale clients\n", obd->obd_name, + obd->obd_stale_clients); - CDEBUG(D_CACHE, "oig %p completed, rc %d -> %d via %d, %d now " - "pending (racey)\n", oig, old_rc, oig->oig_rc, rc, - oig->oig_pending); - if (wake) - cfs_waitq_signal(wake); - oig_release(oig); -} -EXPORT_SYMBOL(oig_complete_one); - -static int oig_done(struct obd_io_group *oig) -{ - int rc = 0; - spin_lock(&oig->oig_lock); - if (oig->oig_pending <= 0) - rc = 1; - spin_unlock(&oig->oig_lock); - return rc; -} - -static void interrupted_oig(void *data) -{ - struct obd_io_group *oig = data; - struct oig_callback_context *occ; - - spin_lock(&oig->oig_lock); - /* We need to restart the processing each time we drop the lock, as - * it is possible other threads called oig_complete_one() to remove - * an entry elsewhere in the list while we dropped lock. We need to - * drop the lock because osc_ap_completion() calls oig_complete_one() - * which re-gets this lock ;-) as well as a lock ordering issue. */ -restart: - list_for_each_entry(occ, &oig->oig_occ_list, occ_oig_item) { - if (occ->interrupted) - continue; - occ->interrupted = 1; - spin_unlock(&oig->oig_lock); - occ->occ_interrupted(occ); - spin_lock(&oig->oig_lock); - goto restart; - } - spin_unlock(&oig->oig_lock); -} - -int oig_wait(struct obd_io_group *oig) -{ - struct l_wait_info lwi = LWI_INTR(interrupted_oig, oig); - int rc; - - CDEBUG(D_CACHE, "waiting for oig %p\n", oig); - - do { - rc = l_wait_event(oig->oig_waitq, oig_done(oig), &lwi); - LASSERTF(rc == 0 || rc == -EINTR, "rc: %d\n", rc); - /* we can't continue until the oig has emptied and stopped - * referencing state that the caller will free upon return */ - if (rc == -EINTR) - lwi = (struct l_wait_info){ 0, }; - } while (rc == -EINTR); - - LASSERTF(oig->oig_pending == 0, - "exiting oig_wait(oig = %p) with %d pending\n", oig, - oig->oig_pending); - - CDEBUG(D_CACHE, "done waiting on oig %p rc %d\n", oig, oig->oig_rc); - return oig->oig_rc; + class_disconnect_export_list(&work_list, flags); + EXIT; } -EXPORT_SYMBOL(oig_wait); +EXPORT_SYMBOL(class_disconnect_stale_exports); void class_fail_export(struct obd_export *exp) { @@ -1308,15 +1245,14 @@ int obd_export_evict_by_nid(struct obd_device *obd, const char *nid) lnet_nid_t nid_key = libcfs_str2nid((char *)nid); do { - doomed_exp = lustre_hash_get_object_by_key(obd->obd_nid_hash_body, - &nid_key); + doomed_exp = lustre_hash_lookup(obd->obd_nid_hash, &nid_key); if (doomed_exp == NULL) break; LASSERTF(doomed_exp->exp_connection->c_peer.nid == nid_key, "nid %s found, wanted nid %s, requested nid %s\n", obd_export_nid2str(doomed_exp), - libcfs_nid2str(nid_key), nid); + libcfs_nid2str(nid_key), nid); LASSERTF(doomed_exp != obd->obd_self_export, "self-export is hashed by NID?\n"); exports_evicted++; @@ -1337,17 +1273,16 @@ EXPORT_SYMBOL(obd_export_evict_by_nid); int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid) { struct obd_export *doomed_exp = NULL; - struct obd_uuid doomed; + struct obd_uuid doomed_uuid; int exports_evicted = 0; - obd_str2uuid(&doomed, uuid); - if (obd_uuid_equals(&doomed, &obd->obd_uuid)) { + obd_str2uuid(&doomed_uuid, uuid); + if (obd_uuid_equals(&doomed_uuid, &obd->obd_uuid)) { CERROR("%s: can't evict myself\n", obd->obd_name); return exports_evicted; } - doomed_exp = lustre_hash_get_object_by_key(obd->obd_uuid_hash_body, - &doomed); + doomed_exp = lustre_hash_lookup(obd->obd_uuid_hash, &doomed_uuid); if (doomed_exp == NULL) { CERROR("%s: can't disconnect %s: no exports found\n", @@ -1374,14 +1309,14 @@ void obd_zombie_impexp_cull(void) ENTRY; do { - spin_lock (&obd_zombie_impexp_lock); + spin_lock(&obd_zombie_impexp_lock); import = NULL; if (!list_empty(&obd_zombie_imports)) { import = list_entry(obd_zombie_imports.next, struct obd_import, imp_zombie_chain); - list_del(&import->imp_zombie_chain); + list_del_init(&import->imp_zombie_chain); } export = NULL; @@ -1410,13 +1345,13 @@ static unsigned long obd_zombie_flags; static cfs_waitq_t obd_zombie_waitq; enum { - OBD_ZOMBIE_STOP = 1 + OBD_ZOMBIE_STOP = 1 << 1 }; /** * check for work for kill zombie import/export thread. */ -int obd_zombie_impexp_check(void *arg) +static int obd_zombie_impexp_check(void *arg) { int rc; @@ -1431,6 +1366,33 @@ int obd_zombie_impexp_check(void *arg) } /** + * Add export to the obd_zombe thread and notify it. + */ +static void obd_zombie_export_add(struct obd_export *exp) { + spin_lock(&obd_zombie_impexp_lock); + LASSERT(list_empty(&exp->exp_obd_chain)); + list_add(&exp->exp_obd_chain, &obd_zombie_exports); + spin_unlock(&obd_zombie_impexp_lock); + + if (obd_zombie_impexp_notify != NULL) + obd_zombie_impexp_notify(); +} + +/** + * Add import to the obd_zombe thread and notify it. + */ +static void obd_zombie_import_add(struct obd_import *imp) { + LASSERT(imp->imp_sec == NULL); + spin_lock(&obd_zombie_impexp_lock); + LASSERT(list_empty(&imp->imp_zombie_chain)); + list_add(&imp->imp_zombie_chain, &obd_zombie_imports); + spin_unlock(&obd_zombie_impexp_lock); + + if (obd_zombie_impexp_notify != NULL) + obd_zombie_impexp_notify(); +} + +/** * notify import/export destroy thread about new zombie. */ static void obd_zombie_impexp_notify(void) @@ -1438,6 +1400,31 @@ static void obd_zombie_impexp_notify(void) cfs_waitq_signal(&obd_zombie_waitq); } +/** + * check whether obd_zombie is idle + */ +static int obd_zombie_is_idle(void) +{ + int rc; + + LASSERT(!test_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags)); + spin_lock(&obd_zombie_impexp_lock); + rc = list_empty(&obd_zombie_imports) && + list_empty(&obd_zombie_exports); + spin_unlock(&obd_zombie_impexp_lock); + return rc; +} + +/** + * wait when obd_zombie import/export queues become empty + */ +void obd_zombie_barrier(void) +{ + struct l_wait_info lwi = { 0 }; + l_wait_event(obd_zombie_waitq, obd_zombie_is_idle(), &lwi); +} +EXPORT_SYMBOL(obd_zombie_barrier); + #ifdef __KERNEL__ /** @@ -1457,9 +1444,15 @@ static int obd_zombie_impexp_thread(void *unused) while(!test_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags)) { struct l_wait_info lwi = { 0 }; - l_wait_event(obd_zombie_waitq, !obd_zombie_impexp_check(NULL), &lwi); - + l_wait_event(obd_zombie_waitq, + !obd_zombie_impexp_check(NULL), &lwi); obd_zombie_impexp_cull(); + + /* + * Notify obd_zombie_barrier callers that queues + * may be empty. + */ + cfs_waitq_signal(&obd_zombie_waitq); } complete(&obd_zombie_stop); @@ -1517,7 +1510,6 @@ int obd_zombie_impexp_init(void) liblustre_register_idle_callback("obd_zombi_impexp_check", &obd_zombie_impexp_check, NULL); rc = 0; - #endif RETURN(rc); } @@ -1535,4 +1527,3 @@ void obd_zombie_impexp_stop(void) liblustre_deregister_idle_callback(obd_zombie_impexp_idle_cb); #endif } -