lustre/fld/fld_request.c

   1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/fld/fld_request.c
   5  *  FLD (Fids Location Database)
   6  *
   7  *  Copyright (C) 2006 Cluster File Systems, Inc.
   8  *   Author: Yury Umanets <umka@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28 #ifndef EXPORT_SYMTAB
  29 # define EXPORT_SYMTAB
  30 #endif
  31 #define DEBUG_SUBSYSTEM S_FLD
  32
  33 #ifdef __KERNEL__
  34 # include <libcfs/libcfs.h>
  35 # include <linux/module.h>
  36 # include <linux/jbd.h>
  37 # include <asm/div64.h>
  38 #else /* __KERNEL__ */
  39 # include <liblustre.h>
  40 # include <libcfs/list.h>
  41 #endif
  42
  43 #include <obd.h>
  44 #include <obd_class.h>
  45 #include <lustre_ver.h>
  46 #include <obd_support.h>
  47 #include <lprocfs_status.h>
  48
  49 #include <dt_object.h>
  50 #include <md_object.h>
  51 #include <lustre_req_layout.h>
  52 #include <lustre_fld.h>
  53 #include <lustre_mdc.h>
  54 #include "fld_internal.h"
  55
  56 /* TODO: these 3 functions are copies of flow-control code from mdc_lib.c
  57  * It should be common thing. The same about mdc RPC lock */
  58 static int fld_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw)
  59 {
  60         int rc;
  61         ENTRY;
  62         spin_lock(&cli->cl_loi_list_lock);
  63         rc = list_empty(&mcw->mcw_entry);
  64         spin_unlock(&cli->cl_loi_list_lock);
  65         RETURN(rc);
  66 };
  67
  68 static void fld_enter_request(struct client_obd *cli)
  69 {
  70         struct mdc_cache_waiter mcw;
  71         struct l_wait_info lwi = { 0 };
  72
  73         spin_lock(&cli->cl_loi_list_lock);
  74         if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
  75                 list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters);
  76                 init_waitqueue_head(&mcw.mcw_waitq);
  77                 spin_unlock(&cli->cl_loi_list_lock);
  78                 l_wait_event(mcw.mcw_waitq, fld_req_avail(cli, &mcw), &lwi);
  79         } else {
  80                 cli->cl_r_in_flight++;
  81                 spin_unlock(&cli->cl_loi_list_lock);
  82         }
  83 }
  84
  85 static void fld_exit_request(struct client_obd *cli)
  86 {
  87         struct list_head *l, *tmp;
  88         struct mdc_cache_waiter *mcw;
  89
  90         spin_lock(&cli->cl_loi_list_lock);
  91         cli->cl_r_in_flight--;
  92         list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
  93
  94                 if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
  95                         /* No free request slots anymore */
  96                         break;
  97                 }
  98
  99                 mcw = list_entry(l, struct mdc_cache_waiter, mcw_entry);
 100                 list_del_init(&mcw->mcw_entry);
 101                 cli->cl_r_in_flight++;
 102                 wake_up(&mcw->mcw_waitq);
 103         }
 104         spin_unlock(&cli->cl_loi_list_lock);
 105 }
 106
 107 static int fld_rrb_hash(struct lu_client_fld *fld,
 108                         seqno_t seq)
 109 {
 110         LASSERT(fld->lcf_count > 0);
 111         return do_div(seq, fld->lcf_count);
 112 }
 113
 114 static struct lu_fld_target *
 115 fld_rrb_scan(struct lu_client_fld *fld, seqno_t seq)
 116 {
 117         struct lu_fld_target *target;
 118         int hash;
 119         ENTRY;
 120
 121         hash = fld_rrb_hash(fld, seq);
 122
 123         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
 124                 if (target->ft_idx == hash)
 125                         RETURN(target);
 126         }
 127
 128         CERROR("%s: Can't find target by hash %d (seq "LPX64"). "
 129                "Targets (%d):\n", fld->lcf_name, hash, seq,
 130                fld->lcf_count);
 131
 132         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
 133                 const char *srv_name = target->ft_srv != NULL  ?
 134                         target->ft_srv->lsf_name : "<null>";
 135                 const char *exp_name = target->ft_exp != NULL ?
 136                         (char *)target->ft_exp->exp_obd->obd_uuid.uuid :
 137                         "<null>";
 138
 139                 CERROR("  exp: 0x%p (%s), srv: 0x%p (%s), idx: "LPU64"\n",
 140                        target->ft_exp, exp_name, target->ft_srv,
 141                        srv_name, target->ft_idx);
 142         }
 143
 144         /*
 145          * If target is not found, there is logical error anyway, so here is
 146          * LBUG() to catch this situation.
 147          */
 148         LBUG();
 149         RETURN(NULL);
 150 }
 151
 152 static int fld_dht_hash(struct lu_client_fld *fld,
 153                         seqno_t seq)
 154 {
 155         /* XXX: here should be DHT hash */
 156         return fld_rrb_hash(fld, seq);
 157 }
 158
 159 static struct lu_fld_target *
 160 fld_dht_scan(struct lu_client_fld *fld, seqno_t seq)
 161 {
 162         /* XXX: here should be DHT scan code */
 163         return fld_rrb_scan(fld, seq);
 164 }
 165
 166 struct lu_fld_hash fld_hash[3] = {
 167         {
 168                 .fh_name = "DHT",
 169                 .fh_hash_func = fld_dht_hash,
 170                 .fh_scan_func = fld_dht_scan
 171         },
 172         {
 173                 .fh_name = "RRB",
 174                 .fh_hash_func = fld_rrb_hash,
 175                 .fh_scan_func = fld_rrb_scan
 176         },
 177         {
 178                 0,
 179         }
 180 };
 181
 182 static struct lu_fld_target *
 183 fld_client_get_target(struct lu_client_fld *fld,
 184                       seqno_t seq)
 185 {
 186         struct lu_fld_target *target;
 187         ENTRY;
 188
 189         LASSERT(fld->lcf_hash != NULL);
 190
 191         spin_lock(&fld->lcf_lock);
 192         target = fld->lcf_hash->fh_scan_func(fld, seq);
 193         spin_unlock(&fld->lcf_lock);
 194
 195         if (target != NULL) {
 196                 CDEBUG(D_INFO, "%s: Found target (idx "LPU64
 197                        ") by seq "LPX64"\n", fld->lcf_name,
 198                        target->ft_idx, seq);
 199         }
 200
 201         RETURN(target);
 202 }
 203
 204 /*
 205  * Add export to FLD. This is usually done by CMM and LMV as they are main users
 206  * of FLD module.
 207  */
 208 int fld_client_add_target(struct lu_client_fld *fld,
 209                           struct lu_fld_target *tar)
 210 {
 211         const char *name = fld_target_name(tar);
 212         struct lu_fld_target *target, *tmp;
 213         ENTRY;
 214
 215         LASSERT(tar != NULL);
 216         LASSERT(name != NULL);
 217         LASSERT(tar->ft_srv != NULL || tar->ft_exp != NULL);
 218
 219         if (fld->lcf_flags != LUSTRE_FLD_INIT) {
 220                 CERROR("%s: Attempt to add target %s (idx "LPU64") "
 221                        "on fly - skip it\n", fld->lcf_name, name,
 222                        tar->ft_idx);
 223                 RETURN(0);
 224         } else {
 225                 CDEBUG(D_INFO, "%s: Adding target %s (idx "
 226                        LPU64")\n", fld->lcf_name, name, tar->ft_idx);
 227         }
 228
 229         OBD_ALLOC_PTR(target);
 230         if (target == NULL)
 231                 RETURN(-ENOMEM);
 232
 233         spin_lock(&fld->lcf_lock);
 234         list_for_each_entry(tmp, &fld->lcf_targets, ft_chain) {
 235                 if (tmp->ft_idx == tar->ft_idx) {
 236                         spin_unlock(&fld->lcf_lock);
 237                         OBD_FREE_PTR(target);
 238                         CERROR("Target %s exists in FLD and known as %s:#"LPU64"\n",
 239                                name, fld_target_name(tmp), tmp->ft_idx);
 240                         RETURN(-EEXIST);
 241                 }
 242         }
 243
 244         target->ft_exp = tar->ft_exp;
 245         if (target->ft_exp != NULL)
 246                 class_export_get(target->ft_exp);
 247         target->ft_srv = tar->ft_srv;
 248         target->ft_idx = tar->ft_idx;
 249
 250         list_add_tail(&target->ft_chain,
 251                       &fld->lcf_targets);
 252
 253         fld->lcf_count++;
 254         spin_unlock(&fld->lcf_lock);
 255
 256         RETURN(0);
 257 }
 258 EXPORT_SYMBOL(fld_client_add_target);
 259
 260 /* Remove export from FLD */
 261 int fld_client_del_target(struct lu_client_fld *fld,
 262                           __u64 idx)
 263 {
 264         struct lu_fld_target *target, *tmp;
 265         ENTRY;
 266
 267         spin_lock(&fld->lcf_lock);
 268         list_for_each_entry_safe(target, tmp,
 269                                  &fld->lcf_targets, ft_chain) {
 270                 if (target->ft_idx == idx) {
 271                         fld->lcf_count--;
 272                         list_del(&target->ft_chain);
 273                         spin_unlock(&fld->lcf_lock);
 274
 275                         if (target->ft_exp != NULL)
 276                                 class_export_put(target->ft_exp);
 277
 278                         OBD_FREE_PTR(target);
 279                         RETURN(0);
 280                 }
 281         }
 282         spin_unlock(&fld->lcf_lock);
 283         RETURN(-ENOENT);
 284 }
 285 EXPORT_SYMBOL(fld_client_del_target);
 286
 287 static void fld_client_proc_fini(struct lu_client_fld *fld);
 288
 289 #ifdef LPROCFS
 290 static int fld_client_proc_init(struct lu_client_fld *fld)
 291 {
 292         int rc;
 293         ENTRY;
 294
 295         fld->lcf_proc_dir = lprocfs_register(fld->lcf_name,
 296                                              fld_type_proc_dir,
 297                                              NULL, NULL);
 298
 299         if (IS_ERR(fld->lcf_proc_dir)) {
 300                 CERROR("%s: LProcFS failed in fld-init\n",
 301                        fld->lcf_name);
 302                 rc = PTR_ERR(fld->lcf_proc_dir);
 303                 RETURN(rc);
 304         }
 305
 306         rc = lprocfs_add_vars(fld->lcf_proc_dir,
 307                               fld_client_proc_list, fld);
 308         if (rc) {
 309                 CERROR("%s: Can't init FLD proc, rc %d\n",
 310                        fld->lcf_name, rc);
 311                 GOTO(out_cleanup, rc);
 312         }
 313
 314         RETURN(0);
 315
 316 out_cleanup:
 317         fld_client_proc_fini(fld);
 318         return rc;
 319 }
 320
 321 static void fld_client_proc_fini(struct lu_client_fld *fld)
 322 {
 323         ENTRY;
 324         if (fld->lcf_proc_dir) {
 325                 if (!IS_ERR(fld->lcf_proc_dir))
 326                         lprocfs_remove(&fld->lcf_proc_dir);
 327                 fld->lcf_proc_dir = NULL;
 328         }
 329         EXIT;
 330 }
 331 #else
 332 static int fld_client_proc_init(struct lu_client_fld *fld)
 333 {
 334         return 0;
 335 }
 336
 337 static void fld_client_proc_fini(struct lu_client_fld *fld)
 338 {
 339         return;
 340 }
 341 #endif
 342
 343 static inline int hash_is_sane(int hash)
 344 {
 345         return (hash >= 0 && hash < ARRAY_SIZE(fld_hash));
 346 }
 347
 348 int fld_client_init(struct lu_client_fld *fld,
 349                     const char *prefix, int hash)
 350 {
 351 #ifdef __KERNEL__
 352         int cache_size, cache_threshold;
 353 #endif
 354         int rc;
 355         ENTRY;
 356
 357         LASSERT(fld != NULL);
 358
 359         snprintf(fld->lcf_name, sizeof(fld->lcf_name),
 360                  "cli-%s", prefix);
 361
 362         if (!hash_is_sane(hash)) {
 363                 CERROR("%s: Wrong hash function %#x\n",
 364                        fld->lcf_name, hash);
 365                 RETURN(-EINVAL);
 366         }
 367
 368         fld->lcf_count = 0;
 369         spin_lock_init(&fld->lcf_lock);
 370         fld->lcf_hash = &fld_hash[hash];
 371         fld->lcf_flags = LUSTRE_FLD_INIT;
 372         INIT_LIST_HEAD(&fld->lcf_targets);
 373
 374 #ifdef __KERNEL__
 375         cache_size = FLD_CLIENT_CACHE_SIZE /
 376                 sizeof(struct fld_cache_entry);
 377
 378         cache_threshold = cache_size *
 379                 FLD_CLIENT_CACHE_THRESHOLD / 100;
 380
 381         fld->lcf_cache = fld_cache_init(fld->lcf_name,
 382                                         FLD_CLIENT_HTABLE_SIZE,
 383                                         cache_size, cache_threshold);
 384         if (IS_ERR(fld->lcf_cache)) {
 385                 rc = PTR_ERR(fld->lcf_cache);
 386                 fld->lcf_cache = NULL;
 387                 GOTO(out, rc);
 388         }
 389 #endif
 390
 391         rc = fld_client_proc_init(fld);
 392         if (rc)
 393                 GOTO(out, rc);
 394         EXIT;
 395 out:
 396         if (rc)
 397                 fld_client_fini(fld);
 398         else
 399                 CDEBUG(D_INFO, "%s: Using \"%s\" hash\n",
 400                        fld->lcf_name, fld->lcf_hash->fh_name);
 401         return rc;
 402 }
 403 EXPORT_SYMBOL(fld_client_init);
 404
 405 void fld_client_fini(struct lu_client_fld *fld)
 406 {
 407         struct lu_fld_target *target, *tmp;
 408         ENTRY;
 409
 410         fld_client_proc_fini(fld);
 411
 412         spin_lock(&fld->lcf_lock);
 413         list_for_each_entry_safe(target, tmp,
 414                                  &fld->lcf_targets, ft_chain) {
 415                 fld->lcf_count--;
 416                 list_del(&target->ft_chain);
 417                 if (target->ft_exp != NULL)
 418                         class_export_put(target->ft_exp);
 419                 OBD_FREE_PTR(target);
 420         }
 421         spin_unlock(&fld->lcf_lock);
 422
 423 #ifdef __KERNEL__
 424         if (fld->lcf_cache != NULL) {
 425                 if (!IS_ERR(fld->lcf_cache))
 426                         fld_cache_fini(fld->lcf_cache);
 427                 fld->lcf_cache = NULL;
 428         }
 429 #endif
 430
 431         EXIT;
 432 }
 433 EXPORT_SYMBOL(fld_client_fini);
 434
 435 static int fld_client_rpc(struct obd_export *exp,
 436                           struct md_fld *mf, __u32 fld_op)
 437 {
 438         int size[3] = { sizeof(struct ptlrpc_body),
 439                         sizeof(__u32),
 440                         sizeof(struct md_fld) };
 441         struct ptlrpc_request *req;
 442         struct req_capsule pill;
 443         struct md_fld *pmf;
 444         __u32 *op;
 445         int rc;
 446         ENTRY;
 447
 448         LASSERT(exp != NULL);
 449
 450         req = ptlrpc_prep_req(class_exp2cliimp(exp),
 451                               LUSTRE_MDS_VERSION,
 452                               FLD_QUERY, 3, size,
 453                               NULL);
 454         if (req == NULL)
 455                 RETURN(-ENOMEM);
 456
 457         req_capsule_init(&pill, req, RCL_CLIENT, NULL);
 458         req_capsule_set(&pill, &RQF_FLD_QUERY);
 459
 460         op = req_capsule_client_get(&pill, &RMF_FLD_OPC);
 461         *op = fld_op;
 462
 463         pmf = req_capsule_client_get(&pill, &RMF_FLD_MDFLD);
 464         *pmf = *mf;
 465
 466         size[1] = sizeof(struct md_fld);
 467         ptlrpc_req_set_repsize(req, 2, size);
 468         req->rq_request_portal = FLD_REQUEST_PORTAL;
 469
 470         if (fld_op != FLD_LOOKUP)
 471                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 472         fld_enter_request(&exp->exp_obd->u.cli);
 473         rc = ptlrpc_queue_wait(req);
 474         fld_exit_request(&exp->exp_obd->u.cli);
 475         if (fld_op != FLD_LOOKUP)
 476                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 477         if (rc)
 478                 GOTO(out_req, rc);
 479
 480         pmf = req_capsule_server_get(&pill, &RMF_FLD_MDFLD);
 481         if (pmf == NULL)
 482                 GOTO(out_req, rc = -EFAULT);
 483         *mf = *pmf;
 484         EXIT;
 485 out_req:
 486         req_capsule_fini(&pill);
 487         ptlrpc_req_finished(req);
 488         return rc;
 489 }
 490
 491 int fld_client_create(struct lu_client_fld *fld,
 492                       seqno_t seq, mdsno_t mds,
 493                       const struct lu_env *env)
 494 {
 495         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = mds };
 496         struct lu_fld_target *target;
 497         int rc;
 498         ENTRY;
 499
 500         fld->lcf_flags |= LUSTRE_FLD_RUN;
 501         target = fld_client_get_target(fld, seq);
 502         LASSERT(target != NULL);
 503
 504         CDEBUG(D_INFO, "%s: Create fld entry (seq: "LPX64"; mds: "
 505                LPU64") on target %s (idx "LPU64")\n", fld->lcf_name,
 506                seq, mds, fld_target_name(target), target->ft_idx);
 507
 508 #ifdef __KERNEL__
 509         if (target->ft_srv != NULL) {
 510                 LASSERT(env != NULL);
 511                 rc = fld_server_create(target->ft_srv, env, seq, mds);
 512         } else {
 513 #endif
 514                 rc = fld_client_rpc(target->ft_exp, &md_fld, FLD_CREATE);
 515 #ifdef __KERNEL__
 516         }
 517 #endif
 518
 519         if (rc == 0) {
 520                 /*
 521                  * Do not return result of calling fld_cache_insert()
 522                  * here. First of all because it may return -EEXISTS. Another
 523                  * reason is that, we do not want to stop proceeding because of
 524                  * cache errors.
 525                  */
 526                 fld_cache_insert(fld->lcf_cache, seq, mds);
 527         } else {
 528                 CERROR("%s: Can't create FLD entry, rc %d\n",
 529                        fld->lcf_name, rc);
 530         }
 531
 532         RETURN(rc);
 533 }
 534 EXPORT_SYMBOL(fld_client_create);
 535
 536 int fld_client_delete(struct lu_client_fld *fld, seqno_t seq,
 537                       const struct lu_env *env)
 538 {
 539         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
 540         struct lu_fld_target *target;
 541         int rc;
 542         ENTRY;
 543
 544         fld->lcf_flags |= LUSTRE_FLD_RUN;
 545         fld_cache_delete(fld->lcf_cache, seq);
 546
 547         target = fld_client_get_target(fld, seq);
 548         LASSERT(target != NULL);
 549
 550         CDEBUG(D_INFO, "%s: Delete fld entry (seq: "LPX64") on "
 551                "target %s (idx "LPU64")\n", fld->lcf_name, seq,
 552                fld_target_name(target), target->ft_idx);
 553
 554 #ifdef __KERNEL__
 555         if (target->ft_srv != NULL) {
 556                 LASSERT(env != NULL);
 557                 rc = fld_server_delete(target->ft_srv,
 558                                        env, seq);
 559         } else {
 560 #endif
 561                 rc = fld_client_rpc(target->ft_exp,
 562                                     &md_fld, FLD_DELETE);
 563 #ifdef __KERNEL__
 564         }
 565 #endif
 566
 567         RETURN(rc);
 568 }
 569 EXPORT_SYMBOL(fld_client_delete);
 570
 571 int fld_client_lookup(struct lu_client_fld *fld,
 572                       seqno_t seq, mdsno_t *mds,
 573                       const struct lu_env *env)
 574 {
 575         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
 576         struct lu_fld_target *target;
 577         int rc;
 578         ENTRY;
 579
 580         fld->lcf_flags |= LUSTRE_FLD_RUN;
 581
 582         rc = fld_cache_lookup(fld->lcf_cache, seq, mds);
 583         if (rc == 0)
 584                 RETURN(0);
 585
 586         /* Can not find it in the cache */
 587         target = fld_client_get_target(fld, seq);
 588         LASSERT(target != NULL);
 589
 590         CDEBUG(D_INFO, "%s: Lookup fld entry (seq: "LPX64") on "
 591                "target %s (idx "LPU64")\n", fld->lcf_name, seq,
 592                fld_target_name(target), target->ft_idx);
 593
 594 #ifdef __KERNEL__
 595         if (target->ft_srv != NULL) {
 596                 LASSERT(env != NULL);
 597                 rc = fld_server_lookup(target->ft_srv,
 598                                        env, seq, &md_fld.mf_mds);
 599         } else {
 600 #endif
 601                 /*
 602                  * insert the 'inflight' sequence. No need to protect that,
 603                  * we are trying to reduce numbers of RPC but not restrict
 604                  * to them exactly one
 605                  */
 606                 fld_cache_insert_inflight(fld->lcf_cache, seq);
 607                 rc = fld_client_rpc(target->ft_exp,
 608                                     &md_fld, FLD_LOOKUP);
 609 #ifdef __KERNEL__
 610         }
 611 #endif
 612         if (rc == 0) {
 613                 *mds = md_fld.mf_mds;
 614
 615                 /*
 616                  * Do not return error here as well. See previous comment in
 617                  * same situation in function fld_client_create().
 618                  */
 619                 fld_cache_insert(fld->lcf_cache, seq, *mds);
 620         } else {
 621                 /* remove 'inflight' seq if it exists */
 622                 fld_cache_delete(fld->lcf_cache, seq);
 623         }
 624         RETURN(rc);
 625 }
 626 EXPORT_SYMBOL(fld_client_lookup);
 627
 628 void fld_client_flush(struct lu_client_fld *fld)
 629 {
 630 #ifdef __KERNEL__
 631         fld_cache_flush(fld->lcf_cache);
 632 #endif
 633 }
 634 EXPORT_SYMBOL(fld_client_flush);
 635