lustre/cmm/cmm_split.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/cmm/cmm_split.c
   5  *  Lustre splitting dir
   6  *
   7  *  Copyright (c) 2006 Cluster File Systems, Inc.
   8  *   Author: Alex thomas <alex@clusterfs.com>
   9  *           Wang Di     <wangdi@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
  30 #ifndef EXPORT_SYMTAB
  31 # define EXPORT_SYMTAB
  32 #endif
  33
  34 #define DEBUG_SUBSYSTEM S_MDS
  35
  36 #include <obd_class.h>
  37 #include <lustre_fid.h>
  38 #include <lustre_mds.h>
  39 #include "cmm_internal.h"
  40 #include "mdc_internal.h"
  41
  42 struct cmm_thread_info {
  43         struct md_attr   cti_ma;
  44 };
  45
  46 struct lu_context_key cmm_thread_key;
  47 struct cmm_thread_info *cmm_ctx_info(const struct lu_context *ctx)
  48 {
  49         struct cmm_thread_info *info;
  50
  51         info = lu_context_key_get(ctx, &cmm_thread_key);
  52         LASSERT(info != NULL);
  53         return info;
  54 }
  55
  56 #define CMM_NO_SPLIT_EXPECTED   0
  57 #define CMM_EXPECT_SPLIT        1
  58 #define CMM_NO_SPLITTABLE       2
  59
  60 #define SPLIT_SIZE 64*1024
  61
  62 static int cmm_expect_splitting(const struct lu_context *ctx,
  63                                 struct md_object *mo, struct md_attr *ma)
  64 {
  65         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
  66         ENTRY;
  67
  68         if (cmm->cmm_tgt_count == 1)
  69                 RETURN(CMM_NO_SPLIT_EXPECTED);
  70
  71         if (ma->ma_attr.la_size < SPLIT_SIZE)
  72                 RETURN(CMM_NO_SPLIT_EXPECTED);
  73
  74         if (ma->ma_lmv_size)
  75                 RETURN(CMM_NO_SPLIT_EXPECTED);
  76
  77         RETURN(CMM_EXPECT_SPLIT);
  78 }
  79
  80 static inline struct lu_fid* cmm2_fid(struct cmm_object *obj)
  81 {
  82        return &(obj->cmo_obj.mo_lu.lo_header->loh_fid);
  83 }
  84
  85 #define cmm_md_size(stripes)                            \
  86        (sizeof(struct lmv_stripe_md) + stripes * sizeof(struct lu_fid))
  87
  88 static int cmm_alloc_fid(const struct lu_context *ctx, struct cmm_device *cmm,
  89                          struct lu_fid *fid, int count)
  90 {
  91         struct  mdc_device *mc, *tmp;
  92         int rc = 0, i = 0;
  93
  94         LASSERT(count == cmm->cmm_tgt_count);
  95
  96         /* FIXME: this spin_lock maybe not proper,
  97          * because fid_alloc may need RPC */
  98         spin_lock(&cmm->cmm_tgt_guard);
  99         list_for_each_entry_safe(mc, tmp, &cmm->cmm_targets,
 100                                  mc_linkage) {
 101                 rc = obd_fid_alloc(mc->mc_desc.cl_exp, &fid[i++], NULL);
 102                 if (rc) {
 103                         spin_unlock(&cmm->cmm_tgt_guard);
 104                         RETURN(rc);
 105                 }
 106         }
 107         spin_unlock(&cmm->cmm_tgt_guard);
 108         RETURN(rc);
 109 }
 110
 111 struct cmm_object *cmm_object_find(const struct lu_context *ctxt,
 112                                    struct cmm_device *d,
 113                                    const struct lu_fid *f)
 114 {
 115         struct lu_object *o;
 116         struct cmm_object *m;
 117         ENTRY;
 118
 119         o = lu_object_find(ctxt, d->cmm_md_dev.md_lu_dev.ld_site, f);
 120         if (IS_ERR(o))
 121                 m = (struct cmm_object *)o;
 122         else
 123                 m = lu2cmm_obj(lu_object_locate(o->lo_header,
 124                                d->cmm_md_dev.md_lu_dev.ld_type));
 125         RETURN(m);
 126 }
 127
 128 static int cmm_creat_remote_obj(const struct lu_context *ctx,
 129                                 struct cmm_device *cmm,
 130                                 struct lu_fid *fid, struct md_attr *ma)
 131 {
 132         struct cmm_object *obj;
 133         struct md_create_spec *spec;
 134         int rc;
 135         ENTRY;
 136
 137         obj = cmm_object_find(ctx, cmm, fid);
 138         if (IS_ERR(obj))
 139                 RETURN(PTR_ERR(obj));
 140
 141         OBD_ALLOC_PTR(spec);
 142         rc = mo_object_create(ctx, md_object_next(&obj->cmo_obj),
 143                               spec, ma);
 144         OBD_FREE_PTR(spec);
 145
 146         RETURN(0);
 147 }
 148
 149 static int cmm_create_slave_objects(const struct lu_context *ctx,
 150                                     struct md_object *mo, struct md_attr *ma)
 151 {
 152         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 153         struct lmv_stripe_md *lmv = NULL;
 154         int lmv_size, i, rc;
 155         struct lu_fid *lf = cmm2_fid(md2cmm_obj(mo));
 156         ENTRY;
 157
 158         lmv_size = cmm_md_size(cmm->cmm_tgt_count + 1);
 159
 160         OBD_ALLOC(lmv, lmv_size);
 161         if (!lmv)
 162                 RETURN(-ENOMEM);
 163
 164         lmv->mea_master = -1;
 165         lmv->mea_magic = MEA_MAGIC_ALL_CHARS;
 166         lmv->mea_count = cmm->cmm_tgt_count + 1;
 167
 168         lmv->mea_ids[0] = *lf;
 169
 170         rc = cmm_alloc_fid(ctx, cmm, &lmv->mea_ids[1], cmm->cmm_tgt_count);
 171         if (rc)
 172                 GOTO(cleanup, rc);
 173
 174         for (i = 0; i < cmm->cmm_tgt_count; i ++) {
 175                 rc = cmm_creat_remote_obj(ctx, cmm, &lmv->mea_ids[i], ma);
 176                 if (rc)
 177                         GOTO(cleanup, rc);
 178         }
 179
 180         rc = mo_xattr_set(ctx, md_object_next(mo), lmv, lmv_size,
 181                           MDS_LMV_MD_NAME, 0);
 182 cleanup:
 183         OBD_FREE(lmv, lmv_size);
 184         RETURN(rc);
 185 }
 186
 187 static int cmm_send_split_pages(const struct lu_context *ctx,
 188                                 struct md_object *mo, struct lu_rdpg *rdpg)
 189 {
 190         RETURN(0);
 191 }
 192
 193 static int cmm_split_entries(const struct lu_context *ctx, struct md_object *mo,
 194                              struct lu_rdpg *rdpg)
 195 {
 196         struct lu_dirpage *dp;
 197         __u32 hash_end;
 198         int rc;
 199         ENTRY;
 200
 201         do {
 202                 rc = mo_readpage(ctx, md_object_next(mo), rdpg);
 203                 if (rc)
 204                         RETURN(rc);
 205
 206                 rc = cmm_send_split_pages(ctx, mo, rdpg);
 207                 if (rc)
 208                         RETURN(rc);
 209
 210                 dp = kmap(rdpg->rp_pages[0]);
 211                 hash_end = dp->ldp_hash_end;
 212                 kunmap(rdpg->rp_pages[0]);
 213                 if (hash_end == ~0ul)
 214                         break;
 215         } while (hash_end < rdpg->rp_hash_end);
 216
 217         RETURN(rc);
 218 }
 219
 220 static int cmm_remove_entries(const struct lu_context *ctx,
 221                               struct md_object *mo, __u32 start, __u32 end)
 222 {
 223         RETURN(0);
 224 }
 225 #define MAX_HASH_SIZE 0x3fffffff
 226 #define SPLIT_PAGE_COUNT 1
 227
 228 static int cmm_scan_and_split(const struct lu_context *ctx,
 229                               struct md_object *mo, struct md_attr *ma)
 230 {
 231         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 232         __u32 hash_segement;
 233         struct lu_rdpg   *rdpg = NULL;
 234         int rc = 0, i;
 235
 236         OBD_ALLOC_PTR(rdpg);
 237         if (!rdpg)
 238                 RETURN(-ENOMEM);
 239
 240         rdpg->rp_npages = SPLIT_PAGE_COUNT;
 241         rdpg->rp_count  = CFS_PAGE_SIZE * rdpg->rp_npages;
 242
 243         OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
 244         if (rdpg->rp_pages == NULL)
 245                 GOTO(free_rdpg, rc = -ENOMEM);
 246
 247         for (i = 0; i < rdpg->rp_npages; i++) {
 248                 rdpg->rp_pages[i] = alloc_pages(GFP_KERNEL, 0);
 249                 if (rdpg->rp_pages[i] == NULL)
 250                         GOTO(cleanup, rc = -ENOMEM);
 251         }
 252
 253         hash_segement = MAX_HASH_SIZE / cmm->cmm_tgt_count;
 254         for (i = 1; i < cmm->cmm_tgt_count; i++) {
 255                 rdpg->rp_hash = i * hash_segement;
 256                 rdpg->rp_hash_end = rdpg->rp_hash + hash_segement;
 257                 rc = cmm_split_entries(ctx, mo, rdpg);
 258                 if (rc)
 259                         GOTO(cleanup, rc);
 260                 rc = cmm_remove_entries(ctx, mo, rdpg->rp_hash,
 261                                         rdpg->rp_hash_end);
 262                 if (rc)
 263                         GOTO(cleanup, rc);
 264         }
 265 cleanup:
 266         for (i = 0; i < rdpg->rp_npages; i++)
 267                 if (rdpg->rp_pages[i] != NULL)
 268                         __free_pages(rdpg->rp_pages[i], 0);
 269         if (rdpg->rp_pages)
 270                 OBD_FREE(rdpg->rp_pages, rdpg->rp_npages *
 271                                          sizeof rdpg->rp_pages[0]);
 272 free_rdpg:
 273         if (rdpg)
 274                 OBD_FREE_PTR(rdpg);
 275
 276         RETURN(rc);
 277 }
 278
 279 int cml_try_to_split(const struct lu_context *ctx, struct md_object *mo)
 280 {
 281         struct md_attr *ma;
 282         int rc = 0;
 283         ENTRY;
 284
 285         LASSERT(S_ISDIR(lu_object_attr(&mo->mo_lu)));
 286
 287         OBD_ALLOC_PTR(ma);
 288         if (ma == NULL)
 289                 RETURN(-ENOMEM);
 290
 291         ma->ma_need = MA_INODE;
 292         rc = mo_attr_get(ctx, mo, ma);
 293         if (rc)
 294                 GOTO(cleanup, ma);
 295
 296         /* step1: checking whether the dir need to be splitted */
 297         rc = cmm_expect_splitting(ctx, mo, ma);
 298         if (rc != CMM_EXPECT_SPLIT)
 299                 GOTO(cleanup, rc = 0);
 300
 301         /* step2: create slave objects */
 302         rc = cmm_create_slave_objects(ctx, mo, ma);
 303         if (rc)
 304                 GOTO(cleanup, ma);
 305
 306         /* step3: scan and split the object */
 307         rc = cmm_scan_and_split(ctx, mo, ma);
 308 cleanup:
 309         OBD_FREE_PTR(ma);
 310
 311         RETURN(rc);
 312 }