lustre/obdclass/genops.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  *
  24  * These are the only exported functions, they provide some generic
  25  * infrastructure for managing object devices
  26  */
  27
  28 #define DEBUG_SUBSYSTEM S_CLASS
  29 #ifndef __KERNEL__
  30 #include <liblustre.h>
  31 #endif
  32 #include <obd_ost.h>
  33 #include <obd_class.h>
  34 #include <lprocfs_status.h>
  35
  36 extern struct list_head obd_types;
  37 spinlock_t obd_types_lock;
  38
  39 cfs_mem_cache_t *obd_device_cachep;
  40 cfs_mem_cache_t *obdo_cachep;
  41 EXPORT_SYMBOL(obdo_cachep);
  42 cfs_mem_cache_t *import_cachep;
  43
  44 struct list_head  obd_zombie_imports;
  45 struct list_head  obd_zombie_exports;
  46 spinlock_t        obd_zombie_impexp_lock;
  47 void            (*obd_zombie_impexp_notify)(void) = NULL;
  48 EXPORT_SYMBOL(obd_zombie_impexp_notify);
  49
  50
  51 int (*ptlrpc_put_connection_superhack)(struct ptlrpc_connection *c);
  52
  53 /*
  54  * support functions: we could use inter-module communication, but this
  55  * is more portable to other OS's
  56  */
  57 static struct obd_device *obd_device_alloc(void)
  58 {
  59         struct obd_device *obd;
  60
  61         OBD_SLAB_ALLOC_PTR(obd, obd_device_cachep);
  62         if (obd != NULL) {
  63                 obd->obd_magic = OBD_DEVICE_MAGIC;
  64         }
  65         return obd;
  66 }
  67 EXPORT_SYMBOL(obd_device_alloc);
  68
  69 static void obd_device_free(struct obd_device *obd)
  70 {
  71         LASSERT(obd != NULL);
  72         LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "obd %p obd_magic %08x != %08x\n",
  73                  obd, obd->obd_magic, OBD_DEVICE_MAGIC);
  74         OBD_SLAB_FREE_PTR(obd, obd_device_cachep);
  75 }
  76 EXPORT_SYMBOL(obd_device_free);
  77
  78 struct obd_type *class_search_type(const char *name)
  79 {
  80         struct list_head *tmp;
  81         struct obd_type *type;
  82
  83         spin_lock(&obd_types_lock);
  84         list_for_each(tmp, &obd_types) {
  85                 type = list_entry(tmp, struct obd_type, typ_chain);
  86                 if (strcmp(type->typ_name, name) == 0) {
  87                         spin_unlock(&obd_types_lock);
  88                         return type;
  89                 }
  90         }
  91         spin_unlock(&obd_types_lock);
  92         return NULL;
  93 }
  94
  95 struct obd_type *class_get_type(const char *name)
  96 {
  97         struct obd_type *type = class_search_type(name);
  98
  99 #ifdef CONFIG_KMOD
 100         if (!type) {
 101                 const char *modname = name;
 102                 if (!request_module(modname)) {
 103                         CDEBUG(D_INFO, "Loaded module '%s'\n", modname);
 104                         type = class_search_type(name);
 105                 } else {
 106                         LCONSOLE_ERROR_MSG(0x158, "Can't load module '%s'\n",
 107                                            modname);
 108                 }
 109         }
 110 #endif
 111         if (type) {
 112                 spin_lock(&type->obd_type_lock);
 113                 type->typ_refcnt++;
 114                 try_module_get(type->typ_dt_ops->o_owner);
 115                 spin_unlock(&type->obd_type_lock);
 116         }
 117         return type;
 118 }
 119
 120 void class_put_type(struct obd_type *type)
 121 {
 122         LASSERT(type);
 123         spin_lock(&type->obd_type_lock);
 124         type->typ_refcnt--;
 125         module_put(type->typ_dt_ops->o_owner);
 126         spin_unlock(&type->obd_type_lock);
 127 }
 128
 129 #define CLASS_MAX_NAME 1024
 130
 131 int class_register_type(struct obd_ops *dt_ops, struct md_ops *md_ops,
 132                         struct lprocfs_vars *vars, const char *name,
 133                         struct lu_device_type *ldt)
 134 {
 135         struct obd_type *type;
 136         int rc = 0;
 137         ENTRY;
 138
 139         /* sanity check */
 140         LASSERT(strnlen(name, CLASS_MAX_NAME) < CLASS_MAX_NAME);
 141
 142         if (class_search_type(name)) {
 143                 CDEBUG(D_IOCTL, "Type %s already registered\n", name);
 144                 RETURN(-EEXIST);
 145         }
 146
 147         rc = -ENOMEM;
 148         OBD_ALLOC(type, sizeof(*type));
 149         if (type == NULL)
 150                 RETURN(rc);
 151
 152         OBD_ALLOC_PTR(type->typ_dt_ops);
 153         OBD_ALLOC_PTR(type->typ_md_ops);
 154         OBD_ALLOC(type->typ_name, strlen(name) + 1);
 155
 156         if (type->typ_dt_ops == NULL ||
 157             type->typ_md_ops == NULL ||
 158             type->typ_name == NULL)
 159                 GOTO (failed, rc);
 160
 161         *(type->typ_dt_ops) = *dt_ops;
 162         /* md_ops is optional */
 163         if (md_ops)
 164                 *(type->typ_md_ops) = *md_ops;
 165         strcpy(type->typ_name, name);
 166         spin_lock_init(&type->obd_type_lock);
 167
 168 #ifdef LPROCFS
 169         type->typ_procroot = lprocfs_register(type->typ_name, proc_lustre_root,
 170                                               vars, type);
 171         if (IS_ERR(type->typ_procroot)) {
 172                 rc = PTR_ERR(type->typ_procroot);
 173                 type->typ_procroot = NULL;
 174                 GOTO (failed, rc);
 175         }
 176 #endif
 177         if (ldt != NULL) {
 178                 type->typ_lu = ldt;
 179                 rc = ldt->ldt_ops->ldto_init(ldt);
 180                 if (rc != 0)
 181                         GOTO (failed, rc);
 182         }
 183
 184         spin_lock(&obd_types_lock);
 185         list_add(&type->typ_chain, &obd_types);
 186         spin_unlock(&obd_types_lock);
 187
 188         RETURN (0);
 189
 190  failed:
 191         if (type->typ_name != NULL)
 192                 OBD_FREE(type->typ_name, strlen(name) + 1);
 193         if (type->typ_md_ops != NULL)
 194                 OBD_FREE_PTR(type->typ_md_ops);
 195         if (type->typ_dt_ops != NULL)
 196                 OBD_FREE_PTR(type->typ_dt_ops);
 197         OBD_FREE(type, sizeof(*type));
 198         RETURN(rc);
 199 }
 200
 201 int class_unregister_type(const char *name)
 202 {
 203         struct obd_type *type = class_search_type(name);
 204         ENTRY;
 205
 206         if (!type) {
 207                 CERROR("unknown obd type\n");
 208                 RETURN(-EINVAL);
 209         }
 210
 211         if (type->typ_refcnt) {
 212                 CERROR("type %s has refcount (%d)\n", name, type->typ_refcnt);
 213                 /* This is a bad situation, let's make the best of it */
 214                 /* Remove ops, but leave the name for debugging */
 215                 OBD_FREE_PTR(type->typ_dt_ops);
 216                 OBD_FREE_PTR(type->typ_md_ops);
 217                 RETURN(-EBUSY);
 218         }
 219
 220         if (type->typ_procroot) {
 221                 lprocfs_remove(&type->typ_procroot);
 222         }
 223
 224         if (type->typ_lu)
 225                 type->typ_lu->ldt_ops->ldto_fini(type->typ_lu);
 226
 227         spin_lock(&obd_types_lock);
 228         list_del(&type->typ_chain);
 229         spin_unlock(&obd_types_lock);
 230         OBD_FREE(type->typ_name, strlen(name) + 1);
 231         if (type->typ_dt_ops != NULL)
 232                 OBD_FREE_PTR(type->typ_dt_ops);
 233         if (type->typ_md_ops != NULL)
 234                 OBD_FREE_PTR(type->typ_md_ops);
 235         OBD_FREE(type, sizeof(*type));
 236         RETURN(0);
 237 } /* class_unregister_type */
 238
 239 struct obd_device *class_newdev(const char *type_name, const char *name)
 240 {
 241         struct obd_device *result = NULL;
 242         struct obd_device *newdev;
 243         struct obd_type *type = NULL;
 244         int i;
 245         int new_obd_minor = 0;
 246
 247         if (strlen(name) > MAX_OBD_NAME) {
 248                 CERROR("name/uuid must be < %u bytes long\n", MAX_OBD_NAME);
 249                 RETURN(ERR_PTR(-EINVAL));
 250         }
 251
 252         type = class_get_type(type_name);
 253         if (type == NULL){
 254                 CERROR("OBD: unknown type: %s\n", type_name);
 255                 RETURN(ERR_PTR(-ENODEV));
 256         }
 257
 258         newdev = obd_device_alloc();
 259         if (newdev == NULL) {
 260                 class_put_type(type);
 261                 RETURN(ERR_PTR(-ENOMEM));
 262         }
 263         LASSERT(newdev->obd_magic == OBD_DEVICE_MAGIC);
 264
 265         spin_lock(&obd_dev_lock);
 266         for (i = 0; i < class_devno_max(); i++) {
 267                 struct obd_device *obd = class_num2obd(i);
 268                 if (obd && obd->obd_name &&
 269                     (strcmp(name, obd->obd_name) == 0)) {
 270                         CERROR("Device %s already exists, won't add\n", name);
 271                         if (result) {
 272                                 LASSERTF(result->obd_magic == OBD_DEVICE_MAGIC,
 273                                          "%p obd_magic %08x != %08x\n", result,
 274                                          result->obd_magic, OBD_DEVICE_MAGIC);
 275                                 LASSERTF(result->obd_minor == new_obd_minor,
 276                                          "%p obd_minor %d != %d\n", result,
 277                                          result->obd_minor, new_obd_minor);
 278
 279                                 obd_devs[result->obd_minor] = NULL;
 280                                 result->obd_name[0]='\0';
 281                          }
 282                         result = ERR_PTR(-EEXIST);
 283                         break;
 284                 }
 285                 if (!result && !obd) {
 286                         result = newdev;
 287                         result->obd_minor = i;
 288                         new_obd_minor = i;
 289                         result->obd_type = type;
 290                         memcpy(result->obd_name, name, strlen(name));
 291                         obd_devs[i] = result;
 292                 }
 293         }
 294         spin_unlock(&obd_dev_lock);
 295
 296         if (result == NULL && i >= class_devno_max()) {
 297                 CERROR("all %u OBD devices used, increase MAX_OBD_DEVICES\n",
 298                        class_devno_max());
 299                 result = ERR_PTR(-EOVERFLOW);
 300         }
 301
 302         if (IS_ERR(result)) {
 303                 obd_device_free(newdev);
 304                 class_put_type(type);
 305         } else {
 306                 CDEBUG(D_IOCTL, "Adding new device %s (%p)\n",
 307                        result->obd_name, result);
 308         }
 309         return result;
 310 }
 311
 312 void class_release_dev(struct obd_device *obd)
 313 {
 314         struct obd_type *obd_type = obd->obd_type;
 315
 316         LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC, "%p obd_magic %08x != %08x\n",
 317                  obd, obd->obd_magic, OBD_DEVICE_MAGIC);
 318         LASSERTF(obd == obd_devs[obd->obd_minor], "obd %p != obd_devs[%d] %p\n",
 319                  obd, obd->obd_minor, obd_devs[obd->obd_minor]);
 320         LASSERT(obd_type != NULL);
 321
 322         CDEBUG(D_INFO, "Release obd device %s obd_type name =%s\n",
 323                obd->obd_name,obd->obd_type->typ_name);
 324
 325         spin_lock(&obd_dev_lock);
 326         obd_devs[obd->obd_minor] = NULL;
 327         spin_unlock(&obd_dev_lock);
 328         obd_device_free(obd);
 329
 330         class_put_type(obd_type);
 331 }
 332
 333 int class_name2dev(const char *name)
 334 {
 335         int i;
 336
 337         if (!name)
 338                 return -1;
 339
 340         spin_lock(&obd_dev_lock);
 341         for (i = 0; i < class_devno_max(); i++) {
 342                 struct obd_device *obd = class_num2obd(i);
 343                 if (obd && obd->obd_name && strcmp(name, obd->obd_name) == 0) {
 344                         /* Make sure we finished attaching before we give
 345                            out any references */
 346                         LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
 347                         if (obd->obd_attached) {
 348                                 spin_unlock(&obd_dev_lock);
 349                                 return i;
 350                         }
 351                         break;
 352                 }
 353         }
 354         spin_unlock(&obd_dev_lock);
 355
 356         return -1;
 357 }
 358
 359 struct obd_device *class_name2obd(const char *name)
 360 {
 361         int dev = class_name2dev(name);
 362
 363         if (dev < 0 || dev > class_devno_max())
 364                 return NULL;
 365         return class_num2obd(dev);
 366 }
 367
 368 int class_uuid2dev(struct obd_uuid *uuid)
 369 {
 370         int i;
 371
 372         spin_lock(&obd_dev_lock);
 373         for (i = 0; i < class_devno_max(); i++) {
 374                 struct obd_device *obd = class_num2obd(i);
 375                 if (obd && obd_uuid_equals(uuid, &obd->obd_uuid)) {
 376                         LASSERT(obd->obd_magic == OBD_DEVICE_MAGIC);
 377                         spin_unlock(&obd_dev_lock);
 378                         return i;
 379                 }
 380         }
 381         spin_unlock(&obd_dev_lock);
 382
 383         return -1;
 384 }
 385
 386 struct obd_device *class_uuid2obd(struct obd_uuid *uuid)
 387 {
 388         int dev = class_uuid2dev(uuid);
 389         if (dev < 0)
 390                 return NULL;
 391         return class_num2obd(dev);
 392 }
 393
 394 struct obd_device *class_num2obd(int num)
 395 {
 396         struct obd_device *obd = NULL;
 397
 398         if (num < class_devno_max()) {
 399                 obd = obd_devs[num];
 400                 if (obd == NULL) {
 401                         return NULL;
 402                 }
 403
 404                 LASSERTF(obd->obd_magic == OBD_DEVICE_MAGIC,
 405                          "%p obd_magic %08x != %08x\n",
 406                          obd, obd->obd_magic, OBD_DEVICE_MAGIC);
 407                 LASSERTF(obd->obd_minor == num,
 408                          "%p obd_minor %0d != %0d\n",
 409                          obd, obd->obd_minor, num);
 410         }
 411
 412         return obd;
 413 }
 414
 415 void class_obd_list(void)
 416 {
 417         char *status;
 418         int i;
 419
 420         spin_lock(&obd_dev_lock);
 421         for (i = 0; i < class_devno_max(); i++) {
 422                 struct obd_device *obd = class_num2obd(i);
 423                 if (obd == NULL)
 424                         continue;
 425                 if (obd->obd_stopping)
 426                         status = "ST";
 427                 else if (obd->obd_set_up)
 428                         status = "UP";
 429                 else if (obd->obd_attached)
 430                         status = "AT";
 431                 else
 432                         status = "--";
 433                 LCONSOLE(D_CONFIG, "%3d %s %s %s %s %d\n",
 434                          i, status, obd->obd_type->typ_name,
 435                          obd->obd_name, obd->obd_uuid.uuid,
 436                          atomic_read(&obd->obd_refcount));
 437         }
 438         spin_unlock(&obd_dev_lock);
 439         return;
 440 }
 441
 442 /* Search for a client OBD connected to tgt_uuid.  If grp_uuid is
 443    specified, then only the client with that uuid is returned,
 444    otherwise any client connected to the tgt is returned. */
 445 struct obd_device * class_find_client_obd(struct obd_uuid *tgt_uuid,
 446                                           const char * typ_name,
 447                                           struct obd_uuid *grp_uuid)
 448 {
 449         int i;
 450
 451         spin_lock(&obd_dev_lock);
 452         for (i = 0; i < class_devno_max(); i++) {
 453                 struct obd_device *obd = class_num2obd(i);
 454                 if (obd == NULL)
 455                         continue;
 456                 if ((strncmp(obd->obd_type->typ_name, typ_name,
 457                              strlen(typ_name)) == 0)) {
 458                         if (obd_uuid_equals(tgt_uuid,
 459                                             &obd->u.cli.cl_target_uuid) &&
 460                             ((grp_uuid)? obd_uuid_equals(grp_uuid,
 461                                                          &obd->obd_uuid) : 1)) {
 462                                 spin_unlock(&obd_dev_lock);
 463                                 return obd;
 464                         }
 465                 }
 466         }
 467         spin_unlock(&obd_dev_lock);
 468
 469         return NULL;
 470 }
 471
 472 struct obd_device *class_find_client_notype(struct obd_uuid *tgt_uuid,
 473                                             struct obd_uuid *grp_uuid)
 474 {
 475         struct obd_device *obd;
 476
 477         obd = class_find_client_obd(tgt_uuid, LUSTRE_MDC_NAME, NULL);
 478         if (!obd)
 479                 obd = class_find_client_obd(tgt_uuid, LUSTRE_OSC_NAME,
 480                                             grp_uuid);
 481         return obd;
 482 }
 483
 484 /* Iterate the obd_device list looking devices have grp_uuid. Start
 485    searching at *next, and if a device is found, the next index to look
 486    at is saved in *next. If next is NULL, then the first matching device
 487    will always be returned. */
 488 struct obd_device * class_devices_in_group(struct obd_uuid *grp_uuid, int *next)
 489 {
 490         int i;
 491
 492         if (next == NULL)
 493                 i = 0;
 494         else if (*next >= 0 && *next < class_devno_max())
 495                 i = *next;
 496         else
 497                 return NULL;
 498
 499         spin_lock(&obd_dev_lock);
 500         for (; i < class_devno_max(); i++) {
 501                 struct obd_device *obd = class_num2obd(i);
 502                 if (obd == NULL)
 503                         continue;
 504                 if (obd_uuid_equals(grp_uuid, &obd->obd_uuid)) {
 505                         if (next != NULL)
 506                                 *next = i+1;
 507                         spin_unlock(&obd_dev_lock);
 508                         return obd;
 509                 }
 510         }
 511         spin_unlock(&obd_dev_lock);
 512
 513         return NULL;
 514 }
 515
 516
 517 void obd_cleanup_caches(void)
 518 {
 519         int rc;
 520
 521         ENTRY;
 522         if (obd_device_cachep) {
 523                 rc = cfs_mem_cache_destroy(obd_device_cachep);
 524                 LASSERTF(rc == 0, "Cannot destropy ll_obd_device_cache: rc %d\n", rc);
 525                 obd_device_cachep = NULL;
 526         }
 527         if (obdo_cachep) {
 528                 rc = cfs_mem_cache_destroy(obdo_cachep);
 529                 LASSERTF(rc == 0, "Cannot destory ll_obdo_cache\n");
 530                 obdo_cachep = NULL;
 531         }
 532         if (import_cachep) {
 533                 rc = cfs_mem_cache_destroy(import_cachep);
 534                 LASSERTF(rc == 0, "Cannot destory ll_import_cache\n");
 535                 import_cachep = NULL;
 536         }
 537         if (capa_cachep) {
 538                 rc = cfs_mem_cache_destroy(capa_cachep);
 539                 LASSERTF(rc == 0, "Cannot destory capa_cache\n");
 540                 capa_cachep = NULL;
 541         }
 542         EXIT;
 543 }
 544
 545 int obd_init_caches(void)
 546 {
 547         ENTRY;
 548
 549         LASSERT(obd_device_cachep == NULL);
 550         obd_device_cachep = cfs_mem_cache_create("ll_obd_dev_cache",
 551                                                  sizeof(struct obd_device),
 552                                                  0, 0);
 553         if (!obd_device_cachep)
 554                 GOTO(out, -ENOMEM);
 555
 556         LASSERT(obdo_cachep == NULL);
 557         obdo_cachep = cfs_mem_cache_create("ll_obdo_cache", sizeof(struct obdo),
 558                                            0, 0);
 559         if (!obdo_cachep)
 560                 GOTO(out, -ENOMEM);
 561
 562         LASSERT(import_cachep == NULL);
 563         import_cachep = cfs_mem_cache_create("ll_import_cache",
 564                                              sizeof(struct obd_import),
 565                                              0, 0);
 566         if (!import_cachep)
 567                 GOTO(out, -ENOMEM);
 568
 569         LASSERT(capa_cachep == NULL);
 570         capa_cachep = cfs_mem_cache_create("capa_cache",
 571                                            sizeof(struct obd_capa), 0, 0);
 572         if (!capa_cachep)
 573                 GOTO(out, -ENOMEM);
 574
 575         RETURN(0);
 576  out:
 577         obd_cleanup_caches();
 578         RETURN(-ENOMEM);
 579
 580 }
 581
 582 /* map connection to client */
 583 struct obd_export *class_conn2export(struct lustre_handle *conn)
 584 {
 585         struct obd_export *export;
 586         ENTRY;
 587
 588         if (!conn) {
 589                 CDEBUG(D_CACHE, "looking for null handle\n");
 590                 RETURN(NULL);
 591         }
 592
 593         if (conn->cookie == -1) {  /* this means assign a new connection */
 594                 CDEBUG(D_CACHE, "want a new connection\n");
 595                 RETURN(NULL);
 596         }
 597
 598         CDEBUG(D_INFO, "looking for export cookie "LPX64"\n", conn->cookie);
 599         export = class_handle2object(conn->cookie);
 600         RETURN(export);
 601 }
 602
 603 struct obd_device *class_exp2obd(struct obd_export *exp)
 604 {
 605         if (exp)
 606                 return exp->exp_obd;
 607         return NULL;
 608 }
 609
 610 struct obd_device *class_conn2obd(struct lustre_handle *conn)
 611 {
 612         struct obd_export *export;
 613         export = class_conn2export(conn);
 614         if (export) {
 615                 struct obd_device *obd = export->exp_obd;
 616                 class_export_put(export);
 617                 return obd;
 618         }
 619         return NULL;
 620 }
 621
 622 struct obd_import *class_exp2cliimp(struct obd_export *exp)
 623 {
 624         struct obd_device *obd = exp->exp_obd;
 625         if (obd == NULL)
 626                 return NULL;
 627         return obd->u.cli.cl_import;
 628 }
 629
 630 struct obd_import *class_conn2cliimp(struct lustre_handle *conn)
 631 {
 632         struct obd_device *obd = class_conn2obd(conn);
 633         if (obd == NULL)
 634                 return NULL;
 635         return obd->u.cli.cl_import;
 636 }
 637
 638 /* Export management functions */
 639 static void export_handle_addref(void *export)
 640 {
 641         class_export_get(export);
 642 }
 643
 644 void __class_export_put(struct obd_export *exp)
 645 {
 646         if (atomic_dec_and_test(&exp->exp_refcount)) {
 647                 LASSERT (list_empty(&exp->exp_obd_chain));
 648
 649                 CDEBUG(D_IOCTL, "final put %p/%s\n",
 650                        exp, exp->exp_client_uuid.uuid);
 651
 652                 spin_lock(&obd_zombie_impexp_lock);
 653                 list_add(&exp->exp_obd_chain, &obd_zombie_exports);
 654                 spin_unlock(&obd_zombie_impexp_lock);
 655
 656                 if (obd_zombie_impexp_notify != NULL)
 657                         obd_zombie_impexp_notify();
 658         }
 659 }
 660 EXPORT_SYMBOL(__class_export_put);
 661
 662 void class_export_destroy(struct obd_export *exp)
 663 {
 664         struct obd_device *obd = exp->exp_obd;
 665         ENTRY;
 666
 667         LASSERT (atomic_read(&exp->exp_refcount) == 0);
 668
 669         CDEBUG(D_IOCTL, "destroying export %p/%s for %s\n", exp,
 670                exp->exp_client_uuid.uuid, obd->obd_name);
 671
 672         LASSERT(obd != NULL);
 673
 674         /* "Local" exports (lctl, LOV->{mdc,osc}) have no connection. */
 675         if (exp->exp_connection)
 676                 ptlrpc_put_connection_superhack(exp->exp_connection);
 677
 678         LASSERT(list_empty(&exp->exp_outstanding_replies));
 679         obd_destroy_export(exp);
 680
 681         OBD_FREE_RCU(exp, sizeof(*exp), &exp->exp_handle);
 682         class_decref(obd);
 683         EXIT;
 684 }
 685
 686 /* Creates a new export, adds it to the hash table, and returns a
 687  * pointer to it. The refcount is 2: one for the hash reference, and
 688  * one for the pointer returned by this function. */
 689 struct obd_export *class_new_export(struct obd_device *obd,
 690                                     struct obd_uuid *cluuid)
 691 {
 692         struct obd_export *export, *tmp;
 693
 694         OBD_ALLOC(export, sizeof(*export));
 695         if (!export)
 696                 return ERR_PTR(-ENOMEM);
 697
 698         export->exp_conn_cnt = 0;
 699         atomic_set(&export->exp_refcount, 2);
 700         atomic_set(&export->exp_rpc_count, 0);
 701         export->exp_obd = obd;
 702         CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies);
 703         /* XXX this should be in LDLM init */
 704         CFS_INIT_LIST_HEAD(&export->exp_ldlm_data.led_held_locks);
 705         spin_lock_init(&export->exp_ldlm_data.led_lock);
 706
 707         CFS_INIT_LIST_HEAD(&export->exp_handle.h_link);
 708         class_handle_hash(&export->exp_handle, export_handle_addref);
 709         export->exp_last_request_time = CURRENT_SECONDS;
 710         spin_lock_init(&export->exp_lock);
 711
 712         export->exp_client_uuid = *cluuid;
 713         obd_init_export(export);
 714
 715         spin_lock(&obd->obd_dev_lock);
 716         if (!obd_uuid_equals(cluuid, &obd->obd_uuid)) {
 717                 list_for_each_entry(tmp, &obd->obd_exports, exp_obd_chain) {
 718                         if (obd_uuid_equals(cluuid, &tmp->exp_client_uuid)) {
 719                                 spin_unlock(&obd->obd_dev_lock);
 720                                 CWARN("%s: denying duplicate export for %s\n",
 721                                       obd->obd_name, cluuid->uuid);
 722                                 class_handle_unhash(&export->exp_handle);
 723                                 OBD_FREE_PTR(export);
 724                                 return ERR_PTR(-EALREADY);
 725                         }
 726                 }
 727         }
 728         LASSERT(!obd->obd_stopping); /* shouldn't happen, but might race */
 729         class_incref(obd);
 730         list_add(&export->exp_obd_chain, &export->exp_obd->obd_exports);
 731         list_add_tail(&export->exp_obd_chain_timed,
 732                       &export->exp_obd->obd_exports_timed);
 733         export->exp_obd->obd_num_exports++;
 734         spin_unlock(&obd->obd_dev_lock);
 735
 736         return export;
 737 }
 738 EXPORT_SYMBOL(class_new_export);
 739
 740 void class_unlink_export(struct obd_export *exp)
 741 {
 742         class_handle_unhash(&exp->exp_handle);
 743
 744         spin_lock(&exp->exp_obd->obd_dev_lock);
 745         list_del_init(&exp->exp_obd_chain);
 746         list_del_init(&exp->exp_obd_chain_timed);
 747         exp->exp_obd->obd_num_exports--;
 748         spin_unlock(&exp->exp_obd->obd_dev_lock);
 749
 750         class_export_put(exp);
 751 }
 752 EXPORT_SYMBOL(class_unlink_export);
 753
 754 /* Import management functions */
 755 static void import_handle_addref(void *import)
 756 {
 757         class_import_get(import);
 758 }
 759
 760 struct obd_import *class_import_get(struct obd_import *import)
 761 {
 762         LASSERT(atomic_read(&import->imp_refcount) >= 0);
 763         LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a);
 764         atomic_inc(&import->imp_refcount);
 765         CDEBUG(D_INFO, "import %p refcount=%d\n", import,
 766                atomic_read(&import->imp_refcount));
 767         return import;
 768 }
 769 EXPORT_SYMBOL(class_import_get);
 770
 771 void class_import_put(struct obd_import *import)
 772 {
 773         ENTRY;
 774
 775         CDEBUG(D_INFO, "import %p refcount=%d\n", import,
 776                atomic_read(&import->imp_refcount) - 1);
 777
 778         LASSERT(atomic_read(&import->imp_refcount) > 0);
 779         LASSERT(atomic_read(&import->imp_refcount) < 0x5a5a5a);
 780         LASSERT(list_empty(&import->imp_zombie_chain));
 781
 782         if (atomic_dec_and_test(&import->imp_refcount)) {
 783
 784                 CDEBUG(D_INFO, "final put import %p\n", import);
 785
 786                 spin_lock(&obd_zombie_impexp_lock);
 787                 list_add(&import->imp_zombie_chain, &obd_zombie_imports);
 788                 spin_unlock(&obd_zombie_impexp_lock);
 789
 790                 if (obd_zombie_impexp_notify != NULL)
 791                         obd_zombie_impexp_notify();
 792         }
 793
 794         EXIT;
 795 }
 796
 797 void class_import_destroy(struct obd_import *import)
 798 {
 799         ENTRY;
 800
 801         CDEBUG(D_IOCTL, "destroying import %p for %s\n", import,
 802                 import->imp_obd->obd_name);
 803
 804         LASSERT(atomic_read(&import->imp_refcount) == 0);
 805
 806         ptlrpc_put_connection_superhack(import->imp_connection);
 807
 808         while (!list_empty(&import->imp_conn_list)) {
 809                 struct obd_import_conn *imp_conn;
 810
 811                 imp_conn = list_entry(import->imp_conn_list.next,
 812                                       struct obd_import_conn, oic_item);
 813                 list_del(&imp_conn->oic_item);
 814                 ptlrpc_put_connection_superhack(imp_conn->oic_conn);
 815                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 816         }
 817
 818         LASSERT(import->imp_sec == NULL);
 819         class_decref(import->imp_obd);
 820         OBD_FREE_RCU(import, sizeof(*import), &import->imp_handle);
 821         EXIT;
 822 }
 823 EXPORT_SYMBOL(class_import_put);
 824
 825 struct obd_import *class_new_import(struct obd_device *obd)
 826 {
 827         struct obd_import *imp;
 828
 829         OBD_ALLOC(imp, sizeof(*imp));
 830         if (imp == NULL)
 831                 return NULL;
 832
 833         CFS_INIT_LIST_HEAD(&imp->imp_zombie_chain);
 834         CFS_INIT_LIST_HEAD(&imp->imp_replay_list);
 835         CFS_INIT_LIST_HEAD(&imp->imp_sending_list);
 836         CFS_INIT_LIST_HEAD(&imp->imp_delayed_list);
 837         spin_lock_init(&imp->imp_lock);
 838         imp->imp_last_success_conn = 0;
 839         imp->imp_state = LUSTRE_IMP_NEW;
 840         imp->imp_obd = class_incref(obd);
 841         cfs_waitq_init(&imp->imp_recovery_waitq);
 842
 843         atomic_set(&imp->imp_refcount, 2);
 844         atomic_set(&imp->imp_inflight, 0);
 845         atomic_set(&imp->imp_replay_inflight, 0);
 846         atomic_set(&imp->imp_inval_count, 0);
 847         CFS_INIT_LIST_HEAD(&imp->imp_conn_list);
 848         CFS_INIT_LIST_HEAD(&imp->imp_handle.h_link);
 849         class_handle_hash(&imp->imp_handle, import_handle_addref);
 850
 851         /* the default magic is V1, will be used in connect RPC, and
 852          * then adjusted according to the flags in request/reply. */
 853         imp->imp_msg_magic = LUSTRE_MSG_MAGIC_V1;
 854
 855         return imp;
 856 }
 857 EXPORT_SYMBOL(class_new_import);
 858
 859 void class_destroy_import(struct obd_import *import)
 860 {
 861         LASSERT(import != NULL);
 862         LASSERT(import != LP_POISON);
 863
 864         class_handle_unhash(&import->imp_handle);
 865
 866         spin_lock(&import->imp_lock);
 867         import->imp_generation++;
 868         spin_unlock(&import->imp_lock);
 869         class_import_put(import);
 870 }
 871 EXPORT_SYMBOL(class_destroy_import);
 872
 873 /* A connection defines an export context in which preallocation can
 874    be managed. This releases the export pointer reference, and returns
 875    the export handle, so the export refcount is 1 when this function
 876    returns. */
 877 int class_connect(struct lustre_handle *conn, struct obd_device *obd,
 878                   struct obd_uuid *cluuid)
 879 {
 880         struct obd_export *export;
 881         LASSERT(conn != NULL);
 882         LASSERT(obd != NULL);
 883         LASSERT(cluuid != NULL);
 884         ENTRY;
 885
 886         export = class_new_export(obd, cluuid);
 887         if (IS_ERR(export))
 888                 RETURN(PTR_ERR(export));
 889
 890         conn->cookie = export->exp_handle.h_cookie;
 891         class_export_put(export);
 892
 893         CDEBUG(D_IOCTL, "connect: client %s, cookie "LPX64"\n",
 894                cluuid->uuid, conn->cookie);
 895         RETURN(0);
 896 }
 897 EXPORT_SYMBOL(class_connect);
 898
 899 /* if export is involved in recovery then clean up related things */
 900 void class_export_recovery_cleanup(struct obd_export *exp)
 901 {
 902         struct obd_device *obd = exp->exp_obd;
 903
 904         spin_lock_bh(&obd->obd_processing_task_lock);
 905         if (obd->obd_recovering && exp->exp_in_recovery) {
 906                 spin_lock(&exp->exp_lock);
 907                 exp->exp_in_recovery = 0;
 908                 spin_unlock(&exp->exp_lock);
 909                 obd->obd_connected_clients--;
 910                 /* each connected client is counted as recoverable */
 911                 obd->obd_recoverable_clients--;
 912                 if (exp->exp_req_replay_needed) {
 913                         spin_lock(&exp->exp_lock);
 914                         exp->exp_req_replay_needed = 0;
 915                         spin_unlock(&exp->exp_lock);
 916                         LASSERT(atomic_read(&obd->obd_req_replay_clients));
 917                         atomic_dec(&obd->obd_req_replay_clients);
 918                 }
 919                 if (exp->exp_lock_replay_needed) {
 920                         spin_lock(&exp->exp_lock);
 921                         exp->exp_lock_replay_needed = 0;
 922                         spin_unlock(&exp->exp_lock);
 923                         LASSERT(atomic_read(&obd->obd_lock_replay_clients));
 924                         atomic_dec(&obd->obd_lock_replay_clients);
 925                 }
 926         }
 927         spin_unlock_bh(&obd->obd_processing_task_lock);
 928 }
 929
 930 /* This function removes two references from the export: one for the
 931  * hash entry and one for the export pointer passed in.  The export
 932  * pointer passed to this function is destroyed should not be used
 933  * again. */
 934 int class_disconnect(struct obd_export *export)
 935 {
 936         int already_disconnected;
 937         ENTRY;
 938
 939         if (export == NULL) {
 940                 fixme();
 941                 CDEBUG(D_IOCTL, "attempting to free NULL export %p\n", export);
 942                 RETURN(-EINVAL);
 943         }
 944
 945         spin_lock(&export->exp_lock);
 946         already_disconnected = export->exp_disconnected;
 947         export->exp_disconnected = 1;
 948         spin_unlock(&export->exp_lock);
 949
 950         /* class_cleanup(), abort_recovery(), and class_fail_export()
 951          * all end up in here, and if any of them race we shouldn't
 952          * call extra class_export_puts(). */
 953         if (already_disconnected)
 954                 RETURN(0);
 955
 956         CDEBUG(D_IOCTL, "disconnect: cookie "LPX64"\n",
 957                export->exp_handle.h_cookie);
 958
 959         class_export_recovery_cleanup(export);
 960         class_unlink_export(export);
 961         class_export_put(export);
 962         RETURN(0);
 963 }
 964
 965 static void class_disconnect_export_list(struct list_head *list, int flags)
 966 {
 967         int rc;
 968         struct lustre_handle fake_conn;
 969         struct obd_export *fake_exp, *exp;
 970         ENTRY;
 971
 972         /* It's possible that an export may disconnect itself, but
 973          * nothing else will be added to this list. */
 974         while (!list_empty(list)) {
 975                 exp = list_entry(list->next, struct obd_export, exp_obd_chain);
 976                 class_export_get(exp);
 977
 978                 spin_lock(&exp->exp_lock);
 979                 exp->exp_flags = flags;
 980                 spin_unlock(&exp->exp_lock);
 981
 982                 if (obd_uuid_equals(&exp->exp_client_uuid,
 983                                     &exp->exp_obd->obd_uuid)) {
 984                         CDEBUG(D_HA,
 985                                "exp %p export uuid == obd uuid, don't discon\n",
 986                                exp);
 987                         /* Need to delete this now so we don't end up pointing
 988                          * to work_list later when this export is cleaned up. */
 989                         list_del_init(&exp->exp_obd_chain);
 990                         class_export_put(exp);
 991                         continue;
 992                 }
 993
 994                 fake_conn.cookie = exp->exp_handle.h_cookie;
 995                 fake_exp = class_conn2export(&fake_conn);
 996                 if (!fake_exp) {
 997                         class_export_put(exp);
 998                         continue;
 999                 }
1000
1001                 spin_lock(&fake_exp->exp_lock);
1002                 fake_exp->exp_flags = flags;
1003                 spin_unlock(&fake_exp->exp_lock);
1004
1005                 rc = obd_disconnect(fake_exp);
1006                 class_export_put(exp);
1007                 if (rc) {
1008                         CDEBUG(D_HA, "disconnecting export %p failed: %d\n",
1009                                exp, rc);
1010                 } else {
1011                         CDEBUG(D_HA, "export %p disconnected\n", exp);
1012                 }
1013         }
1014         EXIT;
1015 }
1016
1017 static inline int get_exp_flags_from_obd(struct obd_device *obd)
1018 {
1019         return ((obd->obd_fail ? OBD_OPT_FAILOVER : 0) |
1020                 (obd->obd_force ? OBD_OPT_FORCE : 0));
1021 }
1022
1023 void class_disconnect_exports(struct obd_device *obd)
1024 {
1025         struct list_head work_list;
1026         ENTRY;
1027
1028         /* Move all of the exports from obd_exports to a work list, en masse. */
1029         spin_lock(&obd->obd_dev_lock);
1030         list_add(&work_list, &obd->obd_exports);
1031         list_del_init(&obd->obd_exports);
1032         spin_unlock(&obd->obd_dev_lock);
1033
1034         if (!list_empty(&work_list)) {
1035                 CDEBUG(D_HA, "OBD device %d (%p) has exports, "
1036                        "disconnecting them\n", obd->obd_minor, obd);
1037                 class_disconnect_export_list(&work_list,
1038                                              get_exp_flags_from_obd(obd));
1039         } else
1040                 CDEBUG(D_HA, "OBD device %d (%p) has no exports\n",
1041                        obd->obd_minor, obd);
1042         EXIT;
1043 }
1044 EXPORT_SYMBOL(class_disconnect_exports);
1045
1046 /* Remove exports that have not completed recovery.
1047  */
1048 int class_disconnect_stale_exports(struct obd_device *obd,
1049                                    int (*test_export)(struct obd_export *))
1050 {
1051         struct list_head work_list;
1052         struct list_head *pos, *n;
1053         struct obd_export *exp;
1054         int cnt = 0;
1055         ENTRY;
1056
1057         CFS_INIT_LIST_HEAD(&work_list);
1058         spin_lock(&obd->obd_dev_lock);
1059         list_for_each_safe(pos, n, &obd->obd_exports) {
1060                 exp = list_entry(pos, struct obd_export, exp_obd_chain);
1061                 if (test_export(exp))
1062                         continue;
1063
1064                 list_del(&exp->exp_obd_chain);
1065                 list_add(&exp->exp_obd_chain, &work_list);
1066                 /* don't count self-export as client */
1067                 if (obd_uuid_equals(&exp->exp_client_uuid,
1068                                      &exp->exp_obd->obd_uuid))
1069                         continue;
1070
1071                 cnt++;
1072                 CDEBUG(D_ERROR, "%s: disconnect stale client %s@%s\n",
1073                        obd->obd_name, exp->exp_client_uuid.uuid,
1074                        exp->exp_connection == NULL ? "<unknown>" :
1075                        libcfs_nid2str(exp->exp_connection->c_peer.nid));
1076         }
1077         spin_unlock(&obd->obd_dev_lock);
1078
1079         CDEBUG(D_ERROR, "%s: disconnecting %d stale clients\n",
1080                obd->obd_name, cnt);
1081         class_disconnect_export_list(&work_list, get_exp_flags_from_obd(obd));
1082         RETURN(cnt);
1083 }
1084 EXPORT_SYMBOL(class_disconnect_stale_exports);
1085
1086 int oig_init(struct obd_io_group **oig_out)
1087 {
1088         struct obd_io_group *oig;
1089         ENTRY;
1090
1091         OBD_ALLOC(oig, sizeof(*oig));
1092         if (oig == NULL)
1093                 RETURN(-ENOMEM);
1094
1095         spin_lock_init(&oig->oig_lock);
1096         oig->oig_rc = 0;
1097         oig->oig_pending = 0;
1098         atomic_set(&oig->oig_refcount, 1);
1099         cfs_waitq_init(&oig->oig_waitq);
1100         CFS_INIT_LIST_HEAD(&oig->oig_occ_list);
1101
1102         *oig_out = oig;
1103         RETURN(0);
1104 };
1105 EXPORT_SYMBOL(oig_init);
1106
1107 static inline void oig_grab(struct obd_io_group *oig)
1108 {
1109         atomic_inc(&oig->oig_refcount);
1110 }
1111
1112 void oig_release(struct obd_io_group *oig)
1113 {
1114         if (atomic_dec_and_test(&oig->oig_refcount))
1115                 OBD_FREE(oig, sizeof(*oig));
1116 }
1117 EXPORT_SYMBOL(oig_release);
1118
1119 int oig_add_one(struct obd_io_group *oig, struct oig_callback_context *occ)
1120 {
1121         int rc = 0;
1122         CDEBUG(D_CACHE, "oig %p ready to roll\n", oig);
1123         spin_lock(&oig->oig_lock);
1124         if (oig->oig_rc) {
1125                 rc = oig->oig_rc;
1126         } else {
1127                 oig->oig_pending++;
1128                 if (occ != NULL)
1129                         list_add_tail(&occ->occ_oig_item, &oig->oig_occ_list);
1130         }
1131         spin_unlock(&oig->oig_lock);
1132         oig_grab(oig);
1133
1134         return rc;
1135 }
1136 EXPORT_SYMBOL(oig_add_one);
1137
1138 void oig_complete_one(struct obd_io_group *oig,
1139                       struct oig_callback_context *occ, int rc)
1140 {
1141         cfs_waitq_t *wake = NULL;
1142         int old_rc;
1143
1144         spin_lock(&oig->oig_lock);
1145
1146         if (occ != NULL)
1147                 list_del_init(&occ->occ_oig_item);
1148
1149         old_rc = oig->oig_rc;
1150         if (oig->oig_rc == 0 && rc != 0)
1151                 oig->oig_rc = rc;
1152
1153         if (--oig->oig_pending <= 0)
1154                 wake = &oig->oig_waitq;
1155
1156         spin_unlock(&oig->oig_lock);
1157
1158         CDEBUG(D_CACHE, "oig %p completed, rc %d -> %d via %d, %d now "
1159                         "pending (racey)\n", oig, old_rc, oig->oig_rc, rc,
1160                         oig->oig_pending);
1161         if (wake)
1162                 cfs_waitq_signal(wake);
1163         oig_release(oig);
1164 }
1165 EXPORT_SYMBOL(oig_complete_one);
1166
1167 static int oig_done(struct obd_io_group *oig)
1168 {
1169         int rc = 0;
1170         spin_lock(&oig->oig_lock);
1171         if (oig->oig_pending <= 0)
1172                 rc = 1;
1173         spin_unlock(&oig->oig_lock);
1174         return rc;
1175 }
1176
1177 static void interrupted_oig(void *data)
1178 {
1179         struct obd_io_group *oig = data;
1180         struct oig_callback_context *occ;
1181
1182         spin_lock(&oig->oig_lock);
1183         /* We need to restart the processing each time we drop the lock, as
1184          * it is possible other threads called oig_complete_one() to remove
1185          * an entry elsewhere in the list while we dropped lock.  We need to
1186          * drop the lock because osc_ap_completion() calls oig_complete_one()
1187          * which re-gets this lock ;-) as well as a lock ordering issue. */
1188 restart:
1189         list_for_each_entry(occ, &oig->oig_occ_list, occ_oig_item) {
1190                 if (occ->interrupted)
1191                         continue;
1192                 occ->interrupted = 1;
1193                 spin_unlock(&oig->oig_lock);
1194                 occ->occ_interrupted(occ);
1195                 spin_lock(&oig->oig_lock);
1196                 goto restart;
1197         }
1198         spin_unlock(&oig->oig_lock);
1199 }
1200
1201 int oig_wait(struct obd_io_group *oig)
1202 {
1203         struct l_wait_info lwi = LWI_INTR(interrupted_oig, oig);
1204         int rc;
1205
1206         CDEBUG(D_CACHE, "waiting for oig %p\n", oig);
1207
1208         do {
1209                 rc = l_wait_event(oig->oig_waitq, oig_done(oig), &lwi);
1210                 LASSERTF(rc == 0 || rc == -EINTR, "rc: %d\n", rc);
1211                 /* we can't continue until the oig has emptied and stopped
1212                  * referencing state that the caller will free upon return */
1213                 if (rc == -EINTR)
1214                         lwi = (struct l_wait_info){ 0, };
1215         } while (rc == -EINTR);
1216
1217         LASSERTF(oig->oig_pending == 0,
1218                  "exiting oig_wait(oig = %p) with %d pending\n", oig,
1219                  oig->oig_pending);
1220
1221         CDEBUG(D_CACHE, "done waiting on oig %p rc %d\n", oig, oig->oig_rc);
1222         return oig->oig_rc;
1223 }
1224 EXPORT_SYMBOL(oig_wait);
1225
1226 void class_fail_export(struct obd_export *exp)
1227 {
1228         int rc, already_failed;
1229
1230         spin_lock(&exp->exp_lock);
1231         already_failed = exp->exp_failed;
1232         exp->exp_failed = 1;
1233         spin_unlock(&exp->exp_lock);
1234
1235         if (already_failed) {
1236                 CDEBUG(D_HA, "disconnecting dead export %p/%s; skipping\n",
1237                        exp, exp->exp_client_uuid.uuid);
1238                 return;
1239         }
1240
1241         CDEBUG(D_HA, "disconnecting export %p/%s\n",
1242                exp, exp->exp_client_uuid.uuid);
1243
1244         if (obd_dump_on_timeout)
1245                 libcfs_debug_dumplog();
1246
1247         /* Most callers into obd_disconnect are removing their own reference
1248          * (request, for example) in addition to the one from the hash table.
1249          * We don't have such a reference here, so make one. */
1250         class_export_get(exp);
1251         rc = obd_disconnect(exp);
1252         if (rc)
1253                 CERROR("disconnecting export %p failed: %d\n", exp, rc);
1254         else
1255                 CDEBUG(D_HA, "disconnected export %p/%s\n",
1256                        exp, exp->exp_client_uuid.uuid);
1257 }
1258 EXPORT_SYMBOL(class_fail_export);
1259
1260 char *obd_export_nid2str(struct obd_export *exp)
1261 {
1262         if (exp->exp_connection != NULL)
1263                 return libcfs_nid2str(exp->exp_connection->c_peer.nid);
1264
1265         return "(no nid)";
1266 }
1267 EXPORT_SYMBOL(obd_export_nid2str);
1268
1269 #define EVICT_BATCH 32
1270 int obd_export_evict_by_nid(struct obd_device *obd, const char *nid)
1271 {
1272         struct obd_export *doomed_exp[EVICT_BATCH] = { NULL };
1273         struct list_head *p;
1274         int exports_evicted = 0, num_to_evict = 0, i;
1275
1276 search_again:
1277         spin_lock(&obd->obd_dev_lock);
1278         list_for_each(p, &obd->obd_exports) {
1279                 doomed_exp[num_to_evict] = list_entry(p, struct obd_export,
1280                                                       exp_obd_chain);
1281                 if (strcmp(obd_export_nid2str(doomed_exp[num_to_evict]),
1282                            nid) == 0) {
1283                         class_export_get(doomed_exp[num_to_evict]);
1284                         if (++num_to_evict == EVICT_BATCH)
1285                                 break;
1286                 }
1287         }
1288         spin_unlock(&obd->obd_dev_lock);
1289
1290         for (i = 0; i < num_to_evict; i++) {
1291                 exports_evicted++;
1292                 CWARN("%s: evict NID '%s' (%s) #%d at adminstrative request\n",
1293                        obd->obd_name, nid, doomed_exp[i]->exp_client_uuid.uuid,
1294                        exports_evicted);
1295                 class_fail_export(doomed_exp[i]);
1296                 class_export_put(doomed_exp[i]);
1297         }
1298         if (num_to_evict == EVICT_BATCH) {
1299                 num_to_evict = 0;
1300                 goto search_again;
1301         }
1302
1303         if (!exports_evicted)
1304                 CDEBUG(D_HA,"%s: can't disconnect NID '%s': no exports found\n",
1305                        obd->obd_name, nid);
1306         return exports_evicted;
1307 }
1308 EXPORT_SYMBOL(obd_export_evict_by_nid);
1309
1310 int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid)
1311 {
1312         struct obd_export *doomed_exp = NULL;
1313         struct list_head *p;
1314         struct obd_uuid doomed;
1315         int exports_evicted = 0;
1316
1317         obd_str2uuid(&doomed, uuid);
1318         if(obd_uuid_equals(&doomed, &obd->obd_uuid)) {
1319                 CERROR("%s: can't evict myself\n", obd->obd_name);
1320                 return exports_evicted;
1321         }
1322
1323         spin_lock(&obd->obd_dev_lock);
1324         list_for_each(p, &obd->obd_exports) {
1325                 doomed_exp = list_entry(p, struct obd_export, exp_obd_chain);
1326
1327                 if (obd_uuid_equals(&doomed, &doomed_exp->exp_client_uuid)) {
1328                         class_export_get(doomed_exp);
1329                         break;
1330                 }
1331                 doomed_exp = NULL;
1332         }
1333         spin_unlock(&obd->obd_dev_lock);
1334
1335         if (doomed_exp == NULL) {
1336                 CERROR("%s: can't disconnect %s: no exports found\n",
1337                        obd->obd_name, uuid);
1338         } else {
1339                 CWARN("%s: evicting %s at adminstrative request\n",
1340                        obd->obd_name, doomed_exp->exp_client_uuid.uuid);
1341                 class_fail_export(doomed_exp);
1342                 class_export_put(doomed_exp);
1343                 exports_evicted++;
1344         }
1345
1346         return exports_evicted;
1347 }
1348 EXPORT_SYMBOL(obd_export_evict_by_uuid);
1349
1350 void obd_zombie_impexp_cull(void)
1351 {
1352         struct obd_import *import;
1353         struct obd_export *export;
1354         ENTRY;
1355
1356         do {
1357                 spin_lock (&obd_zombie_impexp_lock);
1358
1359                 import = NULL;
1360                 if (!list_empty(&obd_zombie_imports)) {
1361                         import = list_entry(obd_zombie_imports.next,
1362                                             struct obd_import,
1363                                             imp_zombie_chain);
1364                         list_del(&import->imp_zombie_chain);
1365                 }
1366
1367                 export = NULL;
1368                 if (!list_empty(&obd_zombie_exports)) {
1369                         export = list_entry(obd_zombie_exports.next,
1370                                             struct obd_export,
1371                                             exp_obd_chain);
1372                         list_del_init(&export->exp_obd_chain);
1373                 }
1374
1375                 spin_unlock(&obd_zombie_impexp_lock);
1376
1377                 if (import != NULL)
1378                         class_import_destroy(import);
1379
1380                 if (export != NULL)
1381                         class_export_destroy(export);
1382
1383         } while (import != NULL || export != NULL);
1384         EXIT;
1385 }
1386 EXPORT_SYMBOL(obd_zombie_impexp_cull);
1387
1388 void obd_zombie_impexp_init(void)
1389 {
1390         INIT_LIST_HEAD(&obd_zombie_imports);
1391         INIT_LIST_HEAD(&obd_zombie_exports);
1392         spin_lock_init(&obd_zombie_impexp_lock);
1393 }
1394