/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
* vim:expandtab:shiftwidth=8:tabstop=8:
*
- * linux/mdt/mdt_recovery.c
- * Lustre Metadata Target (mdt) recovery-related methods
+ * GPL HEADER START
*
- * Copyright (C) 2002-2006 Cluster File Systems, Inc.
- * Author: Huang Hua <huanghua@clusterfs.com>
- * Author: Pershin Mike <tappro@clusterfs.com>
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
- * This file is part of the Lustre file system, http://www.lustre.org
- * Lustre is a trademark of Cluster File Systems, Inc.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.
*
- * 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.
+ * This program 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 version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).
*
- * 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.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
*
- * 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.
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
+ *
+ * GPL HEADER END
+ */
+/*
+ * Copyright 2008 Sun Microsystems, Inc. All rights reserved
+ * Use is subject to license terms.
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
+ *
+ * lustre/mdt/mdt_recovery.c
+ *
+ * Lustre Metadata Target (mdt) recovery-related methods
+ *
+ * Author: Huang Hua <huanghua@clusterfs.com>
+ * Author: Pershin Mike <tappro@clusterfs.com>
*/
+
#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif
#include "mdt_internal.h"
static int mdt_server_data_update(const struct lu_env *env,
- struct mdt_device *mdt);
+ struct mdt_device *mdt,
+ int need_sync);
struct lu_buf *mdt_buf(const struct lu_env *env, void *area, ssize_t len)
{
LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
LASSERT(th != NULL);
- rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA);
+ rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA, 1);
if (rc == buf->lb_len)
rc = 0;
else if (rc >= 0)
rc = -EFAULT;
return rc;
}
-/* only one record write */
-enum {
- MDT_TXN_LAST_RCVD_WRITE_CREDITS = 3
-};
+static inline int mdt_trans_credit_get(const struct lu_env *env,
+ struct mdt_device *mdt,
+ enum mdt_txn_op op)
+{
+ struct dt_device *dev = mdt->mdt_bottom;
+ int cr;
+ switch (op) {
+ case MDT_TXN_CAPA_KEYS_WRITE_OP:
+ case MDT_TXN_LAST_RCVD_WRITE_OP:
+ cr = dev->dd_ops->dt_credit_get(env,
+ dev,
+ DTO_WRITE_BLOCK);
+ break;
+ default:
+ LBUG();
+ }
+ return cr;
+}
+
+void mdt_trans_credit_init(const struct lu_env *env,
+ struct mdt_device *mdt,
+ enum mdt_txn_op op)
+{
+ struct mdt_thread_info *mti;
+ struct txn_param *p;
+ int cr;
+
+ mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
+ p = &mti->mti_txn_param;
+
+ cr = mdt_trans_credit_get(env, mdt, op);
+ txn_param_init(p, cr);
+}
struct thandle* mdt_trans_start(const struct lu_env *env,
- struct mdt_device *mdt, int credits)
+ struct mdt_device *mdt)
{
struct mdt_thread_info *mti;
struct txn_param *p;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
p = &mti->mti_txn_param;
- txn_param_init(p, credits);
/* export can require sync operations */
if (mti->mti_exp != NULL)
return rc;
}
+static void mdt_client_cb(const struct mdt_device *mdt, __u64 transno,
+ void *data, int err)
+{
+ struct obd_device *obd = mdt2obd_dev(mdt);
+ target_client_add_cb(obd, transno, data, err);
+}
+
static inline int mdt_last_rcvd_header_write(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ int need_sync)
{
struct mdt_thread_info *mti;
struct thandle *th;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ if (mti->mti_exp) {
+ spin_lock(&mti->mti_exp->exp_lock);
+ mti->mti_exp->exp_need_sync = need_sync;
+ spin_unlock(&mti->mti_exp->exp_lock);
+ }
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
- mti->mti_off = 0;
+ mti->mti_off = 0;
lsd_cpu_to_le(&mdt->mdt_lsd, &mti->mti_lsd);
+ if (need_sync && mti->mti_exp)
+ mdt_trans_add_cb(th, mdt_client_cb, mti->mti_exp);
+
rc = mdt_record_write(env, mdt->mdt_last_rcvd,
- mdt_buf_const(env, &mti->mti_lsd, sizeof(mti->mti_lsd)),
+ mdt_buf_const(env, &mti->mti_lsd,
+ sizeof(mti->mti_lsd)),
&mti->mti_off, th);
mdt_trans_stop(env, mdt, th);
CDEBUG(D_INFO, "write last_rcvd header rc = %d:\n"
"uuid = %s\nlast_transno = "LPU64"\n",
rc, mdt->mdt_lsd.lsd_uuid, mdt->mdt_lsd.lsd_last_transno);
-
+
RETURN(rc);
}
"uuid = %s\n"
"last_transno = "LPU64"\n"
"last_xid = "LPU64"\n"
- "last_result = %d\n"
- "last_data = %d\n"
+ "last_result = %u\n"
+ "last_data = %u\n"
"last_close_transno = "LPU64"\n"
"last_close_xid = "LPU64"\n"
- "last_close_result = %d\n",
- (int)*off - sizeof(*tmp),
+ "last_close_result = %u\n",
+ (int)(*off - sizeof(*tmp)),
rc,
lcd->lcd_uuid,
lcd->lcd_last_transno,
"uuid = %s\n"
"last_transno = "LPU64"\n"
"last_xid = "LPU64"\n"
- "last_result = %d\n"
- "last_data = %d\n"
+ "last_result = %u\n"
+ "last_data = %u\n"
"last_close_transno = "LPU64"\n"
"last_close_xid = "LPU64"\n"
- "last_close_result = %d\n",
- (int)*off - sizeof(*tmp),
+ "last_close_result = %u\n",
+ (int)(*off - sizeof(*tmp)),
rc,
lcd->lcd_uuid,
lcd->lcd_last_transno,
return rc;
}
-
static int mdt_clients_data_init(const struct lu_env *env,
struct mdt_device *mdt,
unsigned long last_size)
{
struct lr_server_data *lsd = &mdt->mdt_lsd;
struct lsd_client_data *lcd = NULL;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
loff_t off;
int cl_idx;
int rc = 0;
if (IS_ERR(exp)) {
if (PTR_ERR(exp) == -EALREADY) {
/* export already exists, zero out this one */
- lcd->lcd_uuid[0] = '\0';
+ lcd->lcd_uuid[0] = '\0';
} else
GOTO(err_client, rc = PTR_ERR(exp));
} else {
- struct mdt_thread_info *mti;
+ struct mdt_thread_info *mti;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
LASSERT(mti != NULL);
mti->mti_exp = exp;
exp->exp_mdt_data.med_lcd = lcd;
rc = mdt_client_add(env, mdt, cl_idx);
/* can't fail existing */
- LASSERTF(rc == 0, "rc = %d\n", rc);
+ LASSERTF(rc == 0, "rc = %d\n", rc);
lcd = NULL;
spin_lock(&exp->exp_lock);
exp->exp_connecting = 0;
}
static int mdt_server_data_init(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ struct lustre_sb_info *lsi)
{
struct lr_server_data *lsd = &mdt->mdt_lsd;
struct lsd_client_data *lcd = NULL;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct mdt_thread_info *mti;
struct dt_object *obj;
struct lu_attr *la;
+ struct lustre_disk_data *ldd;
unsigned long last_rcvd_size;
__u64 mount_count;
int rc;
}
mount_count = lsd->lsd_mount_count;
+ ldd = lsi->lsi_ldd;
+
+ if (ldd->ldd_flags & LDD_F_IAM_DIR)
+ lsd->lsd_feature_incompat |= OBD_INCOMPAT_IAM_DIR;
+
lsd->lsd_feature_compat = OBD_COMPAT_MDT;
+ lsd->lsd_feature_incompat |= OBD_INCOMPAT_FID;
spin_lock(&mdt->mdt_transno_lock);
mdt->mdt_last_transno = lsd->lsd_last_transno;
lsd->lsd_mount_count = mdt->mdt_mount_count;
/* save it, so mount count and last_transno is current */
- rc = mdt_server_data_update(env, mdt);
+ rc = mdt_server_data_update(env, mdt, (mti->mti_exp &&
+ mti->mti_exp->exp_need_sync));
if (rc)
GOTO(err_client, rc);
}
static int mdt_server_data_update(const struct lu_env *env,
- struct mdt_device *mdt)
+ struct mdt_device *mdt,
+ int need_sync)
{
int rc = 0;
ENTRY;
* mdt->mdt_last_rcvd may be NULL that time.
*/
if (mdt->mdt_last_rcvd != NULL)
- rc = mdt_last_rcvd_header_write(env, mdt);
+ rc = mdt_last_rcvd_header_write(env, mdt, need_sync);
RETURN(rc);
}
-void mdt_cb_new_client(const struct mdt_device *mdt, __u64 transno,
- void *data, int err)
-{
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
-
- target_client_add_cb(obd, transno, data, err);
-}
-
int mdt_client_new(const struct lu_env *env, struct mdt_device *mdt)
{
unsigned long *bitmap = mdt->mdt_client_bitmap;
struct mdt_export_data *med;
struct lsd_client_data *lcd;
struct lr_server_data *lsd = &mdt->mdt_lsd;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct thandle *th;
loff_t off;
int rc;
init_mutex(&med->med_lcd_lock);
LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
- /* write new client data */
+
+ /* Write new client data. */
off = med->med_lr_off;
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
RETURN(PTR_ERR(th));
- /* until this operations will be committed the sync is needed for this
- * export */
- mdt_trans_add_cb(th, mdt_cb_new_client, mti->mti_exp);
+ /*
+ * Until this operations will be committed the sync is needed
+ * for this export. This should be done _after_ starting the
+ * transaction so that many connecting clients will not bring
+ * server down with lots of sync writes.
+ */
+ mdt_trans_add_cb(th, mdt_client_cb, mti->mti_exp);
spin_lock(&mti->mti_exp->exp_lock);
mti->mti_exp->exp_need_sync = 1;
spin_unlock(&mti->mti_exp->exp_lock);
rc = mdt_last_rcvd_write(env, mdt, lcd, &off, th);
- CDEBUG(D_INFO, "wrote client lcd at idx %u off %llu (len %u)\n",
+ CDEBUG(D_INFO, "wrote client lcd at idx %u off %llu (len "LPSZ")\n",
cl_idx, med->med_lr_off, sizeof(*lcd));
mdt_trans_stop(env, mdt, th);
struct mdt_thread_info *mti;
struct mdt_export_data *med;
unsigned long *bitmap = mdt->mdt_client_bitmap;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct lr_server_data *lsd = &mdt->mdt_lsd;
int rc = 0;
ENTRY;
struct mdt_thread_info *mti;
struct mdt_export_data *med;
struct lsd_client_data *lcd;
- struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
- struct thandle *th;
- loff_t off;
- int rc = 0;
+ struct obd_device *obd = mdt2obd_dev(mdt);
+ struct obd_export *exp;
+ struct thandle *th;
+ int need_sync;
+ loff_t off;
+ int rc = 0;
ENTRY;
mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
LASSERT(mti != NULL);
- med = &mti->mti_exp->exp_mdt_data;
+ exp = mti->mti_exp;
+ med = &exp->exp_mdt_data;
lcd = med->med_lcd;
if (!lcd)
RETURN(0);
/* XXX: If lcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
- if (!strcmp(med->med_lcd->lcd_uuid, obd->obd_uuid.uuid))
+ if (!strcmp(lcd->lcd_uuid, obd->obd_uuid.uuid))
GOTO(free, 0);
CDEBUG(D_INFO, "freeing client at idx %u, offset %lld\n",
LBUG();
}
+ /* Don't force sync on disconnect if aborting recovery,
+ * or it does num_clients * num_osts. b=17194 */
+ need_sync = (!exp->exp_libclient || exp->exp_need_sync) &&
+ !(exp->exp_flags & OBD_OPT_ABORT_RECOV);
+
/*
* This may be called from difficult reply handler path and
* mdt->mdt_last_rcvd may be NULL that time.
*/
if (mdt->mdt_last_rcvd != NULL) {
- th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
+ mdt_trans_credit_init(env, mdt, MDT_TXN_LAST_RCVD_WRITE_OP);
+
+ spin_lock(&exp->exp_lock);
+ exp->exp_need_sync = need_sync;
+ spin_unlock(&exp->exp_lock);
+
+ th = mdt_trans_start(env, mdt);
if (IS_ERR(th))
GOTO(free, rc = PTR_ERR(th));
+ if (need_sync) {
+ /*
+ * Until this operations will be committed the sync
+ * is needed for this export.
+ */
+ mdt_trans_add_cb(th, mdt_client_cb, exp);
+ }
+
mutex_down(&med->med_lcd_lock);
memset(lcd, 0, sizeof *lcd);
}
CDEBUG(rc == 0 ? D_INFO : D_ERROR, "Zeroing out client idx %u in "
- "%s rc %d\n", med->med_lr_idx, LAST_RCVD, rc);
+ "%s %ssync rc %d\n", med->med_lr_idx, LAST_RCVD,
+ need_sync ? "" : "a", rc);
spin_lock(&mdt->mdt_client_bitmap_lock);
clear_bit(med->med_lr_idx, mdt->mdt_client_bitmap);
spin_unlock(&mdt->mdt_client_bitmap_lock);
- /*
- * Make sure the server's last_transno is up to date. Do this after the
- * client is freed so we know all the client's transactions have been
- * committed.
+ /*
+ * Make sure the server's last_transno is up to date. Do this
+ * after the client is freed so we know all the client's
+ * transactions have been committed.
*/
- mdt_server_data_update(env, mdt);
+ mdt_server_data_update(env, mdt, need_sync);
+
EXIT;
free:
OBD_FREE_PTR(lcd);
/*
* When we store zero transno in lcd we can lost last transno value
* because lcd contains 0, but lsd is not yet written
- * The server data should be updated also if the latest
+ * The server data should be updated also if the latest
* transno is rewritten by zero. See the bug 11125 for details.
*/
- if (mti->mti_transno == 0 &&
+ if (mti->mti_transno == 0 &&
*transno_p == mdt->mdt_last_transno)
- mdt_server_data_update(mti->mti_env, mdt);
+ mdt_server_data_update(mti->mti_env, mdt,
+ (mti->mti_exp &&
+ mti->mti_exp->exp_need_sync));
*transno_p = mti->mti_transno;
static int mdt_txn_start_cb(const struct lu_env *env,
struct txn_param *param, void *cookie)
{
- param->tp_credits += MDT_TXN_LAST_RCVD_WRITE_CREDITS;
+ struct mdt_device *mdt = cookie;
+
+ param->tp_credits += mdt_trans_credit_get(env, mdt,
+ MDT_TXN_LAST_RCVD_WRITE_OP);
return 0;
}
if (mti->mti_has_trans) {
/* XXX: currently there are allowed cases, but the wrong cases
* are also possible, so better check is needed here */
- CDEBUG(D_INFO, "More than one transaction "LPU64"\n", mti->mti_transno);
+ CDEBUG(D_INFO, "More than one transaction "LPU64"\n",
+ mti->mti_transno);
return 0;
}
return mdt_last_rcvd_update(mti, txn);
}
-/* commit callback, need to update last_commited value */
+/* commit callback, need to update last_committed value */
static int mdt_txn_commit_cb(const struct lu_env *env,
struct thandle *txn, void *cookie)
{
struct mdt_device *mdt = cookie;
- struct obd_device *obd = md2lu_dev(&mdt->mdt_md_dev)->ld_obd;
+ struct obd_device *obd = mdt2obd_dev(mdt);
struct mdt_txn_info *txi;
int i;
}
int mdt_fs_setup(const struct lu_env *env, struct mdt_device *mdt,
- struct obd_device *obd)
+ struct obd_device *obd,
+ struct lustre_sb_info *lsi)
{
struct lu_fid fid;
struct dt_object *o;
dt_txn_callback_add(mdt->mdt_bottom, &mdt->mdt_txn_cb);
- o = dt_store_open(env, mdt->mdt_bottom, LAST_RCVD, &fid);
+ o = dt_store_open(env, mdt->mdt_bottom, "", LAST_RCVD, &fid);
if (!IS_ERR(o)) {
mdt->mdt_last_rcvd = o;
- rc = mdt_server_data_init(env, mdt);
+ rc = mdt_server_data_init(env, mdt, lsi);
if (rc)
GOTO(put_last_rcvd, rc);
} else {
RETURN(rc);
}
- o = dt_store_open(env, mdt->mdt_bottom, CAPA_KEYS, &fid);
+ o = dt_store_open(env, mdt->mdt_bottom, "", CAPA_KEYS, &fid);
if (!IS_ERR(o)) {
mdt->mdt_ck_obj = o;
rc = mdt_capa_keys_init(env, mdt);
return rc;
}
-
void mdt_fs_cleanup(const struct lu_env *env, struct mdt_device *mdt)
{
ENTRY;
libcfs_nid2str(exp->exp_connection->c_peer.nid));
for (i = 0; i < oldrep->rs_nlocks; i++)
- ptlrpc_save_lock(req,
- &oldrep->rs_locks[i],
- oldrep->rs_modes[i]);
+ ptlrpc_save_lock(req, &oldrep->rs_locks[i],
+ oldrep->rs_modes[i], 0);
oldrep->rs_nlocks = 0;
DEBUG_REQ(D_HA, req, "stole locks for");
+ spin_lock(&oldrep->rs_lock);
ptlrpc_schedule_difficult_reply (oldrep);
+ spin_unlock(&oldrep->rs_lock);
spin_unlock (&svc->srv_lock);
break;
}
body = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
+ mti->mti_attr.ma_need = MA_INODE;
+ mti->mti_attr.ma_valid = 0;
rc = mo_attr_get(mti->mti_env, mdt_object_child(child), &mti->mti_attr);
if (rc == -EREMOTE) {
/* object was created on remote server */
req->rq_status = rc;
body->valid |= OBD_MD_MDS;
}
- mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr, mdt_object_fid(child));
+ mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr,
+ mdt_object_fid(child));
mdt_object_put(mti->mti_env, child);
}
EXIT;
return;
}
+ mti->mti_attr.ma_need = MA_INODE;
+ mti->mti_attr.ma_valid = 0;
mo_attr_get(mti->mti_env, mdt_object_child(obj), &mti->mti_attr);
mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr,
mdt_object_fid(obj));
mdt_object_put(mti->mti_env, obj);
}
-static void mdt_reconstruct_setxattr(struct mdt_thread_info *mti,
- struct mdt_lock_handle *lhc)
-{
- /* reply nothing */
- req_capsule_shrink(mti->mti_pill, &RMF_EADATA, 0, RCL_SERVER);
-}
-
typedef void (*mdt_reconstructor)(struct mdt_thread_info *mti,
struct mdt_lock_handle *lhc);
[REINT_UNLINK] = mdt_reconstruct_generic,
[REINT_RENAME] = mdt_reconstruct_generic,
[REINT_OPEN] = mdt_reconstruct_open,
- [REINT_SETXATTR] = mdt_reconstruct_setxattr
+ [REINT_SETXATTR] = mdt_reconstruct_generic
};
void mdt_reconstruct(struct mdt_thread_info *mti,
git://git.whamcloud.com / fs / lustre-release.git / blobdiff