lustre/fid/fid_handler.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, Whamcloud, Inc.
  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/fid/fid_handler.c
  37  *
  38  * Lustre Sequence Manager
  39  *
  40  * Author: Yury Umanets <umka@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_FID
  44
  45 #ifdef __KERNEL__
  46 # include <libcfs/libcfs.h>
  47 # include <linux/module.h>
  48 #else /* __KERNEL__ */
  49 # include <liblustre.h>
  50 #endif
  51
  52 #include <obd.h>
  53 #include <obd_class.h>
  54 #include <dt_object.h>
  55 #include <md_object.h>
  56 #include <obd_support.h>
  57 #include <lustre_req_layout.h>
  58 #include <lustre_fid.h>
  59 #include "fid_internal.h"
  60
  61 #ifdef __KERNEL__
  62 /* Assigns client to sequence controller node. */
  63 int seq_server_set_cli(struct lu_server_seq *seq,
  64                        struct lu_client_seq *cli,
  65                        const struct lu_env *env)
  66 {
  67         int rc = 0;
  68         ENTRY;
  69
  70         /*
  71          * Ask client for new range, assign that range to ->seq_space and write
  72          * seq state to backing store should be atomic.
  73          */
  74         cfs_mutex_lock(&seq->lss_mutex);
  75
  76         if (cli == NULL) {
  77                 CDEBUG(D_INFO, "%s: Detached sequence client %s\n",
  78                        seq->lss_name, cli->lcs_name);
  79                 seq->lss_cli = cli;
  80                 GOTO(out_up, rc = 0);
  81         }
  82
  83         if (seq->lss_cli != NULL) {
  84                 CERROR("%s: Sequence controller is already "
  85                        "assigned\n", seq->lss_name);
  86                 GOTO(out_up, rc = -EINVAL);
  87         }
  88
  89         CDEBUG(D_INFO, "%s: Attached sequence controller %s\n",
  90                seq->lss_name, cli->lcs_name);
  91
  92         seq->lss_cli = cli;
  93         cli->lcs_space.lsr_index = seq->lss_site->ms_node_id;
  94         EXIT;
  95 out_up:
  96         cfs_mutex_unlock(&seq->lss_mutex);
  97         return rc;
  98 }
  99 EXPORT_SYMBOL(seq_server_set_cli);
 100 /*
 101  * allocate \a w units of sequence from range \a from.
 102  */
 103 static inline void range_alloc(struct lu_seq_range *to,
 104                                struct lu_seq_range *from,
 105                                __u64 width)
 106 {
 107         width = min(range_space(from), width);
 108         to->lsr_start = from->lsr_start;
 109         to->lsr_end = from->lsr_start + width;
 110         from->lsr_start += width;
 111 }
 112
 113 /**
 114  * On controller node, allocate new super sequence for regular sequence server.
 115  * As this super sequence controller, this node suppose to maintain fld
 116  * and update index.
 117  * \a out range always has currect mds node number of requester.
 118  */
 119
 120 static int __seq_server_alloc_super(struct lu_server_seq *seq,
 121                                     struct lu_seq_range *out,
 122                                     const struct lu_env *env)
 123 {
 124         struct lu_seq_range *space = &seq->lss_space;
 125         int rc;
 126         ENTRY;
 127
 128         LASSERT(range_is_sane(space));
 129
 130         if (range_is_exhausted(space)) {
 131                 CERROR("%s: Sequences space is exhausted\n",
 132                        seq->lss_name);
 133                 RETURN(-ENOSPC);
 134         } else {
 135                 range_alloc(out, space, seq->lss_width);
 136         }
 137
 138         rc = seq_store_update(env, seq, out, 1 /* sync */);
 139
 140         CDEBUG(D_INFO, "%s: super-sequence allocation rc = %d "
 141                DRANGE"\n", seq->lss_name, rc, PRANGE(out));
 142
 143         RETURN(rc);
 144 }
 145
 146 int seq_server_alloc_super(struct lu_server_seq *seq,
 147                            struct lu_seq_range *out,
 148                            const struct lu_env *env)
 149 {
 150         int rc;
 151         ENTRY;
 152
 153         cfs_mutex_lock(&seq->lss_mutex);
 154         rc = __seq_server_alloc_super(seq, out, env);
 155         cfs_mutex_unlock(&seq->lss_mutex);
 156
 157         RETURN(rc);
 158 }
 159
 160 static int __seq_set_init(const struct lu_env *env,
 161                             struct lu_server_seq *seq)
 162 {
 163         struct lu_seq_range *space = &seq->lss_space;
 164         int rc;
 165
 166         range_alloc(&seq->lss_lowater_set, space, seq->lss_set_width);
 167         range_alloc(&seq->lss_hiwater_set, space, seq->lss_set_width);
 168
 169         rc = seq_store_update(env, seq, NULL, 1);
 170
 171         return rc;
 172 }
 173
 174 /*
 175  * This function implements new seq allocation algorithm using async
 176  * updates to seq file on disk. ref bug 18857 for details.
 177  * there are four variable to keep track of this process
 178  *
 179  * lss_space; - available lss_space
 180  * lss_lowater_set; - lu_seq_range for all seqs before barrier, i.e. safe to use
 181  * lss_hiwater_set; - lu_seq_range after barrier, i.e. allocated but may be
 182  *                    not yet committed
 183  *
 184  * when lss_lowater_set reaches the end it is replaced with hiwater one and
 185  * a write operation is initiated to allocate new hiwater range.
 186  * if last seq write opearion is still not commited, current operation is
 187  * flaged as sync write op.
 188  */
 189 static int range_alloc_set(const struct lu_env *env,
 190                             struct lu_seq_range *out,
 191                             struct lu_server_seq *seq)
 192 {
 193         struct lu_seq_range *space = &seq->lss_space;
 194         struct lu_seq_range *loset = &seq->lss_lowater_set;
 195         struct lu_seq_range *hiset = &seq->lss_hiwater_set;
 196         int rc = 0;
 197
 198         if (range_is_zero(loset))
 199                 __seq_set_init(env, seq);
 200
 201         if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_ALLOC)) /* exhaust set */
 202                 loset->lsr_start = loset->lsr_end;
 203
 204         if (range_is_exhausted(loset)) {
 205                 /* reached high water mark. */
 206                 struct lu_device *dev = seq->lss_site->ms_lu.ls_top_dev;
 207                 int obd_num_clients = dev->ld_obd->obd_num_exports;
 208                 __u64 set_sz;
 209
 210                 /* calculate new seq width based on number of clients */
 211                 set_sz = max(seq->lss_set_width,
 212                              obd_num_clients * seq->lss_width);
 213                 set_sz = min(range_space(space), set_sz);
 214
 215                 /* Switch to hiwater range now */
 216                 *loset = *hiset;
 217                 /* allocate new hiwater range */
 218                 range_alloc(hiset, space, set_sz);
 219
 220                 /* update ondisk seq with new *space */
 221                 rc = seq_store_update(env, seq, NULL, seq->lss_need_sync);
 222         }
 223
 224         LASSERTF(!range_is_exhausted(loset) || range_is_sane(loset),
 225                  DRANGE"\n", PRANGE(loset));
 226
 227         if (rc == 0)
 228                 range_alloc(out, loset, seq->lss_width);
 229
 230         RETURN(rc);
 231 }
 232
 233 static int __seq_server_alloc_meta(struct lu_server_seq *seq,
 234                                    struct lu_seq_range *out,
 235                                    const struct lu_env *env)
 236 {
 237         struct lu_seq_range *space = &seq->lss_space;
 238         int rc = 0;
 239
 240         ENTRY;
 241
 242         LASSERT(range_is_sane(space));
 243
 244         /* Check if available space ends and allocate new super seq */
 245         if (range_is_exhausted(space)) {
 246                 if (!seq->lss_cli) {
 247                         CERROR("%s: No sequence controller is attached.\n",
 248                                seq->lss_name);
 249                         RETURN(-ENODEV);
 250                 }
 251
 252                 rc = seq_client_alloc_super(seq->lss_cli, env);
 253                 if (rc) {
 254                         CERROR("%s: Can't allocate super-sequence, rc %d\n",
 255                                seq->lss_name, rc);
 256                         RETURN(rc);
 257                 }
 258
 259                 /* Saving new range to allocation space. */
 260                 *space = seq->lss_cli->lcs_space;
 261                 LASSERT(range_is_sane(space));
 262         }
 263
 264         rc = range_alloc_set(env, out, seq);
 265         if (rc == 0) {
 266                 CDEBUG(D_INFO, "%s: Allocated meta-sequence "
 267                        DRANGE"\n", seq->lss_name, PRANGE(out));
 268         }
 269
 270         RETURN(rc);
 271 }
 272
 273 int seq_server_alloc_meta(struct lu_server_seq *seq,
 274                           struct lu_seq_range *out,
 275                           const struct lu_env *env)
 276 {
 277         int rc;
 278         ENTRY;
 279
 280         cfs_mutex_lock(&seq->lss_mutex);
 281         rc = __seq_server_alloc_meta(seq, out, env);
 282         cfs_mutex_unlock(&seq->lss_mutex);
 283
 284         RETURN(rc);
 285 }
 286 EXPORT_SYMBOL(seq_server_alloc_meta);
 287
 288 static int seq_server_handle(struct lu_site *site,
 289                              const struct lu_env *env,
 290                              __u32 opc, struct lu_seq_range *out)
 291 {
 292         int rc;
 293         struct md_site *mite;
 294         ENTRY;
 295
 296         mite = lu_site2md(site);
 297         switch (opc) {
 298         case SEQ_ALLOC_META:
 299                 if (!mite->ms_server_seq) {
 300                         CERROR("Sequence server is not "
 301                                "initialized\n");
 302                         RETURN(-EINVAL);
 303                 }
 304                 rc = seq_server_alloc_meta(mite->ms_server_seq, out, env);
 305                 break;
 306         case SEQ_ALLOC_SUPER:
 307                 if (!mite->ms_control_seq) {
 308                         CERROR("Sequence controller is not "
 309                                "initialized\n");
 310                         RETURN(-EINVAL);
 311                 }
 312                 rc = seq_server_alloc_super(mite->ms_control_seq, out, env);
 313                 break;
 314         default:
 315                 rc = -EINVAL;
 316                 break;
 317         }
 318
 319         RETURN(rc);
 320 }
 321
 322 static int seq_req_handle(struct ptlrpc_request *req,
 323                           const struct lu_env *env,
 324                           struct seq_thread_info *info)
 325 {
 326         struct lu_seq_range *out, *tmp;
 327         struct lu_site *site;
 328         int rc = -EPROTO;
 329         __u32 *opc;
 330         ENTRY;
 331
 332         LASSERT(!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY));
 333         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
 334         LASSERT(site != NULL);
 335
 336         rc = req_capsule_server_pack(info->sti_pill);
 337         if (rc)
 338                 RETURN(err_serious(rc));
 339
 340         opc = req_capsule_client_get(info->sti_pill, &RMF_SEQ_OPC);
 341         if (opc != NULL) {
 342                 out = req_capsule_server_get(info->sti_pill, &RMF_SEQ_RANGE);
 343                 if (out == NULL)
 344                         RETURN(err_serious(-EPROTO));
 345
 346                 tmp = req_capsule_client_get(info->sti_pill, &RMF_SEQ_RANGE);
 347
 348                 /* seq client passed mdt id, we need to pass that using out
 349                  * range parameter */
 350
 351                 out->lsr_index = tmp->lsr_index;
 352                 out->lsr_flags = tmp->lsr_flags;
 353                 rc = seq_server_handle(site, env, *opc, out);
 354         } else
 355                 rc = err_serious(-EPROTO);
 356
 357         RETURN(rc);
 358 }
 359
 360 /* context key constructor/destructor: seq_key_init, seq_key_fini */
 361 LU_KEY_INIT_FINI(seq, struct seq_thread_info);
 362
 363 /* context key: seq_thread_key */
 364 LU_CONTEXT_KEY_DEFINE(seq, LCT_MD_THREAD);
 365
 366 static void seq_thread_info_init(struct ptlrpc_request *req,
 367                                  struct seq_thread_info *info)
 368 {
 369         info->sti_pill = &req->rq_pill;
 370         /* Init request capsule */
 371         req_capsule_init(info->sti_pill, req, RCL_SERVER);
 372         req_capsule_set(info->sti_pill, &RQF_SEQ_QUERY);
 373 }
 374
 375 static void seq_thread_info_fini(struct seq_thread_info *info)
 376 {
 377         req_capsule_fini(info->sti_pill);
 378 }
 379
 380 static int seq_handle(struct ptlrpc_request *req)
 381 {
 382         const struct lu_env *env;
 383         struct seq_thread_info *info;
 384         int rc;
 385
 386         env = req->rq_svc_thread->t_env;
 387         LASSERT(env != NULL);
 388
 389         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
 390         LASSERT(info != NULL);
 391
 392         seq_thread_info_init(req, info);
 393         rc = seq_req_handle(req, env, info);
 394         /* XXX: we don't need replay but MDT assign transno in any case,
 395          * remove it manually before reply*/
 396         lustre_msg_set_transno(req->rq_repmsg, 0);
 397         seq_thread_info_fini(info);
 398
 399         return rc;
 400 }
 401
 402 /*
 403  * Entry point for handling FLD RPCs called from MDT.
 404  */
 405 int seq_query(struct com_thread_info *info)
 406 {
 407         return seq_handle(info->cti_pill->rc_req);
 408 }
 409 EXPORT_SYMBOL(seq_query);
 410
 411 static void seq_server_proc_fini(struct lu_server_seq *seq);
 412
 413 #ifdef LPROCFS
 414 static int seq_server_proc_init(struct lu_server_seq *seq)
 415 {
 416         int rc;
 417         ENTRY;
 418
 419         seq->lss_proc_dir = lprocfs_register(seq->lss_name,
 420                                              seq_type_proc_dir,
 421                                              NULL, NULL);
 422         if (IS_ERR(seq->lss_proc_dir)) {
 423                 rc = PTR_ERR(seq->lss_proc_dir);
 424                 RETURN(rc);
 425         }
 426
 427         rc = lprocfs_add_vars(seq->lss_proc_dir,
 428                               seq_server_proc_list, seq);
 429         if (rc) {
 430                 CERROR("%s: Can't init sequence manager "
 431                        "proc, rc %d\n", seq->lss_name, rc);
 432                 GOTO(out_cleanup, rc);
 433         }
 434
 435         RETURN(0);
 436
 437 out_cleanup:
 438         seq_server_proc_fini(seq);
 439         return rc;
 440 }
 441
 442 static void seq_server_proc_fini(struct lu_server_seq *seq)
 443 {
 444         ENTRY;
 445         if (seq->lss_proc_dir != NULL) {
 446                 if (!IS_ERR(seq->lss_proc_dir))
 447                         lprocfs_remove(&seq->lss_proc_dir);
 448                 seq->lss_proc_dir = NULL;
 449         }
 450         EXIT;
 451 }
 452 #else
 453 static int seq_server_proc_init(struct lu_server_seq *seq)
 454 {
 455         return 0;
 456 }
 457
 458 static void seq_server_proc_fini(struct lu_server_seq *seq)
 459 {
 460         return;
 461 }
 462 #endif
 463
 464
 465 int seq_server_init(struct lu_server_seq *seq,
 466                     struct dt_device *dev,
 467                     const char *prefix,
 468                     enum lu_mgr_type type,
 469                     struct md_site *ms,
 470                     const struct lu_env *env)
 471 {
 472         int rc, is_srv = (type == LUSTRE_SEQ_SERVER);
 473         ENTRY;
 474
 475         LASSERT(dev != NULL);
 476         LASSERT(prefix != NULL);
 477
 478         seq->lss_cli = NULL;
 479         seq->lss_type = type;
 480         seq->lss_site = ms;
 481         range_init(&seq->lss_space);
 482
 483         range_init(&seq->lss_lowater_set);
 484         range_init(&seq->lss_hiwater_set);
 485         seq->lss_set_width = LUSTRE_SEQ_BATCH_WIDTH;
 486
 487         cfs_mutex_init(&seq->lss_mutex);
 488
 489         seq->lss_width = is_srv ?
 490                 LUSTRE_SEQ_META_WIDTH : LUSTRE_SEQ_SUPER_WIDTH;
 491
 492         snprintf(seq->lss_name, sizeof(seq->lss_name),
 493                  "%s-%s", (is_srv ? "srv" : "ctl"), prefix);
 494
 495         rc = seq_store_init(seq, env, dev);
 496         if (rc)
 497                 GOTO(out, rc);
 498         /* Request backing store for saved sequence info. */
 499         rc = seq_store_read(seq, env);
 500         if (rc == -ENODATA) {
 501
 502                 /* Nothing is read, init by default value. */
 503                 seq->lss_space = is_srv ?
 504                         LUSTRE_SEQ_ZERO_RANGE:
 505                         LUSTRE_SEQ_SPACE_RANGE;
 506
 507                 seq->lss_space.lsr_index = ms->ms_node_id;
 508                 CDEBUG(D_INFO, "%s: No data found "
 509                        "on store. Initialize space\n",
 510                        seq->lss_name);
 511
 512                 rc = seq_store_update(env, seq, NULL, 0);
 513                 if (rc) {
 514                         CERROR("%s: Can't write space data, "
 515                                "rc %d\n", seq->lss_name, rc);
 516                 }
 517         } else if (rc) {
 518                 CERROR("%s: Can't read space data, rc %d\n",
 519                        seq->lss_name, rc);
 520                 GOTO(out, rc);
 521         }
 522
 523         if (is_srv) {
 524                 LASSERT(range_is_sane(&seq->lss_space));
 525         } else {
 526                 LASSERT(!range_is_zero(&seq->lss_space) &&
 527                         range_is_sane(&seq->lss_space));
 528         }
 529
 530         rc  = seq_server_proc_init(seq);
 531         if (rc)
 532                 GOTO(out, rc);
 533
 534         EXIT;
 535 out:
 536         if (rc)
 537                 seq_server_fini(seq, env);
 538         return rc;
 539 }
 540 EXPORT_SYMBOL(seq_server_init);
 541
 542 void seq_server_fini(struct lu_server_seq *seq,
 543                      const struct lu_env *env)
 544 {
 545         ENTRY;
 546
 547         seq_server_proc_fini(seq);
 548         seq_store_fini(seq, env);
 549
 550         EXIT;
 551 }
 552 EXPORT_SYMBOL(seq_server_fini);
 553
 554 cfs_proc_dir_entry_t *seq_type_proc_dir = NULL;
 555
 556 static struct lu_local_obj_desc llod_seq_srv = {
 557         .llod_name      = LUSTRE_SEQ_SRV_NAME,
 558         .llod_oid       = FID_SEQ_SRV_OID,
 559         .llod_is_index  = 0,
 560 };
 561
 562 static struct lu_local_obj_desc llod_seq_ctl = {
 563         .llod_name      = LUSTRE_SEQ_CTL_NAME,
 564         .llod_oid       = FID_SEQ_CTL_OID,
 565         .llod_is_index  = 0,
 566 };
 567
 568 static int __init fid_mod_init(void)
 569 {
 570         seq_type_proc_dir = lprocfs_register(LUSTRE_SEQ_NAME,
 571                                              proc_lustre_root,
 572                                              NULL, NULL);
 573         if (IS_ERR(seq_type_proc_dir))
 574                 return PTR_ERR(seq_type_proc_dir);
 575
 576         llo_local_obj_register(&llod_seq_srv);
 577         llo_local_obj_register(&llod_seq_ctl);
 578
 579         LU_CONTEXT_KEY_INIT(&seq_thread_key);
 580         lu_context_key_register(&seq_thread_key);
 581         return 0;
 582 }
 583
 584 static void __exit fid_mod_exit(void)
 585 {
 586         llo_local_obj_unregister(&llod_seq_srv);
 587         llo_local_obj_unregister(&llod_seq_ctl);
 588
 589         lu_context_key_degister(&seq_thread_key);
 590         if (seq_type_proc_dir != NULL && !IS_ERR(seq_type_proc_dir)) {
 591                 lprocfs_remove(&seq_type_proc_dir);
 592                 seq_type_proc_dir = NULL;
 593         }
 594 }
 595
 596 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
 597 MODULE_DESCRIPTION("Lustre FID Module");
 598 MODULE_LICENSE("GPL");
 599
 600 cfs_module(fid, "0.1.0", fid_mod_init, fid_mod_exit);
 601 #endif