lustre/obdclass/genops.c

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