lustre/fld/fld_request.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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/fld/fld_request.c
  37  *
  38  * FLD (Fids Location Database)
  39  *
  40  * Author: Yury Umanets <umka@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_FLD
  44
  45 #ifdef __KERNEL__
  46 # include <libcfs/libcfs.h>
  47 # include <linux/module.h>
  48 # include <linux/math64.h>
  49 #else /* __KERNEL__ */
  50 # include <liblustre.h>
  51 # include <libcfs/list.h>
  52 #endif
  53
  54 #include <obd.h>
  55 #include <obd_class.h>
  56 #include <obd_support.h>
  57 #include <lprocfs_status.h>
  58 #include <lustre_req_layout.h>
  59 #include <lustre_fld.h>
  60 #include <lustre_mdc.h>
  61 #include "fld_internal.h"
  62
  63 static int fld_rrb_hash(struct lu_client_fld *fld,
  64                         seqno_t seq)
  65 {
  66         LASSERT(fld->lcf_count > 0);
  67         return do_div(seq, fld->lcf_count);
  68 }
  69
  70 static struct lu_fld_target *
  71 fld_rrb_scan(struct lu_client_fld *fld, seqno_t seq)
  72 {
  73         struct lu_fld_target *target;
  74         int hash;
  75         ENTRY;
  76
  77         /* Because almost all of special sequence located in MDT0,
  78          * it should go to index 0 directly, instead of calculating
  79          * hash again, and also if other MDTs is not being connected,
  80          * the fld lookup requests(for seq on MDT0) should not be
  81          * blocked because of other MDTs */
  82         if (fid_seq_is_norm(seq))
  83                 hash = fld_rrb_hash(fld, seq);
  84         else
  85                 hash = 0;
  86
  87 again:
  88         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
  89                 if (target->ft_idx == hash)
  90                         RETURN(target);
  91         }
  92
  93         if (hash != 0) {
  94                 /* It is possible the remote target(MDT) are not connected to
  95                  * with client yet, so we will refer this to MDT0, which should
  96                  * be connected during mount */
  97                 hash = 0;
  98                 goto again;
  99         }
 100
 101         CERROR("%s: Can't find target by hash %d (seq "LPX64"). "
 102                "Targets (%d):\n", fld->lcf_name, hash, seq,
 103                fld->lcf_count);
 104
 105         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
 106                 const char *srv_name = target->ft_srv != NULL  ?
 107                         target->ft_srv->lsf_name : "<null>";
 108                 const char *exp_name = target->ft_exp != NULL ?
 109                         (char *)target->ft_exp->exp_obd->obd_uuid.uuid :
 110                         "<null>";
 111
 112                 CERROR("  exp: 0x%p (%s), srv: 0x%p (%s), idx: "LPU64"\n",
 113                        target->ft_exp, exp_name, target->ft_srv,
 114                        srv_name, target->ft_idx);
 115         }
 116
 117         /*
 118          * If target is not found, there is logical error anyway, so here is
 119          * LBUG() to catch this situation.
 120          */
 121         LBUG();
 122         RETURN(NULL);
 123 }
 124
 125 struct lu_fld_hash fld_hash[] = {
 126         {
 127                 .fh_name = "RRB",
 128                 .fh_hash_func = fld_rrb_hash,
 129                 .fh_scan_func = fld_rrb_scan
 130         },
 131         {
 132                 0,
 133         }
 134 };
 135
 136 static struct lu_fld_target *
 137 fld_client_get_target(struct lu_client_fld *fld, seqno_t seq)
 138 {
 139         struct lu_fld_target *target;
 140         ENTRY;
 141
 142         LASSERT(fld->lcf_hash != NULL);
 143
 144         spin_lock(&fld->lcf_lock);
 145         target = fld->lcf_hash->fh_scan_func(fld, seq);
 146         spin_unlock(&fld->lcf_lock);
 147
 148         if (target != NULL) {
 149                 CDEBUG(D_INFO, "%s: Found target (idx "LPU64
 150                        ") by seq "LPX64"\n", fld->lcf_name,
 151                        target->ft_idx, seq);
 152         }
 153
 154         RETURN(target);
 155 }
 156
 157 /*
 158  * Add export to FLD. This is usually done by CMM and LMV as they are main users
 159  * of FLD module.
 160  */
 161 int fld_client_add_target(struct lu_client_fld *fld,
 162                           struct lu_fld_target *tar)
 163 {
 164         const char *name;
 165         struct lu_fld_target *target, *tmp;
 166         ENTRY;
 167
 168         LASSERT(tar != NULL);
 169         name = fld_target_name(tar);
 170         LASSERT(name != NULL);
 171         LASSERT(tar->ft_srv != NULL || tar->ft_exp != NULL);
 172
 173         if (fld->lcf_flags != LUSTRE_FLD_INIT) {
 174                 CERROR("%s: Attempt to add target %s (idx "LPU64") "
 175                        "on fly - skip it\n", fld->lcf_name, name,
 176                        tar->ft_idx);
 177                 RETURN(0);
 178         } else {
 179                 CDEBUG(D_INFO, "%s: Adding target %s (idx "
 180                        LPU64")\n", fld->lcf_name, name, tar->ft_idx);
 181         }
 182
 183         OBD_ALLOC_PTR(target);
 184         if (target == NULL)
 185                 RETURN(-ENOMEM);
 186
 187         spin_lock(&fld->lcf_lock);
 188         list_for_each_entry(tmp, &fld->lcf_targets, ft_chain) {
 189                 if (tmp->ft_idx == tar->ft_idx) {
 190                         spin_unlock(&fld->lcf_lock);
 191                         OBD_FREE_PTR(target);
 192                         CERROR("Target %s exists in FLD and known as %s:#"LPU64"\n",
 193                                name, fld_target_name(tmp), tmp->ft_idx);
 194                         RETURN(-EEXIST);
 195                 }
 196         }
 197
 198         target->ft_exp = tar->ft_exp;
 199         if (target->ft_exp != NULL)
 200                 class_export_get(target->ft_exp);
 201         target->ft_srv = tar->ft_srv;
 202         target->ft_idx = tar->ft_idx;
 203
 204         list_add_tail(&target->ft_chain, &fld->lcf_targets);
 205
 206         fld->lcf_count++;
 207         spin_unlock(&fld->lcf_lock);
 208
 209         RETURN(0);
 210 }
 211 EXPORT_SYMBOL(fld_client_add_target);
 212
 213 /* Remove export from FLD */
 214 int fld_client_del_target(struct lu_client_fld *fld, __u64 idx)
 215 {
 216         struct lu_fld_target *target, *tmp;
 217         ENTRY;
 218
 219         spin_lock(&fld->lcf_lock);
 220         list_for_each_entry_safe(target, tmp, &fld->lcf_targets, ft_chain) {
 221                 if (target->ft_idx == idx) {
 222                         fld->lcf_count--;
 223                         list_del(&target->ft_chain);
 224                         spin_unlock(&fld->lcf_lock);
 225
 226                         if (target->ft_exp != NULL)
 227                                 class_export_put(target->ft_exp);
 228
 229                         OBD_FREE_PTR(target);
 230                         RETURN(0);
 231                 }
 232         }
 233         spin_unlock(&fld->lcf_lock);
 234         RETURN(-ENOENT);
 235 }
 236 EXPORT_SYMBOL(fld_client_del_target);
 237
 238 #ifdef LPROCFS
 239 static int fld_client_proc_init(struct lu_client_fld *fld)
 240 {
 241         int rc;
 242         ENTRY;
 243
 244         fld->lcf_proc_dir = lprocfs_seq_register(fld->lcf_name,
 245                                                  fld_type_proc_dir,
 246                                                  NULL, NULL);
 247         if (IS_ERR(fld->lcf_proc_dir)) {
 248                 CERROR("%s: LProcFS failed in fld-init\n",
 249                        fld->lcf_name);
 250                 rc = PTR_ERR(fld->lcf_proc_dir);
 251                 RETURN(rc);
 252         }
 253
 254         rc = lprocfs_seq_add_vars(fld->lcf_proc_dir,
 255                                   fld_client_proc_list, fld);
 256         if (rc) {
 257                 CERROR("%s: Can't init FLD proc, rc %d\n",
 258                        fld->lcf_name, rc);
 259                 GOTO(out_cleanup, rc);
 260         }
 261
 262         RETURN(0);
 263
 264 out_cleanup:
 265         fld_client_proc_fini(fld);
 266         return rc;
 267 }
 268
 269 void fld_client_proc_fini(struct lu_client_fld *fld)
 270 {
 271         ENTRY;
 272         if (fld->lcf_proc_dir) {
 273                 if (!IS_ERR(fld->lcf_proc_dir))
 274                         lprocfs_remove(&fld->lcf_proc_dir);
 275                 fld->lcf_proc_dir = NULL;
 276         }
 277         EXIT;
 278 }
 279 #else
 280 static int fld_client_proc_init(struct lu_client_fld *fld)
 281 {
 282         return 0;
 283 }
 284
 285 void fld_client_proc_fini(struct lu_client_fld *fld)
 286 {
 287         return;
 288 }
 289 #endif
 290
 291 EXPORT_SYMBOL(fld_client_proc_fini);
 292
 293 static inline int hash_is_sane(int hash)
 294 {
 295         return (hash >= 0 && hash < ARRAY_SIZE(fld_hash));
 296 }
 297
 298 int fld_client_init(struct lu_client_fld *fld,
 299                     const char *prefix, int hash)
 300 {
 301         int cache_size, cache_threshold;
 302         int rc;
 303         ENTRY;
 304
 305         LASSERT(fld != NULL);
 306
 307         snprintf(fld->lcf_name, sizeof(fld->lcf_name),
 308                  "cli-%s", prefix);
 309
 310         if (!hash_is_sane(hash)) {
 311                 CERROR("%s: Wrong hash function %#x\n",
 312                        fld->lcf_name, hash);
 313                 RETURN(-EINVAL);
 314         }
 315
 316         fld->lcf_count = 0;
 317         spin_lock_init(&fld->lcf_lock);
 318         fld->lcf_hash = &fld_hash[hash];
 319         fld->lcf_flags = LUSTRE_FLD_INIT;
 320         INIT_LIST_HEAD(&fld->lcf_targets);
 321
 322         cache_size = FLD_CLIENT_CACHE_SIZE /
 323                 sizeof(struct fld_cache_entry);
 324
 325         cache_threshold = cache_size *
 326                 FLD_CLIENT_CACHE_THRESHOLD / 100;
 327
 328         fld->lcf_cache = fld_cache_init(fld->lcf_name,
 329                                         cache_size, cache_threshold);
 330         if (IS_ERR(fld->lcf_cache)) {
 331                 rc = PTR_ERR(fld->lcf_cache);
 332                 fld->lcf_cache = NULL;
 333                 GOTO(out, rc);
 334         }
 335
 336         rc = fld_client_proc_init(fld);
 337         if (rc)
 338                 GOTO(out, rc);
 339         EXIT;
 340 out:
 341         if (rc)
 342                 fld_client_fini(fld);
 343         else
 344                 CDEBUG(D_INFO, "%s: Using \"%s\" hash\n",
 345                        fld->lcf_name, fld->lcf_hash->fh_name);
 346         return rc;
 347 }
 348 EXPORT_SYMBOL(fld_client_init);
 349
 350 void fld_client_fini(struct lu_client_fld *fld)
 351 {
 352         struct lu_fld_target *target, *tmp;
 353         ENTRY;
 354
 355         spin_lock(&fld->lcf_lock);
 356         list_for_each_entry_safe(target, tmp, &fld->lcf_targets, ft_chain) {
 357                 fld->lcf_count--;
 358                 list_del(&target->ft_chain);
 359                 if (target->ft_exp != NULL)
 360                         class_export_put(target->ft_exp);
 361                 OBD_FREE_PTR(target);
 362         }
 363         spin_unlock(&fld->lcf_lock);
 364
 365         if (fld->lcf_cache != NULL) {
 366                 if (!IS_ERR(fld->lcf_cache))
 367                         fld_cache_fini(fld->lcf_cache);
 368                 fld->lcf_cache = NULL;
 369         }
 370
 371         EXIT;
 372 }
 373 EXPORT_SYMBOL(fld_client_fini);
 374
 375 int fld_client_rpc(struct obd_export *exp,
 376                    struct lu_seq_range *range, __u32 fld_op,
 377                    struct ptlrpc_request **reqp)
 378 {
 379         struct ptlrpc_request *req = NULL;
 380         struct lu_seq_range   *prange;
 381         __u32                 *op;
 382         int                    rc = 0;
 383         struct obd_import     *imp;
 384         ENTRY;
 385
 386         LASSERT(exp != NULL);
 387
 388 again:
 389         imp = class_exp2cliimp(exp);
 390         switch (fld_op) {
 391         case FLD_QUERY:
 392                 req = ptlrpc_request_alloc_pack(imp, &RQF_FLD_QUERY,
 393                                                 LUSTRE_MDS_VERSION, FLD_QUERY);
 394                 if (req == NULL)
 395                         RETURN(-ENOMEM);
 396
 397                 /* XXX: only needed when talking to old server(< 2.6), it should
 398                  * be removed when < 2.6 server is not supported */
 399                 op = req_capsule_client_get(&req->rq_pill, &RMF_FLD_OPC);
 400                 *op = FLD_LOOKUP;
 401
 402                 /* For MDS_MDS seq lookup, it will always use LWP connection,
 403                  * but LWP will be evicted after restart, so cause the error.
 404                  * so we will set no_delay for seq lookup request, once the
 405                  * request fails because of the eviction. always retry here */
 406                 if (imp->imp_connect_flags_orig & OBD_CONNECT_MDS_MDS) {
 407                         req->rq_allow_replay = 1;
 408                         req->rq_no_delay = 1;
 409                 }
 410                 break;
 411         case FLD_READ:
 412                 req = ptlrpc_request_alloc_pack(imp, &RQF_FLD_READ,
 413                                                 LUSTRE_MDS_VERSION, FLD_READ);
 414                 if (req == NULL)
 415                         RETURN(-ENOMEM);
 416
 417                 req_capsule_set_size(&req->rq_pill, &RMF_GENERIC_DATA,
 418                                      RCL_SERVER, PAGE_CACHE_SIZE);
 419                 break;
 420         default:
 421                 rc = -EINVAL;
 422                 break;
 423         }
 424
 425         if (rc != 0)
 426                 RETURN(rc);
 427
 428         prange = req_capsule_client_get(&req->rq_pill, &RMF_FLD_MDFLD);
 429         *prange = *range;
 430         ptlrpc_request_set_replen(req);
 431         req->rq_request_portal = FLD_REQUEST_PORTAL;
 432         req->rq_reply_portal = MDC_REPLY_PORTAL;
 433         ptlrpc_at_set_req_timeout(req);
 434
 435         obd_get_request_slot(&exp->exp_obd->u.cli);
 436         rc = ptlrpc_queue_wait(req);
 437         obd_put_request_slot(&exp->exp_obd->u.cli);
 438         if (rc != 0) {
 439                 if (rc == -EWOULDBLOCK) {
 440                         /* For no_delay req(see above), EWOULDBLOCK means the
 441                          * connection is being evicted, but this seq lookup
 442                          * should not return error, since it would cause
 443                          * unecessary failure of the application, instead
 444                          * it should retry here */
 445                         ptlrpc_req_finished(req);
 446                         rc = 0;
 447                         goto again;
 448                 }
 449                 GOTO(out_req, rc);
 450         }
 451
 452         if (fld_op == FLD_QUERY) {
 453                 prange = req_capsule_server_get(&req->rq_pill,
 454                                                 &RMF_FLD_MDFLD);
 455                 if (prange == NULL)
 456                         GOTO(out_req, rc = -EFAULT);
 457                 *range = *prange;
 458         }
 459
 460         EXIT;
 461 out_req:
 462         if (rc != 0 || reqp == NULL) {
 463                 ptlrpc_req_finished(req);
 464                 req = NULL;
 465         }
 466
 467         if (reqp != NULL)
 468                 *reqp = req;
 469
 470         return rc;
 471 }
 472
 473 int fld_client_lookup(struct lu_client_fld *fld, seqno_t seq, mdsno_t *mds,
 474                       __u32 flags, const struct lu_env *env)
 475 {
 476         struct lu_seq_range res = { 0 };
 477         struct lu_fld_target *target;
 478         int rc;
 479         ENTRY;
 480
 481         fld->lcf_flags |= LUSTRE_FLD_RUN;
 482
 483         rc = fld_cache_lookup(fld->lcf_cache, seq, &res);
 484         if (rc == 0) {
 485                 *mds = res.lsr_index;
 486                 RETURN(0);
 487         }
 488
 489         /* Can not find it in the cache */
 490         target = fld_client_get_target(fld, seq);
 491         LASSERT(target != NULL);
 492
 493         CDEBUG(D_INFO, "%s: Lookup fld entry (seq: "LPX64") on "
 494                "target %s (idx "LPU64")\n", fld->lcf_name, seq,
 495                fld_target_name(target), target->ft_idx);
 496
 497         res.lsr_start = seq;
 498         fld_range_set_type(&res, flags);
 499
 500 #if defined(__KERNEL__) && defined(HAVE_SERVER_SUPPORT)
 501         if (target->ft_srv != NULL) {
 502                 LASSERT(env != NULL);
 503                 rc = fld_server_lookup(env, target->ft_srv, seq, &res);
 504         } else
 505 #endif
 506         {
 507                 rc = fld_client_rpc(target->ft_exp, &res, FLD_QUERY, NULL);
 508         }
 509
 510         if (rc == 0) {
 511                 *mds = res.lsr_index;
 512                 fld_cache_insert(fld->lcf_cache, &res);
 513         }
 514
 515         RETURN(rc);
 516 }
 517 EXPORT_SYMBOL(fld_client_lookup);
 518
 519 void fld_client_flush(struct lu_client_fld *fld)
 520 {
 521         fld_cache_flush(fld->lcf_cache);
 522 }
 523 EXPORT_SYMBOL(fld_client_flush);
 524
 525 #ifdef __KERNEL__
 526
 527 struct proc_dir_entry *fld_type_proc_dir;
 528
 529 static int __init fld_mod_init(void)
 530 {
 531         fld_type_proc_dir = lprocfs_seq_register(LUSTRE_FLD_NAME,
 532                                                  proc_lustre_root,
 533                                                  NULL, NULL);
 534         if (IS_ERR(fld_type_proc_dir))
 535                 return PTR_ERR(fld_type_proc_dir);
 536
 537 #ifdef HAVE_SERVER_SUPPORT
 538         fld_server_mod_init();
 539 #endif
 540
 541         return 0;
 542 }
 543
 544 static void __exit fld_mod_exit(void)
 545 {
 546 #ifdef HAVE_SERVER_SUPPORT
 547         fld_server_mod_exit();
 548 #endif
 549
 550         if (fld_type_proc_dir != NULL && !IS_ERR(fld_type_proc_dir)) {
 551                 lprocfs_remove(&fld_type_proc_dir);
 552                 fld_type_proc_dir = NULL;
 553         }
 554 }
 555
 556 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
 557 MODULE_DESCRIPTION("Lustre FLD");
 558 MODULE_LICENSE("GPL");
 559
 560 cfs_module(mdd, LUSTRE_VERSION_STRING, fld_mod_init, fld_mod_exit);
 561 #endif /* __KERNEL__ */