mgc_request.c

lustre 1.6.5 source code

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  lustre/mgc/mgc_request.c
 *  Lustre Management Client
 *
 *  Copyright (C) 2006 Cluster File Systems, Inc.
 *   Author: Nathan Rutman <[e-mail]>
 *
 *   This file is part of Lustre, http://www.lustre.org
 *
 *   Lustre is free software; you can redistribute it and/or
 *   modify it under the terms of version 2 of the GNU General Public
 *   License as published by the Free Software Foundation.
 *
 *   Lustre is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Lustre; if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 *
 */

#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif
#define DEBUG_SUBSYSTEM S_MGC
#define D_MGC D_CONFIG /*|D_WARNING*/

#ifdef __KERNEL__
# include <linux/module.h>
# include <linux/pagemap.h>
# include <linux/miscdevice.h>
# include <linux/init.h>
#else
# include <liblustre.h>
#endif

#include <obd_class.h>
#include <lustre_dlm.h>
#include <lprocfs_status.h>
#include <lustre_log.h>
#include <lustre_fsfilt.h>
#include <lustre_disk.h>
#include "mgc_internal.h"

static int mgc_name2resid(char *name, int len, struct ldlm_res_id *res_id)
{
        __u64 resname = 0;

        if (len > 8) {
                CERROR("name too long: %s\n", name);
                return -EINVAL;
        }
        if (len <= 0) {
                CERROR("missing name: %s\n", name);
                return -EINVAL;
        }
        memcpy(&resname, name, len);

        memset(res_id, 0, sizeof(*res_id));

        /* Always use the same endianness for the resid */
        res_id->name[0] = cpu_to_le64(resname);
        CDEBUG(D_MGC, "log %s to resid "LPX64"/"LPX64" (%.8s)\n", name,
               res_id->name[0], res_id->name[1], (char *)&res_id->name[0]);
        return 0;
}

int mgc_fsname2resid(char *fsname, struct ldlm_res_id *res_id)
{
        /* fsname is at most 8 chars long, maybe contain "-".
         * e.g. "lustre", "CFS-000" */
        return mgc_name2resid(fsname, strlen(fsname), res_id);
}
EXPORT_SYMBOL(mgc_fsname2resid);

int mgc_logname2resid(char *logname, struct ldlm_res_id *res_id)
{
        char *name_end;
        int len;

        /* logname consists of "fsname-nodetype".
         * e.g. "lustre-MDT0001", "CFS-000-client" */
        name_end = strrchr(logname, '-');
        LASSERT(name_end);
        len = name_end - logname;
        return mgc_name2resid(logname, len, res_id);
}


/********************** config llog list **********************/
static struct list_head config_llog_list = LIST_HEAD_INIT(config_llog_list);
static spinlock_t       config_list_lock = SPIN_LOCK_UNLOCKED;

/* Take a reference to a config log */
static int config_log_get(struct config_llog_data *cld)
{
        ENTRY;
        if (cld->cld_stopping)
                RETURN(1);
        atomic_inc(&cld->cld_refcount);
        CDEBUG(D_INFO, "log %s refs %d\n", cld->cld_logname,
               atomic_read(&cld->cld_refcount));
        RETURN(0);
}

/* Drop a reference to a config log.  When no longer referenced, 
   we can free the config log data */
static void config_log_put(struct config_llog_data *cld)
{
        ENTRY;
        CDEBUG(D_INFO, "log %s refs %d\n", cld->cld_logname,
               atomic_read(&cld->cld_refcount));
        if (atomic_dec_and_test(&cld->cld_refcount)) {
                CDEBUG(D_MGC, "dropping config log %s\n", cld->cld_logname);
                class_export_put(cld->cld_mgcexp);
                spin_lock(&config_list_lock);
                list_del(&cld->cld_list_chain);
                spin_unlock(&config_list_lock);
                OBD_FREE(cld->cld_logname, strlen(cld->cld_logname) + 1);
                if (cld->cld_cfg.cfg_instance != NULL)
                        OBD_FREE(cld->cld_cfg.cfg_instance, 
                                 strlen(cld->cld_cfg.cfg_instance) + 1);
                OBD_FREE(cld, sizeof(*cld));
        }
        EXIT;
}

/* Find a config log by name */
static struct config_llog_data *config_log_find(char *logname, 
                                               struct config_llog_instance *cfg)
{
        struct list_head *tmp;
        struct config_llog_data *cld;
        char *logid = logname;
        int match_instance = 0;
        ENTRY;

        if (cfg && cfg->cfg_instance) {
                match_instance++;
                logid = cfg->cfg_instance;
        }
        if (!logid) {
                CERROR("No log specified\n");
                RETURN(ERR_PTR(-EINVAL));
        }

        spin_lock(&config_list_lock);
        list_for_each(tmp, &config_llog_list) {
                cld = list_entry(tmp, struct config_llog_data, cld_list_chain);
                if (match_instance && cld->cld_cfg.cfg_instance && 
                    strcmp(logid, cld->cld_cfg.cfg_instance) == 0)
                        goto out_found;
                if (!match_instance &&  
                    strcmp(logid, cld->cld_logname) == 0)
                        goto out_found;
        }
        spin_unlock(&config_list_lock);

        CDEBUG(D_CONFIG, "can't get log %s\n", logid);
        RETURN(ERR_PTR(-ENOENT));
out_found:
        atomic_inc(&cld->cld_refcount);
        spin_unlock(&config_list_lock);
        RETURN(cld);
}

/* Add this log to our list of active logs. 
   We have one active log per "mount" - client instance or servername.
   Each instance may be at a different point in the log. */
static int config_log_add(char *logname, struct config_llog_instance *cfg,
                          struct super_block *sb)
{
        struct config_llog_data *cld;
        struct lustre_sb_info *lsi = s2lsi(sb);
        int rc;
        ENTRY;

        CDEBUG(D_MGC, "adding config log %s:%s\n", logname, cfg->cfg_instance);
        
        OBD_ALLOC(cld, sizeof(*cld));
        if (!cld) 
                RETURN(-ENOMEM);
        OBD_ALLOC(cld->cld_logname, strlen(logname) + 1);
        if (!cld->cld_logname) { 
                OBD_FREE(cld, sizeof(*cld));
                RETURN(-ENOMEM);
        }
        strcpy(cld->cld_logname, logname);
        cld->cld_cfg = *cfg;
        cld->cld_cfg.cfg_last_idx = 0;
        cld->cld_cfg.cfg_flags = 0;
        cld->cld_cfg.cfg_sb = sb;
        atomic_set(&cld->cld_refcount, 1);
        
        /* Keep the mgc around until we are done */
        cld->cld_mgcexp = class_export_get(lsi->lsi_mgc->obd_self_export);
        
        if (cfg->cfg_instance != NULL) {
                OBD_ALLOC(cld->cld_cfg.cfg_instance, 
                          strlen(cfg->cfg_instance) + 1);
                strcpy(cld->cld_cfg.cfg_instance, cfg->cfg_instance);
        }
        rc = mgc_logname2resid(logname, &cld->cld_resid);
        spin_lock(&config_list_lock);
        list_add(&cld->cld_list_chain, &config_llog_list);
        spin_unlock(&config_list_lock);
        
        if (rc) {
                config_log_put(cld);
                RETURN(rc);
        }
        
        RETURN(rc);
}

/* Stop watching for updates on this log. */
static int config_log_end(char *logname, struct config_llog_instance *cfg)
{       
        struct config_llog_data *cld;
        int rc = 0;
        ENTRY;
                                       
        cld = config_log_find(logname, cfg);
        if (IS_ERR(cld)) 
                RETURN(PTR_ERR(cld));
        /* drop the ref from the find */
        config_log_put(cld);

        cld->cld_stopping = 1;
        /* drop the start ref */
        config_log_put(cld);
        CDEBUG(D_MGC, "end config log %s (%d)\n", logname ? logname : "client",
               rc);
        RETURN(rc);
}

/* reenqueue any lost locks */
#define RQ_RUNNING 0x1
#define RQ_NOW     0x2
#define RQ_LATER   0x4
#define RQ_STOP    0x8
static int rq_state = 0;
static cfs_waitq_t rq_waitq;

static int mgc_process_log(struct obd_device *mgc, 
                           struct config_llog_data *cld);

static int mgc_requeue_thread(void *data)
{
        struct l_wait_info lwi_now, lwi_later;
        struct config_llog_data *cld, *n;
        char name[] = "ll_cfg_requeue";
        int rc = 0;
        ENTRY;
        
        ptlrpc_daemonize(name);
        
        CDEBUG(D_MGC, "Starting requeue thread\n");

        lwi_later = LWI_TIMEOUT(60 * HZ, NULL, NULL);
        l_wait_event(rq_waitq, rq_state & (RQ_NOW | RQ_STOP), &lwi_later);

        /* Keep trying failed locks periodically */
        spin_lock(&config_list_lock);
        while (rq_state & (RQ_NOW | RQ_LATER)) {
                /* Any new or requeued lostlocks will change the state */
                rq_state &= ~(RQ_NOW | RQ_LATER); 
                spin_unlock(&config_list_lock);

                /* Always wait a few seconds to allow the server who 
                   caused the lock revocation to finish its setup, plus some
                   random so everyone doesn't try to reconnect at once. */
                lwi_now = LWI_TIMEOUT(3 * HZ + (ll_rand() & 0xff) * (HZ / 100),
                                      NULL, NULL);
                l_wait_event(rq_waitq, rq_state & RQ_STOP, &lwi_now);
                
                spin_lock(&config_list_lock);
                list_for_each_entry_safe(cld, n, &config_llog_list,
                                         cld_list_chain) {
                        spin_unlock(&config_list_lock);
                        
                        if (cld->cld_lostlock) {
                                CDEBUG(D_MGC, "updating log %s\n", 
                                       cld->cld_logname);
                                cld->cld_lostlock = 0;
                                rc = mgc_process_log(cld->cld_mgcexp->exp_obd,
                                                     cld);
                                /* Whether we enqueued again or not in 
                                   mgc_process_log, we're done with the ref 
                                   from the old enqueue */        
                                config_log_put(cld);
                        }

                        spin_lock(&config_list_lock);
                }
                spin_unlock(&config_list_lock);
                
                /* Wait a bit to see if anyone else needs a requeue */
                l_wait_event(rq_waitq, rq_state & (RQ_NOW | RQ_STOP),
                             &lwi_later);
                spin_lock(&config_list_lock);
        }
        /* spinlock and while guarantee RQ_NOW and RQ_LATER are not set */
        rq_state &= ~RQ_RUNNING;
        spin_unlock(&config_list_lock);
        
        CDEBUG(D_MGC, "Ending requeue thread\n");
        RETURN(rc);
}

/* Add a cld to the list to requeue.  Start the requeue thread if needed.
   We are responsible for dropping the config log reference from here on out. */
static int mgc_requeue_add(struct config_llog_data *cld, int later)
{
        int rc = 0;

        CDEBUG(D_INFO, "log %s: requeue (l=%d r=%d sp=%d st=%x)\n", 
               cld->cld_logname, later, atomic_read(&cld->cld_refcount),
               cld->cld_stopping, rq_state);
        
        /* Hold lock for rq_state */
        spin_lock(&config_list_lock);
        cld->cld_lostlock = 1;

        if (cld->cld_stopping || (rq_state & RQ_STOP)) {
                spin_unlock(&config_list_lock);
                config_log_put(cld);
                RETURN(0);
        }

        if (!(rq_state & RQ_RUNNING)) {
                LASSERT(rq_state == 0);
                rq_state = RQ_RUNNING | (later ? RQ_LATER : RQ_NOW);
                spin_unlock(&config_list_lock);
                rc = cfs_kernel_thread(mgc_requeue_thread, 0,
                                       CLONE_VM | CLONE_FILES);
                if (rc < 0) {
                        CERROR("log %s: cannot start requeue thread (%d),"
                               "no more log updates!\n", cld->cld_logname, rc);
                        /* Drop the ref, since the rq thread won't */
                        cld->cld_lostlock = 0;
                        config_log_put(cld);
                        rq_state = 0;
                        RETURN(rc);
                }
        } else {
                rq_state |= later ? RQ_LATER : RQ_NOW;
                spin_unlock(&config_list_lock);
                cfs_waitq_signal(&rq_waitq);
        }

        RETURN(0);
}

/********************** class fns **********************/

static int mgc_fs_setup(struct obd_device *obd, struct super_block *sb, 
                        struct vfsmount *mnt)
{
        struct lvfs_run_ctxt saved;
        struct lustre_sb_info *lsi = s2lsi(sb);
        struct client_obd *cli = &obd->u.cli;
        struct dentry *dentry;
        char *label;
        int err = 0;
        ENTRY;

        LASSERT(lsi);
        LASSERT(lsi->lsi_srv_mnt == mnt);

        /* The mgc fs exclusion sem. Only one fs can be setup at a time. */
        down(&cli->cl_mgc_sem);

        cleanup_group_info();

        obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd));
        if (IS_ERR(obd->obd_fsops)) {
                up(&cli->cl_mgc_sem);
                CERROR("No fstype %s rc=%ld\n", MT_STR(lsi->lsi_ldd), 
                       PTR_ERR(obd->obd_fsops));
                RETURN(PTR_ERR(obd->obd_fsops));
        }

        cli->cl_mgc_vfsmnt = mnt;
        fsfilt_setup(obd, mnt->mnt_sb);

        OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
        obd->obd_lvfs_ctxt.pwdmnt = mnt;
        obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
        obd->obd_lvfs_ctxt.fs = get_ds();

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        dentry = lookup_one_len(MOUNT_CONFIGS_DIR, current->fs->pwd,
                                strlen(MOUNT_CONFIGS_DIR));
        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        if (IS_ERR(dentry)) {
                err = PTR_ERR(dentry);
                CERROR("cannot lookup %s directory: rc = %d\n", 
                       MOUNT_CONFIGS_DIR, err);
                GOTO(err_ops, err);
        }
        cli->cl_mgc_configs_dir = dentry;

        /* We take an obd ref to insure that we can't get to mgc_cleanup
           without calling mgc_fs_cleanup first. */
        class_incref(obd);

        label = fsfilt_get_label(obd, mnt->mnt_sb);
        if (label)
                CDEBUG(D_MGC, "MGC using disk labelled=%s\n", label);

        /* We keep the cl_mgc_sem until mgc_fs_cleanup */
        RETURN(0);

err_ops:        
        fsfilt_put_ops(obd->obd_fsops);
        obd->obd_fsops = NULL;
        cli->cl_mgc_vfsmnt = NULL;
        up(&cli->cl_mgc_sem);
        RETURN(err);
}

static int mgc_fs_cleanup(struct obd_device *obd)
{