lustre/fid/fid_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, 2012, 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_request.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 /* mdc RPC locks */
  60 #include <lustre_mdc.h>
  61 #include "fid_internal.h"
  62
  63 static int seq_client_rpc(struct lu_client_seq *seq,
  64                           struct lu_seq_range *output, __u32 opc,
  65                           const char *opcname)
  66 {
  67         struct obd_export     *exp = seq->lcs_exp;
  68         struct ptlrpc_request *req;
  69         struct lu_seq_range   *out, *in;
  70         __u32                 *op;
  71         int                    rc;
  72         ENTRY;
  73
  74         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_SEQ_QUERY,
  75                                         LUSTRE_MDS_VERSION, SEQ_QUERY);
  76         if (req == NULL)
  77                 RETURN(-ENOMEM);
  78
  79         /* Init operation code */
  80         op = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_OPC);
  81         *op = opc;
  82
  83         /* Zero out input range, this is not recovery yet. */
  84         in = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_RANGE);
  85         range_init(in);
  86
  87         ptlrpc_request_set_replen(req);
  88
  89        if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
  90                 req->rq_request_portal = SEQ_METADATA_PORTAL;
  91                 in->lsr_flags = LU_SEQ_RANGE_MDT;
  92         } else {
  93                 LASSERTF(seq->lcs_type == LUSTRE_SEQ_DATA,
  94                          "unknown lcs_type %u\n", seq->lcs_type);
  95                 req->rq_request_portal = SEQ_DATA_PORTAL;
  96                 in->lsr_flags = LU_SEQ_RANGE_OST;
  97         }
  98
  99         if (opc == SEQ_ALLOC_SUPER) {
 100                 /* Update index field of *in, it is required for
 101                  * FLD update on super sequence allocator node. */
 102                 in->lsr_index = seq->lcs_space.lsr_index;
 103                 req->rq_request_portal = SEQ_CONTROLLER_PORTAL;
 104         } else {
 105                 LASSERTF(opc == SEQ_ALLOC_META,
 106                          "unknown opcode %u\n, opc", opc);
 107         }
 108
 109         ptlrpc_at_set_req_timeout(req);
 110
 111         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 112         rc = ptlrpc_queue_wait(req);
 113         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
 114
 115         if (rc)
 116                 GOTO(out_req, rc);
 117
 118         out = req_capsule_server_get(&req->rq_pill, &RMF_SEQ_RANGE);
 119         *output = *out;
 120
 121         if (!range_is_sane(output)) {
 122                 CERROR("%s: Invalid range received from server: "
 123                        DRANGE"\n", seq->lcs_name, PRANGE(output));
 124                 GOTO(out_req, rc = -EINVAL);
 125         }
 126
 127         if (range_is_exhausted(output)) {
 128                 CERROR("%s: Range received from server is exhausted: "
 129                        DRANGE"]\n", seq->lcs_name, PRANGE(output));
 130                 GOTO(out_req, rc = -EINVAL);
 131         }
 132
 133         CDEBUG(D_INFO, "%s: Allocated %s-sequence "DRANGE"]\n",
 134                seq->lcs_name, opcname, PRANGE(output));
 135
 136         EXIT;
 137 out_req:
 138         ptlrpc_req_finished(req);
 139         return rc;
 140 }
 141
 142 /* Request sequence-controller node to allocate new super-sequence. */
 143 int seq_client_alloc_super(struct lu_client_seq *seq,
 144                            const struct lu_env *env)
 145 {
 146         int rc;
 147         ENTRY;
 148
 149         cfs_mutex_lock(&seq->lcs_mutex);
 150
 151 #ifdef __KERNEL__
 152         if (seq->lcs_srv) {
 153                 LASSERT(env != NULL);
 154                 rc = seq_server_alloc_super(seq->lcs_srv, &seq->lcs_space,
 155                                             env);
 156         } else {
 157 #endif
 158                 rc = seq_client_rpc(seq, &seq->lcs_space,
 159                                     SEQ_ALLOC_SUPER, "super");
 160 #ifdef __KERNEL__
 161         }
 162 #endif
 163         cfs_mutex_unlock(&seq->lcs_mutex);
 164         RETURN(rc);
 165 }
 166
 167 /* Request sequence-controller node to allocate new meta-sequence. */
 168 static int seq_client_alloc_meta(const struct lu_env *env,
 169                                  struct lu_client_seq *seq)
 170 {
 171         int rc;
 172         ENTRY;
 173
 174 #ifdef __KERNEL__
 175         if (seq->lcs_srv) {
 176                 LASSERT(env != NULL);
 177                 rc = seq_server_alloc_meta(seq->lcs_srv, &seq->lcs_space, env);
 178         } else {
 179 #endif
 180                 rc = seq_client_rpc(seq, &seq->lcs_space,
 181                                     SEQ_ALLOC_META, "meta");
 182 #ifdef __KERNEL__
 183         }
 184 #endif
 185         RETURN(rc);
 186 }
 187
 188 /* Allocate new sequence for client. */
 189 static int seq_client_alloc_seq(const struct lu_env *env,
 190                                 struct lu_client_seq *seq, seqno_t *seqnr)
 191 {
 192         int rc;
 193         ENTRY;
 194
 195         LASSERT(range_is_sane(&seq->lcs_space));
 196
 197         if (range_is_exhausted(&seq->lcs_space)) {
 198                 rc = seq_client_alloc_meta(env, seq);
 199                 if (rc) {
 200                         CERROR("%s: Can't allocate new meta-sequence,"
 201                                "rc %d\n", seq->lcs_name, rc);
 202                         RETURN(rc);
 203                 } else {
 204                         CDEBUG(D_INFO, "%s: New range - "DRANGE"\n",
 205                                seq->lcs_name, PRANGE(&seq->lcs_space));
 206                 }
 207         } else {
 208                 rc = 0;
 209         }
 210
 211         LASSERT(!range_is_exhausted(&seq->lcs_space));
 212         *seqnr = seq->lcs_space.lsr_start;
 213         seq->lcs_space.lsr_start += 1;
 214
 215         CDEBUG(D_INFO, "%s: Allocated sequence ["LPX64"]\n", seq->lcs_name,
 216                *seqnr);
 217
 218         RETURN(rc);
 219 }
 220
 221 static int seq_fid_alloc_prep(struct lu_client_seq *seq,
 222                               cfs_waitlink_t *link)
 223 {
 224         if (seq->lcs_update) {
 225                 cfs_waitq_add(&seq->lcs_waitq, link);
 226                 cfs_set_current_state(CFS_TASK_UNINT);
 227                 cfs_mutex_unlock(&seq->lcs_mutex);
 228
 229                 cfs_waitq_wait(link, CFS_TASK_UNINT);
 230
 231                 cfs_mutex_lock(&seq->lcs_mutex);
 232                 cfs_waitq_del(&seq->lcs_waitq, link);
 233                 cfs_set_current_state(CFS_TASK_RUNNING);
 234                 return -EAGAIN;
 235         }
 236         ++seq->lcs_update;
 237         cfs_mutex_unlock(&seq->lcs_mutex);
 238         return 0;
 239 }
 240
 241 static void seq_fid_alloc_fini(struct lu_client_seq *seq)
 242 {
 243         LASSERT(seq->lcs_update == 1);
 244         cfs_mutex_lock(&seq->lcs_mutex);
 245         --seq->lcs_update;
 246         cfs_waitq_signal(&seq->lcs_waitq);
 247 }
 248
 249 /* Allocate the whole seq to the caller*/
 250 int seq_client_get_seq(const struct lu_env *env,
 251                        struct lu_client_seq *seq, seqno_t *seqnr)
 252 {
 253         cfs_waitlink_t link;
 254         int rc;
 255
 256         LASSERT(seqnr != NULL);
 257         cfs_mutex_lock(&seq->lcs_mutex);
 258         cfs_waitlink_init(&link);
 259
 260         while (1) {
 261                 rc = seq_fid_alloc_prep(seq, &link);
 262                 if (rc == 0)
 263                         break;
 264         }
 265
 266         rc = seq_client_alloc_seq(env, seq, seqnr);
 267         if (rc) {
 268                 CERROR("%s: Can't allocate new sequence, "
 269                        "rc %d\n", seq->lcs_name, rc);
 270                 seq_fid_alloc_fini(seq);
 271                 cfs_mutex_unlock(&seq->lcs_mutex);
 272                 return rc;
 273         }
 274
 275         CDEBUG(D_INFO, "%s: allocate sequence "
 276                "[0x%16.16"LPF64"x]\n", seq->lcs_name, *seqnr);
 277
 278         /*Since the caller require the whole seq,
 279          *so marked this seq to be used*/
 280         seq->lcs_fid.f_oid = LUSTRE_SEQ_MAX_WIDTH;
 281         seq->lcs_fid.f_seq = *seqnr;
 282         seq->lcs_fid.f_ver = 0;
 283
 284         /*
 285          * Inform caller that sequence switch is performed to allow it
 286          * to setup FLD for it.
 287          */
 288         seq_fid_alloc_fini(seq);
 289         cfs_mutex_unlock(&seq->lcs_mutex);
 290
 291         return rc;
 292 }
 293 EXPORT_SYMBOL(seq_client_get_seq);
 294
 295 /* Allocate new fid on passed client @seq and save it to @fid. */
 296 int seq_client_alloc_fid(const struct lu_env *env,
 297                          struct lu_client_seq *seq, struct lu_fid *fid)
 298 {
 299         cfs_waitlink_t link;
 300         int rc;
 301         ENTRY;
 302
 303         LASSERT(seq != NULL);
 304         LASSERT(fid != NULL);
 305
 306         cfs_waitlink_init(&link);
 307         cfs_mutex_lock(&seq->lcs_mutex);
 308
 309         while (1) {
 310                 seqno_t seqnr;
 311
 312                 if (!fid_is_zero(&seq->lcs_fid) &&
 313                     fid_oid(&seq->lcs_fid) < seq->lcs_width) {
 314                         /* Just bump last allocated fid and return to caller. */
 315                         seq->lcs_fid.f_oid += 1;
 316                         rc = 0;
 317                         break;
 318                 }
 319
 320                 rc = seq_fid_alloc_prep(seq, &link);
 321                 if (rc)
 322                         continue;
 323
 324                 rc = seq_client_alloc_seq(env, seq, &seqnr);
 325                 if (rc) {
 326                         CERROR("%s: Can't allocate new sequence, "
 327                                "rc %d\n", seq->lcs_name, rc);
 328                         seq_fid_alloc_fini(seq);
 329                         cfs_mutex_unlock(&seq->lcs_mutex);
 330                         RETURN(rc);
 331                 }
 332
 333                 CDEBUG(D_INFO, "%s: Switch to sequence "
 334                        "[0x%16.16"LPF64"x]\n", seq->lcs_name, seqnr);
 335
 336                 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID;
 337                 seq->lcs_fid.f_seq = seqnr;
 338                 seq->lcs_fid.f_ver = 0;
 339
 340                 /*
 341                  * Inform caller that sequence switch is performed to allow it
 342                  * to setup FLD for it.
 343                  */
 344                 rc = 1;
 345
 346                 seq_fid_alloc_fini(seq);
 347                 break;
 348         }
 349
 350         *fid = seq->lcs_fid;
 351         cfs_mutex_unlock(&seq->lcs_mutex);
 352
 353         CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name,  PFID(fid));
 354         RETURN(rc);
 355 }
 356 EXPORT_SYMBOL(seq_client_alloc_fid);
 357
 358 /*
 359  * Finish the current sequence due to disconnect.
 360  * See mdc_import_event()
 361  */
 362 void seq_client_flush(struct lu_client_seq *seq)
 363 {
 364         cfs_waitlink_t link;
 365
 366         LASSERT(seq != NULL);
 367         cfs_waitlink_init(&link);
 368         cfs_mutex_lock(&seq->lcs_mutex);
 369
 370         while (seq->lcs_update) {
 371                 cfs_waitq_add(&seq->lcs_waitq, &link);
 372                 cfs_set_current_state(CFS_TASK_UNINT);
 373                 cfs_mutex_unlock(&seq->lcs_mutex);
 374
 375                 cfs_waitq_wait(&link, CFS_TASK_UNINT);
 376
 377                 cfs_mutex_lock(&seq->lcs_mutex);
 378                 cfs_waitq_del(&seq->lcs_waitq, &link);
 379                 cfs_set_current_state(CFS_TASK_RUNNING);
 380         }
 381
 382         fid_zero(&seq->lcs_fid);
 383         /**
 384          * this id shld not be used for seq range allocation.
 385          * set to -1 for dgb check.
 386          */
 387
 388         seq->lcs_space.lsr_index = -1;
 389
 390         range_init(&seq->lcs_space);
 391         cfs_mutex_unlock(&seq->lcs_mutex);
 392 }
 393 EXPORT_SYMBOL(seq_client_flush);
 394
 395 static void seq_client_proc_fini(struct lu_client_seq *seq);
 396
 397 #ifdef LPROCFS
 398 static int seq_client_proc_init(struct lu_client_seq *seq)
 399 {
 400         int rc;
 401         ENTRY;
 402
 403         seq->lcs_proc_dir = lprocfs_register(seq->lcs_name,
 404                                              seq_type_proc_dir,
 405                                              NULL, NULL);
 406
 407         if (IS_ERR(seq->lcs_proc_dir)) {
 408                 CERROR("%s: LProcFS failed in seq-init\n",
 409                        seq->lcs_name);
 410                 rc = PTR_ERR(seq->lcs_proc_dir);
 411                 RETURN(rc);
 412         }
 413
 414         rc = lprocfs_add_vars(seq->lcs_proc_dir,
 415                               seq_client_proc_list, seq);
 416         if (rc) {
 417                 CERROR("%s: Can't init sequence manager "
 418                        "proc, rc %d\n", seq->lcs_name, rc);
 419                 GOTO(out_cleanup, rc);
 420         }
 421
 422         RETURN(0);
 423
 424 out_cleanup:
 425         seq_client_proc_fini(seq);
 426         return rc;
 427 }
 428
 429 static void seq_client_proc_fini(struct lu_client_seq *seq)
 430 {
 431         ENTRY;
 432         if (seq->lcs_proc_dir) {
 433                 if (!IS_ERR(seq->lcs_proc_dir))
 434                         lprocfs_remove(&seq->lcs_proc_dir);
 435                 seq->lcs_proc_dir = NULL;
 436         }
 437         EXIT;
 438 }
 439 #else
 440 static int seq_client_proc_init(struct lu_client_seq *seq)
 441 {
 442         return 0;
 443 }
 444
 445 static void seq_client_proc_fini(struct lu_client_seq *seq)
 446 {
 447         return;
 448 }
 449 #endif
 450
 451 int seq_client_init(struct lu_client_seq *seq,
 452                     struct obd_export *exp,
 453                     enum lu_cli_type type,
 454                     const char *prefix,
 455                     struct lu_server_seq *srv)
 456 {
 457         int rc;
 458         ENTRY;
 459
 460         LASSERT(seq != NULL);
 461         LASSERT(prefix != NULL);
 462
 463         seq->lcs_exp = exp;
 464         seq->lcs_srv = srv;
 465         seq->lcs_type = type;
 466         cfs_mutex_init(&seq->lcs_mutex);
 467         seq->lcs_width = LUSTRE_SEQ_MAX_WIDTH;
 468         cfs_waitq_init(&seq->lcs_waitq);
 469
 470         /* Make sure that things are clear before work is started. */
 471         seq_client_flush(seq);
 472
 473         if (exp == NULL) {
 474                 LASSERT(seq->lcs_srv != NULL);
 475         } else {
 476                 LASSERT(seq->lcs_exp != NULL);
 477                 seq->lcs_exp = class_export_get(seq->lcs_exp);
 478         }
 479
 480         snprintf(seq->lcs_name, sizeof(seq->lcs_name),
 481                  "cli-%s", prefix);
 482
 483         rc = seq_client_proc_init(seq);
 484         if (rc)
 485                 seq_client_fini(seq);
 486         RETURN(rc);
 487 }
 488 EXPORT_SYMBOL(seq_client_init);
 489
 490 void seq_client_fini(struct lu_client_seq *seq)
 491 {
 492         ENTRY;
 493
 494         seq_client_proc_fini(seq);
 495
 496         if (seq->lcs_exp != NULL) {
 497                 class_export_put(seq->lcs_exp);
 498                 seq->lcs_exp = NULL;
 499         }
 500
 501         seq->lcs_srv = NULL;
 502         EXIT;
 503 }
 504 EXPORT_SYMBOL(seq_client_fini);