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