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  *  mdd/mdd_orphans.c
   5  *
   6  *  Orphan handling code
   7  *
   8  *  Copyright (C) 2006 Cluster File Systems, Inc.
   9  *   Author: Mike Pershin <tappro@clusterfs.com>
  10  *
  11  *   This file is part of the Lustre file system, http://www.lustre.org
  12  *   Lustre is a trademark of Cluster File Systems, Inc.
  13  *
  14  *   You may have signed or agreed to another license before downloading
  15  *   this software.  If so, you are bound by the terms and conditions
  16  *   of that agreement, and the following does not apply to you.  See the
  17  *   LICENSE file included with this distribution for more information.
  18  *
  19  *   If you did not agree to a different license, then this copy of Lustre
  20  *   is open source software; you can redistribute it and/or modify it
  21  *   under the terms of version 2 of the GNU General Public License as
  22  *   published by the Free Software Foundation.
  23  *
  24  *   In either case, Lustre is distributed in the hope that it will be
  25  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  26  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  27  *   license text for more details.
  28  */
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_MDS
  33
  34 #include <obd.h>
  35 #include <obd_class.h>
  36 #include <lustre_ver.h>
  37 #include <obd_support.h>
  38 #include <lustre_fid.h>
  39 #include "mdd_internal.h"
  40
  41 const char orph_index_name[] = "orphans";
  42
  43 static const struct dt_index_features orph_index_features = {
  44         .dif_flags       = DT_IND_UPDATE,
  45         .dif_keysize_min = sizeof(struct orph_key),
  46         .dif_keysize_max = sizeof(struct orph_key),
  47         .dif_recsize_min = sizeof(loff_t),
  48         .dif_recsize_max = sizeof(loff_t)
  49 };
  50
  51 enum {
  52         ORPH_OP_UNLINK,
  53         ORPH_OP_TRUNCATE
  54 };
  55
  56 static struct orph_key *orph_key_fill(const struct lu_env *env,
  57                                       const struct lu_fid *lf, __u32 op)
  58 {
  59         struct orph_key *key = &mdd_env_info(env)->mti_orph_key;
  60         LASSERT(key);
  61         fid_cpu_to_be(&key->ok_fid, lf);
  62         key->ok_op = cpu_to_be32(op);
  63         return key;
  64 }
  65
  66 static int orph_index_insert(const struct lu_env *env,
  67                              struct mdd_object *obj, __u32 op,
  68                              loff_t *offset, struct thandle *th)
  69 {
  70         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
  71         struct dt_object *dor = mdd->mdd_orphans;
  72         struct orph_key *key = orph_key_fill(env, mdo2fid(obj), op);
  73         int rc;
  74         ENTRY;
  75
  76         rc = dor->do_index_ops->dio_insert(env, dor, (struct dt_rec *)offset,
  77                                            (struct dt_key *)key, th,
  78                                            BYPASS_CAPA);
  79         RETURN(rc);
  80 }
  81
  82 static int orph_index_delete(const struct lu_env *env,
  83                              struct mdd_object *obj, __u32 op,
  84                              struct thandle *th)
  85 {
  86         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
  87         struct dt_object *dor = mdd->mdd_orphans;
  88         struct orph_key *key = orph_key_fill(env, mdo2fid(obj), op);
  89         int rc;
  90         ENTRY;
  91         LASSERT(dor);
  92         rc = dor->do_index_ops->dio_delete(env, dor,
  93                                            (struct dt_key *)key, th,
  94                                            BYPASS_CAPA);
  95         RETURN(rc);
  96
  97 }
  98
  99 static inline struct orph_key *orph_key_empty(const struct lu_env *env,
 100                                               __u32 op)
 101 {
 102         struct orph_key *key = &mdd_env_info(env)->mti_orph_key;
 103         LASSERT(key);
 104         fid_zero(&key->ok_fid);
 105         key->ok_op = cpu_to_be32(op);
 106         return key;
 107 }
 108
 109 static void orph_key_test_and_del(const struct lu_env *env,
 110                                   struct mdd_device *mdd,
 111                                   const struct orph_key *key)
 112 {
 113         struct mdd_object *mdo;
 114
 115         mdo = mdd_object_find(env, mdd, &key->ok_fid);
 116         if (IS_ERR(mdo))
 117                 CERROR("Invalid orphan!\n");
 118         else {
 119                 mdd_write_lock(env, mdo);
 120                 if (mdo->mod_count == 0) {
 121                         /* non-opened orphan, let's delete it */
 122                         struct md_attr *ma = &mdd_env_info(env)->mti_ma;
 123                         CWARN("Found orphan!\n");
 124                         mdd_object_kill(env, mdo, ma);
 125                         /* TODO: now handle OST objects */
 126                         //mdd_ost_objects_destroy(env, ma);
 127                         /* TODO: destroy index entry */
 128                 }
 129                 mdd_write_unlock(env, mdo);
 130                 mdd_object_put(env, mdo);
 131         }
 132 }
 133
 134 static int orph_index_iterate(const struct lu_env *env,
 135                               struct mdd_device *mdd)
 136 {
 137         struct dt_object *dt_obj = mdd->mdd_orphans;
 138         struct dt_it     *it;
 139         struct dt_it_ops *iops;
 140         struct orph_key  *key = orph_key_empty(env, 0);
 141         int result;
 142         ENTRY;
 143
 144         iops = &dt_obj->do_index_ops->dio_it;
 145         it = iops->init(env, dt_obj, 1, BYPASS_CAPA);
 146         if (it != NULL) {
 147                 result = iops->get(env, it, (const void *)key);
 148                 if (result > 0) {
 149                         int i;
 150                         /* main cycle */
 151                         for (result = 0, i = 0; result == +1; ++i) {
 152                                 key = (void *)iops->key(env, it);
 153                                 fid_be_to_cpu(&key->ok_fid, &key->ok_fid);
 154                                 orph_key_test_and_del(env, mdd, key);
 155                                 result = iops->next(env, it);
 156                         }
 157                 } else if (result == 0)
 158                         /* Index contains no zero key? */
 159                         result = -EIO;
 160
 161                 iops->put(env, it);
 162                 iops->fini(env, it);
 163         } else
 164                 result = -ENOMEM;
 165
 166         RETURN(result);
 167 }
 168
 169 int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
 170 {
 171         struct lu_fid fid;
 172         struct dt_object *d;
 173         int rc;
 174         ENTRY;
 175
 176         d = dt_store_open(env, mdd->mdd_child, orph_index_name, &fid);
 177         if (!IS_ERR(d)) {
 178                 mdd->mdd_orphans = d;
 179                 rc = d->do_ops->do_index_try(env, d, &orph_index_features);
 180                 if (rc == 0)
 181                         LASSERT(d->do_index_ops != NULL);
 182                 else
 183                         CERROR("\"%s\" is not an index!\n", orph_index_name);
 184         } else {
 185                 CERROR("cannot find \"%s\" obj %d\n",
 186                        orph_index_name, (int)PTR_ERR(d));
 187                 rc = PTR_ERR(d);
 188         }
 189
 190         RETURN(rc);
 191 }
 192
 193 void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
 194 {
 195         ENTRY;
 196         if (mdd->mdd_orphans != NULL) {
 197                 lu_object_put(env, &mdd->mdd_orphans->do_lu);
 198                 mdd->mdd_orphans = NULL;
 199         }
 200         EXIT;
 201 }
 202
 203 int __mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
 204 {
 205         return orph_index_iterate(env, d);
 206 }
 207
 208 int __mdd_orphan_add(const struct lu_env *env,
 209                      struct mdd_object *obj, struct thandle *th)
 210 {
 211         loff_t offset = 0;
 212         return orph_index_insert(env, obj, ORPH_OP_UNLINK, &offset, th);
 213 }
 214
 215 int __mdd_orphan_del(const struct lu_env *env,
 216                      struct mdd_object *obj, struct thandle *th)
 217 {
 218         return orph_index_delete(env, obj, ORPH_OP_UNLINK, th);
 219 }