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  *           Yury Umanets <umka@clusterfs.com>
  11  *
  12  *   This file is part of the Lustre file system, http://www.lustre.org
  13  *   Lustre is a trademark of Cluster File Systems, Inc.
  14  *
  15  *   You may have signed or agreed to another license before downloading
  16  *   this software.  If so, you are bound by the terms and conditions
  17  *   of that agreement, and the following does not apply to you.  See the
  18  *   LICENSE file included with this distribution for more information.
  19  *
  20  *   If you did not agree to a different license, then this copy of Lustre
  21  *   is open source software; you can redistribute it and/or modify it
  22  *   under the terms of version 2 of the GNU General Public License as
  23  *   published by the Free Software Foundation.
  24  *
  25  *   In either case, Lustre is distributed in the hope that it will be
  26  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  27  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  28  *   license text for more details.
  29  */
  30
  31 #ifndef EXPORT_SYMTAB
  32 # define EXPORT_SYMTAB
  33 #endif
  34
  35 #define DEBUG_SUBSYSTEM S_MDS
  36
  37 #include <obd_class.h>
  38 #include <lustre_fid.h>
  39 #include <lustre_mds.h>
  40 #include <lustre/lustre_idl.h>
  41 #include "cmm_internal.h"
  42 #include "mdc_internal.h"
  43
  44 enum {
  45         CMM_SPLIT_SIZE =  64 * 1024
  46 };
  47
  48 /*
  49  * This function checks if passed @name come to correct server (local MDT). If
  50  * not - return -ERESTART and let client know that dir was split and client
  51  * needs to chose correct stripe.
  52  */
  53 int cmm_split_check(const struct lu_env *env, struct md_object *mp,
  54                     const char *name)
  55 {
  56         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mp));
  57         struct md_attr *ma = &cmm_env_info(env)->cmi_ma;
  58         struct cml_object *clo = md2cml_obj(mp);
  59         struct timeval start;
  60         int rc, lmv_size;
  61         ENTRY;
  62
  63         cmm_lprocfs_time_start(cmm, &start, LPROC_CMM_SPLIT_CHECK);
  64
  65         /* Not split yet */
  66         if (clo->clo_split == CMM_SPLIT_NONE ||
  67             clo->clo_split == CMM_SPLIT_DENIED)
  68                 GOTO(out, rc = 0);
  69
  70         lmv_size = CMM_MD_SIZE(cmm->cmm_tgt_count + 1);
  71
  72         /* Try to get the LMV EA */
  73         memset(ma, 0, sizeof(*ma));
  74
  75         ma->ma_need = MA_LMV;
  76         ma->ma_lmv_size = lmv_size;
  77         OBD_ALLOC(ma->ma_lmv, lmv_size);
  78         if (ma->ma_lmv == NULL)
  79                 GOTO(out, rc = -ENOMEM);
  80
  81         /* Get LMV EA, Note: refresh valid here for getting LMV_EA */
  82         rc = mo_attr_get(env, mp, ma);
  83         if (rc)
  84                 GOTO(cleanup, rc);
  85
  86         /* No LMV just return */
  87         if (!(ma->ma_valid & MA_LMV)) {
  88                 /* update split state if unknown */
  89                 if (clo->clo_split == CMM_SPLIT_UNKNOWN)
  90                         clo->clo_split = CMM_SPLIT_NONE;
  91                 GOTO(cleanup, rc = 0);
  92         }
  93
  94         /* Skip checking the slave dirs (mea_count is 0) */
  95         if (ma->ma_lmv->mea_count != 0) {
  96                 int idx;
  97
  98                 /*
  99                  * Get stripe by name to check the name belongs to master dir,
 100                  * otherwise return the -ERESTART
 101                  */
 102                 idx = mea_name2idx(ma->ma_lmv, name, strlen(name));
 103
 104                 /*
 105                  * Check if name came to correct MDT server. We suppose that if
 106                  * client does not know about split, it sends create operation
 107                  * to master MDT. And this is master job to say it that dir got
 108                  * split and client should orward request to correct MDT. This
 109                  * is why we check here if stripe zero or not. Zero stripe means
 110                  * master stripe. If stripe calculated from name is not zero -
 111                  * return -ERESTART.
 112                  */
 113                 if (idx != 0)
 114                         rc = -ERESTART;
 115
 116                 /* update split state to DONE if unknown */
 117                 if (clo->clo_split == CMM_SPLIT_UNKNOWN)
 118                         clo->clo_split = CMM_SPLIT_DONE;
 119         } else {
 120                 /* split is denied for slave dir */
 121                 clo->clo_split = CMM_SPLIT_DENIED;
 122         }
 123         EXIT;
 124 cleanup:
 125         OBD_FREE(ma->ma_lmv, lmv_size);
 126 out:
 127         cmm_lprocfs_time_end(cmm, &start, LPROC_CMM_SPLIT_CHECK);
 128         return rc;
 129 }
 130
 131 /*
 132  * Return preferable access mode to caller taking into account possible split
 133  * and the fact of existing not splittable dirs in principle.
 134  */
 135 int cmm_split_access(const struct lu_env *env, struct md_object *mo,
 136                      mdl_mode_t lm)
 137 {
 138         struct md_attr *ma = &cmm_env_info(env)->cmi_ma;
 139         int rc, split;
 140         ENTRY;
 141
 142         memset(ma, 0, sizeof(*ma));
 143
 144         /*
 145          * Check only if we need protection from split.  If not - mdt handles
 146          * other cases.
 147          */
 148         rc = cmm_split_expect(env, mo, ma, &split);
 149         if (rc) {
 150                 CERROR("Can't check for possible split, rc %d\n", rc);
 151                 RETURN(MDL_MINMODE);
 152         }
 153
 154         /*
 155          * Do not take PDO lock on non-splittable objects if this is not PW,
 156          * this should speed things up a bit.
 157          */
 158         if (split == CMM_SPLIT_DONE && lm != MDL_PW)
 159                 RETURN(MDL_NL);
 160
 161         /* Protect splitting by exclusive lock. */
 162         if (split == CMM_SPLIT_NEEDED && lm == MDL_PW)
 163                 RETURN(MDL_EX);
 164
 165         /*
 166          * Have no idea about lock mode, let it be what higher layer wants.
 167          */
 168         RETURN(MDL_MINMODE);
 169 }
 170
 171 /* Check if split is expected for current thread. */
 172 int cmm_split_expect(const struct lu_env *env, struct md_object *mo,
 173                      struct md_attr *ma, int *split)
 174 {
 175         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 176         struct cml_object *clo = md2cml_obj(mo);
 177         struct lu_fid root_fid;
 178         int rc;
 179         ENTRY;
 180
 181         if (clo->clo_split == CMM_SPLIT_DONE ||
 182             clo->clo_split == CMM_SPLIT_DENIED) {
 183                 *split = clo->clo_split;
 184                 RETURN(0);
 185         }
 186         /* CMM_SPLIT_UNKNOWN case below */
 187
 188         /* No need to split root object. */
 189         rc = cmm_child_ops(cmm)->mdo_root_get(env, cmm->cmm_child,
 190                                               &root_fid);
 191         if (rc)
 192                 RETURN(rc);
 193
 194         if (lu_fid_eq(&root_fid, cmm2fid(md2cmm_obj(mo)))) {
 195                 /* update split state */
 196                 *split = clo->clo_split == CMM_SPLIT_DENIED;
 197                 RETURN(0);
 198         }
 199
 200         /*
 201          * Assumption: ma_valid = 0 here, we only need get inode and lmv_size
 202          * for this get_attr.
 203          */
 204         LASSERT(ma->ma_valid == 0);
 205         ma->ma_need = MA_INODE | MA_LMV;
 206         rc = mo_attr_get(env, mo, ma);
 207         if (rc)
 208                 RETURN(rc);
 209
 210         /* No need split for already split object */
 211         if (ma->ma_valid & MA_LMV) {
 212                 LASSERT(ma->ma_lmv_size > 0);
 213                 *split = clo->clo_split = CMM_SPLIT_DONE;
 214                 RETURN(0);
 215         }
 216
 217         /* No need split for object whose size < CMM_SPLIT_SIZE */
 218         if (ma->ma_attr.la_size < CMM_SPLIT_SIZE) {
 219                 *split = clo->clo_split = CMM_SPLIT_NONE;
 220                 RETURN(0);
 221         }
 222
 223         *split = clo->clo_split = CMM_SPLIT_NEEDED;
 224         RETURN(0);
 225 }
 226
 227 struct cmm_object *cmm_object_find(const struct lu_env *env,
 228                                    struct cmm_device *d,
 229                                    const struct lu_fid *f)
 230 {
 231         struct lu_object *o;
 232         struct cmm_object *m;
 233         ENTRY;
 234
 235         o = lu_object_find(env, d->cmm_md_dev.md_lu_dev.ld_site, f);
 236         if (IS_ERR(o))
 237                 m = (struct cmm_object *)o;
 238         else
 239                 m = lu2cmm_obj(lu_object_locate(o->lo_header,
 240                                d->cmm_md_dev.md_lu_dev.ld_type));
 241         RETURN(m);
 242 }
 243
 244 static inline void cmm_object_put(const struct lu_env *env,
 245                                   struct cmm_object *o)
 246 {
 247         lu_object_put(env, &o->cmo_obj.mo_lu);
 248 }
 249
 250 /*
 251  * Allocate new on passed @mc for slave object which is going to create there
 252  * soon.
 253  */
 254 static int cmm_split_fid_alloc(const struct lu_env *env,
 255                                struct cmm_device *cmm,
 256                                struct mdc_device *mc,
 257                                struct lu_fid *fid)
 258 {
 259         int rc;
 260         ENTRY;
 261
 262         LASSERT(cmm != NULL && mc != NULL && fid != NULL);
 263
 264         down(&mc->mc_fid_sem);
 265
 266         /* Alloc new fid on @mc. */
 267         rc = obd_fid_alloc(mc->mc_desc.cl_exp, fid, NULL);
 268         if (rc > 0) {
 269                 /* Setup FLD for new sequenceif needed. */
 270                 rc = fld_client_create(cmm->cmm_fld, fid_seq(fid),
 271                                        mc->mc_num, env);
 272                 if (rc)
 273                         CERROR("Can't create fld entry, rc %d\n", rc);
 274         }
 275         up(&mc->mc_fid_sem);
 276
 277         RETURN(rc);
 278 }
 279
 280 /* Allocate new slave object on passed @mc */
 281 static int cmm_split_slave_create(const struct lu_env *env,
 282                                   struct cmm_device *cmm,
 283                                   struct mdc_device *mc,
 284                                   struct lu_fid *fid,
 285                                   struct md_attr *ma,
 286                                   struct lmv_stripe_md *lmv,
 287                                   int lmv_size)
 288 {
 289         struct md_op_spec *spec = &cmm_env_info(env)->cmi_spec;
 290         struct cmm_object *obj;
 291         int rc;
 292         ENTRY;
 293
 294         /* Allocate new fid and store it to @fid */
 295         rc = cmm_split_fid_alloc(env, cmm, mc, fid);
 296         if (rc) {
 297                 CERROR("Can't alloc new fid on "LPU64
 298                        ", rc %d\n", mc->mc_num, rc);
 299                 RETURN(rc);
 300         }
 301
 302         /* Allocate new object on @mc */
 303         obj = cmm_object_find(env, cmm, fid);
 304         if (IS_ERR(obj))
 305                 RETURN(PTR_ERR(obj));
 306
 307         memset(spec, 0, sizeof *spec);
 308         spec->u.sp_ea.fid = fid;
 309         spec->u.sp_ea.eadata = lmv;
 310         spec->u.sp_ea.eadatalen = lmv_size;
 311         spec->sp_cr_flags |= MDS_CREATE_SLAVE_OBJ;
 312         rc = mo_object_create(env, md_object_next(&obj->cmo_obj),
 313                               spec, ma);
 314         cmm_object_put(env, obj);
 315         RETURN(rc);
 316 }
 317
 318 /*
 319  * Create so many slaves as number of stripes. This is called in split time
 320  * before sending pages to slaves.
 321  */
 322 static int cmm_split_slaves_create(const struct lu_env *env,
 323                                    struct md_object *mo,
 324                                    struct md_attr *ma)
 325 {
 326         struct cmm_device    *cmm = cmm_obj2dev(md2cmm_obj(mo));
 327         struct lu_fid        *lf  = cmm2fid(md2cmm_obj(mo));
 328         struct lmv_stripe_md *slave_lmv = &cmm_env_info(env)->cmi_lmv;
 329         struct mdc_device    *mc, *tmp;
 330         struct lmv_stripe_md *lmv;
 331         int i = 1, rc = 0;
 332         ENTRY;
 333
 334         /* Init the split MEA */
 335         lmv = ma->ma_lmv;
 336         lmv->mea_master = cmm->cmm_local_num;
 337         lmv->mea_magic = MEA_MAGIC_HASH_SEGMENT;
 338         lmv->mea_count = cmm->cmm_tgt_count + 1;
 339
 340         /*
 341          * Store master FID to local node idx number. Local node is always
 342          * master and its stripe number if 0.
 343          */
 344         lmv->mea_ids[0] = *lf;
 345
 346         memset(slave_lmv, 0, sizeof *slave_lmv);
 347         slave_lmv->mea_master = cmm->cmm_local_num;
 348         slave_lmv->mea_magic = MEA_MAGIC_HASH_SEGMENT;
 349         slave_lmv->mea_count = 0;
 350
 351         list_for_each_entry_safe(mc, tmp, &cmm->cmm_targets, mc_linkage) {
 352                 rc = cmm_split_slave_create(env, cmm, mc, &lmv->mea_ids[i],
 353                                             ma, slave_lmv, sizeof(*slave_lmv));
 354                 if (rc)
 355                         GOTO(cleanup, rc);
 356                 i++;
 357         }
 358
 359         ma->ma_valid |= MA_LMV;
 360         EXIT;
 361 cleanup:
 362         return rc;
 363 }
 364
 365 static inline int cmm_split_special_entry(struct lu_dirent *ent)
 366 {
 367         if (!strncmp(ent->lde_name, ".", le16_to_cpu(ent->lde_namelen)) ||
 368             !strncmp(ent->lde_name, "..", le16_to_cpu(ent->lde_namelen)))
 369                 return 1;
 370         return 0;
 371 }
 372
 373 /*
 374  * Remove one entry from local MDT. Do not corrupt byte order in page, it will
 375  * be sent to remote MDT.
 376  */
 377 static int cmm_split_remove_entry(const struct lu_env *env,
 378                                   struct md_object *mo,
 379                                   struct lu_dirent *ent)
 380 {
 381         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 382         struct cmm_object *obj;
 383         int is_dir, rc;
 384         char *name;
 385         ENTRY;
 386
 387         if (cmm_split_special_entry(ent))
 388                 RETURN(0);
 389
 390         fid_le_to_cpu(&cmm_env_info(env)->cmi_fid, &ent->lde_fid);
 391         obj = cmm_object_find(env, cmm, &cmm_env_info(env)->cmi_fid);
 392         if (IS_ERR(obj))
 393                 RETURN(PTR_ERR(obj));
 394
 395         if (lu_object_exists(&obj->cmo_obj.mo_lu) > 0)
 396                 is_dir = S_ISDIR(lu_object_attr(&obj->cmo_obj.mo_lu));
 397         else
 398                 /*
 399                  * XXX: These days only cross-ref dirs are possible, so for the
 400                  * sake of simplicity, in split, we suppose that all cross-ref
 401                  * names pint to directory and do not do additional getattr to
 402                  * remote MDT.
 403                  */
 404                 is_dir = 1;
 405
 406         OBD_ALLOC(name, le16_to_cpu(ent->lde_namelen) + 1);
 407         if (!name)
 408                 GOTO(cleanup, rc = -ENOMEM);
 409
 410         memcpy(name, ent->lde_name, le16_to_cpu(ent->lde_namelen));
 411         /* No permission check for name_remove when split */
 412         rc = mdo_name_remove(env, md_object_next(mo),
 413                              name, is_dir);
 414         OBD_FREE(name, le16_to_cpu(ent->lde_namelen) + 1);
 415         if (rc)
 416                 GOTO(cleanup, rc);
 417
 418         /*
 419          * This @ent will be transferred to slave MDS and insert there, so in
 420          * the slave MDS, we should know whether this object is dir or not, so
 421          * use the highest bit of the hash to indicate that (because we do not
 422          * use highest bit of hash).
 423          */
 424         if (is_dir) {
 425                 ent->lde_hash = le32_to_cpu(ent->lde_hash);
 426                 ent->lde_hash = cpu_to_le32(ent->lde_hash | MAX_HASH_HIGHEST_BIT);
 427         }
 428         EXIT;
 429 cleanup:
 430         cmm_object_put(env, obj);
 431         return rc;
 432 }
 433
 434 /*
 435  * Remove all entries from passed page. These entries are going to remote MDT
 436  * and thus should be removed locally.
 437  */
 438 static int cmm_split_remove_page(const struct lu_env *env,
 439                                  struct md_object *mo,
 440                                  struct lu_rdpg *rdpg,
 441                                  __u32 hash_end, __u32 *len)
 442 {
 443         struct lu_dirpage *dp;
 444         struct lu_dirent  *ent;
 445         int rc = 0;
 446         ENTRY;
 447
 448         *len = 0;
 449         kmap(rdpg->rp_pages[0]);
 450         dp = page_address(rdpg->rp_pages[0]);
 451         for (ent = lu_dirent_start(dp);
 452              ent != NULL && le32_to_cpu(ent->lde_hash) < hash_end;
 453              ent = lu_dirent_next(ent)) {
 454                 rc = cmm_split_remove_entry(env, mo, ent);
 455                 if (rc) {
 456                         /*
 457                          * XXX: Error handler to insert remove name back,
 458                          * currently we assumed it will success anyway in
 459                          * verfication test.
 460                          */
 461                         CERROR("Can not del %*.*s, rc %d\n",
 462                                le16_to_cpu(ent->lde_namelen),
 463                                le16_to_cpu(ent->lde_namelen),
 464                                ent->lde_name, rc);
 465                         GOTO(unmap, rc);
 466                 }
 467                 *len += lu_dirent_size(ent);
 468         }
 469
 470         if (ent != lu_dirent_start(dp))
 471                 *len += sizeof(struct lu_dirpage);
 472         EXIT;
 473 unmap:
 474         kunmap(rdpg->rp_pages[0]);
 475         return rc;
 476 }
 477
 478 /* Send one page to remote MDT for creating entries there. */
 479 static int cmm_split_send_page(const struct lu_env *env,
 480                                struct md_object *mo,
 481                                struct lu_rdpg *rdpg,
 482                                struct lu_fid *fid, int len)
 483 {
 484         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 485         struct cmm_object *obj;
 486         int rc = 0;
 487         ENTRY;
 488
 489         obj = cmm_object_find(env, cmm, fid);
 490         if (IS_ERR(obj))
 491                 RETURN(PTR_ERR(obj));
 492
 493         rc = mdc_send_page(cmm, env, md_object_next(&obj->cmo_obj),
 494                            rdpg->rp_pages[0], len);
 495         cmm_object_put(env, obj);
 496         RETURN(rc);
 497 }
 498
 499 /* Read one page of entries from local MDT. */
 500 static int cmm_split_read_page(const struct lu_env *env,
 501                                struct md_object *mo,
 502                                struct lu_rdpg *rdpg)
 503 {
 504         int rc;
 505         ENTRY;
 506         memset(cfs_kmap(rdpg->rp_pages[0]), 0, CFS_PAGE_SIZE);
 507         cfs_kunmap(rdpg->rp_pages[0]);
 508         rc = mo_readpage(env, md_object_next(mo), rdpg);
 509         RETURN(rc);
 510 }
 511
 512 /*
 513  * This function performs migration of all pages with entries which fit into one
 514  * stripe and one hash segment.
 515  */
 516 static int cmm_split_process_stripe(const struct lu_env *env,
 517                                     struct md_object *mo,
 518                                     struct lu_rdpg *rdpg,
 519                                     struct lu_fid *lf,
 520                                     __u32 end)
 521 {
 522         int rc, done = 0;
 523         ENTRY;
 524
 525         LASSERT(rdpg->rp_npages == 1);
 526         do {
 527                 struct lu_dirpage *ldp;
 528                 __u32 len = 0;
 529
 530                 /* Read one page from local MDT. */
 531                 rc = cmm_split_read_page(env, mo, rdpg);
 532                 if (rc) {
 533                         CERROR("Error in readpage: %d\n", rc);
 534                         RETURN(rc);
 535                 }
 536
 537                 /* Remove local entries which are going to remite MDT. */
 538                 rc = cmm_split_remove_page(env, mo, rdpg, end, &len);
 539                 if (rc) {
 540                         CERROR("Error in remove stripe entries: %d\n", rc);
 541                         RETURN(rc);
 542                 }
 543
 544                 /* Send entries page to slave MDT. */
 545                 if (len > 0) {
 546                         rc = cmm_split_send_page(env, mo, rdpg, lf, len);
 547                         if (rc) {
 548                                 CERROR("Error in sending page: %d\n", rc);
 549                                 RETURN(rc);
 550                         }
 551                 }
 552
 553                 kmap(rdpg->rp_pages[0]);
 554                 ldp = page_address(rdpg->rp_pages[0]);
 555                 if (le32_to_cpu(ldp->ldp_hash_end) >= end)
 556                         done = 1;
 557
 558                 rdpg->rp_hash = le32_to_cpu(ldp->ldp_hash_end);
 559                 kunmap(rdpg->rp_pages[0]);
 560         } while (!done);
 561
 562         RETURN(rc);
 563 }
 564
 565 extern int bug11150;
 566
 567 static int cmm_split_process_dir(const struct lu_env *env,
 568                                  struct md_object *mo,
 569                                  struct md_attr *ma)
 570 {
 571         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 572         struct lu_rdpg *rdpg = &cmm_env_info(env)->cmi_rdpg;
 573         __u32 hash_segement;
 574         int rc = 0, i;
 575         ENTRY;
 576
 577         memset(rdpg, 0, sizeof *rdpg);
 578         rdpg->rp_npages = CMM_SPLIT_PAGE_COUNT;
 579         rdpg->rp_count  = CFS_PAGE_SIZE * rdpg->rp_npages;
 580         rdpg->rp_pages  = cmm_env_info(env)->cmi_pages;
 581
 582         for (i = 0; i < rdpg->rp_npages; i++) {
 583                 rdpg->rp_pages[i] = alloc_pages(GFP_KERNEL, 0);
 584                 if (rdpg->rp_pages[i] == NULL)
 585                         GOTO(cleanup, rc = -ENOMEM);
 586         }
 587
 588         LASSERT(ma->ma_valid & MA_LMV);
 589         hash_segement = MAX_HASH_SIZE / (cmm->cmm_tgt_count + 1);
 590         for (i = 1; i < cmm->cmm_tgt_count + 1; i++) {
 591                 struct lu_fid *lf;
 592                 __u32 hash_end;
 593
 594                 lf = &ma->ma_lmv->mea_ids[i];
 595
 596                 rdpg->rp_hash = i * hash_segement;
 597                 if (i == cmm->cmm_tgt_count)
 598                         hash_end = MAX_HASH_SIZE;
 599                 else
 600                         hash_end = rdpg->rp_hash + hash_segement;
 601                 rc = cmm_split_process_stripe(env, mo, rdpg, lf, hash_end);
 602                 if (rc) {
 603                         CERROR("Error (rc = %d) while splitting for %d: fid="
 604                                DFID", %08x:%08x\n", rc, i, PFID(lf),
 605                                rdpg->rp_hash, hash_end);
 606                         bug11150 = 1;
 607                         GOTO(cleanup, rc);
 608                 }
 609         }
 610         EXIT;
 611 cleanup:
 612         for (i = 0; i < rdpg->rp_npages; i++)
 613                 if (rdpg->rp_pages[i] != NULL)
 614                         __free_pages(rdpg->rp_pages[i], 0);
 615         return rc;
 616 }
 617
 618 int cmm_split_dir(const struct lu_env *env, struct md_object *mo)
 619 {
 620         struct cmm_device *cmm = cmm_obj2dev(md2cmm_obj(mo));
 621         struct md_attr    *ma = &cmm_env_info(env)->cmi_ma;
 622         int                rc = 0, split;
 623         struct lu_buf     *buf;
 624         struct timeval     start;
 625         ENTRY;
 626
 627         cmm_lprocfs_time_start(cmm, &start, LPROC_CMM_SPLIT);
 628
 629         LASSERT(S_ISDIR(lu_object_attr(&mo->mo_lu)));
 630         memset(ma, 0, sizeof(*ma));
 631
 632         /* Step1: Checking whether the dir needs to be split. */
 633         rc = cmm_split_expect(env, mo, ma, &split);
 634         if (rc)
 635                 GOTO(out, rc);
 636
 637         if (split != CMM_SPLIT_NEEDED) {
 638                 /* No split is needed, caller may proceed with create. */
 639                 GOTO(out, rc = 0);
 640         }
 641
 642         /* Split should be done now, let's do it. */
 643         CWARN("Dir "DFID" is going to split (size: "LPU64")\n",
 644               PFID(lu_object_fid(&mo->mo_lu)), ma->ma_attr.la_size);
 645
 646         /*
 647          * Disable transacrions for split, since there will be so many trans in
 648          * this one ops, conflict with current recovery design.
 649          */
 650         rc = cmm_upcall(env, &cmm->cmm_md_dev, MD_NO_TRANS);
 651         if (rc) {
 652                 CERROR("Can't disable trans for split, rc %d\n", rc);
 653                 GOTO(out, rc);
 654         }
 655
 656         /* Step2: Prepare the md memory */
 657         ma->ma_lmv_size = CMM_MD_SIZE(cmm->cmm_tgt_count + 1);
 658         OBD_ALLOC(ma->ma_lmv, ma->ma_lmv_size);
 659         if (ma->ma_lmv == NULL)
 660                 GOTO(out, rc = -ENOMEM);
 661
 662         /* Step3: Create slave objects and fill the ma->ma_lmv */
 663         rc = cmm_split_slaves_create(env, mo, ma);
 664         if (rc) {
 665                 CERROR("Can't create slaves for split, rc %d\n", rc);
 666                 GOTO(cleanup, rc);
 667         }
 668
 669         /* Step4: Scan and split the object. */
 670         rc = cmm_split_process_dir(env, mo, ma);
 671         if (rc) {
 672                 CERROR("Can't scan and split, rc %d\n", rc);
 673                 GOTO(cleanup, rc);
 674         }
 675
 676         /* Step5: Set mea to the master object. */
 677         LASSERT(ma->ma_valid & MA_LMV);
 678         buf = cmm_buf_get(env, ma->ma_lmv, ma->ma_lmv_size);
 679         rc = mo_xattr_set(env, md_object_next(mo), buf,
 680                           MDS_LMV_MD_NAME, 0);
 681         if (rc) {
 682                 CERROR("Can't set MEA to master dir, " "rc %d\n", rc);
 683                 GOTO(cleanup, rc);
 684         }
 685
 686         /* set flag in cmm_object */
 687         md2cml_obj(mo)->clo_split = CMM_SPLIT_DONE;
 688
 689         /*
 690          * Finally, split succeed, tell client to repeat opetartion on correct
 691          * MDT.
 692          */
 693         CWARN("Dir "DFID" has been split\n", PFID(lu_object_fid(&mo->mo_lu)));
 694         rc = -ERESTART;
 695         EXIT;
 696 cleanup:
 697         OBD_FREE(ma->ma_lmv, ma->ma_lmv_size);
 698 out:
 699         cmm_lprocfs_time_end(cmm, &start, LPROC_CMM_SPLIT);
 700         return rc;
 701 }