lustre/fid/fid_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/fid/fid_request.c
   5  *  Lustre Sequence Manager
   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
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_FID
  33
  34 #ifdef __KERNEL__
  35 # include <libcfs/libcfs.h>
  36 # include <linux/module.h>
  37 #else /* __KERNEL__ */
  38 # include <liblustre.h>
  39 #endif
  40
  41 #include <obd.h>
  42 #include <obd_class.h>
  43 #include <dt_object.h>
  44 #include <md_object.h>
  45 #include <obd_support.h>
  46 #include <lustre_req_layout.h>
  47 #include <lustre_fid.h>
  48 #include "fid_internal.h"
  49
  50 static int seq_client_rpc(struct lu_client_seq *seq,
  51                           struct lu_range *range,
  52                           __u32 opc, const char *opcname)
  53 {
  54         int rc, size[3] = { sizeof(struct ptlrpc_body),
  55                             sizeof(__u32),
  56                             sizeof(struct lu_range) };
  57         struct obd_export *exp = seq->lcs_exp;
  58         struct ptlrpc_request *req;
  59         struct lu_range *out, *in;
  60         struct req_capsule pill;
  61         __u32 *op;
  62         ENTRY;
  63
  64         req = ptlrpc_prep_req(class_exp2cliimp(exp),
  65                               LUSTRE_MDS_VERSION,
  66                               SEQ_QUERY, 3, size,
  67                               NULL);
  68         if (req == NULL)
  69                 RETURN(-ENOMEM);
  70
  71         req_capsule_init(&pill, req, RCL_CLIENT, NULL);
  72         req_capsule_set(&pill, &RQF_SEQ_QUERY);
  73
  74         /* init operation code */
  75         op = req_capsule_client_get(&pill, &RMF_SEQ_OPC);
  76         *op = opc;
  77
  78         /* zero out input range, this is not recovery yet. */
  79         in = req_capsule_client_get(&pill, &RMF_SEQ_RANGE);
  80         range_zero(in);
  81
  82         size[1] = sizeof(struct lu_range);
  83         ptlrpc_req_set_repsize(req, 2, size);
  84
  85         if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
  86                 req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
  87                         SEQ_CONTROLLER_PORTAL : SEQ_METADATA_PORTAL;
  88         } else {
  89                 req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
  90                         SEQ_CONTROLLER_PORTAL : SEQ_DATA_PORTAL;
  91         }
  92
  93         rc = ptlrpc_queue_wait(req);
  94         if (rc)
  95                 GOTO(out_req, rc);
  96
  97         out = req_capsule_server_get(&pill, &RMF_SEQ_RANGE);
  98         *range = *out;
  99
 100         if (!range_is_sane(range)) {
 101                 CERROR("%s: Invalid range received from server: "
 102                        DRANGE"\n", seq->lcs_name, PRANGE(range));
 103                 GOTO(out_req, rc = -EINVAL);
 104         }
 105
 106         if (range_is_exhausted(range)) {
 107                 CERROR("%s: Range received from server is exhausted: "
 108                        DRANGE"]\n", seq->lcs_name, PRANGE(range));
 109                 GOTO(out_req, rc = -EINVAL);
 110         }
 111
 112         /* Save server out to request for recovery case. */
 113         *in = *out;
 114
 115         CDEBUG(D_INFO, "%s: Allocated %s-sequence "DRANGE"]\n",
 116                seq->lcs_name, opcname, PRANGE(range));
 117
 118         EXIT;
 119 out_req:
 120         req_capsule_fini(&pill);
 121         ptlrpc_req_finished(req);
 122         return rc;
 123 }
 124
 125 /* request sequence-controller node to allocate new super-sequence. */
 126 static int __seq_client_alloc_super(struct lu_client_seq *seq,
 127                                     const struct lu_env *env)
 128 {
 129         int rc;
 130
 131 #ifdef __KERNEL__
 132         if (seq->lcs_srv) {
 133                 LASSERT(env != NULL);
 134                 rc = seq_server_alloc_super(seq->lcs_srv, NULL,
 135                                             &seq->lcs_range,
 136                                             env);
 137         } else {
 138 #endif
 139                 rc = seq_client_rpc(seq, &seq->lcs_range,
 140                                     SEQ_ALLOC_SUPER, "super");
 141 #ifdef __KERNEL__
 142         }
 143 #endif
 144         return rc;
 145 }
 146
 147 int seq_client_alloc_super(struct lu_client_seq *seq,
 148                            const struct lu_env *env)
 149 {
 150         int rc;
 151         ENTRY;
 152
 153         down(&seq->lcs_sem);
 154         rc = __seq_client_alloc_super(seq, env);
 155         up(&seq->lcs_sem);
 156
 157         RETURN(rc);
 158 }
 159 EXPORT_SYMBOL(seq_client_alloc_super);
 160
 161 /* request sequence-controller node to allocate new meta-sequence. */
 162 static int __seq_client_alloc_meta(struct lu_client_seq *seq,
 163                                    const struct lu_env *env)
 164 {
 165         int rc;
 166
 167 #ifdef __KERNEL__
 168         if (seq->lcs_srv) {
 169                 LASSERT(env != NULL);
 170                 rc = seq_server_alloc_meta(seq->lcs_srv, NULL,
 171                                            &seq->lcs_range,
 172                                            env);
 173         } else {
 174 #endif
 175                 rc = seq_client_rpc(seq, &seq->lcs_range,
 176                                     SEQ_ALLOC_META, "meta");
 177 #ifdef __KERNEL__
 178         }
 179 #endif
 180         return rc;
 181 }
 182
 183 int seq_client_alloc_meta(struct lu_client_seq *seq,
 184                           const struct lu_env *env)
 185 {
 186         int rc;
 187         ENTRY;
 188
 189         down(&seq->lcs_sem);
 190         rc = __seq_client_alloc_meta(seq, env);
 191         up(&seq->lcs_sem);
 192
 193         RETURN(rc);
 194 }
 195 EXPORT_SYMBOL(seq_client_alloc_meta);
 196
 197 /* allocate new sequence for client (llite or MDC are expected to use this) */
 198 static int __seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
 199 {
 200         int rc = 0;
 201         ENTRY;
 202
 203         LASSERT(range_is_sane(&seq->lcs_range));
 204
 205         /* if we still have free sequences in meta-sequence we allocate new seq
 206          * from given range, if not - allocate new meta-sequence. */
 207         if (range_space(&seq->lcs_range) == 0) {
 208                 rc = __seq_client_alloc_meta(seq, NULL);
 209                 if (rc) {
 210                         CERROR("%s: Can't allocate new meta-sequence, "
 211                                "rc %d\n", seq->lcs_name, rc);
 212                         RETURN(rc);
 213                 }
 214         }
 215
 216         LASSERT(range_space(&seq->lcs_range) > 0);
 217         *seqnr = seq->lcs_range.lr_start;
 218         seq->lcs_range.lr_start++;
 219
 220         CDEBUG(D_INFO, "%s: Allocated sequence ["LPX64"]\n",
 221                seq->lcs_name, *seqnr);
 222         RETURN(rc);
 223 }
 224
 225 int seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
 226 {
 227         int rc = 0;
 228         ENTRY;
 229
 230         down(&seq->lcs_sem);
 231         rc = __seq_client_alloc_seq(seq, seqnr);
 232         up(&seq->lcs_sem);
 233
 234         RETURN(rc);
 235 }
 236 EXPORT_SYMBOL(seq_client_alloc_seq);
 237
 238 int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
 239 {
 240         int rc;
 241         ENTRY;
 242
 243         LASSERT(fid != NULL);
 244
 245         down(&seq->lcs_sem);
 246
 247         if (!fid_is_sane(&seq->lcs_fid) ||
 248             fid_oid(&seq->lcs_fid) >= seq->lcs_width)
 249         {
 250                 seqno_t seqnr;
 251
 252                 /* allocate new sequence for case client has no sequence at all
 253                  * or sequence is exhausted and should be switched. */
 254                 rc = __seq_client_alloc_seq(seq, &seqnr);
 255                 if (rc) {
 256                         CERROR("%s: Can't allocate new sequence, "
 257                                "rc %d\n", seq->lcs_name, rc);
 258                         GOTO(out, rc);
 259                 }
 260
 261                 /* init new fid */
 262                 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID;
 263                 seq->lcs_fid.f_seq = seqnr;
 264                 seq->lcs_fid.f_ver = 0;
 265
 266                 /* inform caller that sequence switch is performed to allow it
 267                  * to setup FLD for it. */
 268                 rc = 1;
 269         } else {
 270                 seq->lcs_fid.f_oid++;
 271                 rc = 0;
 272         }
 273
 274         *fid = seq->lcs_fid;
 275         LASSERT(fid_is_sane(fid));
 276
 277         CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n",
 278                seq->lcs_name, PFID(fid));
 279
 280         EXIT;
 281 out:
 282         up(&seq->lcs_sem);
 283         return rc;
 284 }
 285 EXPORT_SYMBOL(seq_client_alloc_fid);
 286
 287 static void seq_client_proc_fini(struct lu_client_seq *seq);
 288
 289 #ifdef LPROCFS
 290 static int seq_client_proc_init(struct lu_client_seq *seq)
 291 {
 292         int rc;
 293         ENTRY;
 294
 295         seq->lcs_proc_dir = lprocfs_register(seq->lcs_name,
 296                                              seq_type_proc_dir,
 297                                              NULL, NULL);
 298
 299         if (IS_ERR(seq->lcs_proc_dir)) {
 300                 CERROR("%s: LProcFS failed in seq-init\n",
 301                        seq->lcs_name);
 302                 rc = PTR_ERR(seq->lcs_proc_dir);
 303                 RETURN(rc);
 304         }
 305
 306         rc = lprocfs_add_vars(seq->lcs_proc_dir,
 307                               seq_client_proc_list, seq);
 308         if (rc) {
 309                 CERROR("%s: Can't init sequence manager "
 310                        "proc, rc %d\n", seq->lcs_name, rc);
 311                 GOTO(out_cleanup, rc);
 312         }
 313
 314         RETURN(0);
 315
 316 out_cleanup:
 317         seq_client_proc_fini(seq);
 318         return rc;
 319 }
 320
 321 static void seq_client_proc_fini(struct lu_client_seq *seq)
 322 {
 323         ENTRY;
 324         if (seq->lcs_proc_dir) {
 325                 if (!IS_ERR(seq->lcs_proc_dir))
 326                         lprocfs_remove(seq->lcs_proc_dir);
 327                 seq->lcs_proc_dir = NULL;
 328         }
 329         EXIT;
 330 }
 331 #else
 332 static int seq_client_proc_init(struct lu_client_seq *seq)
 333 {
 334         return 0;
 335 }
 336
 337 static void seq_client_proc_fini(struct lu_client_seq *seq)
 338 {
 339         return;
 340 }
 341 #endif
 342
 343 int seq_client_init(struct lu_client_seq *seq,
 344                     struct obd_export *exp,
 345                     enum lu_cli_type type,
 346                     const char *prefix,
 347                     struct lu_server_seq *srv)
 348 {
 349         int rc;
 350         ENTRY;
 351
 352         LASSERT(seq != NULL);
 353         LASSERT(prefix != NULL);
 354
 355         seq->lcs_exp = exp;
 356         seq->lcs_srv = srv;
 357         seq->lcs_type = type;
 358         fid_zero(&seq->lcs_fid);
 359         range_zero(&seq->lcs_range);
 360         sema_init(&seq->lcs_sem, 1);
 361         seq->lcs_width = LUSTRE_SEQ_MAX_WIDTH;
 362
 363         if (exp == NULL) {
 364                 LASSERT(seq->lcs_srv != NULL);
 365         } else {
 366                 LASSERT(seq->lcs_exp != NULL);
 367                 seq->lcs_exp = class_export_get(seq->lcs_exp);
 368         }
 369
 370         snprintf(seq->lcs_name, sizeof(seq->lcs_name),
 371                  "cli-%s", prefix);
 372
 373         rc = seq_client_proc_init(seq);
 374         if (rc)
 375                 seq_client_fini(seq);
 376         RETURN(rc);
 377 }
 378 EXPORT_SYMBOL(seq_client_init);
 379
 380 void seq_client_fini(struct lu_client_seq *seq)
 381 {
 382         ENTRY;
 383
 384         seq_client_proc_fini(seq);
 385
 386         if (seq->lcs_exp != NULL) {
 387                 class_export_put(seq->lcs_exp);
 388                 seq->lcs_exp = NULL;
 389         }
 390
 391         seq->lcs_srv = NULL;
 392         EXIT;
 393 }
 394 EXPORT_SYMBOL(seq_client_fini);