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