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 /* client seq mgr interface */
  51 static int
  52 seq_client_rpc(struct lu_client_seq *seq,
  53                struct lu_range *range,
  54                __u32 opc)
  55 {
  56         struct obd_export *exp = seq->seq_exp;
  57         int repsize = sizeof(struct lu_range);
  58         int rc, reqsize = sizeof(__u32);
  59         struct ptlrpc_request *req;
  60         struct lu_range *ran;
  61         __u32 *op;
  62         ENTRY;
  63
  64         req = ptlrpc_prep_req(class_exp2cliimp(exp),
  65                               LUSTRE_MDS_VERSION,
  66                               SEQ_QUERY, 1, &reqsize,
  67                               NULL);
  68         if (req == NULL)
  69                 RETURN(-ENOMEM);
  70
  71         op = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*op));
  72         *op = opc;
  73
  74         req->rq_replen = lustre_msg_size(1, &repsize);
  75
  76         req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
  77                 SEQ_CTLR_PORTAL : SEQ_SRV_PORTAL;
  78
  79         rc = ptlrpc_queue_wait(req);
  80         if (rc)
  81                 GOTO(out_req, rc);
  82
  83         ran = lustre_swab_repbuf(req, 0, sizeof(*ran),
  84                                  lustre_swab_lu_range);
  85
  86         if (ran == NULL) {
  87                 CERROR("invalid range is returned\n");
  88                 GOTO(out_req, rc = -EPROTO);
  89         }
  90         *range = *ran;
  91
  92         LASSERT(range_is_sane(range));
  93         LASSERT(!range_is_exhausted(range));
  94
  95         EXIT;
  96 out_req:
  97         ptlrpc_req_finished(req);
  98         return rc;
  99 }
 100
 101 static int
 102 __seq_client_alloc_opc(struct lu_client_seq *seq,
 103                        int opc, const char *opcname)
 104 {
 105         int rc;
 106         ENTRY;
 107
 108         rc = seq_client_rpc(seq, &seq->seq_range, opc);
 109         if (rc == 0) {
 110                 CDEBUG(D_INFO, "SEQ-MGR(cli): allocated "
 111                        "%s-sequence ["LPX64"-"LPX64"]\n",
 112                        opcname, seq->seq_range.lr_start,
 113                        seq->seq_range.lr_end);
 114         }
 115         RETURN(rc);
 116 }
 117
 118 /* request sequence-controller node to allocate new super-sequence. */
 119 static int
 120 __seq_client_alloc_super(struct lu_client_seq *seq)
 121 {
 122         ENTRY;
 123         RETURN(__seq_client_alloc_opc(seq, SEQ_ALLOC_SUPER, "super"));
 124 }
 125
 126 int
 127 seq_client_alloc_super(struct lu_client_seq *seq)
 128 {
 129         int rc;
 130         ENTRY;
 131
 132         down(&seq->seq_sem);
 133         rc = __seq_client_alloc_super(seq);
 134         up(&seq->seq_sem);
 135
 136         RETURN(rc);
 137 }
 138 EXPORT_SYMBOL(seq_client_alloc_super);
 139
 140 /* request sequence-controller node to allocate new meta-sequence. */
 141 static int
 142 __seq_client_alloc_meta(struct lu_client_seq *seq)
 143 {
 144         ENTRY;
 145         RETURN(__seq_client_alloc_opc(seq, SEQ_ALLOC_META, "meta"));
 146 }
 147
 148 int
 149 seq_client_alloc_meta(struct lu_client_seq *seq)
 150 {
 151         int rc;
 152         ENTRY;
 153
 154         down(&seq->seq_sem);
 155         rc = __seq_client_alloc_meta(seq);
 156         up(&seq->seq_sem);
 157
 158         RETURN(rc);
 159 }
 160 EXPORT_SYMBOL(seq_client_alloc_meta);
 161
 162 /* allocate new sequence for client (llite or MDC are expected to use this) */
 163 static int
 164 __seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
 165 {
 166         int rc = 0;
 167         ENTRY;
 168
 169         LASSERT(range_is_sane(&seq->seq_range));
 170
 171         /* if we still have free sequences in meta-sequence we allocate new seq
 172          * from given range, if not - allocate new meta-sequence. */
 173         if (range_space(&seq->seq_range) == 0) {
 174                 rc = __seq_client_alloc_meta(seq);
 175                 if (rc) {
 176                         CERROR("can't allocate new meta-sequence, "
 177                                "rc %d\n", rc);
 178                         RETURN(rc);
 179                 }
 180         }
 181
 182         *seqnr = seq->seq_range.lr_start;
 183         seq->seq_range.lr_start += 1;
 184
 185         if (rc == 0) {
 186                 CDEBUG(D_INFO, "SEQ-MGR(cli): allocated "
 187                        "sequence ["LPX64"]\n", *seqnr);
 188         }
 189         RETURN(rc);
 190 }
 191
 192 int
 193 seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
 194 {
 195         int rc = 0;
 196         ENTRY;
 197
 198         down(&seq->seq_sem);
 199         rc = __seq_client_alloc_seq(seq, seqnr);
 200         up(&seq->seq_sem);
 201
 202         RETURN(rc);
 203 }
 204 EXPORT_SYMBOL(seq_client_alloc_seq);
 205
 206 int
 207 seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
 208 {
 209         seqno_t seqnr = 0;
 210         int rc;
 211         ENTRY;
 212
 213         LASSERT(fid != NULL);
 214
 215         down(&seq->seq_sem);
 216
 217         if (!fid_is_sane(&seq->seq_fid) ||
 218             fid_oid(&seq->seq_fid) >= seq->seq_width)
 219         {
 220                 /* allocate new sequence for case client hass no sequence at all
 221                  * or sequnece is exhausted and should be switched. */
 222                 rc = __seq_client_alloc_seq(seq, &seqnr);
 223                 if (rc) {
 224                         CERROR("can't allocate new sequence, "
 225                                "rc %d\n", rc);
 226                         GOTO(out, rc);
 227                 }
 228
 229                 /* init new fid */
 230                 seq->seq_fid.f_oid = LUSTRE_FID_INIT_OID;
 231                 seq->seq_fid.f_seq = seqnr;
 232                 seq->seq_fid.f_ver = 0;
 233
 234                 /* inform caller that sequnece switch is performed to allow it
 235                  * to setup FLD for it. */
 236                 rc = -ERESTART;
 237         } else {
 238                 seq->seq_fid.f_oid += 1;
 239                 rc = 0;
 240         }
 241
 242         *fid = seq->seq_fid;
 243         LASSERT(fid_is_sane(fid));
 244
 245         CDEBUG(D_INFO, "SEQ-MGR(cli): allocated FID "DFID3"\n",
 246                PFID3(fid));
 247
 248         EXIT;
 249 out:
 250         up(&seq->seq_sem);
 251         return rc;
 252 }
 253 EXPORT_SYMBOL(seq_client_alloc_fid);
 254
 255 #ifdef LPROCFS
 256 static int
 257 seq_client_proc_init(struct lu_client_seq *seq)
 258 {
 259         int rc;
 260         ENTRY;
 261
 262         seq->seq_proc_dir = lprocfs_register(seq->seq_name,
 263                                              proc_lustre_root,
 264                                              NULL, NULL);
 265
 266         if (IS_ERR(seq->seq_proc_dir)) {
 267                 CERROR("LProcFS failed in seq-init\n");
 268                 rc = PTR_ERR(seq->seq_proc_dir);
 269                 GOTO(err, rc);
 270         }
 271
 272         rc = lprocfs_add_vars(seq->seq_proc_dir,
 273                               seq_client_proc_list, seq);
 274         if (rc) {
 275                 CERROR("can't init sequence manager "
 276                        "proc, rc %d\n", rc);
 277                 GOTO(err, rc);
 278         }
 279
 280         RETURN(0);
 281
 282 err:
 283         seq->seq_proc_dir = NULL;
 284         return rc;
 285 }
 286
 287 static void
 288 seq_client_proc_fini(struct lu_client_seq *seq)
 289 {
 290         ENTRY;
 291         if (seq->seq_proc_dir) {
 292                 lprocfs_remove(seq->seq_proc_dir);
 293                 seq->seq_proc_dir = NULL;
 294         }
 295         EXIT;
 296 }
 297 #endif
 298
 299 int
 300 seq_client_init(struct lu_client_seq *seq,
 301                 const char *uuid,
 302                 struct obd_export *exp)
 303 {
 304         int rc = 0;
 305         ENTRY;
 306
 307         LASSERT(exp != NULL);
 308
 309         fid_zero(&seq->seq_fid);
 310         range_zero(&seq->seq_range);
 311         sema_init(&seq->seq_sem, 1);
 312         seq->seq_exp = class_export_get(exp);
 313         seq->seq_width = LUSTRE_SEQ_MAX_WIDTH;
 314
 315         snprintf(seq->seq_name, sizeof(seq->seq_name),
 316                  "%s-%s", LUSTRE_SEQ_NAME, uuid);
 317
 318 #ifdef LPROCFS
 319         rc = seq_client_proc_init(seq);
 320 #endif
 321
 322         if (rc)
 323                 seq_client_fini(seq);
 324         else
 325                 CDEBUG(D_INFO|D_WARNING,
 326                        "Client Sequence Manager\n");
 327         RETURN(rc);
 328 }
 329 EXPORT_SYMBOL(seq_client_init);
 330
 331 void seq_client_fini(struct lu_client_seq *seq)
 332 {
 333         ENTRY;
 334
 335 #ifdef LPROCFS
 336         seq_client_proc_fini(seq);
 337 #endif
 338
 339         if (seq->seq_exp != NULL) {
 340                 class_export_put(seq->seq_exp);
 341                 seq->seq_exp = NULL;
 342         }
 343
 344         CDEBUG(D_INFO|D_WARNING, "Client Sequence Manager\n");
 345
 346         EXIT;
 347 }
 348 EXPORT_SYMBOL(seq_client_fini);