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