lov_obd.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,866 行 · 第 1/5 页

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2002-2006 Cluster File Systems, Inc.
 * Author: Phil Schwan <[e-mail]>
 *         Peter Braam <[e-mail]>
 *         Mike Shaver <[e-mail]>
 *         Nathan Rutman <[e-mail]>
 *
 *   This file is part of the Lustre file system, http://www.lustre.org
 *   Lustre is a trademark of Cluster File Systems, Inc.
 *
 *   You may have signed or agreed to another license before downloading
 *   this software.  If so, you are bound by the terms and conditions
 *   of that agreement, and the following does not apply to you.  See the
 *   LICENSE file included with this distribution for more information.
 *
 *   If you did not agree to a different license, then this copy of Lustre
 *   is open source 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.
 *
 *   In either case, 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
 *   license text for more details.
 */

#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif
#define DEBUG_SUBSYSTEM S_LOV
#ifdef __KERNEL__
#include <libcfs/libcfs.h>
#else
#include <liblustre.h>
#endif

#include <obd_support.h>
#include <lustre_lib.h>
#include <lustre_net.h>
#include <lustre/lustre_idl.h>
#include <lustre_dlm.h>
#include <lustre_mds.h>
#include <lustre_debug.h>
#include <obd_class.h>
#include <obd_lov.h>
#include <obd_ost.h>
#include <lprocfs_status.h>
#include <lustre_param.h>
#include <lustre_cache.h>

#include "lov_internal.h"


/* Keep a refcount of lov->tgt usage to prevent racing with addition/deletion.
   Any function that expects lov_tgts to remain stationary must take a ref. */
void lov_getref(struct obd_device *obd)
{
        struct lov_obd *lov = &obd->u.lov;

        /* nobody gets through here until lov_putref is done */
        mutex_down(&lov->lov_lock);
        atomic_inc(&lov->lov_refcount);
        mutex_up(&lov->lov_lock);
        return;
}

static void __lov_del_obd(struct obd_device *obd, __u32 index);

void lov_putref(struct obd_device *obd)
{
        struct lov_obd *lov = &obd->u.lov;
        mutex_down(&lov->lov_lock);
        /* ok to dec to 0 more than once -- ltd_exp's will be null */
        if (atomic_dec_and_test(&lov->lov_refcount) && lov->lov_death_row) {
                int i;
                CDEBUG(D_CONFIG, "destroying %d lov targets\n", 
                       lov->lov_death_row);
                for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                        if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_reap)
                                continue;
                        /* Disconnect and delete from list */
                        __lov_del_obd(obd, i);
                        lov->lov_death_row--;
                }
        }
        mutex_up(&lov->lov_lock);
}

static int lov_register_page_removal_cb(struct obd_export *exp,
                                        obd_page_removal_cb_t func,
                                        obd_pin_extent_cb pin_cb)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int i, rc = 0;

        if (lov->lov_page_removal_cb && lov->lov_page_removal_cb != func)
                return -EBUSY;

        if (lov->lov_page_pin_cb && lov->lov_page_pin_cb != pin_cb)
                return -EBUSY;

        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
                        continue;
                rc |= obd_register_page_removal_cb(lov->lov_tgts[i]->ltd_exp,
                                                   func, pin_cb);
        }

        lov->lov_page_removal_cb = func;
        lov->lov_page_pin_cb = pin_cb;

        return rc;
}

static int lov_unregister_page_removal_cb(struct obd_export *exp,
                                        obd_page_removal_cb_t func)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int i, rc = 0;

        if (lov->lov_page_removal_cb && lov->lov_page_removal_cb != func)
                return -EINVAL;

        lov->lov_page_removal_cb = NULL;
        lov->lov_page_pin_cb = NULL;

        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
                        continue;
                rc |= obd_unregister_page_removal_cb(lov->lov_tgts[i]->ltd_exp,
                                                     func);
        }

        return rc;
}

static int lov_register_lock_cancel_cb(struct obd_export *exp,
                                         obd_lock_cancel_cb func)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int i, rc = 0;

        if (lov->lov_lock_cancel_cb && lov->lov_lock_cancel_cb != func)
                return -EBUSY;

        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
                        continue;
                rc |= obd_register_lock_cancel_cb(lov->lov_tgts[i]->ltd_exp,
                                                  func);
        }

        lov->lov_lock_cancel_cb = func;

        return rc;
}

static int lov_unregister_lock_cancel_cb(struct obd_export *exp,
                                         obd_lock_cancel_cb func)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int i, rc = 0;

        if (lov->lov_lock_cancel_cb && lov->lov_lock_cancel_cb != func)
                return -EINVAL;

        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
                        continue;
                rc |= obd_unregister_lock_cancel_cb(lov->lov_tgts[i]->ltd_exp,
                                                    func);
        }
        lov->lov_lock_cancel_cb = NULL;
        return rc;
}

#define MAX_STRING_SIZE 128
static int lov_connect_obd(struct obd_device *obd, __u32 index, int activate, 
                           struct obd_connect_data *data)
{
        struct lov_obd *lov = &obd->u.lov;
        struct obd_uuid tgt_uuid;
        struct obd_device *tgt_obd;
        struct obd_uuid lov_osc_uuid = { "LOV_OSC_UUID" };
        struct lustre_handle conn = {0, };
        struct obd_import *imp;

#ifdef __KERNEL__
        cfs_proc_dir_entry_t *lov_proc_dir;
#endif
        int rc;
        ENTRY;

        if (!lov->lov_tgts[index])
                RETURN(-EINVAL);

        tgt_uuid = lov->lov_tgts[index]->ltd_uuid;
        tgt_obd = class_find_client_obd(&tgt_uuid, LUSTRE_OSC_NAME,
                                        &obd->obd_uuid);

        if (!tgt_obd) {
                CERROR("Target %s not attached\n", obd_uuid2str(&tgt_uuid));
                RETURN(-EINVAL);
        }
        
        if (!tgt_obd->obd_set_up) {
                CERROR("Target %s not set up\n", obd_uuid2str(&tgt_uuid));
                RETURN(-EINVAL);
        }

        if (data && (data->ocd_connect_flags & OBD_CONNECT_INDEX))
                data->ocd_index = index;

        /*
         * Divine LOV knows that OBDs under it are OSCs.
         */
        imp = tgt_obd->u.cli.cl_import;

        if (activate) {
                tgt_obd->obd_no_recov = 0;
                /* FIXME this is probably supposed to be 
                   ptlrpc_set_import_active.  Horrible naming. */
                ptlrpc_activate_import(imp);
        }

        if (imp->imp_invalid) {
                CERROR("not connecting OSC %s; administratively "
                       "disabled\n", obd_uuid2str(&tgt_uuid));
                rc = obd_register_observer(tgt_obd, obd);
                if (rc) {
                        CERROR("Target %s register_observer error %d; "
                               "will not be able to reactivate\n",
                               obd_uuid2str(&tgt_uuid), rc);
                }
                RETURN(0);
        }

        rc = obd_connect(&conn, tgt_obd, &lov_osc_uuid, data, NULL);
        if (rc) {
                CERROR("Target %s connect error %d\n",
                       obd_uuid2str(&tgt_uuid), rc);
                RETURN(rc);
        }
        lov->lov_tgts[index]->ltd_exp = class_conn2export(&conn);
        if (!lov->lov_tgts[index]->ltd_exp) {
                CERROR("Target %s: null export!\n", obd_uuid2str(&tgt_uuid));
                RETURN(-ENODEV);
        }

        rc = obd_register_page_removal_cb(lov->lov_tgts[index]->ltd_exp,
                                          lov->lov_page_removal_cb,
                                          lov->lov_page_pin_cb);
        if (rc) {
                obd_disconnect(lov->lov_tgts[index]->ltd_exp);
                lov->lov_tgts[index]->ltd_exp = NULL;
                RETURN(rc);
        }

        rc = obd_register_lock_cancel_cb(lov->lov_tgts[index]->ltd_exp,
                                         lov->lov_lock_cancel_cb);
        if (rc) {
                obd_unregister_page_removal_cb(lov->lov_tgts[index]->ltd_exp,
                                               lov->lov_page_removal_cb);
                obd_disconnect(lov->lov_tgts[index]->ltd_exp);
                lov->lov_tgts[index]->ltd_exp = NULL;
                RETURN(rc);
        }

        rc = obd_register_observer(tgt_obd, obd);
        if (rc) {
                CERROR("Target %s register_observer error %d\n",
                       obd_uuid2str(&tgt_uuid), rc);
                obd_unregister_lock_cancel_cb(lov->lov_tgts[index]->ltd_exp,
                                              lov->lov_lock_cancel_cb);
                obd_unregister_page_removal_cb(lov->lov_tgts[index]->ltd_exp,
                                               lov->lov_page_removal_cb);
                obd_disconnect(lov->lov_tgts[index]->ltd_exp);
                lov->lov_tgts[index]->ltd_exp = NULL;
                RETURN(rc);
        }

        lov->lov_tgts[index]->ltd_reap = 0;
        if (activate) {
                lov->lov_tgts[index]->ltd_active = 1;
                lov->desc.ld_active_tgt_count++;
                lov->lov_tgts[index]->ltd_exp->exp_obd->obd_inactive = 0;
        }
        CDEBUG(D_CONFIG, "Connected tgt idx %d %s (%s) %sactive\n", index,
               obd_uuid2str(&tgt_uuid), tgt_obd->obd_name, activate ? "":"in");

#ifdef __KERNEL__
        lov_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
        if (lov_proc_dir) {
                struct obd_device *osc_obd = class_conn2obd(&conn);
                cfs_proc_dir_entry_t *osc_symlink;
                char name[MAX_STRING_SIZE];

                LASSERT(osc_obd != NULL);
                LASSERT(osc_obd->obd_magic == OBD_DEVICE_MAGIC);
                LASSERT(osc_obd->obd_type->typ_name != NULL);
                snprintf(name, MAX_STRING_SIZE, "../../../%s/%s",
                         osc_obd->obd_type->typ_name,
                         osc_obd->obd_name);
                osc_symlink = proc_symlink(osc_obd->obd_name, lov_proc_dir,
                                           name);
                if (osc_symlink == NULL) {
                        CERROR("could not register LOV target "
                               "/proc/fs/lustre/%s/%s/target_obds/%s.",
                               obd->obd_type->typ_name, obd->obd_name,
                               osc_obd->obd_name);
                        lprocfs_remove(&lov_proc_dir);
                }
        }
#endif

        rc = qos_add_tgt(obd, index);
        if (rc) 
                CERROR("qos_add_tgt failed %d\n", rc);

        RETURN(0);
}

static int lov_connect(struct lustre_handle *conn, struct obd_device *obd,
                       struct obd_uuid *cluuid, struct obd_connect_data *data,
                       void *localdata)
{
        struct lov_obd *lov = &obd->u.lov;
        struct lov_tgt_desc *tgt;
        int i, rc;
        ENTRY;

        CDEBUG(D_CONFIG, "connect #%d\n", lov->lov_connects);

        rc = class_connect(conn, obd, cluuid);
        if (rc)
                RETURN(rc);

        /* Why should there ever be more than 1 connect? */
        lov->lov_connects++;
        LASSERT(lov->lov_connects == 1);
        
        memset(&lov->lov_ocd, 0, sizeof(lov->lov_ocd));
        if (data)
                lov->lov_ocd = *data;

        lov_getref(obd);
        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                tgt = lov->lov_tgts[i];
                if (!tgt || obd_uuid_empty(&tgt->ltd_uuid))
                        continue;
                /* Flags will be lowest common denominator */
                rc = lov_connect_obd(obd, i, lov->lov_tgts[i]->ltd_activate,
                                     &lov->lov_ocd);
                if (rc) {
                        CERROR("%s: lov connect tgt %d failed: %d\n", 
                               obd->obd_name, i, rc);
                        continue;
                }
        }
        lov_putref(obd);
        
        RETURN(0);
}

static int lov_disconnect_obd(struct obd_device *obd, __u32 index)
{
        cfs_proc_dir_entry_t *lov_proc_dir;
        struct lov_obd *lov = &obd->u.lov;
        struct obd_device *osc_obd =
                class_exp2obd(lov->lov_tgts[index]->ltd_exp);
        int rc;