lustre/obdclass/genops.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 1999, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/obdclass/genops.c
  37  *
  38  * These are the only exported functions, they provide some generic
  39  * infrastructure for managing object devices
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_CLASS
  43 #ifndef __KERNEL__
  44 #include <liblustre.h>
  45 #endif
  46 #include <obd_ost.h>
  47 #include <obd_class.h>
  48 #include <lprocfs_status.h>
  49
  50 extern cfs_list_t obd_types;
  51 cfs_spinlock_t obd_types_lock;
  52
  53 cfs_mem_cache_t *obd_device_cachep;
  54 cfs_mem_cache_t *obdo_cachep;
  55 EXPORT_SYMBOL(obdo_cachep);
  56 cfs_mem_cache_t *import_cachep;
  57
  58 cfs_list_t      obd_zombie_imports;
  59 cfs_list_t      obd_zombie_exports;
  60 cfs_spinlock_t  obd_zombie_impexp_lock;
  61 static void obd_zombie_impexp_notify(void);
  62 static void obd_zombie_export_add(struct obd_export *exp);
  63 static void obd_zombie_import_add(struct obd_import *imp);
  64 static void print_export_data(struct obd_export *exp,
  65                               const char *status, int locks);
  66
  67 int (*ptlrpc_put_connection_superhack)(struct ptlrpc_connection *c);
  68
  69 /*
  70  * support functions: we could use inter-module communication, but this
  71  * is more portable to other OS's
  72  */
  73 static struct obd_device *obd_device_alloc(void)
  74 {
  75         struct obd_device *obd;
  76
  77         OBD_SLAB_ALLOC_PTR_GFP(obd, obd_device_cachep, CFS_ALLOC_IO);
  78         if (obd != NULL) {
  79                 obd->obd_magic = OBD_DEVICE_MAGIC;
  80         }
  81         return obd;
  82 }
  83 EXPORT_SYMBOL(obd_device_alloc);
  84
  85 static void obd_device_free(struct obd_device *obd)
  86 {
  87         LASSERT(obd != NULL);
  88         LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "obd %p obd_magic %08x != %08x\n",
  89                  obd, obd->obd_magic, OBD_DEVICE_MAGIC);
  90         if (obd->obd_namespace != NULL) {
  91                 CERROR("obd %p: namespace %p was not properly cleaned up (obd_force=%d)!\n",
  92                        obd, obd->obd_namespace, obd->obd_force);
  93                 LBUG();
  94         }
  95         lu_ref_fini(&obd->obd_reference);
  96         OBD_SLAB_FREE_PTR(obd, obd_device_cachep);
  97 }
  98
  99 struct obd_type *class_search_type(const char *name)
 100 {
 101         cfs_list_t *tmp;
 102         struct obd_type *type;
 103
 104         cfs_spin_lock(&obd_types_lock);
 105         cfs_list_for_each(tmp, &obd_types) {
 106                 type = cfs_list_entry(tmp, struct obd_type, typ_chain);
 107                 if (strcmp(type->typ_name, name) == 0) {
 108                         cfs_spin_unlock(&obd_types_lock);
 109                         return type;
 110                 }
 111         }
 112         cfs_spin_unlock(&obd_types_lock);
 113         return NULL;
 114 }
 115
 116 struct obd_type *class_get_type(const char *name)
 117 {
 118         struct obd_type *type = class_search_type(name);
 119
 120 #ifdef HAVE_MODULE_LOADING_SUPPORT
 121         if (!type) {
 122                 const char *modname = name;
 123                 if (!cfs_request_module("%s", modname)) {
 124                         CDEBUG(D_INFO, "Loaded module '%s'\n", modname);
 125                         type = class_search_type(name);
 126                 } else {
 127                         LCONSOLE_ERROR_MSG(0x158, "Can't load module '%s'\n",
 128                                            modname);
 129                 }
 130         }
 131 #endif
 132         if (type) {
 133                 cfs_spin_lock(&type->obd_type_lock);
 134                 type->typ_refcnt++;
 135                 cfs_try_module_get(type->typ_dt_ops->o_owner);
 136                 cfs_spin_unlock(&type->obd_type_lock);
 137         }
 138         return type;
 139 }
 140
 141 void class_put_type(struct obd_type *type)
 142 {
 143         LASSERT(type);
 144         cfs_spin_lock(&type->obd_type_lock);
 145         type->typ_refcnt--;
 146         cfs_module_put(type->typ_dt_ops->o_owner);
 147         cfs_spin_unlock(&type->obd_type_lock);
 148 }
 149
 150 #define CLASS_MAX_NAME 1024
 151
 152 int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops,
 153                         struct lprocfs_vars *vars, const char *name,
 154                         struct lu_device_type *ldt)
 155 {
 156         struct obd_type *type;
 157         int rc = 0;
 158         ENTRY;
 159
 160         /* sanity check */
 161         LASSERT(strnlen(name, CLASS_MAX_NAME) < CLASS_MAX_NAME);
 162
 163         if (class_search_type(name)) {
 164                 CDEBUG(D_IOCTL, "Type %s already registered\n", name);
 165                 RETURN(-EEXIST);
 166         }
 167
 168         rc = -ENOMEM;
 169         OBD_ALLOC(type, sizeof(*type));
 170         if (type == NULL)
 171                 RETURN(rc);
 172
 173         OBD_ALLOC_PTR(type->typ_dt_ops);
 174         OBD_ALLOC_PTR(type->typ_md_ops);
 175         OBD_ALLOC(type->typ_name, strlen(name) + 1);
 176
 177         if (type->typ_dt_ops == NULL ||
 178             type->typ_md_ops == NULL ||
 179             type->typ_name == NULL)
 180                 GOTO (failed, rc);
 181
 182         *(type->typ_dt_ops) = *dt_ops;
 183         /* md_ops is optional */
 184         if (md_ops)
 185                 *(type->typ_md_ops) = *md_ops;
 186         strcpy(type->typ_name, name);
 187         cfs_spin_lock_init(&type->obd_type_lock);
 188
 189 #ifdef LPROCFS
 190         type->typ_procroot = lprocfs_register(type->typ_name, proc_lustre_root,
 191                                               vars, type);
 192         if (IS_ERR(type->typ_procroot)) {
 193                 rc = PTR_ERR(type->typ_procroot);
 194                 type->typ_procroot = NULL;
 195                 GOTO (failed, rc);
 196         }
 197 #endif
 198         if (ldt != NULL) {
 199                 type->typ_lu = ldt;
 200                 rc = lu_device_type_init(ldt);
 201                 if (rc != 0)
 202                         GOTO (failed, rc);
 203         }
 204
 205         cfs_spin_lock(&obd_types_lock);
 206         cfs_list_add(&type->typ_chain, &obd_types);
 207         cfs_spin_unlock(&obd_types_lock);
 208
 209         RETURN (0);
 210
 211  failed:
 212         if (type->typ_name != NULL)
 213                 OBD_FREE(type->typ_name, strlen(name) + 1);
 214         if (type->typ_md_ops != NULL)
 215                 OBD_FREE_PTR(type->typ_md_ops);
 216         if (type->typ_dt_ops != NULL)
 217                 OBD_FREE_PTR(type->typ_dt_ops);
 218         OBD_FREE(type, sizeof(*type));
 219         RETURN(rc);
 220 }
 221
 222 int class_unregister_type(const char *name)
 223 {
 224         struct obd_type *type = class_search_type(name);
 225         ENTRY;
 226
 227         if (!type) {
 228                 CERROR("unknown obd type\n");
 229                 RETURN(-EINVAL);
 230         }
 231
 232         if (type->typ_refcnt) {
 233                 CERROR("type %s has refcount (%d)\n", name, type->typ_refcnt);
 234                 /* This is a bad situation, let's make the best of it */
 235                 /* Remove ops, but leave the name for debugging */
 236                 OBD_FREE_PTR(type->typ_dt_ops);
 237                 OBD_FREE_PTR(type->typ_md_ops);
 238                 RETURN(-EBUSY);
 239         }
 240
 241         if (type->typ_procroot) {
 242                 lprocfs_remove(&type->typ_procroot);
 243         }
 244
 245         if (type->typ_lu)
 246                 lu_device_type_fini(type->typ_lu);
 247
 248         cfs_spin_lock(&obd_types_lock);
 249         cfs_list_del(&type->typ_chain);
 250         cfs_spin_unlock(&obd_types_lock);
 251         OBD_FREE(type->typ_name, strlen(name) + 1);
 252         if (type->typ_dt_ops != NULL)
 253                 OBD_FREE_PTR(type->typ_dt_ops);
 254         if (type->typ_md_ops != NULL)
 255                 OBD_FREE_PTR(type->typ_md_ops);
 256         OBD_FREE(type, sizeof(*type));
 257         RETURN(0);
 258 } /* class_unregister_type */
 259
 260 /**
 261  * Create a new obd device.
 262  *
 263  * Find an empty slot in ::obd_devs[], create a new obd device in it.
 264  *
 265  * \param[in] type_name obd device type string.
 266  * \param[in] name      obd device name.
 267  *
 268  * \retval NULL if create fails, otherwise return the obd device
 269  *         pointer created.
 270  */
 271 struct obd_device *class_newdev(const char *type_name, const char *name)
 272 {
 273         struct obd_device *result = NULL;
 274         struct obd_device *newdev;
 275         struct obd_type *type = NULL;
 276         int i;
 277         int new_obd_minor = 0;
 278
 279         if (strlen(name) >= MAX_OBD_NAME) {
 280                 CERROR("name/uuid must be < %u bytes long\n", MAX_OBD_NAME);
 281                 RETURN(ERR_PTR(-EINVAL));
 282         }
 283
 284         type = class_get_type(type_name);
 285         if (type == NULL){
 286                 CERROR("OBD: unknown type: %s\n", type_name);
 287                 RETURN(ERR_PTR(-ENODEV));
 288         }
 289
 290         newdev = obd_device_alloc();
 291         if (newdev == NULL) {
 292                 class_put_type(type);
 293                 RETURN(ERR_PTR(-ENOMEM));
 294         }
 295         LASSERT(newdev->obd_magic == OBD_DEVICE_MAGIC);
 296
 297         cfs_spin_lock(&obd_dev_lock);
 298         for (i = 0; i < class_devno_max(); i++) {
 299                 struct obd_device *obd = class_num2obd(i);
 300                 if (obd && obd->obd_name &&
 301                     (strcmp(name, obd->obd_name) == 0)) {
 302                         CERROR("Device %s already exists, won't add\n", name);
 303                         if (result) {
 304                                 LASSERTF(result->obd_magic == OBD_DEVICE_MAGIC,
 305                                          "%p obd_magic %08x != %08x\n", result,
 306                                          result->obd_magic, OBD_DEVICE_MAGIC);
 307                                 LASSERTF(result->obd_minor == new_obd_minor,
 308                                          "%p obd_minor %d != %d\n", result,
 309                                          result->obd_minor, new_obd_minor);
 310
 311                                 obd_devs[result->obd_minor] = NULL;
 312                                 result->obd_name[0]='\0';
 313                          }
 314                         result = ERR_PTR(-EEXIST);
 315                         break;
 316                 }
 317                 if (!result && !obd) {
 318                         result = newdev;
 319                         result->obd_minor = i;
 320                         new_obd_minor = i;
 321                         result->obd_type = type;
 322                         strncpy(result->obd_name, name,
 323                                 sizeof(result->obd_name) - 1);
 324                         obd_devs[i] = result;
 325                 }
 326         }
 327         cfs_spin_unlock(&obd_dev_lock);
 328
 329         if (result == NULL && i >= class_devno_max()) {
 330                 CERROR("all %u OBD devices used, increase MAX_OBD_DEVICES\n",
 331                        class_devno_max());
 332                 result = ERR_PTR(-EOVERFLOW);
 333         }
 334
 335         if (IS_ERR(result)) {
 336                 obd_device_free(newdev);
 337                 class_put_type(type);
 338         } else {
 339                 CDEBUG(D_IOCTL, "Adding new device %s (%p)\n",
 340                        result->obd_name, result);
 341         }
 342         return result;
 343 }
 344
 345 void class_release_dev(struct obd_device *obd)
 346 {
 347         struct obd_type *obd_type = obd->obd_type;
 348
 349         LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "%p obd_magic %08x != %08x\n",
 350                  obd, obd->obd_magic, OBD_DEVICE_MAGIC);
 351         LASSERTF(obd == obd_devs[obd->obd_minor], "obd %p != obd_devs[%d] %p\n",
 352                  obd, obd->obd_minor, obd_devs[obd->obd_minor]);
 353         LASSERT(obd_type != NULL);
 354
 355         CDEBUG(D_INFO, "Release obd device %s obd_type name =%s\n",
 356                obd->obd_name,obd->obd_type->typ_name);
 357
 358         cfs_spin_lock(&obd_dev_lock);
 359         obd_devs[obd->obd_minor] = NULL;
 360         cfs_spin_unlock(&obd_dev_lock);
 361         obd_device_free(obd);
 362
 363         class_put_type(obd_type);
 364 }
 365
 366 int class_name2dev(const char *name)
 367 {
 368         int i;
 369
 370         if (!name)
 371                 return -1;
 372
 373         cfs_spin_lock(&obd_dev_lock);
 374         for (i = 0; i < class_devno_max(); i++) {
 375                 struct obd_device *obd = class_num2obd(i);
 376                 if (obd && obd->obd_name && strcmp(name, obd->obd_name) == 0) {
 377                         /* Make sure we finished attaching before we give
 378                            out any references */
 379                         LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
 380                         if (obd->obd_attached) {
 381                                 cfs_spin_unlock(&obd_dev_lock);
 382                                 return i;
 383                         }
 384                         break;
 385                 }
 386         }
 387         cfs_spin_unlock(&obd_dev_lock);
 388
 389         return -1;
 390 }
 391
 392 struct obd_device *class_name2obd(const char *name)
 393 {
 394         int dev = class_name2dev(name);
 395
 396         if (dev < 0 || dev > class_devno_max())
 397                 return NULL;
 398         return class_num2obd(dev);
 399 }
 400
 401 int class_uuid2dev(struct obd_uuid *uuid)
 402 {
 403         int i;
 404
 405         cfs_spin_lock(&obd_dev_lock);
 406         for (i = 0; i < class_devno_max(); i++) {
 407                 struct obd_device *obd = class_num2obd(i);
 408                 if (obd && obd_uuid_equals(uuid, &obd->obd_uuid)) {
 409                         LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
 410                         cfs_spin_unlock(&obd_dev_lock);
 411                         return i;
 412                 }
 413         }
 414         cfs_spin_unlock(&obd_dev_lock);
 415
 416         return -1;
 417 }
 418
 419 struct obd_device *class_uuid2obd(struct obd_uuid *uuid)
 420 {
 421         int dev = class_uuid2dev(uuid);
 422         if (dev < 0)
 423                 return NULL;
 424         return class_num2obd(dev);
 425 }
 426
 427 /**
 428  * Get obd device from ::obd_devs[]
 429  *
 430  * \param num [in] array index
 431  *
 432  * \retval NULL if ::obd_devs[\a num] does not contains an obd device
 433  *         otherwise return the obd device there.
 434  */
 435 struct obd_device *class_num2obd(int num)
 436 {
 437         struct obd_device *obd = NULL;
 438
 439         if (num < class_devno_max()) {
 440                 obd = obd_devs[num];
 441                 if (obd == NULL)
 442                         return NULL;
 443
 444                 LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC,
 445                          "%p obd_magic %08x != %08x\n",
 446                          obd, obd->obd_magic, OBD_DEVICE_MAGIC);
 447                 LASSERTF(obd->obd_minor == num,
 448                          "%p obd_minor %0d != %0d\n",
 449                          obd, obd->obd_minor, num);
 450         }
 451
 452         return obd;
 453 }
 454
 455 void class_obd_list(void)
 456 {
 457         char *status;
 458         int i;
 459
 460         cfs_spin_lock(&obd_dev_lock);
 461         for (i = 0; i < class_devno_max(); i++) {
 462                 struct obd_device *obd = class_num2obd(i);
 463                 if (obd == NULL)
 464                         continue;
 465                 if (obd->obd_stopping)
 466                         status = "ST";
 467                 else if (obd->obd_set_up)
 468                         status = "UP";
 469                 else if (obd->obd_attached)
 470                         status = "AT";
 471                 else
 472                         status = "--";
 473                 LCONSOLE(D_CONFIG, "%3d %s %s %s %s %d\n",
 474                          i, status, obd->obd_type->typ_name,
 475                          obd->obd_name, obd->obd_uuid.uuid,
 476                          cfs_atomic_read(&obd->obd_refcount));
 477         }
 478         cfs_spin_unlock(&obd_dev_lock);
 479         return;
 480 }
 481
 482 /* Search for a client OBD connected to tgt_uuid.  If grp_uuid is
 483    specified, then only the client with that uuid is returned,
 484    otherwise any client connected to the tgt is returned. */
 485 struct obd_device * class_find_client_obd(struct obd_uuid *tgt_uuid,
 486                                           const char * typ_name,
 487                                           struct obd_uuid *grp_uuid)
 488 {
 489         int i;
 490
 491         cfs_spin_lock(&obd_dev_lock);
 492         for (i = 0; i < class_devno_max(); i++) {
 493                 struct obd_device *obd = class_num2obd(i);
 494                 if (obd == NULL)
 495                         continue;
 496                 if ((strncmp(obd->obd_type->typ_name, typ_name,
 497                              strlen(typ_name)) == 0)) {
 498                         if (obd_uuid_equals(tgt_uuid,
 499                                             &obd->u.cli.cl_target_uuid) &&
 500                             ((grp_uuid)? obd_uuid_equals(grp_uuid,
 501                                                          &obd->obd_uuid) : 1)) {
 502                                 cfs_spin_unlock(&obd_dev_lock);
 503                                 return obd;
 504                         }
 505                 }
 506         }
 507         cfs_spin_unlock(&obd_dev_lock);
 508
 509         return NULL;
 510 }
 511
 512 /* Iterate the obd_device list looking devices have grp_uuid. Start
 513    searching at *next, and if a device is found, the next index to look
 514    at is saved in *next. If next is NULL, then the first matching device
 515    will always be returned. */
 516 struct obd_device * class_devices_in_group(struct obd_uuid *grp_uuid, int *next)
 517 {
 518         int i;
 519
 520         if (next == NULL)
 521                 i = 0;
 522         else if (*next >= 0 && *next < class_devno_max())
 523                 i = *next;
 524         else
 525                 return NULL;
 526
 527         cfs_spin_lock(&obd_dev_lock);
 528         for (; i < class_devno_max(); i++) {
 529                 struct obd_device *obd = class_num2obd(i);
 530                 if (obd == NULL)
 531                         continue;
 532                 if (obd_uuid_equals(grp_uuid, &obd->obd_uuid)) {
 533                         if (next != NULL)
 534                                 *next = i+1;
 535                         cfs_spin_unlock(&obd_dev_lock);
 536                         return obd;
 537                 }
 538         }
 539         cfs_spin_unlock(&obd_dev_lock);
 540
 541         return NULL;
 542 }
 543
 544 /**
 545  * to notify sptlrpc log for \a fsname has changed, let every relevant OBD
 546  * adjust sptlrpc settings accordingly.
 547  */
 548 int class_notify_sptlrpc_conf(const char *fsname, int namelen)
 549 {
 550         struct obd_device  *obd;
 551         const char         *type;
 552         int                 i, rc = 0, rc2;
 553
 554         LASSERT(namelen > 0);
 555
 556         cfs_spin_lock(&obd_dev_lock);
 557         for (i = 0; i < class_devno_max(); i++) {
 558                 obd = class_num2obd(i);
 559
 560                 if (obd == NULL || obd->obd_set_up == 0 || obd->obd_stopping)
 561                         continue;
 562
 563                 /* only notify mdc, osc, mdt, ost */
 564                 type = obd->obd_type->typ_name;
 565                 if (strcmp(type, LUSTRE_MDC_NAME) != 0 &&
 566                     strcmp(type, LUSTRE_OSC_NAME) != 0 &&
 567                     strcmp(type, LUSTRE_MDT_NAME) != 0 &&
 568                     strcmp(type, LUSTRE_OST_NAME) != 0)
 569                         continue;
 570
 571                 if (strncmp(obd->obd_name, fsname, namelen))
 572                         continue;
 573
 574                 class_incref(obd, __FUNCTION__, obd);
 575                 cfs_spin_unlock(&obd_dev_lock);
 576                 rc2 = obd_set_info_async(obd->obd_self_export,
 577                                          sizeof(KEY_SPTLRPC_CONF),
 578                                          KEY_SPTLRPC_CONF, 0, NULL, NULL);
 579                 rc = rc ? rc : rc2;
 580                 class_decref(obd, __FUNCTION__, obd);
 581                 cfs_spin_lock(&obd_dev_lock);
 582         }
 583         cfs_spin_unlock(&obd_dev_lock);
 584         return rc;
 585 }
 586 EXPORT_SYMBOL(class_notify_sptlrpc_conf);
 587
 588 void obd_cleanup_caches(void)
 589 {
 590         int rc;
 591
 592         ENTRY;
 593         if (obd_device_cachep) {
 594                 rc = cfs_mem_cache_destroy(obd_device_cachep);
 595                 LASSERTF(rc == 0, "Cannot destropy ll_obd_device_cache: rc %d\n", rc);
 596                 obd_device_cachep = NULL;
 597         }
 598         if (obdo_cachep) {
 599                 rc = cfs_mem_cache_destroy(obdo_cachep);
 600                 LASSERTF(rc == 0, "Cannot destory ll_obdo_cache\n");
 601                 obdo_cachep = NULL;
 602         }
 603         if (import_cachep) {
 604                 rc = cfs_mem_cache_destroy(import_cachep);
 605                 LASSERTF(rc == 0, "Cannot destory ll_import_cache\n");
 606                 import_cachep = NULL;
 607         }
 608         if (capa_cachep) {
 609                 rc = cfs_mem_cache_destroy(capa_cachep);
 610                 LASSERTF(rc == 0, "Cannot destory capa_cache\n");
 611                 capa_cachep = NULL;
 612         }
 613         EXIT;
 614 }
 615
 616 int obd_init_caches(void)
 617 {
 618         ENTRY;
 619
 620         LASSERT(obd_device_cachep == NULL);
 621         obd_device_cachep = cfs_mem_cache_create("ll_obd_dev_cache",
 622                                                  sizeof(struct obd_device),
 623                                                  0, 0);
 624         if (!obd_device_cachep)
 625                 GOTO(out, -ENOMEM);
 626
 627         LASSERT(obdo_cachep == NULL);
 628         obdo_cachep = cfs_mem_cache_create("ll_obdo_cache", sizeof(struct obdo),
 629                                            0, 0);
 630         if (!obdo_cachep)
 631                 GOTO(out, -ENOMEM);
 632
 633         LASSERT(import_cachep == NULL);
 634         import_cachep = cfs_mem_cache_create("ll_import_cache",
 635                                              sizeof(struct obd_import),
 636                                              0, 0);
 637         if (!import_cachep)
 638                 GOTO(out, -ENOMEM);
 639
 640         LASSERT(capa_cachep == NULL);
 641         capa_cachep = cfs_mem_cache_create("capa_cache",
 642                                            sizeof(struct obd_capa), 0, 0);
 643         if (!capa_cachep)
 644                 GOTO(out, -ENOMEM);
 645
 646         RETURN(0);
 647  out:
 648         obd_cleanup_caches();
 649         RETURN(-ENOMEM);
 650
 651 }
 652
 653 /* map connection to client */
 654 struct obd_export *class_conn2export(struct lustre_handle *conn)
 655 {
 656         struct obd_export *export;
 657         ENTRY;
 658
 659         if (!conn) {
 660                 CDEBUG(D_CACHE, "looking for null handle\n");
 661                 RETURN(NULL);
 662         }
 663
 664         if (conn->cookie == -1) {  /* this means assign a new connection */
 665                 CDEBUG(D_CACHE, "want a new connection\n");
 666                 RETURN(NULL);
 667         }
 668
 669         CDEBUG(D_INFO, "looking for export cookie "LPX64"\n", conn->cookie);
 670         export = class_handle2object(conn->cookie);
 671         RETURN(export);
 672 }
 673
 674 struct obd_device *class_exp2obd(struct obd_export *exp)
 675 {
 676         if (exp)
 677                 return exp->exp_obd;
 678         return NULL;
 679 }
 680
 681 struct obd_device *class_conn2obd(struct lustre_handle *conn)
 682 {
 683         struct obd_export *export;
 684         export = class_conn2export(conn);
 685         if (export) {
 686                 struct obd_device *obd = export->exp_obd;
 687                 class_export_put(export);
 688                 return obd;
 689         }
 690         return NULL;
 691 }
 692
 693 struct obd_import *class_exp2cliimp(struct obd_export *exp)
 694 {
 695         struct obd_device *obd = exp->exp_obd;
 696         if (obd == NULL)
 697                 return NULL;
 698         return obd->u.cli.cl_import;
 699 }
 700
 701 struct obd_import *class_conn2cliimp(struct lustre_handle *conn)
 702 {
 703         struct obd_device *obd = class_conn2obd(conn);
 704         if (obd == NULL)
 705                 return NULL;
 706         return obd->u.cli.cl_import;
 707 }
 708
 709 /* Export management functions */
 710 static void class_export_destroy(struct obd_export *exp)
 711 {
 712         struct obd_device *obd = exp->exp_obd;
 713         ENTRY;
 714
 715         LASSERT (cfs_atomic_read(&exp->exp_refcount) == 0);
 716
 717         CDEBUG(D_IOCTL, "destroying export %p/%s for %s\n", exp,
 718                exp->exp_client_uuid.uuid, obd->obd_name);
 719
 720         LASSERT(obd != NULL);
 721
 722         /* "Local" exports (lctl, LOV->{mdc,osc}) have no connection. */
 723         if (exp->exp_connection)
 724                 ptlrpc_put_connection_superhack(exp->exp_connection);
 725
 726         LASSERT(cfs_list_empty(&exp->exp_outstanding_replies));
 727         LASSERT(cfs_list_empty(&exp->exp_uncommitted_replies));
 728         LASSERT(cfs_list_empty(&exp->exp_req_replay_queue));
 729         LASSERT(cfs_list_empty(&exp->exp_queued_rpc));
 730         obd_destroy_export(exp);
 731         class_decref(obd, "export", exp);
 732
 733         OBD_FREE_RCU(exp, sizeof(*exp), &exp->exp_handle);
 734         EXIT;
 735 }
 736
 737 static void export_handle_addref(void *export)
 738 {
 739         class_export_get(export);
 740 }
 741
 742 struct obd_export *class_export_get(struct obd_export *exp)
 743 {
 744         cfs_atomic_inc(&exp->exp_refcount);
 745         CDEBUG(D_INFO, "GETting export %p : new refcount %d\n", exp,
 746                cfs_atomic_read(&exp->exp_refcount));
 747         return exp;
 748 }
 749 EXPORT_SYMBOL(class_export_get);
 750
 751 void class_export_put(struct obd_export *exp)
 752 {
 753         LASSERT(exp != NULL);
 754         CDEBUG(D_INFO, "PUTting export %p : new refcount %d\n", exp,
 755                cfs_atomic_read(&exp->exp_refcount) - 1);
 756         LASSERT(cfs_atomic_read(&exp->exp_refcount) > 0);
 757         LASSERT(cfs_atomic_read(&exp->exp_refcount) < 0x5a5a5a);
 758
 759         if (cfs_atomic_dec_and_test(&exp->exp_refcount)) {
 760                 LASSERT(!cfs_list_empty(&exp->exp_obd_chain));
 761                 CDEBUG(D_IOCTL, "final put %p/%s\n",
 762                        exp, exp->exp_client_uuid.uuid);
 763                 obd_zombie_export_add(exp);
 764         }
 765 }
 766 EXPORT_SYMBOL(class_export_put);
 767
 768 /* Creates a new export, adds it to the hash table, and returns a
 769  * pointer to it. The refcount is 2: one for the hash reference, and
 770  * one for the pointer returned by this function. */
 771 struct obd_export *class_new_export(struct obd_device *obd,
 772                                     struct obd_uuid *cluuid)
 773 {
 774         struct obd_export *export;
 775         cfs_hash_t *hash;
 776         int rc = 0;
 777         ENTRY;
 778
 779         OBD_ALLOC_PTR(export);
 780         if (!export)
 781                 return ERR_PTR(-ENOMEM);
 782
 783         export->exp_conn_cnt = 0;
 784         export->exp_lock_hash = NULL;
 785         cfs_atomic_set(&export->exp_refcount, 2);
 786         cfs_atomic_set(&export->exp_rpc_count, 0);
 787         cfs_atomic_set(&export->exp_cb_count, 0);
 788         cfs_atomic_set(&export->exp_locks_count, 0);
 789 #if LUSTRE_TRACKS_LOCK_EXP_REFS
 790         CFS_INIT_LIST_HEAD(&export->exp_locks_list);
 791         cfs_spin_lock_init(&export->exp_locks_list_guard);
 792 #endif
 793         cfs_atomic_set(&export->exp_replay_count, 0);
 794         export->exp_obd = obd;
 795         CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies);
 796         cfs_spin_lock_init(&export->exp_uncommitted_replies_lock);
 797         CFS_INIT_LIST_HEAD(&export->exp_uncommitted_replies);
 798         CFS_INIT_LIST_HEAD(&export->exp_req_replay_queue);
 799         CFS_INIT_LIST_HEAD(&export->exp_handle.h_link);
 800         CFS_INIT_LIST_HEAD(&export->exp_queued_rpc);
 801         class_handle_hash(&export->exp_handle, export_handle_addref);
 802         export->exp_last_request_time = cfs_time_current_sec();
 803         cfs_spin_lock_init(&export->exp_lock);
 804         CFS_INIT_HLIST_NODE(&export->exp_uuid_hash);
 805         CFS_INIT_HLIST_NODE(&export->exp_nid_hash);
 806
 807         export->exp_sp_peer = LUSTRE_SP_ANY;
 808         export->exp_flvr.sf_rpc = SPTLRPC_FLVR_INVALID;
 809         export->exp_client_uuid = *cluuid;
 810         obd_init_export(export);
 811
 812         cfs_spin_lock(&obd->obd_dev_lock);
 813          /* shouldn't happen, but might race */
 814         if (obd->obd_stopping)
 815                 GOTO(exit_unlock, rc = -ENODEV);
 816
 817         hash = cfs_hash_getref(obd->obd_uuid_hash);
 818         if (hash == NULL)
 819                 GOTO(exit_unlock, rc = -ENODEV);
 820         cfs_spin_unlock(&obd->obd_dev_lock);
 821
 822         if (!obd_uuid_equals(cluuid, &obd->obd_uuid)) {
 823                 rc = cfs_hash_add_unique(hash, cluuid, &export->exp_uuid_hash);
 824                 if (rc != 0) {
 825                         LCONSOLE_WARN("%s: denying duplicate export for %s, %d\n",
 826                                       obd->obd_name, cluuid->uuid, rc);
 827                         GOTO(exit_err, rc = -EALREADY);
 828                 }
 829         }
 830
 831         cfs_hash_putref(hash);
 832
 833         cfs_spin_lock(&obd->obd_dev_lock);
 834         if (obd->obd_stopping)
 835                 GOTO(exit_unlock, rc = -ENODEV);
 836
 837         class_incref(obd, "export", export);
 838         cfs_list_add(&export->exp_obd_chain, &export->exp_obd->obd_exports);
 839         cfs_list_add_tail(&export->exp_obd_chain_timed,
 840                           &export->exp_obd->obd_exports_timed);
 841         export->exp_obd->obd_num_exports++;
 842         cfs_spin_unlock(&obd->obd_dev_lock);
 843         RETURN(export);
 844
 845 exit_unlock:
 846         cfs_spin_unlock(&obd->obd_dev_lock);
 847 exit_err:
 848         class_handle_unhash(&export->exp_handle);
 849         LASSERT(cfs_hlist_unhashed(&export->exp_uuid_hash));
 850         obd_destroy_export(export);
 851         OBD_FREE_PTR(export);
 852         return ERR_PTR(rc);
 853 }
 854 EXPORT_SYMBOL(class_new_export);
 855
 856 void class_unlink_export(struct obd_export *exp)
 857 {
 858         class_handle_unhash(&exp->exp_handle);
 859
 860         cfs_spin_lock(&exp->exp_obd->obd_dev_lock);
 861         /* delete an uuid-export hashitem from hashtables */
 862         if (!cfs_hlist_unhashed(&exp->exp_uuid_hash))
 863                 cfs_hash_del(exp->exp_obd->obd_uuid_hash,
 864                              &exp->exp_client_uuid,
 865                              &exp->exp_uuid_hash);
 866
 867         cfs_list_move(&exp->exp_obd_chain, &exp->exp_obd->obd_unlinked_exports);
 868         cfs_list_del_init(&exp->exp_obd_chain_timed);
 869         exp->exp_obd->obd_num_exports--;
 870         cfs_spin_unlock(&exp->exp_obd->obd_dev_lock);
 871         class_export_put(exp);
 872 }
 873 EXPORT_SYMBOL(class_unlink_export);
 874
 875 /* Import management functions */
 876 void class_import_destroy(struct obd_import *imp)
 877 {
 878         ENTRY;
 879
 880         CDEBUG(D_IOCTL, "destroying import %p for %s\n", imp,
 881                 imp->imp_obd->obd_name);
 882
 883         LASSERT(cfs_atomic_read(&imp->imp_refcount) == 0);
 884
 885         ptlrpc_put_connection_superhack(imp->imp_connection);
 886
 887         while (!cfs_list_empty(&imp->imp_conn_list)) {
 888                 struct obd_import_conn *imp_conn;
 889
 890                 imp_conn = cfs_list_entry(imp->imp_conn_list.next,
 891                                           struct obd_import_conn, oic_item);
 892                 cfs_list_del_init(&imp_conn->oic_item);
 893                 ptlrpc_put_connection_superhack(imp_conn->oic_conn);
 894                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 895         }
 896
 897         LASSERT(imp->imp_sec == NULL);
 898         class_decref(imp->imp_obd, "import", imp);
 899         OBD_FREE_RCU(imp, sizeof(*imp), &imp->imp_handle);
 900         EXIT;
 901 }
 902
 903 static void import_handle_addref(void *import)
 904 {
 905         class_import_get(import);
 906 }
 907
 908 struct obd_import *class_import_get(struct obd_import *import)
 909 {
 910         LASSERT(cfs_atomic_read(&import->imp_refcount) >= 0);
 911         LASSERT(cfs_atomic_read(&import->imp_refcount) < 0x5a5a5a);
 912         cfs_atomic_inc(&import->imp_refcount);
 913         CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", import,
 914                cfs_atomic_read(&import->imp_refcount),
 915                import->imp_obd->obd_name);
 916         return import;
 917 }
 918 EXPORT_SYMBOL(class_import_get);
 919
 920 void class_import_put(struct obd_import *imp)
 921 {
 922         ENTRY;
 923
 924         LASSERT(cfs_atomic_read(&imp->imp_refcount) > 0);
 925         LASSERT(cfs_atomic_read(&imp->imp_refcount) < 0x5a5a5a);
 926         LASSERT(cfs_list_empty(&imp->imp_zombie_chain));
 927
 928         CDEBUG(D_INFO, "import %p refcount=%d obd=%s\n", imp,
 929                cfs_atomic_read(&imp->imp_refcount) - 1,
 930                imp->imp_obd->obd_name);
 931
 932         if (cfs_atomic_dec_and_test(&imp->imp_refcount)) {
 933                 CDEBUG(D_INFO, "final put import %p\n", imp);
 934                 obd_zombie_import_add(imp);
 935         }
 936
 937         EXIT;
 938 }
 939 EXPORT_SYMBOL(class_import_put);
 940
 941 static void init_imp_at(struct imp_at *at) {
 942         int i;
 943         at_init(&at->iat_net_latency, 0, 0);
 944         for (i = 0; i < IMP_AT_MAX_PORTALS; i++) {
 945                 /* max service estimates are tracked on the server side, so
 946                    don't use the AT history here, just use the last reported
 947                    val. (But keep hist for proc histogram, worst_ever) */
 948                 at_init(&at->iat_service_estimate[i], INITIAL_CONNECT_TIMEOUT,
 949                         AT_FLG_NOHIST);
 950         }
 951 }
 952
 953 struct obd_import *class_new_import(struct obd_device *obd)
 954 {
 955         struct obd_import *imp;
 956
 957         OBD_ALLOC(imp, sizeof(*imp));
 958         if (imp == NULL)
 959                 return NULL;
 960
 961         CFS_INIT_LIST_HEAD(&imp->imp_zombie_chain);
 962         CFS_INIT_LIST_HEAD(&imp->imp_replay_list);
 963         CFS_INIT_LIST_HEAD(&imp->imp_sending_list);
 964         CFS_INIT_LIST_HEAD(&imp->imp_delayed_list);
 965         cfs_spin_lock_init(&imp->imp_lock);
 966         imp->imp_last_success_conn = 0;
 967         imp->imp_state = LUSTRE_IMP_NEW;
 968         imp->imp_obd = class_incref(obd, "import", imp);
 969         cfs_sema_init(&imp->imp_sec_mutex, 1);
 970         cfs_waitq_init(&imp->imp_recovery_waitq);
 971
 972         cfs_atomic_set(&imp->imp_refcount, 2);
 973         cfs_atomic_set(&imp->imp_unregistering, 0);
 974         cfs_atomic_set(&imp->imp_inflight, 0);
 975         cfs_atomic_set(&imp->imp_replay_inflight, 0);
 976         cfs_atomic_set(&imp->imp_inval_count, 0);
 977         CFS_INIT_LIST_HEAD(&imp->imp_conn_list);
 978         CFS_INIT_LIST_HEAD(&imp->imp_handle.h_link);
 979         class_handle_hash(&imp->imp_handle, import_handle_addref);
 980         init_imp_at(&imp->imp_at);
 981
 982         /* the default magic is V2, will be used in connect RPC, and
 983          * then adjusted according to the flags in request/reply. */
 984         imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
 985
 986         return imp;
 987 }
 988 EXPORT_SYMBOL(class_new_import);
 989
 990 void class_destroy_import(struct obd_import *import)
 991 {
 992         LASSERT(import != NULL);
 993         LASSERT(import != LP_POISON);
 994
 995         class_handle_unhash(&import->imp_handle);
 996
 997         cfs_spin_lock(&import->imp_lock);
 998         import->imp_generation++;
 999         cfs_spin_unlock(&import->imp_lock);
1000         class_import_put(import);
1001 }
1002 EXPORT_SYMBOL(class_destroy_import);
1003
1004 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1005
1006 void __class_export_add_lock_ref(struct obd_export *exp, struct ldlm_lock *lock)
1007 {
1008         cfs_spin_lock(&exp->exp_locks_list_guard);
1009
1010         LASSERT(lock->l_exp_refs_nr >= 0);
1011
1012         if (lock->l_exp_refs_target != NULL &&
1013             lock->l_exp_refs_target != exp) {
1014                 LCONSOLE_WARN("setting export %p for lock %p which already has export %p\n",
1015                               exp, lock, lock->l_exp_refs_target);
1016         }
1017         if ((lock->l_exp_refs_nr ++) == 0) {
1018                 cfs_list_add(&lock->l_exp_refs_link, &exp->exp_locks_list);
1019                 lock->l_exp_refs_target = exp;
1020         }
1021         CDEBUG(D_INFO, "lock = %p, export = %p, refs = %u\n",
1022                lock, exp, lock->l_exp_refs_nr);
1023         cfs_spin_unlock(&exp->exp_locks_list_guard);
1024 }
1025 EXPORT_SYMBOL(__class_export_add_lock_ref);
1026
1027 void __class_export_del_lock_ref(struct obd_export *exp, struct ldlm_lock *lock)
1028 {
1029         cfs_spin_lock(&exp->exp_locks_list_guard);
1030         LASSERT(lock->l_exp_refs_nr > 0);
1031         if (lock->l_exp_refs_target != exp) {
1032                 LCONSOLE_WARN("lock %p, "
1033                               "mismatching export pointers: %p, %p\n",
1034                               lock, lock->l_exp_refs_target, exp);
1035         }
1036         if (-- lock->l_exp_refs_nr == 0) {
1037                 cfs_list_del_init(&lock->l_exp_refs_link);
1038                 lock->l_exp_refs_target = NULL;
1039         }
1040         CDEBUG(D_INFO, "lock = %p, export = %p, refs = %u\n",
1041                lock, exp, lock->l_exp_refs_nr);
1042         cfs_spin_unlock(&exp->exp_locks_list_guard);
1043 }
1044 EXPORT_SYMBOL(__class_export_del_lock_ref);
1045 #endif
1046
1047 /* A connection defines an export context in which preallocation can
1048    be managed. This releases the export pointer reference, and returns
1049    the export handle, so the export refcount is 1 when this function
1050    returns. */
1051 int class_connect(struct lustre_handle *conn, struct obd_device *obd,
1052                   struct obd_uuid *cluuid)
1053 {
1054         struct obd_export *export;
1055         LASSERT(conn != NULL);
1056         LASSERT(obd != NULL);
1057         LASSERT(cluuid != NULL);
1058         ENTRY;
1059
1060         export = class_new_export(obd, cluuid);
1061         if (IS_ERR(export))
1062                 RETURN(PTR_ERR(export));
1063
1064         conn->cookie = export->exp_handle.h_cookie;
1065         class_export_put(export);
1066
1067         CDEBUG(D_IOCTL, "connect: client %s, cookie "LPX64"\n",
1068                cluuid->uuid, conn->cookie);
1069         RETURN(0);
1070 }
1071 EXPORT_SYMBOL(class_connect);
1072
1073 /* if export is involved in recovery then clean up related things */
1074 void class_export_recovery_cleanup(struct obd_export *exp)
1075 {
1076         struct obd_device *obd = exp->exp_obd;
1077
1078         cfs_spin_lock_bh(&obd->obd_processing_task_lock);
1079         if (exp->exp_delayed)
1080                 obd->obd_delayed_clients--;
1081         if (obd->obd_recovering && exp->exp_in_recovery) {
1082                 cfs_spin_lock(&exp->exp_lock);
1083                 exp->exp_in_recovery = 0;
1084                 cfs_spin_unlock(&exp->exp_lock);
1085                 LASSERT(obd->obd_connected_clients);
1086                 obd->obd_connected_clients--;
1087         }
1088         /** Cleanup req replay fields */
1089         if (exp->exp_req_replay_needed) {
1090                 cfs_spin_lock(&exp->exp_lock);
1091                 exp->exp_req_replay_needed = 0;
1092                 cfs_spin_unlock(&exp->exp_lock);
1093                 LASSERT(cfs_atomic_read(&obd->obd_req_replay_clients));
1094                 cfs_atomic_dec(&obd->obd_req_replay_clients);
1095         }
1096         /** Cleanup lock replay data */
1097         if (exp->exp_lock_replay_needed) {
1098                 cfs_spin_lock(&exp->exp_lock);
1099                 exp->exp_lock_replay_needed = 0;
1100                 cfs_spin_unlock(&exp->exp_lock);
1101                 LASSERT(cfs_atomic_read(&obd->obd_lock_replay_clients));
1102                 cfs_atomic_dec(&obd->obd_lock_replay_clients);
1103         }
1104         cfs_spin_unlock_bh(&obd->obd_processing_task_lock);
1105 }
1106
1107 /* This function removes 1-3 references from the export:
1108  * 1 - for export pointer passed
1109  * and if disconnect really need
1110  * 2 - removing from hash
1111  * 3 - in client_unlink_export
1112  * The export pointer passed to this function can destroyed */
1113 int class_disconnect(struct obd_export *export)
1114 {
1115         int already_disconnected;
1116         ENTRY;
1117
1118         if (export == NULL) {
1119                 fixme();
1120                 CDEBUG(D_IOCTL, "attempting to free NULL export %p\n", export);
1121                 RETURN(-EINVAL);
1122         }
1123
1124         cfs_spin_lock(&export->exp_lock);
1125         already_disconnected = export->exp_disconnected;
1126         export->exp_disconnected = 1;
1127         cfs_spin_unlock(&export->exp_lock);
1128
1129         /* class_cleanup(), abort_recovery(), and class_fail_export()
1130          * all end up in here, and if any of them race we shouldn't
1131          * call extra class_export_puts(). */
1132         if (already_disconnected) {
1133                 LASSERT(cfs_hlist_unhashed(&export->exp_nid_hash));
1134                 GOTO(no_disconn, already_disconnected);
1135         }
1136
1137         CDEBUG(D_IOCTL, "disconnect: cookie "LPX64"\n",
1138                export->exp_handle.h_cookie);
1139
1140         if (!cfs_hlist_unhashed(&export->exp_nid_hash))
1141                 cfs_hash_del(export->exp_obd->obd_nid_hash,
1142                              &export->exp_connection->c_peer.nid,
1143                              &export->exp_nid_hash);
1144
1145         class_export_recovery_cleanup(export);
1146         class_unlink_export(export);
1147 no_disconn:
1148         class_export_put(export);
1149         RETURN(0);
1150 }
1151
1152 /* Return non-zero for a fully connected export */
1153 int class_connected_export(struct obd_export *exp)
1154 {
1155         if (exp) {
1156                 int connected;
1157                 cfs_spin_lock(&exp->exp_lock);
1158                 connected = (exp->exp_conn_cnt > 0);
1159                 cfs_spin_unlock(&exp->exp_lock);
1160                 return connected;
1161         }
1162         return 0;
1163 }
1164 EXPORT_SYMBOL(class_connected_export);
1165
1166 static void class_disconnect_export_list(cfs_list_t *list,
1167                                          enum obd_option flags)
1168 {
1169         int rc;
1170         struct obd_export *exp;
1171         ENTRY;
1172
1173         /* It's possible that an export may disconnect itself, but
1174          * nothing else will be added to this list. */
1175         while (!cfs_list_empty(list)) {
1176                 exp = cfs_list_entry(list->next, struct obd_export,
1177                                      exp_obd_chain);
1178                 /* need for safe call CDEBUG after obd_disconnect */
1179                 class_export_get(exp);
1180
1181                 cfs_spin_lock(&exp->exp_lock);
1182                 exp->exp_flags = flags;
1183                 cfs_spin_unlock(&exp->exp_lock);
1184
1185                 if (obd_uuid_equals(&exp->exp_client_uuid,
1186                                     &exp->exp_obd->obd_uuid)) {
1187                         CDEBUG(D_HA,
1188                                "exp %p export uuid == obd uuid, don't discon\n",
1189                                exp);
1190                         /* Need to delete this now so we don't end up pointing
1191                          * to work_list later when this export is cleaned up. */
1192                         cfs_list_del_init(&exp->exp_obd_chain);
1193                         class_export_put(exp);
1194                         continue;
1195                 }
1196
1197                 class_export_get(exp);
1198                 CDEBUG(D_HA, "%s: disconnecting export at %s (%p), "
1199                        "last request at "CFS_TIME_T"\n",
1200                        exp->exp_obd->obd_name, obd_export_nid2str(exp),
1201                        exp, exp->exp_last_request_time);
1202                 /* release one export reference anyway */
1203                 rc = obd_disconnect(exp);
1204
1205                 CDEBUG(D_HA, "disconnected export at %s (%p): rc %d\n",
1206                        obd_export_nid2str(exp), exp, rc);
1207                 class_export_put(exp);
1208         }
1209         EXIT;
1210 }
1211
1212 void class_disconnect_exports(struct obd_device *obd)
1213 {
1214         cfs_list_t work_list;
1215         ENTRY;
1216
1217         /* Move all of the exports from obd_exports to a work list, en masse. */
1218         CFS_INIT_LIST_HEAD(&work_list);
1219         cfs_spin_lock(&obd->obd_dev_lock);
1220         cfs_list_splice_init(&obd->obd_exports, &work_list);
1221         cfs_list_splice_init(&obd->obd_delayed_exports, &work_list);
1222         cfs_spin_unlock(&obd->obd_dev_lock);
1223
1224         if (!cfs_list_empty(&work_list)) {
1225                 CDEBUG(D_HA, "OBD device %d (%p) has exports, "
1226                        "disconnecting them\n", obd->obd_minor, obd);
1227                 class_disconnect_export_list(&work_list,
1228                                              exp_flags_from_obd(obd));
1229         } else
1230                 CDEBUG(D_HA, "OBD device %d (%p) has no exports\n",
1231                        obd->obd_minor, obd);
1232         EXIT;
1233 }
1234 EXPORT_SYMBOL(class_disconnect_exports);
1235
1236 /* Remove exports that have not completed recovery.
1237  */
1238 void class_disconnect_stale_exports(struct obd_device *obd,
1239                                     int (*test_export)(struct obd_export *))
1240 {
1241         cfs_list_t work_list;
1242         cfs_list_t *pos, *n;
1243         struct obd_export *exp;
1244         int evicted = 0;
1245         ENTRY;
1246
1247         CFS_INIT_LIST_HEAD(&work_list);
1248         cfs_spin_lock(&obd->obd_dev_lock);
1249         cfs_list_for_each_safe(pos, n, &obd->obd_exports) {
1250                 exp = cfs_list_entry(pos, struct obd_export, exp_obd_chain);
1251                 if (test_export(exp))
1252                         continue;
1253
1254                 /* don't count self-export as client */
1255                 if (obd_uuid_equals(&exp->exp_client_uuid,
1256                                     &exp->exp_obd->obd_uuid))
1257                         continue;
1258
1259                 cfs_list_move(&exp->exp_obd_chain, &work_list);
1260                 evicted++;
1261                 CDEBUG(D_ERROR, "%s: disconnect stale client %s@%s\n",
1262                        obd->obd_name, exp->exp_client_uuid.uuid,
1263                        exp->exp_connection == NULL ? "<unknown>" :
1264                        libcfs_nid2str(exp->exp_connection->c_peer.nid));
1265                 print_export_data(exp, "EVICTING", 0);
1266         }
1267         cfs_spin_unlock(&obd->obd_dev_lock);
1268
1269         if (evicted) {
1270                 CDEBUG(D_HA, "%s: disconnecting %d stale clients\n",
1271                        obd->obd_name, evicted);
1272                 obd->obd_stale_clients += evicted;
1273         }
1274         class_disconnect_export_list(&work_list, exp_flags_from_obd(obd) |
1275                                                  OBD_OPT_ABORT_RECOV);
1276         EXIT;
1277 }
1278 EXPORT_SYMBOL(class_disconnect_stale_exports);
1279
1280 void class_fail_export(struct obd_export *exp)
1281 {
1282         int rc, already_failed;
1283
1284         cfs_spin_lock(&exp->exp_lock);
1285         already_failed = exp->exp_failed;
1286         exp->exp_failed = 1;
1287         cfs_spin_unlock(&exp->exp_lock);
1288
1289         if (already_failed) {
1290                 CDEBUG(D_HA, "disconnecting dead export %p/%s; skipping\n",
1291                        exp, exp->exp_client_uuid.uuid);
1292                 return;
1293         }
1294
1295         CDEBUG(D_HA, "disconnecting export %p/%s\n",
1296                exp, exp->exp_client_uuid.uuid);
1297
1298         if (obd_dump_on_timeout)
1299                 libcfs_debug_dumplog();
1300
1301         /* Most callers into obd_disconnect are removing their own reference
1302          * (request, for example) in addition to the one from the hash table.
1303          * We don't have such a reference here, so make one. */
1304         class_export_get(exp);
1305         rc = obd_disconnect(exp);
1306         if (rc)
1307                 CERROR("disconnecting export %p failed: %d\n", exp, rc);
1308         else
1309                 CDEBUG(D_HA, "disconnected export %p/%s\n",
1310                        exp, exp->exp_client_uuid.uuid);
1311 }
1312 EXPORT_SYMBOL(class_fail_export);
1313
1314 char *obd_export_nid2str(struct obd_export *exp)
1315 {
1316         if (exp->exp_connection != NULL)
1317                 return libcfs_nid2str(exp->exp_connection->c_peer.nid);
1318
1319         return "(no nid)";
1320 }
1321 EXPORT_SYMBOL(obd_export_nid2str);
1322
1323 int obd_export_evict_by_nid(struct obd_device *obd, const char *nid)
1324 {
1325         struct obd_export *doomed_exp = NULL;
1326         int exports_evicted = 0;
1327
1328         lnet_nid_t nid_key = libcfs_str2nid((char *)nid);
1329
1330         do {
1331                 doomed_exp = cfs_hash_lookup(obd->obd_nid_hash, &nid_key);
1332                 if (doomed_exp == NULL)
1333                         break;
1334
1335                 LASSERTF(doomed_exp->exp_connection->c_peer.nid == nid_key,
1336                          "nid %s found, wanted nid %s, requested nid %s\n",
1337                          obd_export_nid2str(doomed_exp),
1338                          libcfs_nid2str(nid_key), nid);
1339                 LASSERTF(doomed_exp != obd->obd_self_export,
1340                          "self-export is hashed by NID?\n");
1341                 exports_evicted++;
1342                 CWARN("%s: evict NID '%s' (%s) #%d at adminstrative request\n",
1343                        obd->obd_name, nid, doomed_exp->exp_client_uuid.uuid,
1344                        exports_evicted);
1345                 class_fail_export(doomed_exp);
1346                 class_export_put(doomed_exp);
1347         } while (1);
1348
1349         if (!exports_evicted)
1350                 CDEBUG(D_HA,"%s: can't disconnect NID '%s': no exports found\n",
1351                        obd->obd_name, nid);
1352         return exports_evicted;
1353 }
1354 EXPORT_SYMBOL(obd_export_evict_by_nid);
1355
1356 int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid)
1357 {
1358         struct obd_export *doomed_exp = NULL;
1359         struct obd_uuid doomed_uuid;
1360         int exports_evicted = 0;
1361
1362         obd_str2uuid(&doomed_uuid, uuid);
1363         if (obd_uuid_equals(&doomed_uuid, &obd->obd_uuid)) {
1364                 CERROR("%s: can't evict myself\n", obd->obd_name);
1365                 return exports_evicted;
1366         }
1367
1368         doomed_exp = cfs_hash_lookup(obd->obd_uuid_hash, &doomed_uuid);
1369
1370         if (doomed_exp == NULL) {
1371                 CERROR("%s: can't disconnect %s: no exports found\n",
1372                        obd->obd_name, uuid);
1373         } else {
1374                 CWARN("%s: evicting %s at adminstrative request\n",
1375                        obd->obd_name, doomed_exp->exp_client_uuid.uuid);
1376                 class_fail_export(doomed_exp);
1377                 class_export_put(doomed_exp);
1378                 exports_evicted++;
1379         }
1380
1381         return exports_evicted;
1382 }
1383 EXPORT_SYMBOL(obd_export_evict_by_uuid);
1384
1385 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1386 void (*class_export_dump_hook)(struct obd_export*) = NULL;
1387 EXPORT_SYMBOL(class_export_dump_hook);
1388 #endif
1389
1390 static void print_export_data(struct obd_export *exp, const char *status,
1391                               int locks)
1392 {
1393         struct ptlrpc_reply_state *rs;
1394         struct ptlrpc_reply_state *first_reply = NULL;
1395         int nreplies = 0;
1396
1397         cfs_spin_lock(&exp->exp_lock);
1398         cfs_list_for_each_entry(rs, &exp->exp_outstanding_replies,
1399                                 rs_exp_list) {
1400                 if (nreplies == 0)
1401                         first_reply = rs;
1402                 nreplies++;
1403         }
1404         cfs_spin_unlock(&exp->exp_lock);
1405
1406         CDEBUG(D_HA, "%s: %s %p %s %s %d (%d %d %d) %d %d %d %d: %p %s "LPU64"\n",
1407                exp->exp_obd->obd_name, status, exp, exp->exp_client_uuid.uuid,
1408                obd_export_nid2str(exp), cfs_atomic_read(&exp->exp_refcount),
1409                cfs_atomic_read(&exp->exp_rpc_count),
1410                cfs_atomic_read(&exp->exp_cb_count),
1411                cfs_atomic_read(&exp->exp_locks_count),
1412                exp->exp_disconnected, exp->exp_delayed, exp->exp_failed,
1413                nreplies, first_reply, nreplies > 3 ? "..." : "",
1414                exp->exp_last_committed);
1415 #if LUSTRE_TRACKS_LOCK_EXP_REFS
1416         if (locks && class_export_dump_hook != NULL)
1417                 class_export_dump_hook(exp);
1418 #endif
1419 }
1420
1421 void dump_exports(struct obd_device *obd, int locks)
1422 {
1423         struct obd_export *exp;
1424
1425         cfs_spin_lock(&obd->obd_dev_lock);
1426         cfs_list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain)
1427                 print_export_data(exp, "ACTIVE", locks);
1428         cfs_list_for_each_entry(exp, &obd->obd_unlinked_exports, exp_obd_chain)
1429                 print_export_data(exp, "UNLINKED", locks);
1430         cfs_list_for_each_entry(exp, &obd->obd_delayed_exports, exp_obd_chain)
1431                 print_export_data(exp, "DELAYED", locks);
1432         cfs_spin_unlock(&obd->obd_dev_lock);
1433         cfs_spin_lock(&obd_zombie_impexp_lock);
1434         cfs_list_for_each_entry(exp, &obd_zombie_exports, exp_obd_chain)
1435                 print_export_data(exp, "ZOMBIE", locks);
1436         cfs_spin_unlock(&obd_zombie_impexp_lock);
1437 }
1438 EXPORT_SYMBOL(dump_exports);
1439
1440 void obd_exports_barrier(struct obd_device *obd)
1441 {
1442         int waited = 2;
1443         LASSERT(cfs_list_empty(&obd->obd_exports));
1444         cfs_spin_lock(&obd->obd_dev_lock);
1445         while (!cfs_list_empty(&obd->obd_unlinked_exports)) {
1446                 cfs_spin_unlock(&obd->obd_dev_lock);
1447                 cfs_schedule_timeout_and_set_state(CFS_TASK_UNINT,
1448                                                    cfs_time_seconds(waited));
1449                 if (waited > 5 && IS_PO2(waited)) {
1450                         LCONSOLE_WARN("%s is waiting for obd_unlinked_exports "
1451                                       "more than %d seconds. "
1452                                       "The obd refcount = %d. Is it stuck?\n",
1453                                       obd->obd_name, waited,
1454                                       cfs_atomic_read(&obd->obd_refcount));
1455                         dump_exports(obd, 1);
1456                 }
1457                 waited *= 2;
1458                 cfs_spin_lock(&obd->obd_dev_lock);
1459         }
1460         cfs_spin_unlock(&obd->obd_dev_lock);
1461 }
1462 EXPORT_SYMBOL(obd_exports_barrier);
1463
1464 /**
1465  * kill zombie imports and exports
1466  */
1467 void obd_zombie_impexp_cull(void)
1468 {
1469         struct obd_import *import;
1470         struct obd_export *export;
1471         ENTRY;
1472
1473         do {
1474                 cfs_spin_lock(&obd_zombie_impexp_lock);
1475
1476                 import = NULL;
1477                 if (!cfs_list_empty(&obd_zombie_imports)) {
1478                         import = cfs_list_entry(obd_zombie_imports.next,
1479                                                 struct obd_import,
1480                                                 imp_zombie_chain);
1481                         cfs_list_del_init(&import->imp_zombie_chain);
1482                 }
1483
1484                 export = NULL;
1485                 if (!cfs_list_empty(&obd_zombie_exports)) {
1486                         export = cfs_list_entry(obd_zombie_exports.next,
1487                                                 struct obd_export,
1488                                                 exp_obd_chain);
1489                         cfs_list_del_init(&export->exp_obd_chain);
1490                 }
1491
1492                 cfs_spin_unlock(&obd_zombie_impexp_lock);
1493
1494                 if (import != NULL)
1495                         class_import_destroy(import);
1496
1497                 if (export != NULL)
1498                         class_export_destroy(export);
1499
1500                 cfs_cond_resched();
1501         } while (import != NULL || export != NULL);
1502         EXIT;
1503 }
1504
1505 static cfs_completion_t         obd_zombie_start;
1506 static cfs_completion_t         obd_zombie_stop;
1507 static unsigned long            obd_zombie_flags;
1508 static cfs_waitq_t              obd_zombie_waitq;
1509 static pid_t                    obd_zombie_pid;
1510
1511 enum {
1512         OBD_ZOMBIE_STOP   = 1 << 1
1513 };
1514
1515 /**
1516  * check for work for kill zombie import/export thread.
1517  */
1518 static int obd_zombie_impexp_check(void *arg)
1519 {
1520         int rc;
1521
1522         cfs_spin_lock(&obd_zombie_impexp_lock);
1523         rc = cfs_list_empty(&obd_zombie_imports) &&
1524              cfs_list_empty(&obd_zombie_exports) &&
1525              !cfs_test_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags);
1526
1527         cfs_spin_unlock(&obd_zombie_impexp_lock);
1528
1529         RETURN(rc);
1530 }
1531
1532 /**
1533  * Add export to the obd_zombe thread and notify it.
1534  */
1535 static void obd_zombie_export_add(struct obd_export *exp) {
1536         cfs_spin_lock(&exp->exp_obd->obd_dev_lock);
1537         LASSERT(!cfs_list_empty(&exp->exp_obd_chain));
1538         cfs_list_del_init(&exp->exp_obd_chain);
1539         cfs_spin_unlock(&exp->exp_obd->obd_dev_lock);
1540         cfs_spin_lock(&obd_zombie_impexp_lock);
1541         cfs_list_add(&exp->exp_obd_chain, &obd_zombie_exports);
1542         cfs_spin_unlock(&obd_zombie_impexp_lock);
1543
1544         if (obd_zombie_impexp_notify != NULL)
1545                 obd_zombie_impexp_notify();
1546 }
1547
1548 /**
1549  * Add import to the obd_zombe thread and notify it.
1550  */
1551 static void obd_zombie_import_add(struct obd_import *imp) {
1552         LASSERT(imp->imp_sec == NULL);
1553         cfs_spin_lock(&obd_zombie_impexp_lock);
1554         LASSERT(cfs_list_empty(&imp->imp_zombie_chain));
1555         cfs_list_add(&imp->imp_zombie_chain, &obd_zombie_imports);
1556         cfs_spin_unlock(&obd_zombie_impexp_lock);
1557
1558         if (obd_zombie_impexp_notify != NULL)
1559                 obd_zombie_impexp_notify();
1560 }
1561
1562 /**
1563  * notify import/export destroy thread about new zombie.
1564  */
1565 static void obd_zombie_impexp_notify(void)
1566 {
1567         cfs_waitq_signal(&obd_zombie_waitq);
1568 }
1569
1570 /**
1571  * check whether obd_zombie is idle
1572  */
1573 static int obd_zombie_is_idle(void)
1574 {
1575         int rc;
1576
1577         LASSERT(!cfs_test_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags));
1578         cfs_spin_lock(&obd_zombie_impexp_lock);
1579         rc = cfs_list_empty(&obd_zombie_imports) &&
1580              cfs_list_empty(&obd_zombie_exports);
1581         cfs_spin_unlock(&obd_zombie_impexp_lock);
1582         return rc;
1583 }
1584
1585 /**
1586  * wait when obd_zombie import/export queues become empty
1587  */
1588 void obd_zombie_barrier(void)
1589 {
1590         struct l_wait_info lwi = { 0 };
1591
1592         if (obd_zombie_pid == cfs_curproc_pid())
1593                 /* don't wait for myself */
1594                 return;
1595         l_wait_event(obd_zombie_waitq, obd_zombie_is_idle(), &lwi);
1596 }
1597 EXPORT_SYMBOL(obd_zombie_barrier);
1598
1599 #ifdef __KERNEL__
1600
1601 /**
1602  * destroy zombie export/import thread.
1603  */
1604 static int obd_zombie_impexp_thread(void *unused)
1605 {
1606         int rc;
1607
1608         if ((rc = cfs_daemonize_ctxt("obd_zombid"))) {
1609                 cfs_complete(&obd_zombie_start);
1610                 RETURN(rc);
1611         }
1612
1613         cfs_complete(&obd_zombie_start);
1614
1615         obd_zombie_pid = cfs_curproc_pid();
1616
1617         while(!cfs_test_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags)) {
1618                 struct l_wait_info lwi = { 0 };
1619
1620                 l_wait_event(obd_zombie_waitq,
1621                              !obd_zombie_impexp_check(NULL), &lwi);
1622                 obd_zombie_impexp_cull();
1623
1624                 /*
1625                  * Notify obd_zombie_barrier callers that queues
1626                  * may be empty.
1627                  */
1628                 cfs_waitq_signal(&obd_zombie_waitq);
1629         }
1630
1631         cfs_complete(&obd_zombie_stop);
1632
1633         RETURN(0);
1634 }
1635
1636 #else /* ! KERNEL */
1637
1638 static cfs_atomic_t zombie_recur = CFS_ATOMIC_INIT(0);
1639 static void *obd_zombie_impexp_work_cb;
1640 static void *obd_zombie_impexp_idle_cb;
1641
1642 int obd_zombie_impexp_kill(void *arg)
1643 {
1644         int rc = 0;
1645
1646         if (cfs_atomic_inc_return(&zombie_recur) == 1) {
1647                 obd_zombie_impexp_cull();
1648                 rc = 1;
1649         }
1650         cfs_atomic_dec(&zombie_recur);
1651         return rc;
1652 }
1653
1654 #endif
1655
1656 /**
1657  * start destroy zombie import/export thread
1658  */
1659 int obd_zombie_impexp_init(void)
1660 {
1661         int rc;
1662
1663         CFS_INIT_LIST_HEAD(&obd_zombie_imports);
1664         CFS_INIT_LIST_HEAD(&obd_zombie_exports);
1665         cfs_spin_lock_init(&obd_zombie_impexp_lock);
1666         cfs_init_completion(&obd_zombie_start);
1667         cfs_init_completion(&obd_zombie_stop);
1668         cfs_waitq_init(&obd_zombie_waitq);
1669         obd_zombie_pid = 0;
1670
1671 #ifdef __KERNEL__
1672         rc = cfs_kernel_thread(obd_zombie_impexp_thread, NULL, 0);
1673         if (rc < 0)
1674                 RETURN(rc);
1675
1676         cfs_wait_for_completion(&obd_zombie_start);
1677 #else
1678
1679         obd_zombie_impexp_work_cb =
1680                 liblustre_register_wait_callback("obd_zombi_impexp_kill",
1681                                                  &obd_zombie_impexp_kill, NULL);
1682
1683         obd_zombie_impexp_idle_cb =
1684                 liblustre_register_idle_callback("obd_zombi_impexp_check",
1685                                                  &obd_zombie_impexp_check, NULL);
1686         rc = 0;
1687 #endif
1688         RETURN(rc);
1689 }
1690 /**
1691  * stop destroy zombie import/export thread
1692  */
1693 void obd_zombie_impexp_stop(void)
1694 {
1695         cfs_set_bit(OBD_ZOMBIE_STOP, &obd_zombie_flags);
1696         obd_zombie_impexp_notify();
1697 #ifdef __KERNEL__
1698         cfs_wait_for_completion(&obd_zombie_stop);
1699 #else
1700         liblustre_deregister_wait_callback(obd_zombie_impexp_work_cb);
1701         liblustre_deregister_idle_callback(obd_zombie_impexp_idle_cb);
1702 #endif
1703 }
1704