X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fquota%2Fqsd_lib.c;h=d7e8e5c4d160ecaac98e1e4b7783b56015b19dfa;hb=af46e574eeebe23c5a6146a134fa00ee91931de3;hp=2514a63b66f6717be76ec2c19d4c7af14665e092;hpb=e88b3bf26a477a2b42182c4b451bc2fe55fa88fa;p=fs%2Flustre-release.git diff --git a/lustre/quota/qsd_lib.c b/lustre/quota/qsd_lib.c index 2514a63..d7e8e5c 100644 --- a/lustre/quota/qsd_lib.c +++ b/lustre/quota/qsd_lib.c @@ -21,7 +21,7 @@ * GPL HEADER END */ /* - * Copyright (c) 2012 Intel, Inc. + * Copyright (c) 2012, 2014, Intel Corporation. * Use is subject to license terms. * * Author: Johann Lombardi @@ -49,16 +49,12 @@ * freed via qsd_fini(). */ -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif - #define DEBUG_SUBSYSTEM S_LQUOTA #include #include "qsd_internal.h" -cfs_mem_cache_t *upd_kmem; +struct kmem_cache *upd_kmem; struct lu_kmem_descr qsd_caches[] = { { @@ -77,10 +73,9 @@ LU_CONTEXT_KEY_DEFINE(qsd, LCT_MD_THREAD | LCT_DT_THREAD | LCT_LOCAL); LU_KEY_INIT_GENERIC(qsd); /* some procfs helpers */ -static int lprocfs_qsd_rd_state(char *page, char **start, off_t off, - int count, int *eof, void *data) +static int qsd_state_seq_show(struct seq_file *m, void *data) { - struct qsd_instance *qsd = (struct qsd_instance *)data; + struct qsd_instance *qsd = m->private; char enabled[5]; int rc; @@ -94,33 +89,41 @@ static int lprocfs_qsd_rd_state(char *page, char **start, off_t off, if (strlen(enabled) == 0) strcat(enabled, "none"); - rc = snprintf(page, count, - "target name: %s\n" - "pool ID: %d\n" - "type: %s\n" - "quota enabled: %s\n" - "conn to master: %s\n", - qsd->qsd_svname, qsd->qsd_pool_id, - qsd->qsd_is_md ? "md" : "dt", enabled, - qsd->qsd_exp_valid ? "setup" : "not setup yet"); - - if (qsd->qsd_prepared) - rc += snprintf(page + rc, count - rc, + rc = seq_printf(m, "target name: %s\n" + "pool ID: %d\n" + "type: %s\n" + "quota enabled: %s\n" + "conn to master: %s\n", + qsd->qsd_svname, qsd->qsd_pool_id, + qsd->qsd_is_md ? "md" : "dt", enabled, + qsd->qsd_exp_valid ? "setup" : "not setup yet"); + + if (qsd->qsd_prepared) { + memset(enabled, 0, sizeof(enabled)); + if (qsd->qsd_type_array[USRQUOTA]->qqi_acct_obj != NULL) + strcat(enabled, "u"); + if (qsd->qsd_type_array[GRPQUOTA]->qqi_acct_obj != NULL) + strcat(enabled, "g"); + if (strlen(enabled) == 0) + strcat(enabled, "none"); + rc += seq_printf(m, "space acct: %s\n" "user uptodate: glb[%d],slv[%d],reint[%d]\n" "group uptodate: glb[%d],slv[%d],reint[%d]\n", + enabled, qsd->qsd_type_array[USRQUOTA]->qqi_glb_uptodate, qsd->qsd_type_array[USRQUOTA]->qqi_slv_uptodate, qsd->qsd_type_array[USRQUOTA]->qqi_reint, qsd->qsd_type_array[GRPQUOTA]->qqi_glb_uptodate, qsd->qsd_type_array[GRPQUOTA]->qqi_slv_uptodate, qsd->qsd_type_array[GRPQUOTA]->qqi_reint); + } return rc; } +LPROC_SEQ_FOPS_RO(qsd_state); -static int lprocfs_qsd_rd_enabled(char *page, char **start, off_t off, - int count, int *eof, void *data) +static int qsd_enabled_seq_show(struct seq_file *m, void *data) { - struct qsd_instance *qsd = (struct qsd_instance *)data; + struct qsd_instance *qsd = m->private; char enabled[5]; LASSERT(qsd != NULL); @@ -133,20 +136,22 @@ static int lprocfs_qsd_rd_enabled(char *page, char **start, off_t off, if (strlen(enabled) == 0) strcat(enabled, "none"); - return snprintf(page, count, "%s\n", enabled); + return seq_printf(m, "%s\n", enabled); } +LPROC_SEQ_FOPS_RO(qsd_enabled); /* force reintegration procedure to be executed. * Used for test/debugging purpose */ -static int lprocfs_qsd_wr_force_reint(struct file *file, const char *buffer, - unsigned long count, void *data) +static ssize_t +lprocfs_force_reint_seq_write(struct file *file, const char *buffer, + size_t count, loff_t *off) { - struct qsd_instance *qsd = (struct qsd_instance *)data; - int rc = 0, qtype; + struct qsd_instance *qsd = ((struct seq_file *)file->private_data)->private; + int rc = 0, qtype; LASSERT(qsd != NULL); - cfs_write_lock(&qsd->qsd_lock); + write_lock(&qsd->qsd_lock); if (qsd->qsd_stopping) { /* don't mess up with shutdown procedure, it is already * complicated enough */ @@ -160,7 +165,7 @@ static int lprocfs_qsd_wr_force_reint(struct file *file, const char *buffer, qsd->qsd_type_array[qtype]->qqi_slv_uptodate = false; } } - cfs_write_unlock(&qsd->qsd_lock); + write_unlock(&qsd->qsd_lock); if (rc) return rc; @@ -173,21 +178,22 @@ static int lprocfs_qsd_wr_force_reint(struct file *file, const char *buffer, } return rc == 0 ? count : rc; } +LPROC_SEQ_FOPS_WO_TYPE(qsd, force_reint); -static int lprocfs_qsd_rd_timeout(char *page, char **start, off_t off, - int count, int *eof, void *data) +static int qsd_timeout_seq_show(struct seq_file *m, void *data) { - struct qsd_instance *qsd = (struct qsd_instance *)data; + struct qsd_instance *qsd = m->private; LASSERT(qsd != NULL); - return snprintf(page, count, "%d\n", qsd_wait_timeout(qsd)); + return seq_printf(m, "%d\n", qsd_wait_timeout(qsd)); } -static int lprocfs_qsd_wr_timeout(struct file *file, const char *buffer, - unsigned long count, void *data) +static ssize_t +qsd_timeout_seq_write(struct file *file, const char *buffer, + size_t count, loff_t *off) { - struct qsd_instance *qsd = (struct qsd_instance *)data; - int timeout, rc; + struct qsd_instance *qsd = ((struct seq_file *)file->private_data)->private; + int timeout, rc; LASSERT(qsd != NULL); rc = lprocfs_write_helper(buffer, count, &timeout); @@ -199,12 +205,17 @@ static int lprocfs_qsd_wr_timeout(struct file *file, const char *buffer, qsd->qsd_timeout = timeout; return count; } +LPROC_SEQ_FOPS(qsd_timeout); static struct lprocfs_vars lprocfs_quota_qsd_vars[] = { - { "info", lprocfs_qsd_rd_state, 0, 0}, - { "enabled", lprocfs_qsd_rd_enabled, 0, 0}, - { "force_reint", 0, lprocfs_qsd_wr_force_reint, 0}, - { "timeout", lprocfs_qsd_rd_timeout, lprocfs_qsd_wr_timeout, 0}, + { .name = "info", + .fops = &qsd_state_fops }, + { .name = "enabled", + .fops = &qsd_enabled_fops }, + { .name = "force_reint", + .fops = &qsd_force_reint_fops }, + { .name = "timeout", + .fops = &qsd_timeout_fops }, { NULL } }; @@ -225,14 +236,12 @@ static int qsd_conn_callback(void *data) /* qsd_exp should now be valid */ LASSERT(qsd->qsd_exp); - /* grab reference on namespace */ - ldlm_namespace_get(class_exp2obd(qsd->qsd_exp)->obd_namespace); qsd->qsd_ns = class_exp2obd(qsd->qsd_exp)->obd_namespace; - cfs_write_lock(&qsd->qsd_lock); + write_lock(&qsd->qsd_lock); /* notify that qsd_exp is now valid */ qsd->qsd_exp_valid = true; - cfs_write_unlock(&qsd->qsd_lock); + write_unlock(&qsd->qsd_lock); /* Now that the connection to master is setup, we can initiate the * reintegration procedure for quota types which are enabled. @@ -243,7 +252,7 @@ static int qsd_conn_callback(void *data) * step 3) will have to wait for qsd_start() to be called */ for (type = USRQUOTA; type < MAXQUOTAS; type++) { struct qsd_qtype_info *qqi = qsd->qsd_type_array[type]; - cfs_waitq_signal(&qqi->qqi_reint_thread.t_ctl_waitq); + wake_up(&qqi->qqi_reint_thread.t_ctl_waitq); } RETURN(0); @@ -263,6 +272,7 @@ static void qsd_qtype_fini(const struct lu_env *env, struct qsd_instance *qsd, int qtype) { struct qsd_qtype_info *qqi; + int repeat = 0; ENTRY; if (qsd->qsd_type_array[qtype] == NULL) @@ -271,8 +281,8 @@ static void qsd_qtype_fini(const struct lu_env *env, struct qsd_instance *qsd, qsd->qsd_type_array[qtype] = NULL; /* all deferred work lists should be empty */ - LASSERT(cfs_list_empty(&qqi->qqi_deferred_glb)); - LASSERT(cfs_list_empty(&qqi->qqi_deferred_slv)); + LASSERT(list_empty(&qqi->qqi_deferred_glb)); + LASSERT(list_empty(&qqi->qqi_deferred_slv)); /* shutdown lquota site */ if (qqi->qqi_site != NULL && !IS_ERR(qqi->qqi_site)) { @@ -280,8 +290,31 @@ static void qsd_qtype_fini(const struct lu_env *env, struct qsd_instance *qsd, qqi->qqi_site = NULL; } + /* The qqi may still be holding by global locks which are being + * canceled asynchronously (LU-4365), see the following steps: + * + * - On server umount, we try to clear all quota locks first by + * disconnecting LWP (which will invalidate import and cleanup + * all locks on it), however, if quota reint process is holding + * the global lock for reintegration at that time, global lock + * will fail to be cleared on LWP disconnection. + * + * - Umount process goes on and stops reint process, the global + * lock will be dropped on reint process exit, however, the lock + * cancel in done in asynchronous way, so the + * qsd_glb_blocking_ast() might haven't been called yet when we + * get here. + */ + while (atomic_read(&qqi->qqi_ref) > 1) { + CDEBUG(D_QUOTA, "qqi reference count %u, repeat: %d\n", + atomic_read(&qqi->qqi_ref), repeat); + repeat++; + set_current_state(TASK_INTERRUPTIBLE); + schedule_timeout(cfs_time_seconds(1)); + } + /* by now, all qqi users should have gone away */ - LASSERT(cfs_atomic_read(&qqi->qqi_ref) == 1); + LASSERT(atomic_read(&qqi->qqi_ref) == 1); lu_ref_fini(&qqi->qqi_reference); /* release accounting object */ @@ -337,7 +370,7 @@ static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd, if (qqi == NULL) RETURN(-ENOMEM); qsd->qsd_type_array[qtype] = qqi; - cfs_atomic_set(&qqi->qqi_ref, 1); /* referenced from qsd */ + atomic_set(&qqi->qqi_ref, 1); /* referenced from qsd */ /* set backpointer and other parameters */ qqi->qqi_qsd = qsd; @@ -348,21 +381,20 @@ static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd, qqi->qqi_glb_uptodate = false; qqi->qqi_slv_uptodate = false; qqi->qqi_reint = false; - cfs_waitq_init(&qqi->qqi_reint_thread.t_ctl_waitq); + init_waitqueue_head(&qqi->qqi_reint_thread.t_ctl_waitq); thread_set_flags(&qqi->qqi_reint_thread, SVC_STOPPED); - CFS_INIT_LIST_HEAD(&qqi->qqi_deferred_glb); - CFS_INIT_LIST_HEAD(&qqi->qqi_deferred_slv); - memset(&qqi->qqi_lockh, 0, sizeof(qqi->qqi_lockh)); + INIT_LIST_HEAD(&qqi->qqi_deferred_glb); + INIT_LIST_HEAD(&qqi->qqi_deferred_slv); /* open accounting object */ LASSERT(qqi->qqi_acct_obj == NULL); qqi->qqi_acct_obj = acct_obj_lookup(env, qsd->qsd_dev, qtype); - if (qqi->qqi_acct_obj == NULL) { - LCONSOLE_ERROR("%s: No %s space accounting support. Please use " - "tunefs.lustre --quota option to enable quota " - "accounting.\n", - qsd->qsd_svname, QTYPE_NAME(qtype)); - GOTO(out, rc = -ENOENT); + if (IS_ERR(qqi->qqi_acct_obj)) { + CDEBUG(D_QUOTA, "%s: no %s space accounting support rc:%ld\n", + qsd->qsd_svname, QTYPE_NAME(qtype), + PTR_ERR(qqi->qqi_acct_obj)); + qqi->qqi_acct_obj = NULL; + qsd->qsd_acct_failed = true; } /* open global index copy */ @@ -401,7 +433,7 @@ static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd, GOTO(out, rc = PTR_ERR(qqi->qqi_site)); } - /* register proc entry for accounting object */ + /* register proc entry for accounting & global index copy objects */ rc = lprocfs_seq_create(qsd->qsd_proc, qtype == USRQUOTA ? "acct_user" : "acct_group", 0444, &lprocfs_quota_seq_fops, @@ -411,6 +443,16 @@ static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd, qsd->qsd_svname, rc); GOTO(out, rc); } + + rc = lprocfs_seq_create(qsd->qsd_proc, + qtype == USRQUOTA ? "limit_user" : "limit_group", + 0444, &lprocfs_quota_seq_fops, + qqi->qqi_glb_obj); + if (rc) { + CERROR("%s: can't add procfs entry for global index copy %d\n", + qsd->qsd_svname, rc); + GOTO(out, rc); + } EXIT; out: if (rc) @@ -436,17 +478,9 @@ void qsd_fini(const struct lu_env *env, struct qsd_instance *qsd) RETURN_EXIT; CDEBUG(D_QUOTA, "%s: initiating QSD shutdown\n", qsd->qsd_svname); - cfs_write_lock(&qsd->qsd_lock); + write_lock(&qsd->qsd_lock); qsd->qsd_stopping = true; - cfs_write_unlock(&qsd->qsd_lock); - - /* remove from the list of fsinfo */ - if (!cfs_list_empty(&qsd->qsd_link)) { - LASSERT(qsd->qsd_fsinfo != NULL); - cfs_down(&qsd->qsd_fsinfo->qfs_sem); - cfs_list_del_init(&qsd->qsd_link); - cfs_up(&qsd->qsd_fsinfo->qfs_sem); - } + write_unlock(&qsd->qsd_lock); /* remove qsd proc entry */ if (qsd->qsd_proc != NULL) { @@ -464,9 +498,7 @@ void qsd_fini(const struct lu_env *env, struct qsd_instance *qsd) qsd_stop_reint_thread(qsd->qsd_type_array[qtype]); } - /* release reference on namespace */ if (qsd->qsd_ns != NULL) { - ldlm_namespace_put(qsd->qsd_ns); qsd->qsd_ns = NULL; } @@ -476,11 +508,17 @@ void qsd_fini(const struct lu_env *env, struct qsd_instance *qsd) /* deregister connection to the quota master */ qsd->qsd_exp_valid = false; - lustre_deregister_osp_item(&qsd->qsd_exp); + lustre_deregister_lwp_item(&qsd->qsd_exp); /* release per-filesystem information */ - if (qsd->qsd_fsinfo != NULL) + if (qsd->qsd_fsinfo != NULL) { + mutex_lock(&qsd->qsd_fsinfo->qfs_mutex); + /* remove from the list of fsinfo */ + list_del_init(&qsd->qsd_link); + mutex_unlock(&qsd->qsd_fsinfo->qfs_mutex); qsd_put_fsinfo(qsd->qsd_fsinfo); + qsd->qsd_fsinfo = NULL; + } /* release quota root directory */ if (qsd->qsd_root != NULL) { @@ -516,7 +554,7 @@ EXPORT_SYMBOL(qsd_fini); */ struct qsd_instance *qsd_init(const struct lu_env *env, char *svname, struct dt_device *dev, - cfs_proc_dir_entry_t *osd_proc) + struct proc_dir_entry *osd_proc) { struct qsd_thread_info *qti = qsd_info(env); struct qsd_instance *qsd; @@ -534,18 +572,20 @@ struct qsd_instance *qsd_init(const struct lu_env *env, char *svname, RETURN(ERR_PTR(-ENOMEM)); /* generic initializations */ - cfs_rwlock_init(&qsd->qsd_lock); - CFS_INIT_LIST_HEAD(&qsd->qsd_link); + rwlock_init(&qsd->qsd_lock); + INIT_LIST_HEAD(&qsd->qsd_link); thread_set_flags(&qsd->qsd_upd_thread, SVC_STOPPED); - cfs_waitq_init(&qsd->qsd_upd_thread.t_ctl_waitq); - CFS_INIT_LIST_HEAD(&qsd->qsd_upd_list); - cfs_spin_lock_init(&qsd->qsd_adjust_lock); - CFS_INIT_LIST_HEAD(&qsd->qsd_adjust_list); + init_waitqueue_head(&qsd->qsd_upd_thread.t_ctl_waitq); + INIT_LIST_HEAD(&qsd->qsd_upd_list); + spin_lock_init(&qsd->qsd_adjust_lock); + INIT_LIST_HEAD(&qsd->qsd_adjust_list); qsd->qsd_prepared = false; qsd->qsd_started = false; /* copy service name */ - strncpy(qsd->qsd_svname, svname, MAX_OBD_NAME); + if (strlcpy(qsd->qsd_svname, svname, sizeof(qsd->qsd_svname)) + >= sizeof(qsd->qsd_svname)) + GOTO(out, rc = -E2BIG); /* grab reference on osd device */ lu_device_get(&dev->dd_lu_dev); @@ -572,9 +612,9 @@ struct qsd_instance *qsd_init(const struct lu_env *env, char *svname, } /* add in the list of lquota_fsinfo */ - cfs_down(&qsd->qsd_fsinfo->qfs_sem); + mutex_lock(&qsd->qsd_fsinfo->qfs_mutex); list_add_tail(&qsd->qsd_link, &qsd->qsd_fsinfo->qfs_qsd_list); - cfs_up(&qsd->qsd_fsinfo->qfs_sem); + mutex_unlock(&qsd->qsd_fsinfo->qfs_mutex); /* register procfs directory */ qsd->qsd_proc = lprocfs_register(QSD_DIR, osd_proc, @@ -620,12 +660,12 @@ int qsd_prepare(const struct lu_env *env, struct qsd_instance *qsd) if (unlikely(qsd == NULL)) RETURN(0); - cfs_read_lock(&qsd->qsd_lock); + read_lock(&qsd->qsd_lock); if (qsd->qsd_prepared) { CERROR("%s: qsd instance already prepared\n", qsd->qsd_svname); rc = -EALREADY; } - cfs_read_unlock(&qsd->qsd_lock); + read_unlock(&qsd->qsd_lock); if (rc) RETURN(rc); @@ -657,14 +697,23 @@ int qsd_prepare(const struct lu_env *env, struct qsd_instance *qsd) } /* pools successfully setup, mark the qsd as prepared */ - cfs_write_lock(&qsd->qsd_lock); + write_lock(&qsd->qsd_lock); qsd->qsd_prepared = true; - cfs_write_unlock(&qsd->qsd_lock); + write_unlock(&qsd->qsd_lock); /* start reintegration thread for each type, if required */ for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) { struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype]; + if (qsd_type_enabled(qsd, qtype) && qsd->qsd_acct_failed) { + LCONSOLE_ERROR("%s: can't enable quota enforcement " + "since space accounting isn't functional" + ". Please run tunefs.lustre --quota on " + "an unmounted filesystem if not done " + "already\n", qsd->qsd_svname); + break; + } + rc = qsd_start_reint_thread(qqi); if (rc) { CERROR("%s: failed to start reint thread for type %s " @@ -683,7 +732,8 @@ int qsd_prepare(const struct lu_env *env, struct qsd_instance *qsd) } /* generate osp name */ - rc = tgt_name2ospname((char *)qsd->qsd_svname, qti->qti_buf); + rc = tgt_name2lwp_name(qsd->qsd_svname, qti->qti_buf, + MTI_NAME_MAXLEN, 0); if (rc) { CERROR("%s: failed to generate ospname (%d)\n", qsd->qsd_svname, rc); @@ -692,7 +742,7 @@ int qsd_prepare(const struct lu_env *env, struct qsd_instance *qsd) /* the connection callback will start the reintegration * procedure if quota is enabled */ - rc = lustre_register_osp_item(qti->qti_buf, &qsd->qsd_exp, + rc = lustre_register_lwp_item(qti->qti_buf, &qsd->qsd_exp, qsd_conn_callback, (void *)qsd); if (rc) { CERROR("%s: fail to get connection to master (%d)\n", @@ -721,9 +771,9 @@ int qsd_start(const struct lu_env *env, struct qsd_instance *qsd) if (unlikely(qsd == NULL)) RETURN(0); - cfs_write_lock(&qsd->qsd_lock); + write_lock(&qsd->qsd_lock); if (!qsd->qsd_prepared) { - CERROR("%s: can't start qsd instance since it was properly " + CERROR("%s: can't start qsd instance since it wasn't properly " "initialized\n", qsd->qsd_svname); rc = -EFAULT; } else if (qsd->qsd_started) { @@ -733,7 +783,7 @@ int qsd_start(const struct lu_env *env, struct qsd_instance *qsd) /* notify that the qsd_instance is now started */ qsd->qsd_started = true; } - cfs_write_unlock(&qsd->qsd_lock); + write_unlock(&qsd->qsd_lock); if (rc) RETURN(rc); @@ -742,7 +792,7 @@ int qsd_start(const struct lu_env *env, struct qsd_instance *qsd) * up to usage; If usage < granted, release down to usage. */ for (type = USRQUOTA; type < MAXQUOTAS; type++) { struct qsd_qtype_info *qqi = qsd->qsd_type_array[type]; - cfs_waitq_signal(&qqi->qqi_reint_thread.t_ctl_waitq); + wake_up(&qqi->qqi_reint_thread.t_ctl_waitq); } RETURN(rc);