lustre/quota/qsd_lib.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, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012 Whamcloud, Inc.
  25  * Use is subject to license terms.
  26  *
  27  * Author: Johann Lombardi <johann@whamcloud.com>
  28  * Author: Niu    Yawei    <niu@whamcloud.com>
  29  */
  30
  31 /*
  32  * Quota Slave Driver (QSD) management.
  33  */
  34
  35 #ifndef EXPORT_SYMTAB
  36 # define EXPORT_SYMTAB
  37 #endif
  38
  39 #define DEBUG_SUBSYSTEM S_LQUOTA
  40
  41 #include "qsd_internal.h"
  42
  43 /* define qsd thread key */
  44 LU_KEY_INIT_FINI(qsd, struct qsd_thread_info);
  45 LU_CONTEXT_KEY_DEFINE(qsd, LCT_MD_THREAD | LCT_DT_THREAD | LCT_LOCAL);
  46 LU_KEY_INIT_GENERIC(qsd);
  47
  48 /* some procfs helpers */
  49 static int lprocfs_qsd_rd_state(char *page, char **start, off_t off,
  50                                 int count, int *eof, void *data)
  51 {
  52         struct qsd_instance     *qsd = (struct qsd_instance *)data;
  53
  54         LASSERT(qsd != NULL);
  55
  56         return snprintf(page, count,
  57                         "target name:    %s\n"
  58                         "pool ID:        %d\n"
  59                         "type:           %s\n"
  60                         "quota enabled:  none\n",
  61                         qsd->qsd_svname, qsd->qsd_pool_id,
  62                         qsd->qsd_is_md ? "md" : "dt");
  63 }
  64
  65 static struct lprocfs_vars lprocfs_quota_qsd_vars[] = {
  66         { "info", lprocfs_qsd_rd_state, 0, 0},
  67         { NULL }
  68 };
  69
  70 /*
  71  * Release qsd_qtype_info structure which contains data associated with a
  72  * given quota type. This releases the accounting objects.
  73  * It's called on OSD cleanup when the qsd instance is released.
  74  *
  75  * \param env - is the environment passed by the caller
  76  * \param qsd - is the qsd instance managing the qsd_qtype_info structure
  77  *              to be released
  78  * \param qtype - is the quota type to be shutdown
  79  */
  80 static void qsd_qtype_fini(const struct lu_env *env, struct qsd_instance *qsd,
  81                            int qtype)
  82 {
  83         struct qsd_qtype_info   *qqi;
  84         ENTRY;
  85
  86         if (qsd->qsd_type_array[qtype] == NULL)
  87                 RETURN_EXIT;
  88         qqi = qsd->qsd_type_array[qtype];
  89         qsd->qsd_type_array[qtype] = NULL;
  90
  91         /* release accounting object */
  92         if (qqi->qqi_acct_obj != NULL && !IS_ERR(qqi->qqi_acct_obj)) {
  93                 lu_object_put(env, &qqi->qqi_acct_obj->do_lu);
  94                 qqi->qqi_acct_obj = NULL;
  95         }
  96
  97         /* release slv index */
  98         if (qqi->qqi_slv_obj != NULL && !IS_ERR(qqi->qqi_slv_obj)) {
  99                 lu_object_put(env, &qqi->qqi_slv_obj->do_lu);
 100                 qqi->qqi_slv_obj = NULL;
 101                 qqi->qqi_slv_ver = 0;
 102         }
 103
 104         /* release global index */
 105         if (qqi->qqi_glb_obj != NULL && !IS_ERR(qqi->qqi_glb_obj)) {
 106                 lu_object_put(env, &qqi->qqi_glb_obj->do_lu);
 107                 qqi->qqi_glb_obj = NULL;
 108                 qqi->qqi_glb_ver = 0;
 109         }
 110
 111         OBD_FREE_PTR(qqi);
 112         EXIT;
 113 }
 114
 115 /*
 116  * Allocate and initialize a qsd_qtype_info structure for quota type \qtype.
 117  * This opens the accounting object and initializes the proc file.
 118  * It's called on OSD start when the qsd instance is created.
 119  *
 120  * \param env  - the environment passed by the caller
 121  * \param qsd  - is the qsd instance which will be in charge of the new
 122  *               qsd_qtype_info instance.
 123  * \param qtype - is quota type to set up
 124  *
 125  * \retval - 0 on success and qsd->qsd_type_array[qtype] is allocated,
 126  *           appropriate error on failure
 127  */
 128 static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd,
 129                           int qtype)
 130 {
 131         struct qsd_qtype_info   *qqi;
 132         int                      rc;
 133         struct obd_uuid          uuid;
 134         ENTRY;
 135
 136         LASSERT(qsd->qsd_type_array[qtype] == NULL);
 137
 138         /* allocate structure for this quota type */
 139         OBD_ALLOC_PTR(qqi);
 140         if (qqi == NULL)
 141                 RETURN(-ENOMEM);
 142         qsd->qsd_type_array[qtype] = qqi;
 143
 144         /* set backpointer and other parameters */
 145         qqi->qqi_qsd   = qsd;
 146         qqi->qqi_qtype = qtype;
 147         lquota_generate_fid(&qqi->qqi_fid, qsd->qsd_pool_id, QSD_RES_TYPE(qsd),
 148                             qtype);
 149
 150         /* open accounting object */
 151         LASSERT(qqi->qqi_acct_obj == NULL);
 152         qqi->qqi_acct_obj = acct_obj_lookup(env, qsd->qsd_dev,
 153                                             qtype == USRQUOTA ? ACCT_USER_OID
 154                                                               : ACCT_GROUP_OID);
 155         /* don't print any error message on failure in order not to confuse
 156          * non-OFD user (e.g. 2.3 MDT stack) */
 157         if (IS_ERR(qqi->qqi_acct_obj))
 158                 qqi->qqi_acct_obj = NULL;
 159
 160         /* open global index copy */
 161         LASSERT(qqi->qqi_glb_obj == NULL);
 162         qqi->qqi_glb_obj = lquota_disk_glb_find_create(env, qsd->qsd_dev,
 163                                                        qsd->qsd_root,
 164                                                        &qqi->qqi_fid, true);
 165         if (IS_ERR(qqi->qqi_glb_obj)) {
 166                 CERROR("%s: can't open global index copy "DFID" %ld\n",
 167                        qsd->qsd_svname, PFID(&qqi->qqi_fid),
 168                        PTR_ERR(qqi->qqi_glb_obj));
 169                 GOTO(out, rc = PTR_ERR(qqi->qqi_glb_obj));
 170         }
 171         qqi->qqi_glb_ver = dt_version_get(env, qqi->qqi_glb_obj);
 172
 173         /* open slave index copy */
 174         LASSERT(qqi->qqi_slv_obj == NULL);
 175         obd_str2uuid(&uuid, qsd->qsd_svname);
 176         qqi->qqi_slv_obj = lquota_disk_slv_find_create(env, qsd->qsd_dev,
 177                                                        qsd->qsd_root,
 178                                                        &qqi->qqi_fid, &uuid,
 179                                                        true);
 180         if (IS_ERR(qqi->qqi_slv_obj)) {
 181                 CERROR("%s: can't open slave index copy "DFID" %ld\n",
 182                        qsd->qsd_svname, PFID(&qqi->qqi_fid),
 183                        PTR_ERR(qqi->qqi_slv_obj));
 184                 GOTO(out, rc = PTR_ERR(qqi->qqi_slv_obj));
 185         }
 186         qqi->qqi_slv_ver = dt_version_get(env, qqi->qqi_slv_obj);
 187
 188         /* register proc entry for accounting object */
 189         rc = lprocfs_seq_create(qsd->qsd_proc,
 190                                 qtype == USRQUOTA ? "acct_user" : "acct_group",
 191                                 0444, &lprocfs_quota_seq_fops,
 192                                 qqi->qqi_acct_obj);
 193         if (rc) {
 194                 CWARN("%s: can't add procfs entry for accounting file %d\n",
 195                       qsd->qsd_svname, rc);
 196                 GOTO(out, rc);
 197         }
 198
 199         EXIT;
 200 out:
 201         if (rc)
 202                 qsd_qtype_fini(env, qsd, qtype);
 203         return rc;
 204 }
 205
 206 /*
 207  * Release a qsd_instance. Companion of qsd_init(). This releases all data
 208  * structures associated with the quota slave.
 209  * This function should be called when the OSD is shutting down.
 210  *
 211  * \param env - is the environment passed by the caller
 212  * \param qsd - is the qsd instance to shutdown
 213  */
 214 void qsd_fini(const struct lu_env *env, struct qsd_instance *qsd)
 215 {
 216         int     qtype;
 217         ENTRY;
 218
 219         CDEBUG(D_QUOTA, "%s: initiating QSD shutdown\n", qsd->qsd_svname);
 220         qsd->qsd_stopping = true;
 221
 222         /* remove qsd proc entry */
 223         if (qsd->qsd_proc != NULL && !IS_ERR(qsd->qsd_proc)) {
 224                 lprocfs_remove(&qsd->qsd_proc);
 225                 qsd->qsd_proc = NULL;
 226         }
 227
 228         /* free per-quota type data */
 229         for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++)
 230                 qsd_qtype_fini(env, qsd, qtype);
 231
 232         /* release quota root directory */
 233         if (qsd->qsd_root != NULL && !IS_ERR(qsd->qsd_root)) {
 234                 lu_object_put(env, &qsd->qsd_root->do_lu);
 235                 qsd->qsd_root = NULL;
 236         }
 237
 238         /* release reference on dt_device */
 239         if (qsd->qsd_dev != NULL) {
 240                 lu_ref_del(&qsd->qsd_dev->dd_lu_dev.ld_reference, "qsd", qsd);
 241                 lu_device_put(&qsd->qsd_dev->dd_lu_dev);
 242                 qsd->qsd_dev = NULL;
 243         }
 244
 245         OBD_FREE_PTR(qsd);
 246         EXIT;
 247 }
 248 EXPORT_SYMBOL(qsd_fini);
 249
 250 /*
 251  * Create a new qsd_instance to be associated with backend osd device
 252  * identified by \dev. For now, this function just create procfs files which
 253  * dumps the accounting information
 254  *
 255  * \param env    - the environment passed by the caller
 256  * \param svname - is the service name of the OSD device creating this instance
 257  * \param dev    - is the dt_device where to store quota index files
 258  * \param osd_proc - is the procfs parent directory where to create procfs file
 259  *                   related to this new qsd instance
 260  *
 261  * \retval - pointer to new qsd_instance associated with dev \dev on success,
 262  *           appropriate error on failure
 263  */
 264 struct qsd_instance *qsd_init(const struct lu_env *env, char *svname,
 265                               struct dt_device *dev,
 266                               cfs_proc_dir_entry_t *osd_proc)
 267 {
 268         struct qsd_instance     *qsd;
 269         int                      rc, qtype;
 270         ENTRY;
 271
 272         /* allocate qsd instance */
 273         OBD_ALLOC_PTR(qsd);
 274         if (qsd == NULL)
 275                 RETURN(ERR_PTR(-ENOMEM));
 276
 277         /* copy service name */
 278         strncpy(qsd->qsd_svname, svname, MAX_OBD_NAME);
 279
 280         /* grab reference on osd device */
 281         lu_device_get(&dev->dd_lu_dev);
 282         lu_ref_add(&dev->dd_lu_dev.ld_reference, "qsd", qsd);
 283         qsd->qsd_dev = dev;
 284
 285         /* we only support pool ID 0 (default data or metadata pool) for the
 286          * time being. A different pool ID could be assigned to this target via
 287          * the configuration log in the future */
 288         qsd->qsd_pool_id  = 0;
 289
 290         /* Record whether this qsd instance is managing quota enforcement for a
 291          * MDT (i.e. inode quota) or OST (block quota) */
 292         qsd->qsd_is_md = lu_device_is_md(dev->dd_lu_dev.ld_site->ls_top_dev);
 293
 294         /* look-up on-disk directory for the quota slave */
 295         qsd->qsd_root = lquota_disk_dir_find_create(env, dev, NULL, QSD_DIR);
 296         if (IS_ERR(qsd->qsd_root)) {
 297                 rc = PTR_ERR(qsd->qsd_root);
 298                 CERROR("%s: failed to create quota slave root dir (%d)\n",
 299                        svname, rc);
 300                 GOTO(out, rc);
 301         }
 302
 303         /* register procfs directory */
 304         qsd->qsd_proc = lprocfs_register(QSD_DIR, osd_proc,
 305                                          lprocfs_quota_qsd_vars, qsd);
 306         if (IS_ERR(qsd->qsd_proc)) {
 307                 rc = PTR_ERR(qsd->qsd_proc);
 308                 CERROR("%s: fail to create quota slave proc entry (%d)\n",
 309                        svname, rc);
 310                 GOTO(out, rc);
 311         }
 312
 313         /* initialize per-quota type data */
 314         for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
 315                 rc = qsd_qtype_init(env, qsd, qtype);
 316                 if (rc)
 317                         GOTO(out, rc);
 318         }
 319 out:
 320         if (rc) {
 321                 qsd_fini(env, qsd);
 322                 return ERR_PTR(rc);
 323         }
 324         RETURN(qsd);
 325 }
 326 EXPORT_SYMBOL(qsd_init);
 327
 328 /*
 329  * Global initialization performed at module load time
 330  */
 331 int qsd_glb_init(void)
 332 {
 333         qsd_key_init_generic(&qsd_thread_key, NULL);
 334         lu_context_key_register(&qsd_thread_key);
 335         return 0;
 336 }
 337
 338 /*
 339  * Companion of qsd_glb_init() called at module unload time
 340  */
 341 void qsd_glb_fini(void)
 342 {
 343         lu_context_key_degister(&qsd_thread_key);
 344 }