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  2008 Sun Microsystems, Inc. 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  */
  42
  43 #ifndef EXPORT_SYMTAB
  44 # define EXPORT_SYMTAB
  45 #endif
  46 #define DEBUG_SUBSYSTEM S_MDS
  47
  48 #include <obd.h>
  49 #include <obd_class.h>
  50 #include <lustre_ver.h>
  51 #include <obd_support.h>
  52 #include <lustre_fid.h>
  53 #include "mdd_internal.h"
  54
  55 const char orph_index_name[] = "orphans";
  56
  57 static const struct dt_index_features orph_index_features = {
  58         .dif_flags       = DT_IND_UPDATE,
  59         .dif_keysize_min = sizeof(struct orph_key),
  60         .dif_keysize_max = sizeof(struct orph_key),
  61         .dif_recsize_min = sizeof(loff_t),
  62         .dif_recsize_max = sizeof(loff_t)
  63 };
  64
  65 enum {
  66         ORPH_OP_UNLINK,
  67         ORPH_OP_TRUNCATE
  68 };
  69
  70 static struct orph_key *orph_key_fill(const struct lu_env *env,
  71                                       const struct lu_fid *lf, __u32 op)
  72 {
  73         struct orph_key *key = &mdd_env_info(env)->mti_orph_key;
  74         LASSERT(key);
  75         fid_cpu_to_be(&key->ok_fid, lf);
  76         key->ok_op = cpu_to_be32(op);
  77         return key;
  78 }
  79
  80 static int orph_index_insert(const struct lu_env *env,
  81                              struct mdd_object *obj, __u32 op,
  82                              loff_t *offset, struct thandle *th)
  83 {
  84         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
  85         struct dt_object *dor = mdd->mdd_orphans;
  86         struct orph_key *key = orph_key_fill(env, mdo2fid(obj), op);
  87         int rc;
  88         ENTRY;
  89
  90         rc = dor->do_index_ops->dio_insert(env, dor, (struct dt_rec *)offset,
  91                                            (struct dt_key *)key, th,
  92                                            BYPASS_CAPA);
  93         RETURN(rc);
  94 }
  95
  96 static int orph_index_delete(const struct lu_env *env,
  97                              struct mdd_object *obj, __u32 op,
  98                              struct thandle *th)
  99 {
 100         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
 101         struct dt_object *dor = mdd->mdd_orphans;
 102         struct orph_key *key = orph_key_fill(env, mdo2fid(obj), op);
 103         int rc;
 104         ENTRY;
 105         LASSERT(dor);
 106         rc = dor->do_index_ops->dio_delete(env, dor,
 107                                            (struct dt_key *)key, th,
 108                                            BYPASS_CAPA);
 109         RETURN(rc);
 110
 111 }
 112
 113 static inline struct orph_key *orph_key_empty(const struct lu_env *env,
 114                                               __u32 op)
 115 {
 116         struct orph_key *key = &mdd_env_info(env)->mti_orph_key;
 117         LASSERT(key);
 118         fid_zero(&key->ok_fid);
 119         key->ok_op = cpu_to_be32(op);
 120         return key;
 121 }
 122
 123 static void orph_key_test_and_del(const struct lu_env *env,
 124                                   struct mdd_device *mdd,
 125                                   const struct orph_key *key)
 126 {
 127         struct mdd_object *mdo;
 128
 129         mdo = mdd_object_find(env, mdd, &key->ok_fid);
 130         if (IS_ERR(mdo))
 131                 CERROR("Invalid orphan!\n");
 132         else {
 133                 mdd_write_lock(env, mdo);
 134                 if (mdo->mod_count == 0) {
 135                         /* non-opened orphan, let's delete it */
 136                         struct md_attr *ma = &mdd_env_info(env)->mti_ma;
 137                         CWARN("Found orphan!\n");
 138                         mdd_object_kill(env, mdo, ma);
 139                         /* TODO: now handle OST objects */
 140                         //mdd_ost_objects_destroy(env, ma);
 141                         /* TODO: destroy index entry */
 142                 }
 143                 mdd_write_unlock(env, mdo);
 144                 mdd_object_put(env, mdo);
 145         }
 146 }
 147
 148 static int orph_index_iterate(const struct lu_env *env,
 149                               struct mdd_device *mdd)
 150 {
 151         struct dt_object *dt_obj = mdd->mdd_orphans;
 152         struct dt_it     *it;
 153         struct dt_it_ops *iops;
 154         struct orph_key  *key = orph_key_empty(env, 0);
 155         int result;
 156         ENTRY;
 157
 158         iops = &dt_obj->do_index_ops->dio_it;
 159         it = iops->init(env, dt_obj, 1, BYPASS_CAPA);
 160         if (it != NULL) {
 161                 result = iops->get(env, it, (const void *)key);
 162                 if (result > 0) {
 163                         int i;
 164                         /* main cycle */
 165                         for (result = 0, i = 0; result == +1; ++i) {
 166                                 key = (void *)iops->key(env, it);
 167                                 fid_be_to_cpu(&key->ok_fid, &key->ok_fid);
 168                                 orph_key_test_and_del(env, mdd, key);
 169                                 result = iops->next(env, it);
 170                         }
 171                 } else if (result == 0)
 172                         /* Index contains no zero key? */
 173                         result = -EIO;
 174
 175                 iops->put(env, it);
 176                 iops->fini(env, it);
 177         } else
 178                 result = -ENOMEM;
 179
 180         RETURN(result);
 181 }
 182
 183 int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
 184 {
 185         struct lu_fid fid;
 186         struct dt_object *d;
 187         int rc;
 188         ENTRY;
 189
 190         d = dt_store_open(env, mdd->mdd_child, orph_index_name, &fid);
 191         if (!IS_ERR(d)) {
 192                 mdd->mdd_orphans = d;
 193                 rc = d->do_ops->do_index_try(env, d, &orph_index_features);
 194                 if (rc == 0)
 195                         LASSERT(d->do_index_ops != NULL);
 196                 else
 197                         CERROR("\"%s\" is not an index!\n", orph_index_name);
 198         } else {
 199                 CERROR("cannot find \"%s\" obj %d\n",
 200                        orph_index_name, (int)PTR_ERR(d));
 201                 rc = PTR_ERR(d);
 202         }
 203
 204         RETURN(rc);
 205 }
 206
 207 void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
 208 {
 209         ENTRY;
 210         if (mdd->mdd_orphans != NULL) {
 211                 lu_object_put(env, &mdd->mdd_orphans->do_lu);
 212                 mdd->mdd_orphans = NULL;
 213         }
 214         EXIT;
 215 }
 216
 217 int __mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
 218 {
 219         return orph_index_iterate(env, d);
 220 }
 221
 222 int __mdd_orphan_add(const struct lu_env *env,
 223                      struct mdd_object *obj, struct thandle *th)
 224 {
 225         loff_t offset = 0;
 226         return orph_index_insert(env, obj, ORPH_OP_UNLINK, &offset, th);
 227 }
 228
 229 int __mdd_orphan_del(const struct lu_env *env,
 230                      struct mdd_object *obj, struct thandle *th)
 231 {
 232         return orph_index_delete(env, obj, ORPH_OP_UNLINK, th);
 233 }