lustre/mdd/mdd_orphans.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mdd/mdd_orphans.c
  37  *
  38  * Orphan handling code
  39  *
  40  * Author: Mike Pershin <tappro@clusterfs.com>
  41  *         Pravin B Shelar <pravin.shelar@sun.com>
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include <obd.h>
  50 #include <obd_class.h>
  51 #include <lustre_ver.h>
  52 #include <obd_support.h>
  53 #include <lustre_fid.h>
  54 #include "mdd_internal.h"
  55
  56 const char orph_index_name[] = "PENDING";
  57
  58 enum {
  59         ORPH_OP_UNLINK,
  60         ORPH_OP_TRUNCATE
  61 };
  62
  63 #define ORPHAN_FILE_NAME_FORMAT         "%016llx:%08x:%08x:%2x"
  64 #define ORPHAN_FILE_NAME_FORMAT_18      "%llx:%08x"
  65
  66 static struct dt_key* orph_key_fill(const struct lu_env *env,
  67                                     const struct lu_fid *lf, __u32 op)
  68 {
  69         char *key = mdd_env_info(env)->mti_orph_key;
  70         int rc;
  71
  72         LASSERT(key);
  73         rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT,
  74                       (long long unsigned int)fid_seq(lf),
  75                       fid_oid(lf), fid_ver(lf), op);
  76         if (rc > 0)
  77                 return (struct dt_key*) key;
  78         else
  79                 return ERR_PTR(rc);
  80 }
  81
  82 static struct dt_key* orph_key_fill_18(const struct lu_env *env,
  83                                        const struct lu_fid *lf)
  84 {
  85         char *key = mdd_env_info(env)->mti_orph_key;
  86         int rc;
  87
  88         LASSERT(key);
  89         rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT_18,
  90                       (unsigned long long)fid_seq(lf), fid_oid(lf));
  91         if (rc > 0)
  92                 return (struct dt_key*) key;
  93         else
  94                 return ERR_PTR(rc);
  95 }
  96
  97 static int orphan_key_to_fid(char *key, struct lu_fid *lf)
  98 {
  99         int rc = 0;
 100         unsigned int op;
 101
 102         rc = sscanf(key, ORPHAN_FILE_NAME_FORMAT,
 103                     (long long unsigned int *)&lf->f_seq, &lf->f_oid,
 104                     &lf->f_ver, &op);
 105         if (rc == 4)
 106                 return 0;
 107
 108         /* build igif */
 109         rc = sscanf(key, ORPHAN_FILE_NAME_FORMAT_18,
 110                     (long long unsigned int *)&lf->f_seq, &lf->f_oid);
 111         if (rc == 2) {
 112                 lf->f_ver = 0;
 113                 return 0;
 114         }
 115
 116         CERROR("can not parse orphan file name %s\n",key);
 117         return -EINVAL;
 118 }
 119
 120 static inline void mdd_orphan_write_lock(const struct lu_env *env,
 121                                     struct mdd_device *mdd)
 122 {
 123
 124         struct dt_object        *dor    = mdd->mdd_orphans;
 125         dor->do_ops->do_write_lock(env, dor, MOR_TGT_ORPHAN);
 126 }
 127
 128 static inline void mdd_orphan_write_unlock(const struct lu_env *env,
 129                                            struct mdd_device *mdd)
 130 {
 131
 132         struct dt_object        *dor    = mdd->mdd_orphans;
 133         dor->do_ops->do_write_unlock(env, dor);
 134 }
 135
 136 static inline int mdd_orphan_insert_obj(const struct lu_env *env,
 137                                         struct mdd_device *mdd,
 138                                         struct mdd_object *obj,
 139                                         __u32 op,
 140                                         struct thandle *th)
 141 {
 142         struct dt_object        *dor    = mdd->mdd_orphans;
 143         const struct lu_fid     *lf     = mdo2fid(obj);
 144         struct dt_key           *key    = orph_key_fill(env, lf, op);
 145         ENTRY;
 146
 147         return  dor->do_index_ops->dio_insert(env, dor,
 148                                               (struct dt_rec *)lf,
 149                                               key, th,
 150                                               BYPASS_CAPA, 1);
 151 }
 152
 153 static inline int mdd_orphan_delete_obj(const struct lu_env *env,
 154                                         struct mdd_device  *mdd ,
 155                                         struct dt_key *key,
 156                                         struct thandle *th)
 157 {
 158         struct dt_object        *dor    = mdd->mdd_orphans;
 159
 160         return  dor->do_index_ops->dio_delete(env, dor,
 161                                               key, th,
 162                                               BYPASS_CAPA);
 163 }
 164
 165 static inline void mdd_orphan_ref_add(const struct lu_env *env,
 166                                  struct mdd_device *mdd,
 167                                  struct thandle *th)
 168 {
 169         struct dt_object        *dor    = mdd->mdd_orphans;
 170         dor->do_ops->do_ref_add(env, dor, th);
 171 }
 172
 173 static inline void mdd_orphan_ref_del(const struct lu_env *env,
 174                                  struct mdd_device *mdd,
 175                                  struct thandle *th)
 176 {
 177         struct dt_object        *dor    = mdd->mdd_orphans;
 178         dor->do_ops->do_ref_del(env, dor, th);
 179 }
 180
 181
 182 static int orph_index_insert(const struct lu_env *env,
 183                              struct mdd_object *obj,
 184                              __u32 op,
 185                              struct thandle *th)
 186 {
 187         struct mdd_device       *mdd    = mdo2mdd(&obj->mod_obj);
 188         struct dt_object        *dor    = mdd->mdd_orphans;
 189         const struct lu_fid     *lf_dor = lu_object_fid(&dor->do_lu);
 190         struct dt_object        *next   = mdd_object_child(obj);
 191         const struct dt_key     *dotdot = (const struct dt_key *) "..";
 192         int rc;
 193         ENTRY;
 194
 195         LASSERT(mdd_write_locked(env, obj) != 0);
 196         LASSERT(!(obj->mod_flags & ORPHAN_OBJ));
 197         LASSERT(obj->mod_count > 0);
 198
 199         mdd_orphan_write_lock(env, mdd);
 200
 201         rc = mdd_orphan_insert_obj(env, mdd, obj, op, th);
 202         if (rc)
 203                 GOTO(out, rc);
 204
 205         mdo_ref_add(env, obj, th);
 206         if (!S_ISDIR(mdd_object_type(obj)))
 207                 goto out;
 208
 209         mdo_ref_add(env, obj, th);
 210         mdd_orphan_ref_add(env, mdd, th);
 211
 212         /* try best to fixup directory, dont return errors
 213          * from here */
 214         if (!dt_try_as_dir(env, next))
 215                 goto out;
 216         next->do_index_ops->dio_delete(env, next,
 217                                        dotdot, th, BYPASS_CAPA);
 218
 219         next->do_index_ops->dio_insert(env, next,
 220                                        (struct dt_rec *) lf_dor,
 221                                        dotdot, th, BYPASS_CAPA, 1);
 222
 223 out:
 224         if (rc == 0)
 225                 obj->mod_flags |= ORPHAN_OBJ;
 226
 227         mdd_orphan_write_unlock(env, mdd);
 228
 229         RETURN(rc);
 230 }
 231
 232 /**
 233  * destroy osd object on mdd and associated ost objects.
 234  *
 235  * \param obj orphan object
 236  * \param mdd used for sending llog msg to osts
 237  *
 238  * \retval  0   success
 239  * \retval -ve  error
 240  */
 241 static int orphan_object_kill(const struct lu_env *env,
 242                               struct mdd_object *obj,
 243                               struct mdd_device *mdd,
 244                               struct thandle *th)
 245 {
 246         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 247         int rc = 0;
 248         ENTRY;
 249
 250         /* No need to lock this object as its recovery phase, and
 251          * no other thread can access it. But we need to lock it
 252          * as its precondition for osd api we using. */
 253
 254         mdo_ref_del(env, obj, th);
 255         if (S_ISDIR(mdd_object_type(obj))) {
 256                 mdo_ref_del(env, obj, th);
 257                 mdd_orphan_ref_del(env, mdd, th);
 258         } else {
 259                 /* regular file , cleanup linked ost objects */
 260                 rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 261                 if (rc == 0)
 262                         rc = mdd_lov_destroy(env, mdd, obj, la);
 263         }
 264         RETURN(rc);
 265 }
 266
 267 static int orph_index_delete(const struct lu_env *env,
 268                              struct mdd_object *obj,
 269                              __u32 op,
 270                              struct thandle *th)
 271 {
 272         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
 273         struct dt_object *dor = mdd->mdd_orphans;
 274         struct dt_key *key;
 275         int rc;
 276
 277         ENTRY;
 278
 279         LASSERT(mdd_write_locked(env, obj) != 0);
 280         LASSERT(obj->mod_flags & ORPHAN_OBJ);
 281         LASSERT(obj->mod_count == 0);
 282
 283         LASSERT(dor);
 284
 285         key = orph_key_fill(env, mdo2fid(obj), op);
 286         mdd_orphan_write_lock(env, mdd);
 287
 288         rc = mdd_orphan_delete_obj(env, mdd, key, th);
 289
 290         if (rc == -ENOENT) {
 291                 key = orph_key_fill_18(env, mdo2fid(obj));
 292                 rc = mdd_orphan_delete_obj(env, mdd, key, th);
 293         }
 294
 295         if (!rc) {
 296                 /* lov objects will be destroyed by caller */
 297                 mdo_ref_del(env, obj, th);
 298                 if (S_ISDIR(mdd_object_type(obj))) {
 299                         mdo_ref_del(env, obj, th);
 300                         mdd_orphan_ref_del(env, mdd, th);
 301                 }
 302                 obj->mod_flags &= ~ORPHAN_OBJ;
 303         } else {
 304                 CERROR("could not delete object: rc = %d\n",rc);
 305         }
 306
 307         mdd_orphan_write_unlock(env, mdd);
 308         RETURN(rc);
 309 }
 310
 311
 312 static int orphan_object_destroy(const struct lu_env *env,
 313                                  struct mdd_object *obj,
 314                                  struct dt_key *key)
 315 {
 316         struct thandle *th = NULL;
 317         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
 318         int rc = 0;
 319         ENTRY;
 320
 321         mdd_txn_param_build(env, mdd, MDD_TXN_UNLINK_OP);
 322         th = mdd_trans_start(env, mdd);
 323         if (IS_ERR(th)) {
 324                 CERROR("Cannot get thandle\n");
 325                 RETURN(-ENOMEM);
 326         }
 327
 328         mdd_write_lock(env, obj, MOR_TGT_CHILD);
 329         if (likely(obj->mod_count == 0)) {
 330                 mdd_orphan_write_lock(env, mdd);
 331                 rc = mdd_orphan_delete_obj(env, mdd, key, th);
 332                 if (!rc)
 333                         orphan_object_kill(env, obj, mdd, th);
 334                 else
 335                         CERROR("could not delete object: rc = %d\n",rc);
 336                 mdd_orphan_write_unlock(env, mdd);
 337         }
 338         mdd_write_unlock(env, obj);
 339         mdd_trans_stop(env, mdd, 0, th);
 340
 341         RETURN(rc);
 342 }
 343
 344 static int orph_key_test_and_del(const struct lu_env *env,
 345                                  struct mdd_device *mdd,
 346                                  struct lu_fid *lf,
 347                                  struct dt_key *key)
 348 {
 349         struct mdd_object *mdo;
 350         int rc;
 351
 352         mdo = mdd_object_find(env, mdd, lf);
 353
 354         if (IS_ERR(mdo))
 355                 return PTR_ERR(mdo);
 356
 357         rc = -EBUSY;
 358         if (mdo->mod_count == 0) {
 359                 CWARN("Found orphan! Delete it\n");
 360                 rc = orphan_object_destroy(env, mdo, key);
 361         } else {
 362                 mdd_write_lock(env, mdo, MOR_TGT_CHILD);
 363                 if (likely(mdo->mod_count > 0)) {
 364                         CDEBUG(D_HA, "Found orphan, open count = %d\n",
 365                                mdo->mod_count);
 366                         mdo->mod_flags |= ORPHAN_OBJ;
 367                 }
 368                 mdd_write_unlock(env, mdo);
 369         }
 370
 371         mdd_object_put(env, mdo);
 372         return rc;
 373 }
 374
 375 static int orph_index_iterate(const struct lu_env *env,
 376                               struct mdd_device *mdd)
 377 {
 378         struct dt_object *dor = mdd->mdd_orphans;
 379         char             *mti_key = mdd_env_info(env)->mti_orph_key;
 380         const struct dt_it_ops *iops;
 381         struct dt_it     *it;
 382         char             *key;
 383         struct lu_fid     fid;
 384         int               result = 0;
 385         int               key_sz = 0;
 386         int               rc;
 387         __u64             cookie;
 388         ENTRY;
 389
 390         /* In recovery phase, do not need for any lock here */
 391
 392         iops = &dor->do_index_ops->dio_it;
 393         it = iops->init(env, dor, LUDA_64BITHASH, BYPASS_CAPA);
 394         if (!IS_ERR(it)) {
 395                 result = iops->load(env, it, 0);
 396                 if (result > 0) {
 397                         /* main cycle */
 398                         do {
 399
 400                                 key = (void *)iops->key(env, it);
 401                                 if (IS_ERR(key)) {
 402                                         CERROR("key failed when clean pending.\n");
 403                                         goto next;
 404                                 }
 405                                 key_sz = iops->key_size(env, it);
 406
 407                                 /* filter out "." and ".." entries from
 408                                  * PENDING dir. */
 409                                 if (key_sz < 8)
 410                                         goto next;
 411
 412                                 memcpy(mti_key, key, key_sz);
 413                                 mti_key[key_sz] = 0;
 414
 415                                 if (orphan_key_to_fid(mti_key, &fid))
 416                                         goto next;
 417                                 if (!fid_is_sane(&fid)) {
 418                                         CERROR("fid is not sane when clean pending.\n");
 419                                         goto next;
 420                                 }
 421
 422                                 /* kill orphan object */
 423                                 cookie =  iops->store(env, it);
 424                                 iops->put(env, it);
 425                                 rc = orph_key_test_and_del(env, mdd, &fid,
 426                                                 (struct dt_key *)mti_key);
 427
 428                                 /* after index delete reset iterator */
 429                                 if (!rc)
 430                                         result = iops->get(env, it,
 431                                                            (const void *)"");
 432                                 else
 433                                         result = iops->load(env, it, cookie);
 434 next:
 435                                 result = iops->next(env, it);
 436                         } while (result == 0);
 437                         result = 0;
 438                 } else if (result == 0) {
 439                         CERROR("Input/Output for clean pending.\n");
 440                         /* Index contains no zero key? */
 441                         result = -EIO;
 442                 }
 443                 iops->put(env, it);
 444                 iops->fini(env, it);
 445         } else {
 446                 result = PTR_ERR(it);
 447                 CERROR("Cannot clean pending (%d).\n", result);
 448         }
 449
 450         RETURN(result);
 451 }
 452
 453 int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
 454 {
 455         struct lu_fid fid;
 456         struct dt_object *d;
 457         int rc = 0;
 458         ENTRY;
 459
 460         d = dt_store_open(env, mdd->mdd_child, "", orph_index_name, &fid);
 461         if (!IS_ERR(d)) {
 462                 mdd->mdd_orphans = d;
 463                 if (!dt_try_as_dir(env, d)) {
 464                         rc = -ENOTDIR;
 465                         CERROR("\"%s\" is not an index! : rc = %d\n",
 466                                         orph_index_name, rc);
 467                 }
 468         } else {
 469                 CERROR("cannot find \"%s\" obj %d\n",
 470                        orph_index_name, (int)PTR_ERR(d));
 471                 rc = PTR_ERR(d);
 472         }
 473
 474         RETURN(rc);
 475 }
 476
 477 void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
 478 {
 479         ENTRY;
 480         if (mdd->mdd_orphans != NULL) {
 481                 lu_object_put(env, &mdd->mdd_orphans->do_lu);
 482                 mdd->mdd_orphans = NULL;
 483         }
 484         EXIT;
 485 }
 486
 487 /**
 488  *  Iterate orphan index to cleanup orphan objects in case of recovery.
 489  *  \param d   mdd device in recovery.
 490  *
 491  */
 492
 493 int __mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
 494 {
 495         return orph_index_iterate(env, d);
 496 }
 497
 498 /**
 499  *  delete an orphan \a obj from orphan index.
 500  *  \param obj file or directory.
 501  *  \param th  transaction for index insert.
 502  *
 503  *  \pre obj nlink == 0 && obj->mod_count != 0
 504  *
 505  *  \retval 0  success
 506  *  \retval  -ve index operation error.
 507  */
 508
 509 int __mdd_orphan_add(const struct lu_env *env,
 510                      struct mdd_object *obj, struct thandle *th)
 511 {
 512         return orph_index_insert(env, obj, ORPH_OP_UNLINK, th);
 513 }
 514
 515 /**
 516  *  delete an orphan \a obj from orphan index.
 517  *  \param obj file or directory.
 518  *  \param th  transaction for index deletion and object destruction.
 519  *
 520  *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
 521  *
 522  *  \retval 0  success
 523  *  \retval  -ve index operation error.
 524  */
 525
 526 int __mdd_orphan_del(const struct lu_env *env,
 527                      struct mdd_object *obj, struct thandle *th)
 528 {
 529         return orph_index_delete(env, obj, ORPH_OP_UNLINK, th);
 530 }