lustre/ptlrpc/nodemap_storage.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (C) 2015, Trustees of Indiana University
  24  *
  25  * Copyright (c) 2014, Intel Corporation.
  26  *
  27  * Author: Joshua Walgenbach <jjw@iu.edu>
  28  * Author: Kit Westneat <cwestnea@iu.edu>
  29  *
  30  * Implements the storage functionality for the nodemap configuration. Functions
  31  * in this file prepare, store, and load nodemap configuration data. Targets
  32  * using nodemap services should register a configuration file object. Nodemap
  33  * configuration changes that need to persist should call the appropriate
  34  * storage function for the data being modified.
  35  *
  36  * There are several index types as defined in enum nodemap_idx_type:
  37  *      NODEMAP_CLUSTER_IDX     stores the data found on the lu_nodemap struct,
  38  *                              like root squash and config flags, as well as
  39  *                              the name.
  40  *      NODEMAP_RANGE_IDX       stores NID range information for a nodemap
  41  *      NODEMAP_UIDMAP_IDX      stores a fs/client UID mapping pair
  42  *      NODEMAP_GIDMAP_IDX      stores a fs/client GID mapping pair
  43  *      NODEMAP_GLOBAL_IDX      stores whether or not nodemaps are active
  44  */
  45
  46 #include <libcfs/libcfs.h>
  47 #include <linux/err.h>
  48 #include <linux/kernel.h>
  49 #include <linux/list.h>
  50 #include <linux/mutex.h>
  51 #include <linux/string.h>
  52 #include <linux/types.h>
  53 #include <lnet/types.h>
  54 #include <lustre/lustre_idl.h>
  55 #include <dt_object.h>
  56 #include <lu_object.h>
  57 #include <lustre_net.h>
  58 #include <lustre_nodemap.h>
  59 #include <obd_class.h>
  60 #include <obd_support.h>
  61 #include "nodemap_internal.h"
  62
  63 /* list of registered nodemap index files, except MGS */
  64 static LIST_HEAD(ncf_list_head);
  65 static DEFINE_MUTEX(ncf_list_lock);
  66
  67 /* MGS index is different than others, others are listeners to MGS idx */
  68 static struct nm_config_file *nodemap_mgs_ncf;
  69
  70 /* lu_nodemap flags */
  71 enum nm_flag_shifts {
  72         NM_FL_ALLOW_ROOT_ACCESS = 0x1,
  73         NM_FL_TRUST_CLIENT_IDS = 0x2,
  74         NM_FL_DENY_UNKNOWN = 0x4,
  75 };
  76
  77 static void nodemap_cluster_key_init(struct nodemap_key *nk, unsigned int nm_id)
  78 {
  79         nk->nk_nodemap_id = cpu_to_le32(nm_idx_set_type(nm_id,
  80                                                         NODEMAP_CLUSTER_IDX));
  81         nk->nk_unused = 0;
  82 }
  83
  84 static void nodemap_cluster_rec_init(union nodemap_rec *nr,
  85                                      const struct lu_nodemap *nodemap)
  86 {
  87         CLASSERT(sizeof(nr->ncr.ncr_name) == sizeof(nodemap->nm_name));
  88
  89         strncpy(nr->ncr.ncr_name, nodemap->nm_name, sizeof(nodemap->nm_name));
  90         nr->ncr.ncr_squash_uid = cpu_to_le32(nodemap->nm_squash_uid);
  91         nr->ncr.ncr_squash_gid = cpu_to_le32(nodemap->nm_squash_gid);
  92         nr->ncr.ncr_flags = cpu_to_le32(
  93                 (nodemap->nmf_trust_client_ids ?
  94                         NM_FL_TRUST_CLIENT_IDS : 0) |
  95                 (nodemap->nmf_allow_root_access ?
  96                         NM_FL_ALLOW_ROOT_ACCESS : 0) |
  97                 (nodemap->nmf_deny_unknown ?
  98                         NM_FL_DENY_UNKNOWN : 0));
  99 }
 100
 101 static void nodemap_idmap_key_init(struct nodemap_key *nk, unsigned int nm_id,
 102                                    enum nodemap_id_type id_type,
 103                                    u32 id_client)
 104 {
 105         enum nodemap_idx_type idx_type;
 106
 107         if (id_type == NODEMAP_UID)
 108                 idx_type = NODEMAP_UIDMAP_IDX;
 109         else
 110                 idx_type = NODEMAP_GIDMAP_IDX;
 111
 112         nk->nk_nodemap_id = cpu_to_le32(nm_idx_set_type(nm_id, idx_type));
 113         nk->nk_id_client = cpu_to_le32(id_client);
 114 }
 115
 116 static void nodemap_idmap_rec_init(union nodemap_rec *nr, u32 id_fs)
 117 {
 118         nr->nir.nir_id_fs = cpu_to_le32(id_fs);
 119 }
 120
 121 static void nodemap_range_key_init(struct nodemap_key *nk, unsigned int nm_id,
 122                                    unsigned int rn_id)
 123 {
 124         nk->nk_nodemap_id = cpu_to_le32(nm_idx_set_type(nm_id,
 125                                                         NODEMAP_RANGE_IDX));
 126         nk->nk_range_id = cpu_to_le32(rn_id);
 127 }
 128
 129 static void nodemap_range_rec_init(union nodemap_rec *nr,
 130                                    const lnet_nid_t nid[2])
 131 {
 132         nr->nrr.nrr_start_nid = cpu_to_le64(nid[0]);
 133         nr->nrr.nrr_end_nid = cpu_to_le64(nid[1]);
 134 }
 135
 136 static void nodemap_global_key_init(struct nodemap_key *nk)
 137 {
 138         nk->nk_nodemap_id = cpu_to_le32(nm_idx_set_type(0, NODEMAP_GLOBAL_IDX));
 139         nk->nk_unused = 0;
 140 }
 141
 142 static void nodemap_global_rec_init(union nodemap_rec *nr, bool active)
 143 {
 144         nr->ngr.ngr_is_active = active;
 145 }
 146
 147 /* should be called with dt_write lock */
 148 static void nodemap_inc_version(const struct lu_env *env,
 149                                 struct dt_object *nodemap_idx,
 150                                 struct thandle *th)
 151 {
 152         u64 ver = dt_version_get(env, nodemap_idx);
 153         dt_version_set(env, nodemap_idx, ver + 1, th);
 154 }
 155
 156 enum ncfc_find_create {
 157         NCFC_CREATE_NEW = 1,
 158 };
 159
 160 static struct dt_object *nodemap_cache_find_create(const struct lu_env *env,
 161                                                    struct dt_device *dev,
 162                                                    struct local_oid_storage *los,
 163                                                    enum ncfc_find_create create_new)
 164 {
 165         struct lu_fid root_fid;
 166         struct dt_object *root_obj;
 167         struct dt_object *nm_obj;
 168         int rc = 0;
 169
 170         rc = dt_root_get(env, dev, &root_fid);
 171         if (rc < 0)
 172                 GOTO(out, nm_obj = ERR_PTR(rc));
 173
 174         root_obj = dt_locate(env, dev, &root_fid);
 175         if (unlikely(IS_ERR(root_obj)))
 176                 GOTO(out, nm_obj = root_obj);
 177
 178 again:
 179         /* if loading index fails the first time, create new index */
 180         if (create_new == NCFC_CREATE_NEW) {
 181                 CDEBUG(D_INFO, "removing old index, creating new one\n");
 182                 rc = local_object_unlink(env, dev, root_obj,
 183                                          LUSTRE_NODEMAP_NAME);
 184                 if (rc < 0) {
 185                         /* XXX not sure the best way to get obd name. */
 186                         CERROR("cannot destroy nodemap index: rc = %d\n",
 187                                rc);
 188                         GOTO(out_root, nm_obj = ERR_PTR(rc));
 189                 }
 190         }
 191
 192         nm_obj = local_index_find_or_create(env, los, root_obj,
 193                                                 LUSTRE_NODEMAP_NAME,
 194                                                 S_IFREG | S_IRUGO | S_IWUSR,
 195                                                 &dt_nodemap_features);
 196         if (IS_ERR(nm_obj))
 197                 GOTO(out_root, nm_obj);
 198
 199         if (nm_obj->do_index_ops == NULL) {
 200                 rc = nm_obj->do_ops->do_index_try(env, nm_obj,
 201                                                       &dt_nodemap_features);
 202                 /* even if loading from tgt fails, connecting to MGS will
 203                  * rewrite the config
 204                  */
 205                 if (rc < 0) {
 206                         lu_object_put(env, &nm_obj->do_lu);
 207
 208                         if (create_new == NCFC_CREATE_NEW)
 209                                 GOTO(out_root, nm_obj = ERR_PTR(rc));
 210
 211                         CERROR("cannot load nodemap index from disk, creating "
 212                                "new index: rc = %d\n", rc);
 213                         create_new = NCFC_CREATE_NEW;
 214                         goto again;
 215                 }
 216         }
 217
 218 out_root:
 219         lu_object_put(env, &root_obj->do_lu);
 220 out:
 221         return nm_obj;
 222 }
 223
 224 static int nodemap_idx_insert(const struct lu_env *env,
 225                               struct dt_object *idx,
 226                               const struct nodemap_key *nk,
 227                               const union nodemap_rec *nr)
 228 {
 229         struct thandle          *th;
 230         struct dt_device        *dev = lu2dt_dev(idx->do_lu.lo_dev);
 231         int                      rc;
 232
 233         CLASSERT(sizeof(union nodemap_rec) == 32);
 234
 235         th = dt_trans_create(env, dev);
 236
 237         if (IS_ERR(th))
 238                 GOTO(out, rc = PTR_ERR(th));
 239
 240         rc = dt_declare_insert(env, idx,
 241                                (const struct dt_rec *)nr,
 242                                (const struct dt_key *)nk, th);
 243         if (rc != 0)
 244                 GOTO(out, rc);
 245
 246         rc = dt_declare_version_set(env, idx, th);
 247         if (rc != 0)
 248                 GOTO(out, rc);
 249
 250         rc = dt_trans_start_local(env, dev, th);
 251         if (rc != 0)
 252                 GOTO(out, rc);
 253
 254         dt_write_lock(env, idx, 0);
 255
 256         rc = dt_insert(env, idx, (const struct dt_rec *)nr,
 257                        (const struct dt_key *)nk, th, 1);
 258
 259         nodemap_inc_version(env, idx, th);
 260         dt_write_unlock(env, idx);
 261 out:
 262         dt_trans_stop(env, dev, th);
 263
 264         return rc;
 265 }
 266
 267 static int nodemap_idx_update(const struct lu_env *env,
 268                               struct dt_object *idx,
 269                               const struct nodemap_key *nk,
 270                               const union nodemap_rec *nr)
 271 {
 272         struct thandle          *th;
 273         struct dt_device        *dev = lu2dt_dev(idx->do_lu.lo_dev);
 274         int                      rc = 0;
 275
 276         th = dt_trans_create(env, dev);
 277
 278         if (IS_ERR(th))
 279                 GOTO(out, rc = PTR_ERR(th));
 280
 281         rc = dt_declare_delete(env, idx, (const struct dt_key *)nk, th);
 282         if (rc != 0)
 283                 GOTO(out, rc);
 284
 285         rc = dt_declare_insert(env, idx, (const struct dt_rec *)nr,
 286                                (const struct dt_key *)nk, th);
 287         if (rc != 0)
 288                 GOTO(out, rc);
 289
 290         rc = dt_declare_version_set(env, idx, th);
 291         if (rc != 0)
 292                 GOTO(out, rc);
 293
 294         rc = dt_trans_start_local(env, dev, th);
 295         if (rc != 0)
 296                 GOTO(out, rc);
 297
 298         dt_write_lock(env, idx, 0);
 299
 300         rc = dt_delete(env, idx, (const struct dt_key *)nk, th);
 301         if (rc != 0)
 302                 GOTO(out_lock, rc);
 303
 304         rc = dt_insert(env, idx, (const struct dt_rec *)nr,
 305                        (const struct dt_key *)nk, th, 1);
 306         if (rc != 0)
 307                 GOTO(out_lock, rc);
 308
 309         nodemap_inc_version(env, idx, th);
 310 out_lock:
 311         dt_write_unlock(env, idx);
 312 out:
 313         dt_trans_stop(env, dev, th);
 314
 315         return rc;
 316 }
 317
 318 static int nodemap_idx_delete(const struct lu_env *env,
 319                               struct dt_object *idx,
 320                               const struct nodemap_key *nk,
 321                               const union nodemap_rec *unused)
 322 {
 323         struct thandle          *th;
 324         struct dt_device        *dev = lu2dt_dev(idx->do_lu.lo_dev);
 325         int                      rc = 0;
 326
 327         th = dt_trans_create(env, dev);
 328
 329         if (IS_ERR(th))
 330                 GOTO(out, rc = PTR_ERR(th));
 331
 332         rc = dt_declare_delete(env, idx, (const struct dt_key *)nk, th);
 333         if (rc != 0)
 334                 GOTO(out, rc);
 335
 336         rc = dt_declare_version_set(env, idx, th);
 337         if (rc != 0)
 338                 GOTO(out, rc);
 339
 340         rc = dt_trans_start_local(env, dev, th);
 341         if (rc != 0)
 342                 GOTO(out, rc);
 343
 344         dt_write_lock(env, idx, 0);
 345
 346         rc = dt_delete(env, idx, (const struct dt_key *)nk, th);
 347
 348         nodemap_inc_version(env, idx, th);
 349
 350         dt_write_unlock(env, idx);
 351 out:
 352         dt_trans_stop(env, dev, th);
 353
 354         return rc;
 355 }
 356
 357 enum nm_add_update {
 358         NM_ADD = 0,
 359         NM_UPDATE = 1,
 360 };
 361
 362 static int nodemap_idx_nodemap_add_update(const struct lu_nodemap *nodemap,
 363                                           struct dt_object *idx,
 364                                           enum nm_add_update update)
 365 {
 366         struct nodemap_key nk;
 367         union nodemap_rec nr;
 368         struct lu_env env;
 369         int rc = 0;
 370
 371         ENTRY;
 372
 373         rc = lu_env_init(&env, LCT_LOCAL);
 374         if (rc)
 375                 RETURN(rc);
 376
 377         nodemap_cluster_key_init(&nk, nodemap->nm_id);
 378         nodemap_cluster_rec_init(&nr, nodemap);
 379
 380         if (update == NM_UPDATE)
 381                 rc = nodemap_idx_update(&env, idx, &nk, &nr);
 382         else
 383                 rc = nodemap_idx_insert(&env, idx, &nk, &nr);
 384
 385         lu_env_fini(&env);
 386
 387         RETURN(rc);
 388 }
 389
 390 int nodemap_idx_nodemap_add(const struct lu_nodemap *nodemap)
 391 {
 392         if (nodemap_mgs_ncf == NULL) {
 393                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 394                 return -EINVAL;
 395         }
 396
 397         return nodemap_idx_nodemap_add_update(nodemap, nodemap_mgs_ncf->ncf_obj,
 398                                               NM_ADD);
 399 }
 400
 401 int nodemap_idx_nodemap_update(const struct lu_nodemap *nodemap)
 402 {
 403         if (nodemap_mgs_ncf == NULL) {
 404                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 405                 return -EINVAL;
 406         }
 407
 408         return nodemap_idx_nodemap_add_update(nodemap, nodemap_mgs_ncf->ncf_obj,
 409                                               NM_UPDATE);
 410 }
 411
 412 int nodemap_idx_nodemap_del(const struct lu_nodemap *nodemap)
 413 {
 414         struct rb_root           root;
 415         struct lu_idmap         *idmap;
 416         struct lu_idmap         *temp;
 417         struct lu_nid_range     *range;
 418         struct lu_nid_range     *range_temp;
 419         struct nodemap_key       nk;
 420         struct lu_env            env;
 421         int                      rc = 0;
 422         int                      rc2 = 0;
 423
 424         ENTRY;
 425
 426         if (nodemap_mgs_ncf == NULL) {
 427                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 428                 return -EINVAL;
 429         }
 430
 431         rc = lu_env_init(&env, LCT_LOCAL);
 432         if (rc != 0)
 433                 RETURN(rc);
 434
 435         root = nodemap->nm_fs_to_client_uidmap;
 436         nm_rbtree_postorder_for_each_entry_safe(idmap, temp, &root,
 437                                                 id_fs_to_client) {
 438                 nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_UID,
 439                                        idmap->id_client);
 440                 rc2 = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj,
 441                                          &nk, NULL);
 442                 if (rc2 < 0)
 443                         rc = rc2;
 444         }
 445
 446         root = nodemap->nm_client_to_fs_gidmap;
 447         nm_rbtree_postorder_for_each_entry_safe(idmap, temp, &root,
 448                                                 id_client_to_fs) {
 449                 nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_GID,
 450                                        idmap->id_client);
 451                 rc2 = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj,
 452                                          &nk, NULL);
 453                 if (rc2 < 0)
 454                         rc = rc2;
 455         }
 456
 457         list_for_each_entry_safe(range, range_temp, &nodemap->nm_ranges,
 458                                  rn_list) {
 459                 nodemap_range_key_init(&nk, nodemap->nm_id, range->rn_id);
 460                 rc2 = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj,
 461                                          &nk, NULL);
 462                 if (rc2 < 0)
 463                         rc = rc2;
 464         }
 465
 466         nodemap_cluster_key_init(&nk, nodemap->nm_id);
 467         rc2 = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj, &nk, NULL);
 468         if (rc2 < 0)
 469                 rc = rc2;
 470
 471         lu_env_fini(&env);
 472
 473         RETURN(rc);
 474 }
 475
 476 int nodemap_idx_range_add(const struct lu_nid_range *range,
 477                           const lnet_nid_t nid[2])
 478 {
 479         struct nodemap_key       nk;
 480         union nodemap_rec        nr;
 481         struct lu_env            env;
 482         int                      rc = 0;
 483         ENTRY;
 484
 485         if (nodemap_mgs_ncf == NULL) {
 486                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 487                 return -EINVAL;
 488         }
 489
 490         rc = lu_env_init(&env, LCT_LOCAL);
 491         if (rc != 0)
 492                 RETURN(rc);
 493
 494         nodemap_range_key_init(&nk, range->rn_nodemap->nm_id, range->rn_id);
 495         nodemap_range_rec_init(&nr, nid);
 496
 497         rc = nodemap_idx_insert(&env, nodemap_mgs_ncf->ncf_obj, &nk, &nr);
 498         lu_env_fini(&env);
 499
 500         RETURN(rc);
 501 }
 502
 503 int nodemap_idx_range_del(const struct lu_nid_range *range)
 504 {
 505         struct nodemap_key       nk;
 506         struct lu_env            env;
 507         int                      rc = 0;
 508         ENTRY;
 509
 510         if (nodemap_mgs_ncf == NULL) {
 511                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 512                 return -EINVAL;
 513         }
 514
 515         rc = lu_env_init(&env, LCT_LOCAL);
 516         if (rc != 0)
 517                 RETURN(rc);
 518
 519         nodemap_range_key_init(&nk, range->rn_nodemap->nm_id, range->rn_id);
 520
 521         rc = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj, &nk, NULL);
 522         lu_env_fini(&env);
 523
 524         RETURN(rc);
 525 }
 526
 527 int nodemap_idx_idmap_add(const struct lu_nodemap *nodemap,
 528                           enum nodemap_id_type id_type,
 529                           const u32 map[2])
 530 {
 531         struct nodemap_key       nk;
 532         union nodemap_rec        nr;
 533         struct lu_env            env;
 534         int                      rc = 0;
 535         ENTRY;
 536
 537         if (nodemap_mgs_ncf == NULL) {
 538                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 539                 return -EINVAL;
 540         }
 541
 542         rc = lu_env_init(&env, LCT_LOCAL);
 543         if (rc != 0)
 544                 RETURN(rc);
 545
 546         nodemap_idmap_key_init(&nk, nodemap->nm_id, id_type, map[0]);
 547         nodemap_idmap_rec_init(&nr, map[1]);
 548
 549         rc = nodemap_idx_insert(&env, nodemap_mgs_ncf->ncf_obj, &nk, &nr);
 550         lu_env_fini(&env);
 551
 552         RETURN(rc);
 553 }
 554
 555 int nodemap_idx_idmap_del(const struct lu_nodemap *nodemap,
 556                           enum nodemap_id_type id_type,
 557                           const u32 map[2])
 558 {
 559         struct nodemap_key       nk;
 560         struct lu_env            env;
 561         int                      rc = 0;
 562         ENTRY;
 563
 564         if (nodemap_mgs_ncf == NULL) {
 565                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 566                 return -EINVAL;
 567         }
 568
 569         rc = lu_env_init(&env, LCT_LOCAL);
 570         if (rc != 0)
 571                 RETURN(rc);
 572
 573         nodemap_idmap_key_init(&nk, nodemap->nm_id, id_type, map[0]);
 574
 575         rc = nodemap_idx_delete(&env, nodemap_mgs_ncf->ncf_obj, &nk, NULL);
 576         lu_env_fini(&env);
 577
 578         RETURN(rc);
 579 }
 580
 581 static int nodemap_idx_global_add_update(bool value, enum nm_add_update update)
 582 {
 583         struct nodemap_key       nk;
 584         union nodemap_rec        nr;
 585         struct lu_env            env;
 586         int                      rc = 0;
 587         ENTRY;
 588
 589         if (nodemap_mgs_ncf == NULL) {
 590                 CERROR("cannot add nodemap config to non-existing MGS.\n");
 591                 return -EINVAL;
 592         }
 593
 594         rc = lu_env_init(&env, LCT_LOCAL);
 595         if (rc != 0)
 596                 RETURN(rc);
 597
 598         nodemap_global_key_init(&nk);
 599         nodemap_global_rec_init(&nr, value);
 600
 601         if (update == NM_UPDATE)
 602                 rc = nodemap_idx_update(&env, nodemap_mgs_ncf->ncf_obj,
 603                                         &nk, &nr);
 604         else
 605                 rc = nodemap_idx_insert(&env, nodemap_mgs_ncf->ncf_obj,
 606                                         &nk, &nr);
 607
 608         lu_env_fini(&env);
 609
 610         RETURN(rc);
 611 }
 612
 613 int nodemap_idx_nodemap_activate(bool value)
 614 {
 615         return nodemap_idx_global_add_update(value, NM_UPDATE);
 616 }
 617
 618 /**
 619  * Process a key/rec pair and modify the new configuration.
 620  *
 621  * \param       config          configuration to update with this key/rec data
 622  * \param       key             key of the record that was loaded
 623  * \param       rec             record that was loaded
 624  * \param       recent_nodemap  last referenced nodemap
 625  * \retval      type of record processed, see enum #nodemap_idx_type
 626  * \retval      -ENOENT         range or map loaded before nodemap record
 627  * \retval      -EINVAL         duplicate nodemap cluster records found with
 628  *                              different IDs, or nodemap has invalid name
 629  * \retval      -ENOMEM
 630  */
 631 static int nodemap_process_keyrec(struct nodemap_config *config,
 632                                   const struct nodemap_key *key,
 633                                   const union nodemap_rec *rec,
 634                                   struct lu_nodemap **recent_nodemap)
 635 {
 636         struct lu_nodemap       *nodemap = NULL;
 637         enum nodemap_idx_type    type;
 638         enum nodemap_id_type     id_type;
 639         u8                       flags;
 640         u32                      nodemap_id;
 641         lnet_nid_t               nid[2];
 642         u32                      map[2];
 643         int                      rc;
 644
 645         CLASSERT(sizeof(union nodemap_rec) == 32);
 646
 647         nodemap_id = le32_to_cpu(key->nk_nodemap_id);
 648         type = nm_idx_get_type(nodemap_id);
 649         nodemap_id = nm_idx_set_type(nodemap_id, 0);
 650
 651         CDEBUG(D_INFO, "found config entry, nm_id %d type %d\n",
 652                nodemap_id, type);
 653
 654         /* find the correct nodemap in the load list */
 655         if (type == NODEMAP_RANGE_IDX || type == NODEMAP_UIDMAP_IDX ||
 656             type == NODEMAP_GIDMAP_IDX) {
 657                 struct lu_nodemap *tmp = NULL;
 658
 659                 nodemap = *recent_nodemap;
 660
 661                 if (nodemap == NULL)
 662                         GOTO(out, rc = -ENOENT);
 663
 664                 if (nodemap->nm_id != nodemap_id) {
 665                         list_for_each_entry(tmp, &nodemap->nm_list, nm_list)
 666                                 if (tmp->nm_id == nodemap_id) {
 667                                         nodemap = tmp;
 668                                         break;
 669                                 }
 670
 671                         if (nodemap->nm_id != nodemap_id)
 672                                 GOTO(out, rc = -ENOENT);
 673                 }
 674
 675                 /* update most recently used nodemap if necessay */
 676                 if (nodemap != *recent_nodemap)
 677                         *recent_nodemap = nodemap;
 678         }
 679
 680         switch (type) {
 681         case NODEMAP_EMPTY_IDX:
 682                 if (nodemap_id != 0)
 683                         CWARN("Found nodemap config record without type field, "
 684                               " nodemap_id=%d. nodemap config file corrupt?\n",
 685                               nodemap_id);
 686                 break;
 687         case NODEMAP_CLUSTER_IDX:
 688                 nodemap = cfs_hash_lookup(config->nmc_nodemap_hash,
 689                                           rec->ncr.ncr_name);
 690                 if (nodemap == NULL) {
 691                         if (nodemap_id == LUSTRE_NODEMAP_DEFAULT_ID) {
 692                                 nodemap = nodemap_create(rec->ncr.ncr_name,
 693                                                          config, 1);
 694                                 config->nmc_default_nodemap = nodemap;
 695                         } else {
 696                                 nodemap = nodemap_create(rec->ncr.ncr_name,
 697                                                          config, 0);
 698                         }
 699                         if (IS_ERR(nodemap))
 700                                 GOTO(out, rc = PTR_ERR(nodemap));
 701
 702                         /* we need to override the local ID with the saved ID */
 703                         nodemap->nm_id = nodemap_id;
 704                         if (nodemap_id > config->nmc_nodemap_highest_id)
 705                                 config->nmc_nodemap_highest_id = nodemap_id;
 706
 707                 } else if (nodemap->nm_id != nodemap_id) {
 708                         nodemap_putref(nodemap);
 709                         GOTO(out, rc = -EINVAL);
 710                 }
 711
 712                 nodemap->nm_squash_uid =
 713                                 le32_to_cpu(rec->ncr.ncr_squash_uid);
 714                 nodemap->nm_squash_gid =
 715                                 le32_to_cpu(rec->ncr.ncr_squash_gid);
 716
 717                 flags = le32_to_cpu(rec->ncr.ncr_flags);
 718                 nodemap->nmf_allow_root_access =
 719                                         flags & NM_FL_ALLOW_ROOT_ACCESS;
 720                 nodemap->nmf_trust_client_ids =
 721                                         flags & NM_FL_TRUST_CLIENT_IDS;
 722                 nodemap->nmf_deny_unknown =
 723                                         flags & NM_FL_DENY_UNKNOWN;
 724
 725                 if (*recent_nodemap == NULL) {
 726                         *recent_nodemap = nodemap;
 727                         INIT_LIST_HEAD(&nodemap->nm_list);
 728                 } else {
 729                         list_add(&nodemap->nm_list,
 730                                  &(*recent_nodemap)->nm_list);
 731                 }
 732                 nodemap_putref(nodemap);
 733                 break;
 734         case NODEMAP_RANGE_IDX:
 735                 nid[0] = le64_to_cpu(rec->nrr.nrr_start_nid);
 736                 nid[1] = le64_to_cpu(rec->nrr.nrr_end_nid);
 737
 738                 rc = nodemap_add_range_helper(config, nodemap, nid,
 739                                         le32_to_cpu(key->nk_range_id));
 740                 if (rc != 0)
 741                         GOTO(out, rc);
 742                 break;
 743         case NODEMAP_UIDMAP_IDX:
 744         case NODEMAP_GIDMAP_IDX:
 745                 map[0] = le32_to_cpu(key->nk_id_client);
 746                 map[1] = le32_to_cpu(rec->nir.nir_id_fs);
 747
 748                 if (type == NODEMAP_UIDMAP_IDX)
 749                         id_type = NODEMAP_UID;
 750                 else
 751                         id_type = NODEMAP_GID;
 752
 753                 rc = nodemap_add_idmap_helper(nodemap, id_type, map);
 754                 if (rc != 0)
 755                         GOTO(out, rc);
 756                 break;
 757         case NODEMAP_GLOBAL_IDX:
 758                 config->nmc_nodemap_is_active = rec->ngr.ngr_is_active;
 759                 break;
 760         default:
 761                 CERROR("got keyrec pair for unknown type %d\n", type);
 762                 break;
 763         }
 764
 765         rc = type;
 766
 767 out:
 768         return rc;
 769 }
 770
 771 static int nodemap_load_entries(const struct lu_env *env,
 772                                 struct dt_object *nodemap_idx)
 773 {
 774         const struct dt_it_ops  *iops;
 775         struct dt_it            *it;
 776         struct lu_nodemap       *recent_nodemap = NULL;
 777         struct nodemap_config   *new_config = NULL;
 778         u64                      hash = 0;
 779         bool                     activate_nodemap = false;
 780         bool                     loaded_global_idx = false;
 781         int                      rc = 0;
 782
 783         ENTRY;
 784
 785         iops = &nodemap_idx->do_index_ops->dio_it;
 786
 787         dt_read_lock(env, nodemap_idx, 0);
 788         it = iops->init(env, nodemap_idx, 0);
 789         if (IS_ERR(it))
 790                 GOTO(out, rc = PTR_ERR(it));
 791
 792         rc = iops->load(env, it, hash);
 793         if (rc < 0)
 794                 GOTO(out_iops_fini, rc);
 795
 796         /* rc == 0 means we need to advance to record */
 797         if (rc == 0) {
 798                 rc = iops->next(env, it);
 799
 800                 if (rc < 0)
 801                         GOTO(out_iops_put, rc);
 802                 /* rc > 0 is eof, will be checked in while below */
 803         } else {
 804                 /* rc == 1, we found initial record and can process below */
 805                 rc = 0;
 806         }
 807
 808         new_config = nodemap_config_alloc();
 809         if (IS_ERR(new_config)) {
 810                 rc = PTR_ERR(new_config);
 811                 new_config = NULL;
 812                 GOTO(out_iops_put, rc);
 813         }
 814
 815         /* rc > 0 is eof, check initial iops->next here as well */
 816         while (rc == 0) {
 817                 struct nodemap_key *key;
 818                 union nodemap_rec rec;
 819
 820                 key = (struct nodemap_key *)iops->key(env, it);
 821                 rc = iops->rec(env, it, (struct dt_rec *)&rec, 0);
 822                 if (rc != -ESTALE) {
 823                         if (rc != 0)
 824                                 GOTO(out_nodemap_config, rc);
 825                         rc = nodemap_process_keyrec(new_config, key, &rec,
 826                                                     &recent_nodemap);
 827                         if (rc < 0)
 828                                 GOTO(out_nodemap_config, rc);
 829                         if (rc == NODEMAP_GLOBAL_IDX)
 830                                 loaded_global_idx = true;
 831                 }
 832
 833                 do
 834                         rc = iops->next(env, it);
 835                 while (rc == -ESTALE);
 836         }
 837
 838         if (rc > 0)
 839                 rc = 0;
 840
 841 out_nodemap_config:
 842         if (rc != 0)
 843                 nodemap_config_dealloc(new_config);
 844         else
 845                 /* creating new default needs to be done outside dt read lock */
 846                 activate_nodemap = true;
 847 out_iops_put:
 848         iops->put(env, it);
 849 out_iops_fini:
 850         iops->fini(env, it);
 851 out:
 852         dt_read_unlock(env, nodemap_idx);
 853
 854         if (rc != 0)
 855                 CWARN("%s: failed to load nodemap configuration: rc = %d\n",
 856                       nodemap_idx->do_lu.lo_dev->ld_obd->obd_name, rc);
 857
 858         if (!activate_nodemap)
 859                 RETURN(rc);
 860
 861         if (new_config->nmc_default_nodemap == NULL) {
 862                 /* new MGS won't have a default nm on disk, so create it here */
 863                 new_config->nmc_default_nodemap =
 864                         nodemap_create(DEFAULT_NODEMAP, new_config, 1);
 865                 if (IS_ERR(new_config->nmc_default_nodemap)) {
 866                         rc = PTR_ERR(new_config->nmc_default_nodemap);
 867                 } else {
 868                         rc = nodemap_idx_nodemap_add_update(
 869                                         new_config->nmc_default_nodemap,
 870                                         nodemap_idx,
 871                                         NM_ADD);
 872                         nodemap_putref(new_config->nmc_default_nodemap);
 873                 }
 874         }
 875
 876         /* new nodemap config won't have an active/inactive record */
 877         if (rc == 0 && loaded_global_idx == false) {
 878                 struct nodemap_key       nk;
 879                 union nodemap_rec        nr;
 880
 881                 nodemap_global_key_init(&nk);
 882                 nodemap_global_rec_init(&nr, false);
 883                 rc = nodemap_idx_insert(env, nodemap_idx, &nk, &nr);
 884         }
 885
 886         if (rc == 0)
 887                 nodemap_config_set_active(new_config);
 888         else
 889                 nodemap_config_dealloc(new_config);
 890
 891         RETURN(rc);
 892 }
 893
 894 /**
 895  * Step through active config and write to disk.
 896  */
 897 struct dt_object *nodemap_save_config_cache(const struct lu_env *env,
 898                                             struct dt_device *dev,
 899                                             struct local_oid_storage *los)
 900 {
 901         struct dt_object *o;
 902         struct lu_nodemap *nodemap;
 903         struct lu_nodemap *nm_tmp;
 904         struct lu_nid_range *range;
 905         struct lu_nid_range *range_temp;
 906         struct lu_idmap *idmap;
 907         struct lu_idmap *id_tmp;
 908         struct rb_root root;
 909         struct nodemap_key nk;
 910         union nodemap_rec nr;
 911         LIST_HEAD(nodemap_list_head);
 912         int rc = 0, rc2;
 913
 914         ENTRY;
 915
 916         /* create a new index file to fill with active config */
 917         o = nodemap_cache_find_create(env, dev, los, NCFC_CREATE_NEW);
 918         if (IS_ERR(o))
 919                 GOTO(out, o);
 920
 921         mutex_lock(&active_config_lock);
 922
 923         /* convert hash to list so we don't spin */
 924         cfs_hash_for_each_safe(active_config->nmc_nodemap_hash,
 925                                nm_hash_list_cb, &nodemap_list_head);
 926
 927         list_for_each_entry_safe(nodemap, nm_tmp, &nodemap_list_head, nm_list) {
 928                 nodemap_cluster_key_init(&nk, nodemap->nm_id);
 929                 nodemap_cluster_rec_init(&nr, nodemap);
 930
 931                 rc2 = nodemap_idx_insert(env, o, &nk, &nr);
 932                 if (rc2 < 0) {
 933                         rc = rc2;
 934                         continue;
 935                 }
 936
 937                 down_read(&active_config->nmc_range_tree_lock);
 938                 list_for_each_entry_safe(range, range_temp, &nodemap->nm_ranges,
 939                                          rn_list) {
 940                         lnet_nid_t nid[2] = {
 941                                 range->rn_node.in_extent.start,
 942                                 range->rn_node.in_extent.end
 943                         };
 944                         nodemap_range_key_init(&nk, nodemap->nm_id,
 945                                                range->rn_id);
 946                         nodemap_range_rec_init(&nr, nid);
 947                         rc2 = nodemap_idx_insert(env, o, &nk, &nr);
 948                         if (rc2 < 0)
 949                                 rc = rc2;
 950                 }
 951                 up_read(&active_config->nmc_range_tree_lock);
 952
 953                 /* we don't need to take nm_idmap_lock because active config
 954                  * lock prevents changes from happening to nodemaps
 955                  */
 956                 root = nodemap->nm_client_to_fs_uidmap;
 957                 nm_rbtree_postorder_for_each_entry_safe(idmap, id_tmp, &root,
 958                                                         id_client_to_fs) {
 959                         nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_UID,
 960                                                idmap->id_client);
 961                         nodemap_idmap_rec_init(&nr, idmap->id_fs);
 962                         rc2 = nodemap_idx_insert(env, o, &nk, &nr);
 963                         if (rc2 < 0)
 964                                 rc = rc2;
 965                 }
 966
 967                 root = nodemap->nm_client_to_fs_gidmap;
 968                 nm_rbtree_postorder_for_each_entry_safe(idmap, id_tmp, &root,
 969                                                         id_client_to_fs) {
 970                         nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_GID,
 971                                                idmap->id_client);
 972                         nodemap_idmap_rec_init(&nr, idmap->id_fs);
 973                         rc2 = nodemap_idx_insert(env, o, &nk, &nr);
 974                         if (rc2 < 0)
 975                                 rc = rc2;
 976                 }
 977         }
 978         nodemap_global_key_init(&nk);
 979         nodemap_global_rec_init(&nr, active_config->nmc_nodemap_is_active);
 980         rc2 = nodemap_idx_insert(env, o, &nk, &nr);
 981         if (rc2 < 0)
 982                 rc = rc2;
 983
 984 out:
 985         mutex_unlock(&active_config_lock);
 986
 987         if (rc < 0) {
 988                 lu_object_put(env, &o->do_lu);
 989                 o = ERR_PTR(rc);
 990         }
 991
 992         RETURN(o);
 993 }
 994
 995 static void nodemap_save_all_caches(void)
 996 {
 997         struct nm_config_file   *ncf;
 998         struct lu_env            env;
 999         int                      rc = 0;
1000
1001         /* recreating nodemap cache requires fld_thread_key be in env */
1002         rc = lu_env_init(&env, LCT_MD_THREAD | LCT_DT_THREAD | LCT_MG_THREAD);
1003         if (rc != 0) {
1004                 CWARN("cannot init env for nodemap config: rc = %d\n", rc);
1005                 return;
1006         }
1007
1008         mutex_lock(&ncf_list_lock);
1009         list_for_each_entry(ncf, &ncf_list_head, ncf_list) {
1010                 struct dt_device *dev = lu2dt_dev(ncf->ncf_obj->do_lu.lo_dev);
1011                 struct obd_device *obd = ncf->ncf_obj->do_lu.lo_dev->ld_obd;
1012                 struct dt_object *o;
1013
1014                 /* put current config file so save conf can rewrite it */
1015                 lu_object_put_nocache(&env, &ncf->ncf_obj->do_lu);
1016                 ncf->ncf_obj = NULL;
1017
1018                 o = nodemap_save_config_cache(&env, dev, ncf->ncf_los);
1019                 if (IS_ERR(o))
1020                         CWARN("%s: error writing to nodemap config: rc = %d\n",
1021                               obd->obd_name, rc);
1022                 else
1023                         ncf->ncf_obj = o;
1024         }
1025         mutex_unlock(&ncf_list_lock);
1026
1027         lu_env_fini(&env);
1028 }
1029
1030 /* tracks if config still needs to be loaded, either from disk or network */
1031 static bool nodemap_config_loaded;
1032 static DEFINE_MUTEX(nodemap_config_loaded_lock);
1033
1034 /**
1035  * Ensures that configs loaded over the wire are prioritized over those loaded
1036  * from disk.
1037  *
1038  * \param config        config to set as the active config
1039  */
1040 void nodemap_config_set_active_mgc(struct nodemap_config *config)
1041 {
1042         mutex_lock(&nodemap_config_loaded_lock);
1043         nodemap_config_set_active(config);
1044         nodemap_config_loaded = true;
1045         nodemap_save_all_caches();
1046         mutex_unlock(&nodemap_config_loaded_lock);
1047 }
1048 EXPORT_SYMBOL(nodemap_config_set_active_mgc);
1049
1050 /**
1051  * Register a dt_object representing the config index file. This should be
1052  * called by targets in order to load the nodemap configuration from disk. The
1053  * dt_object should be created with local_index_find_or_create and the index
1054  * features should be enabled with do_index_try.
1055  *
1056  * \param obj   dt_object returned by local_index_find_or_create
1057  *
1058  * \retval      on success: nm_config_file handle for later deregistration
1059  * \retval      -ENOMEM         memory allocation failure
1060  * \retval      -ENOENT         error loading nodemap config
1061  * \retval      -EINVAL         error loading nodemap config
1062  * \retval      -EEXIST         nodemap config already registered for MGS
1063  */
1064 struct nm_config_file *nm_config_file_register_mgs(const struct lu_env *env,
1065                                                    struct dt_object *obj,
1066                                                    struct local_oid_storage *los)
1067 {
1068         struct nm_config_file *ncf;
1069         int rc = 0;
1070         ENTRY;
1071
1072         if (nodemap_mgs_ncf != NULL)
1073                 GOTO(out, ncf = ERR_PTR(-EEXIST));
1074
1075         OBD_ALLOC_PTR(ncf);
1076         if (ncf == NULL)
1077                 GOTO(out, ncf = ERR_PTR(-ENOMEM));
1078
1079         /* if loading from cache, prevent activation of MGS config until cache
1080          * loading is done, so disk config is overwritten by MGS config.
1081          */
1082         mutex_lock(&nodemap_config_loaded_lock);
1083         rc = nodemap_load_entries(env, obj);
1084         if (!rc)
1085                 nodemap_config_loaded = true;
1086         mutex_unlock(&nodemap_config_loaded_lock);
1087
1088         if (rc) {
1089                 OBD_FREE_PTR(ncf);
1090                 GOTO(out, ncf = ERR_PTR(rc));
1091         }
1092
1093         lu_object_get(&obj->do_lu);
1094
1095         ncf->ncf_obj = obj;
1096         ncf->ncf_los = los;
1097
1098         nodemap_mgs_ncf = ncf;
1099
1100 out:
1101         return ncf;
1102 }
1103 EXPORT_SYMBOL(nm_config_file_register_mgs);
1104
1105 struct nm_config_file *nm_config_file_register_tgt(const struct lu_env *env,
1106                                                    struct dt_device *dev,
1107                                                    struct local_oid_storage *los)
1108 {
1109         struct nm_config_file *ncf;
1110         struct dt_object *config_obj = NULL;
1111         int rc = 0;
1112
1113         OBD_ALLOC_PTR(ncf);
1114         if (ncf == NULL)
1115                 RETURN(ERR_PTR(-ENOMEM));
1116
1117         /* don't load from cache if config already loaded */
1118         mutex_lock(&nodemap_config_loaded_lock);
1119         if (!nodemap_config_loaded) {
1120                 config_obj = nodemap_cache_find_create(env, dev, los, 0);
1121                 if (IS_ERR(config_obj))
1122                         rc = PTR_ERR(config_obj);
1123                 else
1124                         rc = nodemap_load_entries(env, config_obj);
1125
1126                 if (!rc)
1127                         nodemap_config_loaded = true;
1128         }
1129         mutex_unlock(&nodemap_config_loaded_lock);
1130         if (rc)
1131                 GOTO(out_ncf, rc);
1132
1133         /* sync on disk caches w/ loaded config in memory, ncf_obj may change */
1134         if (!config_obj) {
1135                 config_obj = nodemap_save_config_cache(env, dev, los);
1136                 if (IS_ERR(config_obj))
1137                         GOTO(out_ncf, rc = PTR_ERR(config_obj));
1138         }
1139
1140         ncf->ncf_obj = config_obj;
1141         ncf->ncf_los = los;
1142
1143         mutex_lock(&ncf_list_lock);
1144         list_add(&ncf->ncf_list, &ncf_list_head);
1145         mutex_unlock(&ncf_list_lock);
1146
1147 out_ncf:
1148         if (rc) {
1149                 OBD_FREE_PTR(ncf);
1150                 RETURN(ERR_PTR(rc));
1151         }
1152
1153         RETURN(ncf);
1154 }
1155 EXPORT_SYMBOL(nm_config_file_register_tgt);
1156
1157 /**
1158  * Deregister a nm_config_file. Should be called by targets during cleanup.
1159  *
1160  * \param ncf   config file to deregister
1161  */
1162 void nm_config_file_deregister_mgs(const struct lu_env *env,
1163                                    struct nm_config_file *ncf)
1164 {
1165         ENTRY;
1166         LASSERT(nodemap_mgs_ncf == ncf);
1167
1168         nodemap_mgs_ncf = NULL;
1169         if (ncf->ncf_obj)
1170                 lu_object_put(env, &ncf->ncf_obj->do_lu);
1171
1172         OBD_FREE_PTR(ncf);
1173
1174         EXIT;
1175 }
1176 EXPORT_SYMBOL(nm_config_file_deregister_mgs);
1177
1178 void nm_config_file_deregister_tgt(const struct lu_env *env,
1179                                    struct nm_config_file *ncf)
1180 {
1181         ENTRY;
1182
1183         if (ncf == NULL)
1184                 return;
1185
1186         mutex_lock(&ncf_list_lock);
1187         list_del(&ncf->ncf_list);
1188         mutex_unlock(&ncf_list_lock);
1189
1190         if (ncf->ncf_obj)
1191                 lu_object_put(env, &ncf->ncf_obj->do_lu);
1192
1193         OBD_FREE_PTR(ncf);
1194
1195         EXIT;
1196 }
1197 EXPORT_SYMBOL(nm_config_file_deregister_tgt);
1198
1199 int nodemap_process_idx_pages(struct nodemap_config *config, union lu_page *lip,
1200                               struct lu_nodemap **recent_nodemap)
1201 {
1202         struct nodemap_key *key;
1203         union nodemap_rec *rec;
1204         char *entry;
1205         int j;
1206         int k;
1207         int rc = 0;
1208         int size = dt_nodemap_features.dif_keysize_max +
1209                    dt_nodemap_features.dif_recsize_max;
1210         ENTRY;
1211
1212         for (j = 0; j < LU_PAGE_COUNT; j++) {
1213                 if (lip->lp_idx.lip_magic != LIP_MAGIC)
1214                         return -EINVAL;
1215
1216                 /* get and process keys and records from page */
1217                 for (k = 0; k < lip->lp_idx.lip_nr; k++) {
1218                         entry = lip->lp_idx.lip_entries + k * size;
1219                         key = (struct nodemap_key *)entry;
1220
1221                         entry += dt_nodemap_features.dif_keysize_max;
1222                         rec = (union nodemap_rec *)entry;
1223
1224                         rc = nodemap_process_keyrec(config, key, rec,
1225                                                     recent_nodemap);
1226                         if (rc < 0)
1227                                 return rc;
1228                 }
1229                 lip++;
1230         }
1231
1232         EXIT;
1233         return 0;
1234 }
1235 EXPORT_SYMBOL(nodemap_process_idx_pages);
1236
1237 int nodemap_index_read(struct lu_env *env,
1238                        struct nm_config_file *ncf,
1239                        struct idx_info *ii,
1240                        const struct lu_rdpg *rdpg)
1241 {
1242         struct dt_object        *nodemap_idx = ncf->ncf_obj;
1243         __u64                    version;
1244         int                      rc = 0;
1245
1246         ii->ii_keysize = dt_nodemap_features.dif_keysize_max;
1247         ii->ii_recsize = dt_nodemap_features.dif_recsize_max;
1248
1249         dt_read_lock(env, nodemap_idx, 0);
1250         version = dt_version_get(env, nodemap_idx);
1251         if (rdpg->rp_hash != 0 && ii->ii_version != version) {
1252                 CDEBUG(D_INFO, "nodemap config changed inflight, old %llu, new %llu\n",
1253                        ii->ii_version,
1254                        version);
1255                 ii->ii_hash_end = 0;
1256         } else {
1257                 rc = dt_index_walk(env, nodemap_idx, rdpg, NULL, ii);
1258                 CDEBUG(D_INFO, "walked index, hashend %llx\n", ii->ii_hash_end);
1259         }
1260
1261         if (rc >= 0)
1262                 ii->ii_version = version;
1263
1264         dt_read_unlock(env, nodemap_idx);
1265         return rc;
1266 }
1267 EXPORT_SYMBOL(nodemap_index_read);
1268
1269 /**
1270  * Returns the current nodemap configuration to MGC by walking the nodemap
1271  * config index and storing it in the response buffer.
1272  *
1273  * \param       req             incoming MGS_CONFIG_READ request
1274  * \retval      0               success
1275  * \retval      -EINVAL         malformed request
1276  * \retval      -ENOTCONN       client evicted/reconnected already
1277  * \retval      -ETIMEDOUT      client timeout or network error
1278  * \retval      -ENOMEM
1279  */
1280 int nodemap_get_config_req(struct obd_device *mgs_obd,
1281                            struct ptlrpc_request *req)
1282 {
1283         struct mgs_config_body *body;
1284         struct mgs_config_res *res;
1285         struct lu_rdpg rdpg;
1286         struct idx_info nodemap_ii;
1287         struct ptlrpc_bulk_desc *desc;
1288         struct l_wait_info lwi;
1289         struct tg_export_data *rqexp_ted = &req->rq_export->exp_target_data;
1290         int i;
1291         int page_count;
1292         int bytes = 0;
1293         int rc = 0;
1294
1295         body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
1296         if (!body)
1297                 RETURN(-EINVAL);
1298
1299         if (body->mcb_type != CONFIG_T_NODEMAP)
1300                 RETURN(-EINVAL);
1301
1302         rdpg.rp_count = (body->mcb_units << body->mcb_bits);
1303         rdpg.rp_npages = (rdpg.rp_count + PAGE_SIZE - 1) >>
1304                 PAGE_SHIFT;
1305         if (rdpg.rp_npages > PTLRPC_MAX_BRW_PAGES)
1306                 RETURN(-EINVAL);
1307
1308         CDEBUG(D_INFO, "reading nodemap log, name '%s', size = %u\n",
1309                body->mcb_name, rdpg.rp_count);
1310
1311         /* allocate pages to store the containers */
1312         OBD_ALLOC(rdpg.rp_pages, sizeof(*rdpg.rp_pages) * rdpg.rp_npages);
1313         if (rdpg.rp_pages == NULL)
1314                 RETURN(-ENOMEM);
1315         for (i = 0; i < rdpg.rp_npages; i++) {
1316                 rdpg.rp_pages[i] = alloc_page(GFP_NOFS);
1317                 if (rdpg.rp_pages[i] == NULL)
1318                         GOTO(out, rc = -ENOMEM);
1319         }
1320
1321         rdpg.rp_hash = body->mcb_offset;
1322         nodemap_ii.ii_magic = IDX_INFO_MAGIC;
1323         nodemap_ii.ii_flags = II_FL_NOHASH;
1324         nodemap_ii.ii_version = rqexp_ted->ted_nodemap_version;
1325
1326         bytes = nodemap_index_read(req->rq_svc_thread->t_env,
1327                                    mgs_obd->u.obt.obt_nodemap_config_file,
1328                                    &nodemap_ii, &rdpg);
1329         if (bytes < 0)
1330                 GOTO(out, rc = bytes);
1331
1332         rqexp_ted->ted_nodemap_version = nodemap_ii.ii_version;
1333
1334         res = req_capsule_server_get(&req->rq_pill, &RMF_MGS_CONFIG_RES);
1335         if (res == NULL)
1336                 GOTO(out, rc = -EINVAL);
1337         res->mcr_offset = nodemap_ii.ii_hash_end;
1338         res->mcr_size = bytes;
1339
1340         page_count = (bytes + PAGE_SIZE - 1) >> PAGE_SHIFT;
1341         LASSERT(page_count <= rdpg.rp_count);
1342         desc = ptlrpc_prep_bulk_exp(req, page_count, 1,
1343                                     PTLRPC_BULK_PUT_SOURCE |
1344                                         PTLRPC_BULK_BUF_KIOV,
1345                                     MGS_BULK_PORTAL,
1346                                     &ptlrpc_bulk_kiov_pin_ops);
1347         if (desc == NULL)
1348                 GOTO(out, rc = -ENOMEM);
1349
1350         for (i = 0; i < page_count && bytes > 0; i++) {
1351                 ptlrpc_prep_bulk_page_pin(desc, rdpg.rp_pages[i], 0,
1352                                           min_t(int, bytes, PAGE_SIZE));
1353                 bytes -= PAGE_SIZE;
1354         }
1355
1356         rc = target_bulk_io(req->rq_export, desc, &lwi);
1357         ptlrpc_free_bulk(desc);
1358
1359 out:
1360         if (rdpg.rp_pages != NULL) {
1361                 for (i = 0; i < rdpg.rp_npages; i++)
1362                         if (rdpg.rp_pages[i] != NULL)
1363                                 __free_page(rdpg.rp_pages[i]);
1364                 OBD_FREE(rdpg.rp_pages,
1365                          rdpg.rp_npages * sizeof(rdpg.rp_pages[0]));
1366         }
1367         return rc;
1368 }
1369 EXPORT_SYMBOL(nodemap_get_config_req);