lustre/fid/fid_store.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/fid/fid_store.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 #ifdef __KERNEL__
  51 enum {
  52         SEQ_TXN_STORE_CREDITS = 20
  53 };
  54
  55 static struct lu_buf *seq_store_buf(struct seq_thread_info *info)
  56 {
  57         struct lu_buf *buf;
  58
  59         buf = &info->sti_buf;
  60         buf->lb_buf = &info->sti_space;
  61         buf->lb_len = sizeof(info->sti_space);
  62         return buf;
  63 }
  64
  65 /* This function implies that caller takes care about locking. */
  66 int seq_store_write(struct lu_server_seq *seq,
  67                     const struct lu_env *env)
  68 {
  69         struct dt_object *dt_obj = seq->lss_obj;
  70         struct seq_thread_info *info;
  71         struct dt_device *dt_dev;
  72         struct thandle *th;
  73         loff_t pos = 0;
  74         int rc;
  75         ENTRY;
  76
  77         dt_dev = lu2dt_dev(seq->lss_obj->do_lu.lo_dev);
  78         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
  79         LASSERT(info != NULL);
  80
  81         /* Stub here, will fix it later. */
  82         txn_param_init(&info->sti_txn, SEQ_TXN_STORE_CREDITS);
  83
  84         th = dt_dev->dd_ops->dt_trans_start(env, dt_dev, &info->sti_txn);
  85         if (!IS_ERR(th)) {
  86                 /* Store ranges in le format. */
  87                 range_cpu_to_le(&info->sti_space, &seq->lss_space);
  88
  89                 rc = dt_obj->do_body_ops->dbo_write(env, dt_obj,
  90                                                     seq_store_buf(info),
  91                                                     &pos, th, BYPASS_CAPA);
  92                 if (rc == sizeof(info->sti_space)) {
  93                         CDEBUG(D_INFO, "%s: Space - "DRANGE"\n",
  94                                seq->lss_name, PRANGE(&seq->lss_space));
  95                         rc = 0;
  96                 } else if (rc >= 0) {
  97                         rc = -EIO;
  98                 }
  99
 100                 dt_dev->dd_ops->dt_trans_stop(env, th);
 101         } else {
 102                 rc = PTR_ERR(th);
 103         }
 104
 105         RETURN(rc);
 106 }
 107
 108 /*
 109  * This function implies that caller takes care about locking or locking is not
 110  * needed (init time).
 111  */
 112 int seq_store_read(struct lu_server_seq *seq,
 113                    const struct lu_env *env)
 114 {
 115         struct dt_object *dt_obj = seq->lss_obj;
 116         struct seq_thread_info *info;
 117         loff_t pos = 0;
 118         int rc;
 119         ENTRY;
 120
 121         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
 122         LASSERT(info != NULL);
 123
 124         rc = dt_obj->do_body_ops->dbo_read(env, dt_obj, seq_store_buf(info),
 125                                            &pos, BYPASS_CAPA);
 126
 127         if (rc == sizeof(info->sti_space)) {
 128                 range_le_to_cpu(&seq->lss_space, &info->sti_space);
 129                 CDEBUG(D_INFO, "%s: Space - "DRANGE"\n",
 130                        seq->lss_name, PRANGE(&seq->lss_space));
 131                 rc = 0;
 132         } else if (rc == 0) {
 133                 rc = -ENODATA;
 134         } else if (rc >= 0) {
 135                 CERROR("%s: Read only %d bytes of %d\n", seq->lss_name,
 136                        rc, sizeof(info->sti_space));
 137                 rc = -EIO;
 138         }
 139
 140         RETURN(rc);
 141 }
 142
 143 int seq_store_init(struct lu_server_seq *seq,
 144                    const struct lu_env *env,
 145                    struct dt_device *dt)
 146 {
 147         struct dt_object *dt_obj;
 148         struct lu_fid fid;
 149         const char *name;
 150         int rc;
 151         ENTRY;
 152
 153         name = seq->lss_type == LUSTRE_SEQ_SERVER ?
 154                 LUSTRE_SEQ_SRV_NAME : LUSTRE_SEQ_CTL_NAME;
 155
 156         dt_obj = dt_store_open(env, dt, name, &fid);
 157         if (!IS_ERR(dt_obj)) {
 158                 seq->lss_obj = dt_obj;
 159                 rc = 0;
 160         } else {
 161                 CERROR("%s: Can't find \"%s\" obj %d\n",
 162                        seq->lss_name, name, (int)PTR_ERR(dt_obj));
 163                 rc = PTR_ERR(dt_obj);
 164         }
 165
 166         RETURN(rc);
 167 }
 168
 169 void seq_store_fini(struct lu_server_seq *seq,
 170                     const struct lu_env *env)
 171 {
 172         ENTRY;
 173
 174         if (seq->lss_obj != NULL) {
 175                 if (!IS_ERR(seq->lss_obj))
 176                         lu_object_put(env, &seq->lss_obj->do_lu);
 177                 seq->lss_obj = NULL;
 178         }
 179
 180         EXIT;
 181 }
 182 #endif