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