X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fobdfilter%2Ffilter.c;h=dfb7d2fa4b04507cc341c00eb874c7b1d0a54deb;hp=4169e199d72c143987647355780a906a51bfe2e0;hb=6ad1581bf6003631d87e8365c01c03b06be136de;hpb=2ececcfb12ea2d5ce7fd26c0c702f0c5be555d1f diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c index 4169e19..dfb7d2f 100644 --- a/lustre/obdfilter/filter.c +++ b/lustre/obdfilter/filter.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -26,8 +24,10 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2012, Whamcloud, Inc. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -120,9 +120,9 @@ int filter_version_get_check(struct obd_export *exp, int filter_finish_transno(struct obd_export *exp, struct inode *inode, struct obd_trans_info *oti, int rc, int force_sync) { - struct filter_obd *filter = &exp->exp_obd->u.filter; - struct filter_export_data *fed = &exp->exp_filter_data; - struct lr_server_data *fsd = class_server_data(exp->exp_obd); + struct obd_device_target *obt = &exp->exp_obd->u.obt; + struct tg_export_data *ted = &exp->exp_target_data; + struct lr_server_data *lsd = class_server_data(exp->exp_obd); struct lsd_client_data *lcd; __u64 last_rcvd; loff_t off; @@ -135,12 +135,12 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode, if (!exp->exp_obd->obd_replayable || oti == NULL) RETURN(rc); - cfs_mutex_down(&fed->fed_lcd_lock); - lcd = fed->fed_lcd; + cfs_mutex_lock(&ted->ted_lcd_lock); + lcd = ted->ted_lcd; /* if the export has already been disconnected, we have no last_rcvd slot, * update server data with latest transno then */ if (lcd == NULL) { - cfs_mutex_up(&fed->fed_lcd_lock); + cfs_mutex_unlock(&ted->ted_lcd_lock); CWARN("commit transaction for disconnected client %s: rc %d\n", exp->exp_client_uuid.uuid, rc); err = filter_update_server_data(exp->exp_obd); @@ -148,30 +148,41 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode, } /* we don't allocate new transnos for replayed requests */ - cfs_spin_lock(&filter->fo_translock); + cfs_spin_lock(&obt->obt_lut->lut_translock); if (oti->oti_transno == 0) { - last_rcvd = le64_to_cpu(fsd->lsd_last_transno) + 1; - fsd->lsd_last_transno = cpu_to_le64(last_rcvd); + last_rcvd = le64_to_cpu(lsd->lsd_last_transno) + 1; + lsd->lsd_last_transno = cpu_to_le64(last_rcvd); + LASSERT(last_rcvd >= le64_to_cpu(lcd->lcd_last_transno)); } else { last_rcvd = oti->oti_transno; - if (last_rcvd > le64_to_cpu(fsd->lsd_last_transno)) - fsd->lsd_last_transno = cpu_to_le64(last_rcvd); + if (last_rcvd > le64_to_cpu(lsd->lsd_last_transno)) + lsd->lsd_last_transno = cpu_to_le64(last_rcvd); + if (unlikely(last_rcvd < le64_to_cpu(lcd->lcd_last_transno))) { + CERROR("Trying to overwrite bigger transno, on-disk: " + LPU64", new: "LPU64"\n", + le64_to_cpu(lcd->lcd_last_transno), last_rcvd); + cfs_spin_lock(&exp->exp_lock); + exp->exp_vbr_failed = 1; + cfs_spin_unlock(&exp->exp_lock); + cfs_spin_unlock(&obt->obt_lut->lut_translock); + cfs_mutex_unlock(&ted->ted_lcd_lock); + RETURN(-EOVERFLOW); + } } oti->oti_transno = last_rcvd; - LASSERT(last_rcvd >= le64_to_cpu(lcd->lcd_last_transno)); lcd->lcd_last_transno = cpu_to_le64(last_rcvd); lcd->lcd_pre_versions[0] = cpu_to_le64(oti->oti_pre_version); lcd->lcd_last_xid = cpu_to_le64(oti->oti_xid); - cfs_spin_unlock(&filter->fo_translock); + cfs_spin_unlock(&obt->obt_lut->lut_translock); if (inode) fsfilt_set_version(exp->exp_obd, inode, last_rcvd); - off = fed->fed_lr_off; + off = ted->ted_lr_off; if (off <= 0) { CERROR("%s: client idx %d is %lld\n", exp->exp_obd->obd_name, - fed->fed_lr_idx, fed->fed_lr_off); + ted->ted_lr_idx, ted->ted_lr_off); err = -EINVAL; } else { class_export_cb_get(exp); /* released when the cb is called */ @@ -182,7 +193,7 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode, filter_commit_cb, exp); - err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp, + err = fsfilt_write_record(exp->exp_obd, obt->obt_rcvd_filp, lcd, sizeof(*lcd), &off, force_sync | exp->exp_need_sync); if (force_sync) @@ -195,8 +206,8 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode, } CDEBUG(log_pri, "wrote trans "LPU64" for client %s at #%d: err = %d\n", - last_rcvd, lcd->lcd_uuid, fed->fed_lr_idx, err); - cfs_mutex_up(&fed->fed_lcd_lock); + last_rcvd, lcd->lcd_uuid, ted->ted_lr_idx, err); + cfs_mutex_unlock(&ted->ted_lcd_lock); RETURN(rc); } @@ -282,23 +293,13 @@ static int filter_export_stats_init(struct obd_device *obd, tmp->nid_stats); if (rc) GOTO(clean, rc); - /* Always add in ldlm_stats */ - tmp->nid_ldlm_stats = - lprocfs_alloc_stats(LDLM_LAST_OPC - LDLM_FIRST_OPC, - LPROCFS_STATS_FLAG_NOPERCPU); - if (tmp->nid_ldlm_stats == NULL) - GOTO(clean, rc = -ENOMEM); - - lprocfs_init_ldlm_stats(tmp->nid_ldlm_stats); - rc = lprocfs_register_stats(tmp->nid_proc, "ldlm_stats", - tmp->nid_ldlm_stats); + rc = lprocfs_nid_ldlm_stats_init(tmp); if (rc) GOTO(clean, rc); } RETURN(0); clean: - lprocfs_exp_cleanup(exp); return rc; } @@ -309,10 +310,10 @@ static int filter_export_stats_init(struct obd_device *obd, static int filter_client_add(struct obd_device *obd, struct obd_export *exp, int cl_idx) { - struct filter_obd *filter = &obd->u.filter; - struct filter_export_data *fed = &exp->exp_filter_data; - struct lr_server_data *fsd = class_server_data(obd); - unsigned long *bitmap = filter->fo_last_rcvd_slots; + struct obd_device_target *obt = &obd->u.obt; + struct tg_export_data *ted = &exp->exp_target_data; + struct lr_server_data *lsd = class_server_data(obd); + unsigned long *bitmap = obt->obt_lut->lut_client_bitmap; int new_client = (cl_idx == -1); ENTRY; @@ -321,7 +322,7 @@ static int filter_client_add(struct obd_device *obd, struct obd_export *exp, LASSERTF(cl_idx > -2, "%d\n", cl_idx); /* Self-export */ - if (strcmp(fed->fed_lcd->lcd_uuid, obd->obd_uuid.uuid) == 0) + if (strcmp(ted->ted_lcd->lcd_uuid, obd->obd_uuid.uuid) == 0) RETURN(0); /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so @@ -348,23 +349,23 @@ static int filter_client_add(struct obd_device *obd, struct obd_export *exp, } } - fed->fed_lr_idx = cl_idx; - fed->fed_lr_off = le32_to_cpu(fsd->lsd_client_start) + - cl_idx * le16_to_cpu(fsd->lsd_client_size); - cfs_init_mutex(&fed->fed_lcd_lock); - LASSERTF(fed->fed_lr_off > 0, "fed_lr_off = %llu\n", fed->fed_lr_off); + ted->ted_lr_idx = cl_idx; + ted->ted_lr_off = le32_to_cpu(lsd->lsd_client_start) + + cl_idx * le16_to_cpu(lsd->lsd_client_size); + cfs_mutex_init(&ted->ted_lcd_lock); + LASSERTF(ted->ted_lr_off > 0, "ted_lr_off = %llu\n", ted->ted_lr_off); CDEBUG(D_INFO, "client at index %d (%llu) with UUID '%s' added\n", - fed->fed_lr_idx, fed->fed_lr_off, fed->fed_lcd->lcd_uuid); + ted->ted_lr_idx, ted->ted_lr_off, ted->ted_lcd->lcd_uuid); if (new_client) { struct lvfs_run_ctxt saved; - loff_t off = fed->fed_lr_off; + loff_t off = ted->ted_lr_off; int rc; void *handle; CDEBUG(D_INFO, "writing client lcd at idx %u (%llu) (len %u)\n", - fed->fed_lr_idx,off,(unsigned int)sizeof(*fed->fed_lcd)); + ted->ted_lr_idx,off,(unsigned int)sizeof(*ted->ted_lcd)); if (OBD_FAIL_CHECK(OBD_FAIL_TGT_CLIENT_ADD)) RETURN(-ENOSPC); @@ -372,13 +373,13 @@ static int filter_client_add(struct obd_device *obd, struct obd_export *exp, push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); /* Transaction needed to fix bug 1403 */ handle = fsfilt_start(obd, - filter->fo_rcvd_filp->f_dentry->d_inode, + obt->obt_rcvd_filp->f_dentry->d_inode, FSFILT_OP_SETATTR, NULL); if (IS_ERR(handle)) { rc = PTR_ERR(handle); CERROR("unable to start transaction: rc %d\n", rc); } else { - fed->fed_lcd->lcd_last_epoch = fsd->lsd_start_epoch; + ted->ted_lcd->lcd_last_epoch = lsd->lsd_start_epoch; exp->exp_last_request_time = cfs_time_current_sec(); rc = fsfilt_add_journal_cb(obd, 0, handle, target_client_add_cb, @@ -388,19 +389,19 @@ static int filter_client_add(struct obd_device *obd, struct obd_export *exp, exp->exp_need_sync = 1; cfs_spin_unlock(&exp->exp_lock); } - rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, - fed->fed_lcd, - sizeof(*fed->fed_lcd), + rc = fsfilt_write_record(obd, obt->obt_rcvd_filp, + ted->ted_lcd, + sizeof(*ted->ted_lcd), &off, rc /* sync if no cb */); fsfilt_commit(obd, - filter->fo_rcvd_filp->f_dentry->d_inode, + obt->obt_rcvd_filp->f_dentry->d_inode, handle, 0); } pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); if (rc) { CERROR("error writing %s client idx %u: rc %d\n", - LAST_RCVD, fed->fed_lr_idx, rc); + LAST_RCVD, ted->ted_lr_idx, rc); RETURN(rc); } } @@ -409,62 +410,60 @@ static int filter_client_add(struct obd_device *obd, struct obd_export *exp, static int filter_client_del(struct obd_export *exp) { - struct filter_export_data *fed = &exp->exp_filter_data; - struct filter_obd *filter = &exp->exp_obd->u.filter; - struct obd_device *obd = exp->exp_obd; + struct tg_export_data *ted = &exp->exp_target_data; + struct obd_device_target *obt = &exp->exp_obd->u.obt; struct lvfs_run_ctxt saved; - struct lsd_client_data *lcd = fed->fed_lcd; int rc; loff_t off; ENTRY; - if (lcd == NULL) + if (ted->ted_lcd == NULL) RETURN(0); /* XXX if lcd_uuid were a real obd_uuid, I could use obd_uuid_equals */ - if (strcmp(lcd->lcd_uuid, obd->obd_uuid.uuid ) == 0) + if (strcmp(ted->ted_lcd->lcd_uuid, exp->exp_obd->obd_uuid.uuid ) == 0) GOTO(free, 0); - LASSERT(filter->fo_last_rcvd_slots != NULL); + LASSERT(obt->obt_lut->lut_client_bitmap != NULL); - off = fed->fed_lr_off; + off = ted->ted_lr_off; CDEBUG(D_INFO, "freeing client at idx %u, offset %lld with UUID '%s'\n", - fed->fed_lr_idx, fed->fed_lr_off, lcd->lcd_uuid); + ted->ted_lr_idx, ted->ted_lr_off, ted->ted_lcd->lcd_uuid); - /* Don't clear fed_lr_idx here as it is likely also unset. At worst + /* Don't clear ted_lr_idx here as it is likely also unset. At worst * we leak a client slot that will be cleaned on the next recovery. */ if (off <= 0) { CERROR("%s: client idx %d has med_off %lld\n", - obd->obd_name, fed->fed_lr_idx, off); + exp->exp_obd->obd_name, ted->ted_lr_idx, off); GOTO(free, rc = -EINVAL); } /* Clear the bit _after_ zeroing out the client so we don't race with filter_client_add and zero out new clients.*/ - if (!cfs_test_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) { + if (!cfs_test_bit(ted->ted_lr_idx, obt->obt_lut->lut_client_bitmap)) { CERROR("FILTER client %u: bit already clear in bitmap!!\n", - fed->fed_lr_idx); + ted->ted_lr_idx); LBUG(); } - push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); /* Make sure the server's last_transno is up to date. * This should be done before zeroing client slot so last_transno will * be in server data or in client data in case of failure */ - filter_update_server_data(obd); - - cfs_mutex_down(&fed->fed_lcd_lock); - memset(fed->fed_lcd->lcd_uuid, 0, sizeof fed->fed_lcd->lcd_uuid); - rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, - fed->fed_lcd, - sizeof(*fed->fed_lcd), &off, 0); - cfs_mutex_up(&fed->fed_lcd_lock); - pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + filter_update_server_data(exp->exp_obd); + + cfs_mutex_lock(&ted->ted_lcd_lock); + memset(ted->ted_lcd->lcd_uuid, 0, sizeof ted->ted_lcd->lcd_uuid); + rc = fsfilt_write_record(exp->exp_obd, obt->obt_rcvd_filp, + ted->ted_lcd, + sizeof(*ted->ted_lcd), &off, 0); + cfs_mutex_unlock(&ted->ted_lcd_lock); + pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); CDEBUG(rc == 0 ? D_INFO : D_ERROR, "zero out client %s at idx %u/%llu in %s, rc %d\n", - lcd->lcd_uuid, fed->fed_lr_idx, fed->fed_lr_off, + ted->ted_lcd->lcd_uuid, ted->ted_lr_idx, ted->ted_lr_off, LAST_RCVD, rc); EXIT; free: @@ -532,7 +531,7 @@ void filter_fmd_expire(struct obd_export *exp) * caller must hold fed_lock and take fmd reference itself */ static struct filter_mod_data *filter_fmd_find_nolock(struct filter_obd *filter, struct filter_export_data *fed, - obd_id objid, obd_gr group) + obd_id objid, obd_seq group) { struct filter_mod_data *found = NULL, *fmd; @@ -555,7 +554,7 @@ static struct filter_mod_data *filter_fmd_find_nolock(struct filter_obd *filter, /* Find fmd based on objid and group, or return NULL if not found. */ struct filter_mod_data *filter_fmd_find(struct obd_export *exp, - obd_id objid, obd_gr group) + obd_id objid, obd_seq group) { struct filter_mod_data *fmd; @@ -575,7 +574,7 @@ struct filter_mod_data *filter_fmd_find(struct obd_export *exp, * Currently this is not fatal because any fmd state is transient and * may also be freed when it gets sufficiently old. */ struct filter_mod_data *filter_fmd_get(struct obd_export *exp, - obd_id objid, obd_gr group) + obd_id objid, obd_seq group) { struct filter_export_data *fed = &exp->exp_filter_data; struct filter_mod_data *found = NULL, *fmd_new = NULL; @@ -613,7 +612,7 @@ struct filter_mod_data *filter_fmd_get(struct obd_export *exp, * This isn't so critical because it would in fact only affect the one client * that is doing the unlink and at worst we have an stale entry referencing * an object that should never be used again. */ -static void filter_fmd_drop(struct obd_export *exp, obd_id objid, obd_gr group) +static void filter_fmd_drop(struct obd_export *exp, obd_id objid, obd_seq group) { struct filter_mod_data *found = NULL; @@ -645,6 +644,9 @@ static void filter_fmd_cleanup(struct obd_export *exp) static int filter_init_export(struct obd_export *exp) { + int rc; + ENTRY; + cfs_spin_lock_init(&exp->exp_filter_data.fed_lock); CFS_INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list); @@ -652,13 +654,25 @@ static int filter_init_export(struct obd_export *exp) exp->exp_connecting = 1; cfs_spin_unlock(&exp->exp_lock); - return ldlm_init_export(exp); + /* self-export doesn't need client data and ldlm initialization */ + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + + rc = lut_client_alloc(exp); + if (rc == 0) + rc = ldlm_init_export(exp); + if (rc) + CERROR("%s: Can't initialize export: rc %d\n", + exp->exp_obd->obd_name, rc); + + RETURN(rc); } -static int filter_free_server_data(struct filter_obd *filter) +static int filter_free_server_data(struct obd_device_target *obt) { - lut_fini(NULL, filter->fo_obt.obt_lut); - OBD_FREE_PTR(filter->fo_obt.obt_lut); + lut_fini(NULL, obt->obt_lut); + OBD_FREE_PTR(obt->obt_lut); return 0; } @@ -666,25 +680,25 @@ static int filter_free_server_data(struct filter_obd *filter) int filter_update_server_data(struct obd_device *obd) { struct file *filp = obd->u.obt.obt_rcvd_filp; - struct lr_server_data *fsd = class_server_data(obd); + struct lr_server_data *lsd = class_server_data(obd); loff_t off = 0; int rc; ENTRY; - CDEBUG(D_INODE, "server uuid : %s\n", fsd->lsd_uuid); + CDEBUG(D_INODE, "server uuid : %s\n", lsd->lsd_uuid); CDEBUG(D_INODE, "server last_rcvd : "LPU64"\n", - le64_to_cpu(fsd->lsd_last_transno)); + le64_to_cpu(lsd->lsd_last_transno)); CDEBUG(D_INODE, "server last_mount: "LPU64"\n", - le64_to_cpu(fsd->lsd_mount_count)); + le64_to_cpu(lsd->lsd_mount_count)); - rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off, 0); + rc = fsfilt_write_record(obd, filp, lsd, sizeof(*lsd), &off, 0); if (rc) CERROR("error writing lr_server_data: rc = %d\n", rc); RETURN(rc); } -int filter_update_last_objid(struct obd_device *obd, obd_gr group, +int filter_update_last_objid(struct obd_device *obd, obd_seq group, int force_sync) { struct filter_obd *filter = &obd->u.filter; @@ -694,19 +708,19 @@ int filter_update_last_objid(struct obd_device *obd, obd_gr group, ENTRY; if (filter->fo_last_objid_files[group] == NULL) { - CERROR("Object group "LPU64" not fully setup; not updating " + CERROR("Object seq "LPU64" not fully setup; not updating " "last_objid\n", group); RETURN(-EINVAL); } - CDEBUG(D_INODE, "%s: server last_objid for group "LPU64": "LPU64"\n", - obd->obd_name, group, filter->fo_last_objids[group]); + CDEBUG(D_INODE, "%s: server last_objid for "POSTID"\n", + obd->obd_name, filter->fo_last_objids[group], group); tmp = cpu_to_le64(filter->fo_last_objids[group]); rc = fsfilt_write_record(obd, filter->fo_last_objid_files[group], &tmp, sizeof(tmp), &off, force_sync); if (rc) - CERROR("error writing group "LPU64" last objid: rc = %d\n", + CERROR("error writing seq "LPU64" last objid: rc = %d\n", group, rc); RETURN(rc); } @@ -715,7 +729,7 @@ extern int ost_handle(struct ptlrpc_request *req); static int filter_init_server_data(struct obd_device *obd, struct file * filp) { struct filter_obd *filter = &obd->u.filter; - struct lr_server_data *fsd; + struct lr_server_data *lsd; struct lsd_client_data *lcd = NULL; struct inode *inode = filp->f_dentry->d_inode; unsigned long last_rcvd_size = i_size_read(inode); @@ -728,7 +742,7 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) /* ensure padding in the struct is the correct size */ CLASSERT (offsetof(struct lr_server_data, lsd_padding) + - sizeof(fsd->lsd_padding) == LR_SERVER_SIZE); + sizeof(lsd->lsd_padding) == LR_SERVER_SIZE); CLASSERT (offsetof(struct lsd_client_data, lcd_padding) + sizeof(lcd->lcd_padding) == LR_CLIENT_SIZE); @@ -738,108 +752,106 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) RETURN(-ENOMEM); rc = lut_init(NULL, lut, obd, NULL); if (rc) - GOTO(err_fsd, rc); - fsd = class_server_data(obd); + GOTO(err_lut, rc); + lsd = class_server_data(obd); if (last_rcvd_size == 0) { LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name); - memcpy(fsd->lsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->lsd_uuid)); - fsd->lsd_last_transno = 0; - mount_count = fsd->lsd_mount_count = 0; - fsd->lsd_server_size = cpu_to_le32(LR_SERVER_SIZE); - fsd->lsd_client_start = cpu_to_le32(LR_CLIENT_START); - fsd->lsd_client_size = cpu_to_le16(LR_CLIENT_SIZE); - fsd->lsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT); + memcpy(lsd->lsd_uuid, obd->obd_uuid.uuid,sizeof(lsd->lsd_uuid)); + lsd->lsd_last_transno = 0; + mount_count = lsd->lsd_mount_count = 0; + lsd->lsd_server_size = cpu_to_le32(LR_SERVER_SIZE); + lsd->lsd_client_start = cpu_to_le32(LR_CLIENT_START); + lsd->lsd_client_size = cpu_to_le16(LR_CLIENT_SIZE); + lsd->lsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT); filter->fo_subdir_count = FILTER_SUBDIR_COUNT; /* OBD_COMPAT_OST is set in filter_connect_internal when the * MDS first connects and assigns the OST index number. */ - fsd->lsd_feature_incompat = cpu_to_le32(OBD_INCOMPAT_COMMON_LR| + lsd->lsd_feature_incompat = cpu_to_le32(OBD_INCOMPAT_COMMON_LR| OBD_INCOMPAT_OST); } else { - rc = fsfilt_read_record(obd, filp, fsd, sizeof(*fsd), &off); + rc = fsfilt_read_record(obd, filp, lsd, sizeof(*lsd), &off); if (rc) { CDEBUG(D_INODE,"OBD filter: error reading %s: rc %d\n", LAST_RCVD, rc); - GOTO(err_fsd, rc); + GOTO(err_lut, rc); } - if (strcmp(fsd->lsd_uuid, obd->obd_uuid.uuid) != 0) { + if (strcmp(lsd->lsd_uuid, obd->obd_uuid.uuid) != 0) { LCONSOLE_ERROR_MSG(0x134, "Trying to start OBD %s " "using the wrong disk %s. Were the " "/dev/ assignments rearranged?\n", - obd->obd_uuid.uuid, fsd->lsd_uuid); - GOTO(err_fsd, rc = -EINVAL); + obd->obd_uuid.uuid, lsd->lsd_uuid); + GOTO(err_lut, rc = -EINVAL); } - mount_count = le64_to_cpu(fsd->lsd_mount_count); - filter->fo_subdir_count = le16_to_cpu(fsd->lsd_subdir_count); + mount_count = le64_to_cpu(lsd->lsd_mount_count); + filter->fo_subdir_count = le16_to_cpu(lsd->lsd_subdir_count); /* COMPAT_146 */ /* Assume old last_rcvd format unless I_C_LR is set */ - if (!(fsd->lsd_feature_incompat & + if (!(lsd->lsd_feature_incompat & cpu_to_le32(OBD_INCOMPAT_COMMON_LR))) - fsd->lsd_last_transno = fsd->lsd_compat14; + lsd->lsd_last_transno = lsd->lsd_compat14; /* end COMPAT_146 */ /* OBD_COMPAT_OST is set in filter_connect_internal when the * MDS first connects and assigns the OST index number. */ - fsd->lsd_feature_incompat |= cpu_to_le32(OBD_INCOMPAT_COMMON_LR| + lsd->lsd_feature_incompat |= cpu_to_le32(OBD_INCOMPAT_COMMON_LR| OBD_INCOMPAT_OST); } - if (fsd->lsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) { + if (lsd->lsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) { CERROR("%s: unsupported incompat filesystem feature(s) %x\n", - obd->obd_name, le32_to_cpu(fsd->lsd_feature_incompat) & + obd->obd_name, le32_to_cpu(lsd->lsd_feature_incompat) & ~FILTER_INCOMPAT_SUPP); - GOTO(err_fsd, rc = -EINVAL); + GOTO(err_lut, rc = -EINVAL); } - if (fsd->lsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) { + if (lsd->lsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) { CERROR("%s: unsupported read-only filesystem feature(s) %x\n", - obd->obd_name, le32_to_cpu(fsd->lsd_feature_rocompat) & + obd->obd_name, le32_to_cpu(lsd->lsd_feature_rocompat) & ~FILTER_ROCOMPAT_SUPP); /* Do something like remount filesystem read-only */ - GOTO(err_fsd, rc = -EINVAL); + GOTO(err_lut, rc = -EINVAL); } - start_epoch = le32_to_cpu(fsd->lsd_start_epoch); + start_epoch = le32_to_cpu(lsd->lsd_start_epoch); CDEBUG(D_INODE, "%s: server start_epoch : %#x\n", obd->obd_name, start_epoch); CDEBUG(D_INODE, "%s: server last_transno : "LPX64"\n", - obd->obd_name, le64_to_cpu(fsd->lsd_last_transno)); + obd->obd_name, le64_to_cpu(lsd->lsd_last_transno)); CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n", obd->obd_name, mount_count + 1); CDEBUG(D_INODE, "%s: server data size: %u\n", - obd->obd_name, le32_to_cpu(fsd->lsd_server_size)); + obd->obd_name, le32_to_cpu(lsd->lsd_server_size)); CDEBUG(D_INODE, "%s: per-client data start: %u\n", - obd->obd_name, le32_to_cpu(fsd->lsd_client_start)); + obd->obd_name, le32_to_cpu(lsd->lsd_client_start)); CDEBUG(D_INODE, "%s: per-client data size: %u\n", - obd->obd_name, le32_to_cpu(fsd->lsd_client_size)); + obd->obd_name, le32_to_cpu(lsd->lsd_client_size)); CDEBUG(D_INODE, "%s: server subdir_count: %u\n", - obd->obd_name, le16_to_cpu(fsd->lsd_subdir_count)); + obd->obd_name, le16_to_cpu(lsd->lsd_subdir_count)); CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name, - last_rcvd_size <= le32_to_cpu(fsd->lsd_client_start) ? 0 : - (last_rcvd_size - le32_to_cpu(fsd->lsd_client_start)) / - le16_to_cpu(fsd->lsd_client_size)); + last_rcvd_size <= le32_to_cpu(lsd->lsd_client_start) ? 0 : + (last_rcvd_size - le32_to_cpu(lsd->lsd_client_start)) / + le16_to_cpu(lsd->lsd_client_size)); if (!obd->obd_replayable) { CWARN("%s: recovery support OFF\n", obd->obd_name); GOTO(out, rc = 0); } - for (cl_idx = 0, off = le32_to_cpu(fsd->lsd_client_start); + OBD_ALLOC_PTR(lcd); + if (!lcd) + GOTO(err_client, rc = -ENOMEM); + + for (cl_idx = 0, off = le32_to_cpu(lsd->lsd_client_start); off < last_rcvd_size; cl_idx++) { __u64 last_rcvd; struct obd_export *exp; struct filter_export_data *fed; - if (!lcd) { - OBD_ALLOC_PTR(lcd); - if (!lcd) - GOTO(err_client, rc = -ENOMEM); - } - /* Don't assume off is incremented properly by * fsfilt_read_record(), in case sizeof(*lcd) - * isn't the same as fsd->lsd_client_size. */ - off = le32_to_cpu(fsd->lsd_client_start) + - cl_idx * le16_to_cpu(fsd->lsd_client_size); + * isn't the same as lsd->lsd_client_size. */ + off = le32_to_cpu(lsd->lsd_client_start) + + cl_idx * le16_to_cpu(lsd->lsd_client_size); rc = fsfilt_read_record(obd, filp, lcd, sizeof(*lcd), &off); if (rc) { CERROR("error reading FILT %s idx %d off %llu: rc %d\n", @@ -857,60 +869,51 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) last_rcvd = le64_to_cpu(lcd->lcd_last_transno); + CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64 + " srv lr: "LPU64"\n", lcd->lcd_uuid, cl_idx, + last_rcvd, le64_to_cpu(lsd->lsd_last_transno)); + /* These exports are cleaned up by filter_disconnect(), so they * need to be set up like real exports as filter_connect() does. */ exp = class_new_export(obd, (struct obd_uuid *)lcd->lcd_uuid); - - CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64 - " srv lr: "LPU64"\n", lcd->lcd_uuid, cl_idx, - last_rcvd, le64_to_cpu(fsd->lsd_last_transno)); if (IS_ERR(exp)) { if (PTR_ERR(exp) == -EALREADY) { /* export already exists, zero out this one */ - CERROR("Zeroing out duplicate export due to " - "bug 10479.\n"); - lcd->lcd_uuid[0] = '\0'; - } else { - GOTO(err_client, rc = PTR_ERR(exp)); + CERROR("Duplicate export %s!\n", lcd->lcd_uuid); + continue; } - } else { - fed = &exp->exp_filter_data; - fed->fed_lcd = lcd; - fed->fed_group = 0; /* will be assigned at connect */ - filter_export_stats_init(obd, exp, NULL); - rc = filter_client_add(obd, exp, cl_idx); - /* can't fail for existing client */ - LASSERTF(rc == 0, "rc = %d\n", rc); - - /* VBR: set export last committed */ - exp->exp_last_committed = last_rcvd; - cfs_spin_lock(&exp->exp_lock); - exp->exp_connecting = 0; - exp->exp_in_recovery = 0; - cfs_spin_unlock(&exp->exp_lock); - cfs_spin_lock_bh(&obd->obd_processing_task_lock); - obd->obd_max_recoverable_clients++; - cfs_spin_unlock_bh(&obd->obd_processing_task_lock); - lcd = NULL; - class_export_put(exp); + OBD_FREE_PTR(lcd); + GOTO(err_client, rc = PTR_ERR(exp)); } - /* Need to check last_rcvd even for duplicated exports. */ - CDEBUG(D_OTHER, "client at idx %d has last_rcvd = "LPU64"\n", - cl_idx, last_rcvd); + fed = &exp->exp_filter_data; + *fed->fed_ted.ted_lcd = *lcd; + fed->fed_group = 0; /* will be assigned at connect */ + filter_export_stats_init(obd, exp, NULL); + rc = filter_client_add(obd, exp, cl_idx); + /* can't fail for existing client */ + LASSERTF(rc == 0, "rc = %d\n", rc); + + /* VBR: set export last committed */ + exp->exp_last_committed = last_rcvd; + cfs_spin_lock(&exp->exp_lock); + exp->exp_connecting = 0; + exp->exp_in_recovery = 0; + cfs_spin_unlock(&exp->exp_lock); + obd->obd_max_recoverable_clients++; + class_export_put(exp); - if (last_rcvd > le64_to_cpu(fsd->lsd_last_transno)) - fsd->lsd_last_transno = cpu_to_le64(last_rcvd); + if (last_rcvd > le64_to_cpu(lsd->lsd_last_transno)) + lsd->lsd_last_transno = cpu_to_le64(last_rcvd); } + OBD_FREE_PTR(lcd); - if (lcd) - OBD_FREE_PTR(lcd); - - obd->obd_last_committed = le64_to_cpu(fsd->lsd_last_transno); + obd->obd_last_committed = le64_to_cpu(lsd->lsd_last_transno); out: - filter->fo_mount_count = mount_count + 1; - fsd->lsd_mount_count = cpu_to_le64(filter->fo_mount_count); + obd->u.obt.obt_mount_count = mount_count + 1; + obd->u.obt.obt_instance = (__u32)obd->u.obt.obt_mount_count; + lsd->lsd_mount_count = cpu_to_le64(obd->u.obt.obt_mount_count); /* save it, so mount count and last_transno is current */ rc = filter_update_server_data(obd); @@ -921,8 +924,8 @@ out: err_client: class_disconnect_exports(obd); -err_fsd: - filter_free_server_data(filter); +err_lut: + filter_free_server_data(&obd->u.obt); RETURN(rc); } @@ -1054,8 +1057,8 @@ static int filter_read_group_internal(struct obd_device *obd, int group, RETURN(PTR_ERR(dentry)); } } else { - dentry = simple_mkdir(filter->fo_dentry_O, filter->fo_vfsmnt, - name, 0700, 1); + dentry = simple_mkdir(filter->fo_dentry_O, + obd->u.obt.obt_vfsmnt, name, 0700, 1); if (IS_ERR(dentry)) { CERROR("cannot lookup/create O/%s: rc = %ld\n", name, PTR_ERR(dentry)); @@ -1079,7 +1082,7 @@ static int filter_read_group_internal(struct obd_device *obd, int group, GOTO(cleanup, rc); } - if (filter->fo_subdir_count && filter_group_is_mds(group)) { + if (filter->fo_subdir_count && fid_seq_is_mdt(group)) { OBD_ALLOC(tmp_subdirs, sizeof(*tmp_subdirs)); if (tmp_subdirs == NULL) GOTO(cleanup, rc = -ENOMEM); @@ -1090,7 +1093,7 @@ static int filter_read_group_internal(struct obd_device *obd, int group, snprintf(dir, sizeof(dir), "d%u", i); tmp_subdirs->dentry[i] = simple_mkdir(dentry, - filter->fo_vfsmnt, + obd->u.obt.obt_vfsmnt, dir, 0700, 1); if (IS_ERR(tmp_subdirs->dentry[i])) { rc = PTR_ERR(tmp_subdirs->dentry[i]); @@ -1143,7 +1146,7 @@ static int filter_read_group_internal(struct obd_device *obd, int group, filter->fo_dentry_O_groups[group] = dentry; filter->fo_last_objid_files[group] = filp; - if (filter->fo_subdir_count && filter_group_is_mds(group)) { + if (filter->fo_subdir_count && fid_seq_is_mdt(group)) { filter->fo_dentry_O_sub[group] = *tmp_subdirs; OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs)); } @@ -1172,7 +1175,7 @@ static int filter_read_group_internal(struct obd_device *obd, int group, if (new_files != NULL) OBD_FREE(new_files, len * sizeof(*new_files)); case 3: - if (filter->fo_subdir_count && filter_group_is_mds(group)) { + if (filter->fo_subdir_count && fid_seq_is_mdt(group)) { for (i = 0; i < filter->fo_subdir_count; i++) { if (tmp_subdirs->dentry[i] != NULL) dput(tmp_subdirs->dentry[i]); @@ -1193,14 +1196,14 @@ static int filter_read_groups(struct obd_device *obd, int last_group, struct filter_obd *filter = &obd->u.filter; int old_count, group, rc = 0; - cfs_down(&filter->fo_init_lock); + cfs_mutex_lock(&filter->fo_init_lock); old_count = filter->fo_group_count; for (group = old_count; group <= last_group; group++) { rc = filter_read_group_internal(obd, group, create); if (rc != 0) break; } - cfs_up(&filter->fo_init_lock); + cfs_mutex_unlock(&filter->fo_init_lock); return rc; } @@ -1214,12 +1217,12 @@ static int filter_prep_groups(struct obd_device *obd) loff_t off = 0; ENTRY; - O_dentry = simple_mkdir(current->fs->pwd, filter->fo_vfsmnt, + O_dentry = simple_mkdir(cfs_fs_pwd(current->fs), obd->u.obt.obt_vfsmnt, "O", 0700, 1); - CDEBUG(D_INODE, "got/created O: %p\n", O_dentry); + CDEBUG(D_INODE, "%s: got/created O: %p\n", obd->obd_name, O_dentry); if (IS_ERR(O_dentry)) { rc = PTR_ERR(O_dentry); - CERROR("cannot open/create O: rc = %d\n", rc); + CERROR("%s: cannot open/create O: rc = %d\n", obd->obd_name,rc); GOTO(cleanup, rc); } filter->fo_dentry_O = O_dentry; @@ -1229,22 +1232,24 @@ static int filter_prep_groups(struct obd_device *obd) * clients because they may send create/destroy for any group -bzzz */ filp = filp_open("LAST_GROUP", O_CREAT | O_RDWR, 0700); if (IS_ERR(filp)) { - CERROR("cannot create LAST_GROUP: rc = %ld\n", PTR_ERR(filp)); + CERROR("%s: cannot create LAST_GROUP: rc = %ld\n", + obd->obd_name, PTR_ERR(filp)); GOTO(cleanup, rc = PTR_ERR(filp)); } cleanup_phase = 2; /* filp */ rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off); if (rc) { - CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n", rc); + CERROR("%s: error reading LAST_GROUP: rc %d\n", + obd->obd_name, rc); GOTO(cleanup, rc); } if (off == 0) - last_group = FILTER_GROUP_MDS0; + last_group = FID_SEQ_OST_MDT0; - CWARN("%s: initialize groups [%d,%d]\n", obd->obd_name, - FILTER_GROUP_MDS0, last_group); + CDEBUG(D_INODE, "%s: initialize group %u (max %u)\n", obd->obd_name, + FID_SEQ_OST_MDT0, last_group); filter->fo_committed_group = last_group; rc = filter_read_groups(obd, last_group, 1); if (rc) @@ -1286,7 +1291,7 @@ static int filter_prep(struct obd_device *obd) LAST_RCVD, rc); GOTO(out, rc); } - filter->fo_rcvd_filp = file; + obd->u.obt.obt_rcvd_filp = file; if (!S_ISREG(file->f_dentry->d_inode->i_mode)) { CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD, file->f_dentry->d_inode->i_mode); @@ -1341,11 +1346,11 @@ err_health_check: filter->fo_obt.obt_health_check_filp = NULL; err_server_data: target_recovery_fini(obd); - filter_free_server_data(filter); + filter_free_server_data(&obd->u.obt); err_filp: - if (filp_close(filter->fo_rcvd_filp, 0)) + if (filp_close(obd->u.obt.obt_rcvd_filp, 0)) CERROR("can't close %s after error\n", LAST_RCVD); - filter->fo_rcvd_filp = NULL; + obd->u.obt.obt_rcvd_filp = NULL; goto out; } @@ -1373,8 +1378,8 @@ static void filter_post(struct obd_device *obd) i, rc); } - rc = filp_close(filter->fo_rcvd_filp, 0); - filter->fo_rcvd_filp = NULL; + rc = filp_close(obd->u.obt.obt_rcvd_filp, 0); + obd->u.obt.obt_rcvd_filp = NULL; if (rc) CERROR("error closing %s: rc = %d\n", LAST_RCVD, rc); @@ -1384,7 +1389,7 @@ static void filter_post(struct obd_device *obd) CERROR("error closing %s: rc = %d\n", HEALTH_CHECK, rc); filter_cleanup_groups(obd); - filter_free_server_data(filter); + filter_free_server_data(&obd->u.obt); pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); filter_free_capa_keys(filter); @@ -1392,7 +1397,7 @@ static void filter_post(struct obd_device *obd) } static void filter_set_last_id(struct filter_obd *filter, - obd_id id, obd_gr group) + obd_id id, obd_seq group) { LASSERT(group <= filter->fo_group_count); @@ -1401,7 +1406,7 @@ static void filter_set_last_id(struct filter_obd *filter, cfs_spin_unlock(&filter->fo_objidlock); } -obd_id filter_last_id(struct filter_obd *filter, obd_gr group) +obd_id filter_last_id(struct filter_obd *filter, obd_seq group) { obd_id id; LASSERT(group <= filter->fo_group_count); @@ -1421,13 +1426,15 @@ static int filter_lock_dentry(struct obd_device *obd, struct dentry *dparent) } /* We never dget the object parent, so DON'T dput it either */ -struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid) +struct dentry *filter_parent(struct obd_device *obd, obd_seq group, obd_id objid) { struct filter_obd *filter = &obd->u.filter; struct filter_subdirs *subdirs; - LASSERT(group < filter->fo_group_count); /* FIXME: object groups */ - if (!filter_group_is_mds(group) || filter->fo_subdir_count == 0) + if (group >= filter->fo_group_count) /* FIXME: object groups */ + return ERR_PTR(-EBADF); + + if (!fid_seq_is_mdt(group) || filter->fo_subdir_count == 0) return filter->fo_dentry_O_groups[group]; subdirs = &filter->fo_dentry_O_sub[group]; @@ -1435,7 +1442,7 @@ struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid) } /* We never dget the object parent, so DON'T dput it either */ -struct dentry *filter_parent_lock(struct obd_device *obd, obd_gr group, +struct dentry *filter_parent_lock(struct obd_device *obd, obd_seq group, obd_id objid) { unsigned long now = jiffies; @@ -1467,7 +1474,7 @@ static void filter_parent_unlock(struct dentry *dparent) * internal to the filesystem code. */ struct dentry *filter_fid2dentry(struct obd_device *obd, struct dentry *dir_dentry, - obd_gr group, obd_id id) + obd_seq group, obd_id id) { struct dentry *dparent = dir_dentry; struct dentry *dchild; @@ -1479,7 +1486,7 @@ struct dentry *filter_fid2dentry(struct obd_device *obd, obd->u.filter.fo_destroys_in_progress == 0) { /* don't fail lookups for orphan recovery, it causes * later LBUGs when objects still exist during precreate */ - CDEBUG(D_INFO, "*** obd_fail_loc=%x ***\n",OBD_FAIL_OST_ENOENT); + CDEBUG(D_INFO, "*** cfs_fail_loc=%x ***\n",OBD_FAIL_OST_ENOENT); RETURN(ERR_PTR(-ENOENT)); } if (id == 0) { @@ -1491,7 +1498,7 @@ struct dentry *filter_fid2dentry(struct obd_device *obd, if (dir_dentry == NULL) { dparent = filter_parent_lock(obd, group, id); if (IS_ERR(dparent)) { - CERROR("%s: error getting object "LPU64":"LPU64 + CERROR("%s: error getting object "POSTID " parent: rc %ld\n", obd->obd_name, id, group, PTR_ERR(dparent)); RETURN(dparent); @@ -1499,12 +1506,13 @@ struct dentry *filter_fid2dentry(struct obd_device *obd, } CDEBUG(D_INODE, "looking up object O/%.*s/%s\n", dparent->d_name.len, dparent->d_name.name, name); - dchild = /*ll_*/lookup_one_len(name, dparent, len); + /* dparent is already locked here, so we cannot use ll_lookup_one_len() */ + dchild = lookup_one_len(name, dparent, len); if (dir_dentry == NULL) filter_parent_unlock(dparent); if (IS_ERR(dchild)) { - CERROR("%s: child lookup error %ld\n", obd->obd_name, - PTR_ERR(dchild)); + CERROR("%s: object "LPU64":"LPU64" lookup error: rc %ld\n", + obd->obd_name, id, group, PTR_ERR(dchild)); RETURN(dchild); } @@ -1580,15 +1588,13 @@ int filter_vfs_unlink(struct inode *dir, struct dentry *dentry, GOTO(out, rc = -EPERM); /* check_sticky() */ - if ((dentry->d_inode->i_uid != current->fsuid && + if ((dentry->d_inode->i_uid != cfs_curproc_fsuid() && !cfs_capable(CFS_CAP_FOWNER)) || IS_APPEND(dentry->d_inode) || IS_IMMUTABLE(dentry->d_inode)) GOTO(out, rc = -EPERM); - /* NOTE: This might need to go outside i_mutex, though it isn't clear if - * that was done because of journal_start (which is already done - * here) or some other ordering issue. */ - DQUOT_INIT(dir); + /* Locking order: i_mutex -> journal_lock -> dqptr_sem. LU-952 */ + ll_vfs_dq_init(dir); rc = ll_security_inode_unlink(dir, dentry, mnt); if (rc) @@ -1608,21 +1614,23 @@ out: * Caller must hold child i_mutex, we drop it always. * Caller is also required to ensure that dchild->d_inode exists. */ static int filter_destroy_internal(struct obd_device *obd, obd_id objid, - obd_gr group, struct dentry *dparent, + obd_seq group, struct dentry *dparent, struct dentry *dchild) { struct inode *inode = dchild->d_inode; - struct filter_obd *filter = &obd->u.filter; int rc; - if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) { + /* There should be 2 references to the inode: + * 1) taken by filter_prepare_destroy + * 2) taken by filter_destroy */ + if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 2) { CERROR("destroying objid %.*s ino %lu nlink %lu count %d\n", dchild->d_name.len, dchild->d_name.name, inode->i_ino, (unsigned long)inode->i_nlink, atomic_read(&inode->i_count)); } - rc = filter_vfs_unlink(dparent->d_inode, dchild, filter->fo_vfsmnt); + rc = filter_vfs_unlink(dparent->d_inode, dchild, obd->u.obt.obt_vfsmnt); if (rc) CERROR("error unlinking objid %.*s: rc %d\n", dchild->d_name.len, dchild->d_name.name, rc); @@ -1680,7 +1688,6 @@ static int filter_intent_policy(struct ldlm_namespace *ns, struct ldlm_lock **lockp, void *req_cookie, ldlm_mode_t mode, int flags, void *data) { - CFS_LIST_HEAD(rpc_list); struct ptlrpc_request *req = req_cookie; struct ldlm_lock *lock = *lockp, *l = NULL; struct ldlm_resource *res = lock->l_resource; @@ -1719,29 +1726,23 @@ static int filter_intent_policy(struct ldlm_namespace *ns, * lock, and should not be granted if the lock will be blocked. */ - LASSERT(ns == res->lr_namespace); - lock_res(res); - rc = policy(lock, &tmpflags, 0, &err, &rpc_list); - check_res_locked(res); + if (flags & LDLM_FL_BLOCK_NOWAIT) { + OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_AGL_DELAY, 5); - /* FIXME: we should change the policy function slightly, to not make - * this list at all, since we just turn around and free it */ - while (!cfs_list_empty(&rpc_list)) { - struct ldlm_lock *wlock = - cfs_list_entry(rpc_list.next, struct ldlm_lock, - l_cp_ast); - LASSERT((lock->l_flags & LDLM_FL_AST_SENT) == 0); - LASSERT(lock->l_flags & LDLM_FL_CP_REQD); - lock->l_flags &= ~LDLM_FL_CP_REQD; - cfs_list_del_init(&wlock->l_cp_ast); - LDLM_LOCK_RELEASE(wlock); + if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_AGL_NOLOCK)) + RETURN(ELDLM_LOCK_ABORTED); } + LASSERT(ns == ldlm_res_to_ns(res)); + lock_res(res); + rc = policy(lock, &tmpflags, 0, &err, NULL); + check_res_locked(res); + /* The lock met with no resistance; we're finished. */ if (rc == LDLM_ITER_CONTINUE) { /* do not grant locks to the liblustre clients: they cannot * handle ASTs robustly. We need to do this while still - * holding ns_lock to avoid the lock remaining on the res_link + * holding lr_lock to avoid the lock remaining on the res_link * list (and potentially being added to l_pending_list by an * AST) when we are going to drop this lock ASAP. */ if (lock->l_export->exp_libclient || @@ -1753,6 +1754,12 @@ static int filter_intent_policy(struct ldlm_namespace *ns, } unlock_res(res); RETURN(err); + } else if (flags & LDLM_FL_BLOCK_NOWAIT) { + /* LDLM_FL_BLOCK_NOWAIT means it is for AGL. Do not send glimpse + * callback for glimpse size. The real size user will trigger + * the glimpse callback when necessary. */ + unlock_res(res); + RETURN(ELDLM_LOCK_ABORTED); } /* Do not grant any lock, but instead send GL callbacks. The extent @@ -1764,11 +1771,11 @@ static int filter_intent_policy(struct ldlm_namespace *ns, *reply_lvb = *res_lvb; /* - * ->ns_lock guarantees that no new locks are granted, and, + * lr_lock guarantees that no new locks are granted, and, * therefore, that res->lr_lvb_data cannot increase beyond the * end of already granted lock. As a result, it is safe to * check against "stale" reply_lvb->lvb_size value without - * res->lr_lvb_sem. + * res->lr_lvb_mutex. */ arg.size = reply_lvb->lvb_size; arg.victim = &l; @@ -1817,13 +1824,6 @@ static int filter_intent_policy(struct ldlm_namespace *ns, LASSERTF(l->l_glimpse_ast != NULL, "l == %p", l); rc = l->l_glimpse_ast(l, NULL); /* this will update the LVB */ - /* Update the LVB from disk if the AST failed (this is a legal race) */ - /* - * XXX nikita: situation when ldlm_server_glimpse_ast() failed before - * sending ast is not handled. This can result in lost client writes. - */ - if (rc != 0) - ldlm_res_lvbo_update(res, NULL, 1); lock_res(res); *reply_lvb = *res_lvb; @@ -1921,7 +1921,7 @@ static int filter_iobuf_pool_init(struct filter_obd *filter) OBD_ALLOC_GFP(filter->fo_iobuf_pool, OSS_THREADS_MAX * sizeof(*pool), - GFP_KERNEL); + CFS_ALLOC_KERNEL); if (filter->fo_iobuf_pool == NULL) RETURN(-ENOMEM); @@ -1967,12 +1967,13 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, { struct filter_obd *filter = &obd->u.filter; struct vfsmount *mnt; + struct file_system_type *type; struct lustre_mount_info *lmi; struct obd_uuid uuid; __u8 *uuid_ptr; char *str, *label; char ns_name[48]; - request_queue_t *q; + struct request_queue *q; int rc, i; ENTRY; @@ -1991,9 +1992,14 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, } else { /* old path - used by lctl */ CERROR("Using old MDS mount method\n"); - mnt = ll_kern_mount(lustre_cfg_string(lcfg, 2), - MS_NOATIME|MS_NODIRATIME, - lustre_cfg_string(lcfg, 1), option); + type = get_fs_type(lustre_cfg_string(lcfg, 2)); + if (!type) { + CERROR("get_fs_type failed\n"); + RETURN(-ENODEV); + } + mnt = vfs_kern_mount(type, MS_NOATIME|MS_NODIRATIME, + lustre_cfg_string(lcfg, 1), option); + cfs_module_put(type->owner); if (IS_ERR(mnt)) { rc = PTR_ERR(mnt); LCONSOLE_ERROR_MSG(0x135, "Can't mount disk %s (%d)\n", @@ -2019,6 +2025,9 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, /* failover is the default */ obd->obd_replayable = 1; + /* disable connection until configuration finishes */ + obd->obd_no_conn = 1; + if (lcfg->lcfg_bufcount > 3 && LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) { str = lustre_cfg_string(lcfg, 3); if (strchr(str, 'n')) { @@ -2027,12 +2036,15 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, } } - filter->fo_vfsmnt = mnt; + obd->u.obt.obt_magic = OBT_MAGIC; + obd->u.obt.obt_vfsmnt = mnt; obd->u.obt.obt_sb = mnt->mnt_sb; filter->fo_fstype = mnt->mnt_sb->s_type->name; CDEBUG(D_SUPER, "%s: mnt = %p\n", filter->fo_fstype, mnt); - fsfilt_setup(obd, obd->u.obt.obt_sb); + rc = fsfilt_setup(obd, obd->u.obt.obt_sb); + if (rc) + GOTO(err_ops, rc); OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt); obd->obd_lvfs_ctxt.pwdmnt = mnt; @@ -2040,21 +2052,24 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, obd->obd_lvfs_ctxt.fs = get_ds(); obd->obd_lvfs_ctxt.cb_ops = filter_lvfs_ops; - cfs_init_mutex(&filter->fo_init_lock); + cfs_mutex_init(&filter->fo_init_lock); filter->fo_committed_group = 0; filter->fo_destroys_in_progress = 0; for (i = 0; i < 32; i++) - cfs_sema_init(&filter->fo_create_locks[i], 1); + cfs_mutex_init(&filter->fo_create_locks[i]); cfs_spin_lock_init(&filter->fo_objidlock); CFS_INIT_LIST_HEAD(&filter->fo_export_list); - cfs_sema_init(&filter->fo_alloc_lock, 1); + cfs_mutex_init(&filter->fo_alloc_lock); init_brw_stats(&filter->fo_filter_stats); + cfs_spin_lock_init(&filter->fo_flags_lock); filter->fo_read_cache = 1; /* enable read-only cache by default */ filter->fo_writethrough_cache = 1; /* enable writethrough cache */ filter->fo_readcache_max_filesize = FILTER_MAX_CACHE_SIZE; filter->fo_fmd_max_num = FILTER_FMD_MAX_NUM_DEFAULT; filter->fo_fmd_max_age = FILTER_FMD_MAX_AGE_DEFAULT; + filter->fo_syncjournal = 0; /* Don't sync journals on i/o by default */ + filter_slc_set(filter); /* initialize sync on lock cancel */ rc = filter_prep(obd); if (rc) @@ -2071,8 +2086,10 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, GOTO(err_post, rc = -ENOMEM); sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid); - obd->obd_namespace = ldlm_namespace_new(obd, ns_name, LDLM_NAMESPACE_SERVER, - LDLM_NAMESPACE_GREEDY); + obd->obd_namespace = ldlm_namespace_new(obd, ns_name, + LDLM_NAMESPACE_SERVER, + LDLM_NAMESPACE_GREEDY, + LDLM_NS_TYPE_OST); if (obd->obd_namespace == NULL) GOTO(err_post, rc = -ENOMEM); obd->obd_namespace->ns_lvbp = obd; @@ -2098,13 +2115,13 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, GOTO(err_post, rc); q = bdev_get_queue(mnt->mnt_sb->s_bdev); - if (q->max_sectors < q->max_hw_sectors && - q->max_sectors < PTLRPC_MAX_BRW_SIZE >> 9) + if (queue_max_sectors(q) < queue_max_hw_sectors(q) && + queue_max_sectors(q) < PTLRPC_MAX_BRW_SIZE >> 9) LCONSOLE_INFO("%s: underlying device %s should be tuned " "for larger I/O requests: max_sectors = %u " "could be up to max_hw_sectors=%u\n", obd->obd_name, mnt->mnt_sb->s_id, - q->max_sectors, q->max_hw_sectors); + queue_max_sectors(q), queue_max_hw_sectors(q)); uuid_ptr = fsfilt_uuid(obd, obd->u.obt.obt_sb); if (uuid_ptr != NULL) { @@ -2120,17 +2137,6 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg, lmi ? s2lsi(lmi->lmi_sb)->lsi_lmd->lmd_dev : "", obd->obd_replayable ? "enabled" : "disabled"); - if (obd->obd_recovering) - LCONSOLE_WARN("%s: Will be in recovery for at least %d:%.02d, " - "or until %d client%s reconnect%s\n", - obd->obd_name, - obd->obd_recovery_timeout / 60, - obd->obd_recovery_timeout % 60, - obd->obd_max_recoverable_clients, - (obd->obd_max_recoverable_clients == 1) ? "" : "s", - (obd->obd_max_recoverable_clients == 1) ? "s": ""); - - RETURN(0); err_post: @@ -2147,9 +2153,11 @@ err_mntput: static int filter_setup(struct obd_device *obd, struct lustre_cfg* lcfg) { struct lprocfs_static_vars lvars; + cfs_proc_dir_entry_t *entry; unsigned long addr; struct page *page; int rc; + ENTRY; CLASSERT(offsetof(struct obd_device, u.obt) == offsetof(struct obd_device, u.filter.fo_obt)); @@ -2157,69 +2165,96 @@ static int filter_setup(struct obd_device *obd, struct lustre_cfg* lcfg) if (!LUSTRE_CFG_BUFLEN(lcfg, 1) || !LUSTRE_CFG_BUFLEN(lcfg, 2)) RETURN(-EINVAL); - /* 2.6.9 selinux wants a full option page for do_kern_mount (bug6471) */ - OBD_PAGE_ALLOC(page, CFS_ALLOC_STD); - if (!page) - RETURN(-ENOMEM); - addr = (unsigned long)cfs_page_address(page); - clear_page((void *)addr); - /* lprocfs must be setup before the filter so state can be safely added * to /proc incrementally as the filter is setup */ lprocfs_filter_init_vars(&lvars); - if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0 && - lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST) == 0) { - /* Init obdfilter private stats here */ - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES, - LPROCFS_CNTR_AVGMINMAX, - "read_bytes", "bytes"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES, - LPROCFS_CNTR_AVGMINMAX, - "write_bytes", "bytes"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_GET_PAGE, - LPROCFS_CNTR_AVGMINMAX|LPROCFS_CNTR_STDDEV, - "get_page", "usec"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_NO_PAGE, - LPROCFS_CNTR_AVGMINMAX, - "get_page_failures", "num"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_ACCESS, - LPROCFS_CNTR_AVGMINMAX, - "cache_access", "pages"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_HIT, - LPROCFS_CNTR_AVGMINMAX, - "cache_hit", "pages"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_MISS, - LPROCFS_CNTR_AVGMINMAX, - "cache_miss", "pages"); - - lproc_filter_attach_seqstat(obd); - obd->obd_proc_exports_entry = lprocfs_register("exports", - obd->obd_proc_entry, - NULL, NULL); - if (IS_ERR(obd->obd_proc_exports_entry)) { - rc = PTR_ERR(obd->obd_proc_exports_entry); - CERROR("error %d setting up lprocfs for %s\n", - rc, "exports"); - obd->obd_proc_exports_entry = NULL; - } + rc = lprocfs_obd_setup(obd, lvars.obd_vars); + if (rc) { + CERROR("%s: lprocfs_obd_setup failed: %d.\n", + obd->obd_name, rc); + RETURN(rc); + } + + rc = lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST); + if (rc) { + CERROR("%s: lprocfs_alloc_obd_stats failed: %d.\n", + obd->obd_name, rc); + GOTO(obd_cleanup, rc); } - if (obd->obd_proc_exports_entry) - lprocfs_add_simple(obd->obd_proc_exports_entry, "clear", + + /* Init obdfilter private stats here */ + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES, + LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES, + LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_GET_PAGE, + LPROCFS_CNTR_AVGMINMAX|LPROCFS_CNTR_STDDEV, + "get_page", "usec"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_NO_PAGE, + LPROCFS_CNTR_AVGMINMAX, "get_page_failures", "num"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_ACCESS, + LPROCFS_CNTR_AVGMINMAX, "cache_access", "pages"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_HIT, + LPROCFS_CNTR_AVGMINMAX, "cache_hit", "pages"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_CACHE_MISS, + LPROCFS_CNTR_AVGMINMAX, "cache_miss", "pages"); + + rc = lproc_filter_attach_seqstat(obd); + if (rc) { + CERROR("%s: create seqstat failed: %d.\n", obd->obd_name, rc); + GOTO(free_obd_stats, rc); + } + + entry = lprocfs_register("exports", obd->obd_proc_entry, NULL, NULL); + if (IS_ERR(entry)) { + rc = PTR_ERR(entry); + CERROR("%s: error %d setting up lprocfs for %s\n", + obd->obd_name, rc, "exports"); + GOTO(free_obd_stats, rc); + } + obd->obd_proc_exports_entry = entry; + + entry = lprocfs_add_simple(obd->obd_proc_exports_entry, "clear", lprocfs_nid_stats_clear_read, lprocfs_nid_stats_clear_write, obd, NULL); + if (IS_ERR(entry)) { + rc = PTR_ERR(entry); + CERROR("%s: add proc entry 'clear' failed: %d.\n", + obd->obd_name, rc); + GOTO(free_obd_stats, rc); + } + + rc = lprocfs_job_stats_init(obd, LPROC_FILTER_STATS_LAST, + filter_stats_counter_init); + if (rc) + GOTO(remove_entry_clear, rc); + /* 2.6.9 selinux wants a full option page for do_kern_mount (bug6471) */ + OBD_PAGE_ALLOC(page, CFS_ALLOC_STD); + if (!page) + GOTO(job_stats_fini, rc = -ENOMEM); + addr = (unsigned long)cfs_page_address(page); + clear_page((void *)addr); memcpy((void *)addr, lustre_cfg_buf(lcfg, 4), LUSTRE_CFG_BUFLEN(lcfg, 4)); rc = filter_common_setup(obd, lcfg, (void *)addr); OBD_PAGE_FREE(page); - if (rc) { - lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); - lprocfs_free_per_client_stats(obd); - lprocfs_free_obd_stats(obd); - lprocfs_obd_cleanup(obd); + CERROR("%s: filter_common_setup failed: %d.\n", + obd->obd_name, rc); + GOTO(job_stats_fini, rc); } + RETURN(0); + +job_stats_fini: + lprocfs_job_stats_fini(obd); +remove_entry_clear: + lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); +free_obd_stats: + lprocfs_free_obd_stats(obd); +obd_cleanup: + lprocfs_obd_cleanup(obd); return rc; } @@ -2373,28 +2408,28 @@ static int filter_llog_finish(struct obd_device *obd, int count) * We actually do sync in disconnect time, but disconnect * may not come being marked rq_no_resend = 1. */ - llog_sync(ctxt, NULL); + llog_sync(ctxt, NULL, OBD_LLOG_FL_EXIT); /* * Balance class_import_get() in llog_receptor_accept(). * This is safe to do, as llog is already synchronized * and its import may go. */ - cfs_mutex_down(&ctxt->loc_sem); + cfs_mutex_lock(&ctxt->loc_mutex); if (ctxt->loc_imp) { class_import_put(ctxt->loc_imp); ctxt->loc_imp = NULL; } - cfs_mutex_up(&ctxt->loc_sem); + + if (filter->fo_lcm) { + llog_recov_thread_fini(filter->fo_lcm, obd->obd_force); + filter->fo_lcm = NULL; + } + + cfs_mutex_unlock(&ctxt->loc_mutex); llog_ctxt_put(ctxt); } - if (filter->fo_lcm) { - cfs_mutex_down(&ctxt->loc_sem); - llog_recov_thread_fini(filter->fo_lcm, obd->obd_force); - filter->fo_lcm = NULL; - cfs_mutex_up(&ctxt->loc_sem); - } RETURN(filter_olg_fini(&obd->obd_olg)); } @@ -2408,7 +2443,7 @@ filter_find_olg_internal(struct filter_obd *filter, int group) LASSERT_SPIN_LOCKED(&filter->fo_llog_list_lock); cfs_list_for_each_entry(olg, &filter->fo_llog_list, olg_list) { - if (olg->olg_group == group) + if (olg->olg_seq == group) RETURN(olg); } RETURN(NULL); @@ -2424,7 +2459,7 @@ struct obd_llog_group *filter_find_olg(struct obd_device *obd, int group) filter = &obd->u.filter; - if (group == FILTER_GROUP_LLOG) + if (group == FID_SEQ_LLOG) RETURN(&obd->obd_olg); cfs_spin_lock(&filter->fo_llog_list_lock); @@ -2440,15 +2475,19 @@ struct obd_llog_group *filter_find_olg(struct obd_device *obd, int group) */ struct obd_llog_group *filter_find_create_olg(struct obd_device *obd, int group) { - struct obd_llog_group *olg = NULL; + struct obd_llog_group *olg = NULL, *olg_new = NULL; struct filter_obd *filter; int rc; filter = &obd->u.filter; - if (group == FILTER_GROUP_LLOG) + if (group == FID_SEQ_LLOG) RETURN(&obd->obd_olg); + OBD_ALLOC_PTR(olg_new); + if (olg_new == NULL) + RETURN(ERR_PTR(-ENOMEM)); + cfs_spin_lock(&filter->fo_llog_list_lock); olg = filter_find_olg_internal(filter, group); if (olg) { @@ -2457,10 +2496,11 @@ struct obd_llog_group *filter_find_create_olg(struct obd_device *obd, int group) } else { GOTO(out_unlock, olg); } + } else { + /* set as the newly allocated one */ + olg = olg_new; + olg_new = NULL; } - OBD_ALLOC_PTR(olg); - if (olg == NULL) - GOTO(out_unlock, olg = ERR_PTR(-ENOMEM)); llog_group_init(olg, group); cfs_list_add(&olg->olg_list, &filter->fo_llog_list); @@ -2485,7 +2525,9 @@ out: out_unlock: cfs_spin_unlock(&filter->fo_llog_list_lock); - GOTO(out, olg); + if (olg_new) + OBD_FREE_PTR(olg_new); + goto out; } static int filter_llog_connect(struct obd_export *exp, @@ -2499,12 +2541,12 @@ static int filter_llog_connect(struct obd_export *exp, CDEBUG(D_OTHER, "%s: LLog connect for: "LPX64"/"LPX64":%x\n", obd->obd_name, body->lgdc_logid.lgl_oid, - body->lgdc_logid.lgl_ogr, body->lgdc_logid.lgl_ogen); + body->lgdc_logid.lgl_oseq, body->lgdc_logid.lgl_ogen); - olg = filter_find_olg(obd, body->lgdc_logid.lgl_ogr); + olg = filter_find_olg(obd, body->lgdc_logid.lgl_oseq); if (!olg) { CERROR(" %s: can not find olg of group %d\n", - obd->obd_name, (int)body->lgdc_logid.lgl_ogr); + obd->obd_name, (int)body->lgdc_logid.lgl_oseq); RETURN(-ENOENT); } llog_group_set_export(olg, exp); @@ -2513,13 +2555,13 @@ static int filter_llog_connect(struct obd_export *exp, LASSERTF(ctxt != NULL, "ctxt is not null, ctxt idx %d \n", body->lgdc_ctxt_idx); - CWARN("%s: Recovery from log "LPX64"/"LPX64":%x\n", - obd->obd_name, body->lgdc_logid.lgl_oid, - body->lgdc_logid.lgl_ogr, body->lgdc_logid.lgl_ogen); + CDEBUG(D_HA, "%s: Recovery from log "LPX64"/"LPX64":%x\n", + obd->obd_name, body->lgdc_logid.lgl_oid, + body->lgdc_logid.lgl_oseq, body->lgdc_logid.lgl_ogen); - cfs_spin_lock_bh(&obd->obd_processing_task_lock); + cfs_spin_lock(&obd->u.filter.fo_flags_lock); obd->u.filter.fo_mds_ost_sync = 1; - cfs_spin_unlock_bh(&obd->obd_processing_task_lock); + cfs_spin_unlock(&obd->u.filter.fo_flags_lock); rc = llog_connect(ctxt, &body->lgdc_logid, &body->lgdc_gen, NULL); llog_ctxt_put(ctxt); @@ -2559,7 +2601,7 @@ static int filter_llog_preclean(struct obd_device *obd) rc = filter_olg_fini(olg); if (rc) CERROR("failed to cleanup llogging subsystem for %u\n", - olg->olg_group); + olg->olg_seq); OBD_FREE_PTR(olg); } @@ -2578,7 +2620,17 @@ static int filter_precleanup(struct obd_device *obd, case OBD_CLEANUP_EXPORTS: /* Stop recovery before namespace cleanup. */ target_recovery_fini(obd); + + obd_exports_barrier(obd); + obd_zombie_barrier(); + rc = filter_llog_preclean(obd); + lprocfs_job_stats_fini(obd); + lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); + lprocfs_free_per_client_stats(obd); + lprocfs_obd_cleanup(obd); + lprocfs_free_obd_stats(obd); + lquota_cleanup(filter_quota_interface_ref, obd); break; } RETURN(rc); @@ -2593,15 +2645,6 @@ static int filter_cleanup(struct obd_device *obd) LCONSOLE_WARN("%s: shutting down for failover; client state " "will be preserved.\n", obd->obd_name); - obd_exports_barrier(obd); - obd_zombie_barrier(); - - lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); - lprocfs_free_per_client_stats(obd); - lprocfs_free_obd_stats(obd); - lprocfs_obd_cleanup(obd); - lquota_cleanup(filter_quota_interface_ref, obd); - ldlm_namespace_free(obd->obd_namespace, NULL, obd->obd_force); obd->obd_namespace = NULL; @@ -2612,10 +2655,10 @@ static int filter_cleanup(struct obd_device *obd) filter_post(obd); - LL_DQUOT_OFF(obd->u.obt.obt_sb); + ll_vfs_dq_off(obd->u.obt.obt_sb, 0); shrink_dcache_sb(obd->u.obt.obt_sb); - server_put_mount(obd->obd_name, filter->fo_vfsmnt); + server_put_mount(obd->obd_name, obd->u.obt.obt_vfsmnt); obd->u.obt.obt_sb = NULL; fsfilt_put_ops(obd->obd_fsops); @@ -2646,7 +2689,7 @@ static int filter_connect_internal(struct obd_export *exp, CWARN("!!! This export (nid %s) used object group %d " "earlier; now it's trying to use group %d! This could " "be a bug in the MDS. Please report to " - "http://bugzilla.lustre.org/\n", + "http://bugs.whamcloud.com/\n", obd_export_nid2str(exp), fed->fed_group,data->ocd_group); RETURN(-EPROTO); } @@ -2657,8 +2700,10 @@ static int filter_connect_internal(struct obd_export *exp, data->ocd_version = LUSTRE_VERSION_CODE; /* Kindly make sure the SKIP_ORPHAN flag is from MDS. */ - if (!ergo(data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN, - data->ocd_connect_flags & OBD_CONNECT_MDS)) + if (data->ocd_connect_flags & OBD_CONNECT_MDS) + CDEBUG(D_HA, "%s: Received MDS connection for group %u\n", + exp->exp_obd->obd_name, data->ocd_group); + else if (data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN) RETURN(-EPROTO); if (exp->exp_connect_flags & OBD_CONNECT_GRANT) { @@ -2701,9 +2746,9 @@ static int filter_connect_internal(struct obd_export *exp, data->ocd_index); RETURN(-EBADF); } - /* FIXME: Do the same with the MDS UUID and fsd_peeruuid. + /* FIXME: Do the same with the MDS UUID and lsd_peeruuid. * FIXME: We don't strictly need the COMPAT flag for that, - * FIXME: as fsd_peeruuid[0] will tell us if that is set. + * FIXME: as lsd_peeruuid[0] will tell us if that is set. * FIXME: We needed it for the index, as index 0 is valid. */ } @@ -2712,7 +2757,18 @@ static int filter_connect_internal(struct obd_export *exp, } else if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) { data->ocd_brw_size = min(data->ocd_brw_size, (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT)); - LASSERT(data->ocd_brw_size); + if (data->ocd_brw_size == 0) { + CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64 + " ocd_version: %x ocd_grant: %d ocd_index: %u " + "ocd_brw_size is unexpectedly zero, " + "network data corruption?" + "Refusing connection of this client\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, + exp, data->ocd_connect_flags, data->ocd_version, + data->ocd_grant, data->ocd_index); + RETURN(-EPROTO); + } } if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) { @@ -2721,9 +2777,10 @@ static int filter_connect_internal(struct obd_export *exp, /* The client set in ocd_cksum_types the checksum types it * supports. We have to mask off the algorithms that we don't * support */ - if (cksum_types & OBD_CKSUM_ALL) - data->ocd_cksum_types &= OBD_CKSUM_ALL; - else + data->ocd_cksum_types &= cksum_types_supported(); + + /* 1.6.4- only support CRC32 and didn't set ocd_cksum_types */ + if (unlikely(data->ocd_cksum_types == 0)) data->ocd_cksum_types = OBD_CKSUM_CRC32; CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return " @@ -2739,6 +2796,9 @@ static int filter_connect_internal(struct obd_export *exp, obd_export_nid2str(exp)); } + if (data->ocd_connect_flags & OBD_CONNECT_MAXBYTES) + data->ocd_maxbytes = exp->exp_obd->u.obt.obt_sb->s_maxbytes; + RETURN(0); } @@ -2761,7 +2821,6 @@ static int filter_reconnect(const struct lu_env *env, RETURN(rc); } -/* nearly identical to mds_connect */ static int filter_connect(const struct lu_env *env, struct obd_export **exp, struct obd_device *obd, struct obd_uuid *cluuid, @@ -2770,9 +2829,6 @@ static int filter_connect(const struct lu_env *env, struct lvfs_run_ctxt saved; struct lustre_handle conn = { 0 }; struct obd_export *lexp; - struct filter_export_data *fed; - struct lsd_client_data *lcd = NULL; - __u32 group; int rc; ENTRY; @@ -2785,37 +2841,25 @@ static int filter_connect(const struct lu_env *env, lexp = class_conn2export(&conn); LASSERT(lexp != NULL); - fed = &lexp->exp_filter_data; - rc = filter_connect_internal(lexp, data, 0); if (rc) GOTO(cleanup, rc); filter_export_stats_init(obd, lexp, localdata); if (obd->obd_replayable) { - OBD_ALLOC(lcd, sizeof(*lcd)); - if (!lcd) { - CERROR("filter: out of memory for client data\n"); - GOTO(cleanup, rc = -ENOMEM); - } - + struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; + LASSERT(lcd); memcpy(lcd->lcd_uuid, cluuid, sizeof(lcd->lcd_uuid)); - fed->fed_lcd = lcd; rc = filter_client_add(obd, lexp, -1); if (rc) GOTO(cleanup, rc); } - group = data->ocd_group; - - CWARN("%s: Received MDS connection ("LPX64"); group %d\n", - obd->obd_name, lexp->exp_handle.h_cookie, group); - push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - rc = filter_read_groups(obd, group, 1); + rc = filter_read_groups(obd, data->ocd_group, 1); pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); if (rc != 0) { - CERROR("can't read group %u\n", group); + CERROR("can't read group %u\n", data->ocd_group); GOTO(cleanup, rc); } @@ -2824,7 +2868,6 @@ static int filter_connect(const struct lu_env *env, cleanup: if (rc) { class_disconnect(lexp); - lprocfs_exp_cleanup(lexp); *exp = NULL; } else { *exp = lexp; @@ -2954,11 +2997,12 @@ static int filter_destroy_export(struct obd_export *exp) lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd); target_destroy_export(exp); - ldlm_destroy_export(exp); - if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid)) - RETURN(0); + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + ldlm_destroy_export(exp); lut_client_free(exp); if (!exp->exp_obd->obd_replayable) @@ -2974,7 +3018,7 @@ static int filter_destroy_export(struct obd_export *exp) } if (!(exp->exp_flags & OBD_OPT_FORCE)) - filter_grant_sanity_check(exp->exp_obd, __FUNCTION__); + filter_grant_sanity_check(exp->exp_obd, __func__); RETURN(0); } @@ -2998,31 +3042,31 @@ static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp) group = 1 << 30; cfs_spin_lock(&filter->fo_llog_list_lock); cfs_list_for_each_entry(olg, &filter->fo_llog_list, olg_list) { - if (olg->olg_group <= worked) { + if (olg->olg_seq <= worked) { /* this group is already synced */ continue; } - if (group < olg->olg_group) { + if (group < olg->olg_seq) { /* we have group with smaller number to sync */ continue; } /* store current minimal group */ olg_min = olg; - group = olg->olg_group; + group = olg->olg_seq; } cfs_spin_unlock(&filter->fo_llog_list_lock); if (olg_min == NULL) break; - worked = olg_min->olg_group; + worked = olg_min->olg_seq; if (olg_min->olg_exp && (dexp == olg_min->olg_exp || dexp == NULL)) { int err; ctxt = llog_group_get_ctxt(olg_min, LLOG_MDS_OST_REPL_CTXT); if (ctxt) { - err = llog_sync(ctxt, olg_min->olg_exp); + err = llog_sync(ctxt, olg_min->olg_exp, 0); llog_ctxt_put(ctxt); if (err) { CERROR("error flushing logs to MDS: " @@ -3044,7 +3088,7 @@ static int filter_disconnect(struct obd_export *exp) class_export_get(exp); if (!(exp->exp_flags & OBD_OPT_FORCE)) - filter_grant_sanity_check(obd, __FUNCTION__); + filter_grant_sanity_check(obd, __func__); filter_grant_discard(exp); /* Flush any remaining cancel messages out to the target */ @@ -3078,49 +3122,62 @@ static void filter_revimp_update(struct obd_export *exp) EXIT; } -static int filter_ping(struct obd_export *exp) +static int filter_ping(const struct lu_env *env, struct obd_export *exp) { filter_fmd_expire(exp); return 0; } -struct dentry *__filter_oa2dentry(struct obd_device *obd, struct obdo *oa, +struct dentry *__filter_oa2dentry(struct obd_device *obd, struct ost_id *ostid, const char *what, int quiet) { struct dentry *dchild = NULL; - obd_gr group = 0; - if (oa->o_valid & OBD_MD_FLGROUP) - group = oa->o_gr; - - dchild = filter_fid2dentry(obd, NULL, group, oa->o_id); + dchild = filter_fid2dentry(obd, NULL, ostid->oi_seq, ostid->oi_id); if (IS_ERR(dchild)) { - CERROR("%s error looking up object: "LPU64":"LPU64"\n", - what, group, oa->o_id); + CERROR("%s error looking up object: "POSTID"\n", + what, ostid->oi_id, ostid->oi_seq); RETURN(dchild); } if (dchild->d_inode == NULL) { if (!quiet) - CERROR("%s: %s on non-existent object: "LPU64"\n", - obd->obd_name, what, oa->o_id); + CERROR("%s: %s on non-existent object: "POSTID" \n", + obd->obd_name, what, ostid->oi_id,ostid->oi_seq); f_dput(dchild); RETURN(ERR_PTR(-ENOENT)); } +#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2,7,50,0) + /* Try to correct for a bug in 2.1.0 (LU-221) that caused negative + * timestamps to appear to be in the far future, due old timestamp + * being stored on disk as an unsigned value. This fixes up any + * bad values stored on disk before returning them to the client, + * and ensures any timestamp updates are correct. LU-1042 */ + if (unlikely(LTIME_S(dchild->d_inode->i_atime) == LU221_BAD_TIME)) + LTIME_S(dchild->d_inode->i_atime) = 0; + if (unlikely(LTIME_S(dchild->d_inode->i_mtime) == LU221_BAD_TIME)) + LTIME_S(dchild->d_inode->i_mtime) = 0; + if (unlikely(LTIME_S(dchild->d_inode->i_ctime) == LU221_BAD_TIME)) + LTIME_S(dchild->d_inode->i_ctime) = 0; +#else +#warning "remove old LU-221/LU-1042 workaround code" +#endif + return dchild; } -static int filter_getattr(struct obd_export *exp, struct obd_info *oinfo) +static int filter_getattr(const struct lu_env *env, struct obd_export *exp, + struct obd_info *oinfo) { struct dentry *dentry = NULL; struct obd_device *obd; + __u64 curr_version; int rc = 0; ENTRY; - LASSERT(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP); - rc = filter_auth_capa(exp, NULL, oinfo->oi_oa->o_gr, + rc = filter_auth_capa(exp, NULL, oinfo->oi_oa->o_seq, oinfo_capa(oinfo), CAPA_OPC_META_READ); if (rc) RETURN(rc); @@ -3131,7 +3188,7 @@ static int filter_getattr(struct obd_export *exp, struct obd_info *oinfo) RETURN(-EINVAL); } - dentry = filter_oa2dentry(obd, oinfo->oi_oa); + dentry = filter_oa2dentry(obd, &oinfo->oi_oa->o_oi); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); @@ -3139,6 +3196,13 @@ static int filter_getattr(struct obd_export *exp, struct obd_info *oinfo) oinfo->oi_oa->o_valid = OBD_MD_FLID; obdo_from_inode(oinfo->oi_oa, dentry->d_inode, FILTER_VALID_FLAGS); + /* Store inode version in reply */ + curr_version = fsfilt_get_version(exp->exp_obd, dentry->d_inode); + if ((__s64)curr_version != -EOPNOTSUPP) { + oinfo->oi_oa->o_valid |= OBD_MD_FLDATAVERSION; + oinfo->oi_oa->o_data_version = curr_version; + } + f_dput(dentry); RETURN(rc); } @@ -3156,19 +3220,21 @@ int filter_update_fidea(struct obd_export *exp, struct inode *inode, struct filter_fid ff; if (!(oa->o_valid & OBD_MD_FLGROUP)) - oa->o_gr = 0; + oa->o_seq = 0; /* packing fid and converting it to LE for storing into EA. * Here ->o_stripe_idx should be filled by LOV and rest of * fields - by client. */ - ff.ff_fid.id = cpu_to_le64(oa->o_fid); - ff.ff_fid.f_type = cpu_to_le32(oa->o_stripe_idx); - ff.ff_fid.generation = cpu_to_le32(oa->o_generation); + ff.ff_parent.f_seq = cpu_to_le64(oa->o_parent_seq); + ff.ff_parent.f_oid = cpu_to_le32(oa->o_parent_oid); + /* XXX: we are ignoring o_parent_ver here, since this should + * be the same for all objects in this fileset. */ + ff.ff_parent.f_ver = cpu_to_le32(oa->o_stripe_idx); ff.ff_objid = cpu_to_le64(oa->o_id); - ff.ff_group = cpu_to_le64(oa->o_gr); + ff.ff_seq = cpu_to_le64(oa->o_seq); - CDEBUG(D_INODE, "storing filter fid EA ("LPU64"/%u/%u" - LPU64"/"LPU64")\n", oa->o_fid, oa->o_stripe_idx, - oa->o_generation, oa->o_id, oa->o_gr); + CDEBUG(D_INODE, "storing filter fid EA (parent "DFID" " + LPU64"/"LPU64")\n", PFID(&ff.ff_parent), oa->o_id, + oa->o_seq); rc = fsfilt_set_md(obd, inode, handle, &ff, sizeof(ff), "fid"); if (rc) @@ -3212,13 +3278,14 @@ int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry, *fcc = oa->o_lcookie; } if (ia_valid & (ATTR_SIZE | ATTR_UID | ATTR_GID)) { - DQUOT_INIT(inode); + unsigned long now = jiffies; /* Filter truncates and writes are serialized by * i_alloc_sem, see the comment in * filter_preprw_write.*/ if (ia_valid & ATTR_SIZE) down_write(&inode->i_alloc_sem); LOCK_INODE_MUTEX(inode); + fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem and i_mutex"); old_size = i_size_read(inode); } @@ -3281,6 +3348,11 @@ int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry, if (IS_ERR(handle)) GOTO(out_unlock, rc = PTR_ERR(handle)); } + + /* Locking order: i_mutex -> journal_lock -> dqptr_sem. LU-952 */ + if (ia_valid & (ATTR_SIZE | ATTR_UID | ATTR_GID)) + ll_vfs_dq_init(inode); + if (oa->o_valid & OBD_MD_FLFLAGS) { rc = fsfilt_iocontrol(exp->exp_obd, dentry, FSFILT_IOC_SETFLAGS, (long)&oa->o_flags); @@ -3296,11 +3368,14 @@ int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry, if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_CREDITS)) fsfilt_extend(exp->exp_obd, inode, 0, handle); - /* The truncate might have used up our transaction credits. Make - * sure we have one left for the last_rcvd update. */ - err = fsfilt_extend(exp->exp_obd, inode, 1, handle); + /* The truncate might have used up our transaction credits. Make sure + * we have two left for the last_rcvd and VBR inode version updates. */ + err = fsfilt_extend(exp->exp_obd, inode, 2, handle); + + /* Update inode version only if data has changed => size has changed */ + rc = filter_finish_transno(exp, ia_valid & ATTR_SIZE ? inode : NULL, + oti, rc, sync); - rc = filter_finish_transno(exp, inode, oti, rc, sync); if (sync) { filter_cancel_cookies_cb(exp->exp_obd, 0, fcc, rc); fcc = NULL; @@ -3351,8 +3426,8 @@ out_unlock: } /* this is called from filter_truncate() until we have filter_punch() */ -int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, - struct obd_trans_info *oti) +int filter_setattr(const struct lu_env *env, struct obd_export *exp, + struct obd_info *oinfo, struct obd_trans_info *oti) { struct obdo *oa = oinfo->oi_oa; struct lustre_capa *capa = oinfo_capa(oinfo); @@ -3366,21 +3441,20 @@ int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, int rc; ENTRY; - if (oa->o_valid & OBD_FL_TRUNC) + if (oinfo->oi_flags & OBD_FL_PUNCH) opc |= CAPA_OPC_OSS_TRUNC; - LASSERT(oa->o_valid & OBD_MD_FLGROUP); - rc = filter_auth_capa(exp, NULL, oa->o_gr, capa, opc); + rc = filter_auth_capa(exp, NULL, oa->o_seq, capa, opc); if (rc) RETURN(rc); if (oa->o_valid & (OBD_MD_FLUID | OBD_MD_FLGID)) { - rc = filter_capa_fixoa(exp, oa, oa->o_gr, capa); + rc = filter_capa_fixoa(exp, oa, oa->o_seq, capa); if (rc) RETURN(rc); } - osc_build_res_name(oa->o_id, oa->o_gr, &res_id); + osc_build_res_name(oa->o_id, oa->o_seq, &res_id); /* This would be very bad - accidentally truncating a file when * changing the time or similar - bug 12203. */ if (oa->o_valid & OBD_MD_FLSIZE && @@ -3388,20 +3462,19 @@ int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, static char mdsinum[48]; if (oa->o_valid & OBD_MD_FLFID) - snprintf(mdsinum, sizeof(mdsinum) - 1, - " of inode "LPU64"/%u", oa->o_fid, - oa->o_generation); + snprintf(mdsinum, sizeof(mdsinum) - 1, " of inode "DFID, + oa->o_parent_seq, oa->o_parent_oid, + oa->o_parent_ver); else mdsinum[0] = '\0'; - CERROR("%s: setattr from %s trying to truncate objid "LPU64 - " %s\n", - exp->exp_obd->obd_name, obd_export_nid2str(exp), - oa->o_id, mdsinum); + CERROR("%s: setattr from %s trying to truncate objid "POSTID + "%s\n", exp->exp_obd->obd_name, obd_export_nid2str(exp), + oa->o_id, oa->o_seq, mdsinum); RETURN(-EPERM); } - dentry = __filter_oa2dentry(exp->exp_obd, oa, __FUNCTION__, 1); + dentry = __filter_oa2dentry(exp->exp_obd, &oinfo->oi_oa->o_oi, __func__, 1); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); @@ -3417,8 +3490,10 @@ int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, */ if (oa->o_valid & (OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME)) { + unsigned long now = jiffies; down_write(&dentry->d_inode->i_alloc_sem); - fmd = filter_fmd_get(exp, oa->o_id, oa->o_gr); + fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem"); + fmd = filter_fmd_get(exp, oa->o_id, oa->o_seq); if (fmd && fmd->fmd_mactime_xid < oti->oti_xid) fmd->fmd_mactime_xid = oti->oti_xid; filter_fmd_put(exp, fmd); @@ -3446,6 +3521,8 @@ int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS | OBD_MD_FLUID | OBD_MD_FLGID); + filter_counter_incr(exp, LPROC_FILTER_STATS_SETATTR, + oti ? oti->oti_jobid : NULL, 1); EXIT; out_unlock: f_dput(dentry); @@ -3504,12 +3581,12 @@ static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp, LASSERT((*lsmp)->lsm_object_id); } - (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES; + (*lsmp)->lsm_maxbytes = exp->exp_obd->u.obt.obt_sb->s_maxbytes; RETURN(lsm_size); } -/* caller must hold fo_create_locks[oa->o_gr] */ +/* caller must hold fo_create_locks[oa->o_seq] */ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa, struct filter_obd *filter) { @@ -3519,24 +3596,21 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa, int skip_orphan; ENTRY; - LASSERT(oa); - LASSERT_MDS_GROUP(oa->o_gr); - LASSERT(oa->o_valid & OBD_MD_FLGROUP); - LASSERT(down_trylock(&filter->fo_create_locks[oa->o_gr]) != 0); + LASSERT_MUTEX_LOCKED(&filter->fo_create_locks[oa->o_seq]); memset(&doa, 0, sizeof(doa)); doa.o_valid |= OBD_MD_FLGROUP; - doa.o_gr = oa->o_gr; + doa.o_seq = oa->o_seq; doa.o_mode = S_IFREG; - if (!cfs_test_bit(doa.o_gr, &filter->fo_destroys_in_progress)) { + if (!cfs_test_bit(doa.o_seq, &filter->fo_destroys_in_progress)) { CERROR("%s:["LPU64"] destroys_in_progress already cleared\n", - exp->exp_obd->obd_name, doa.o_gr); + exp->exp_obd->obd_name, doa.o_seq); RETURN(0); } - last = filter_last_id(filter, doa.o_gr); + last = filter_last_id(filter, doa.o_seq); skip_orphan = !!(exp->exp_connect_flags & OBD_CONNECT_SKIP_ORPHAN); @@ -3546,7 +3620,7 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa, for (id = last; id > oa->o_id; id--) { doa.o_id = id; - rc = filter_destroy(exp, &doa, NULL, NULL, NULL, NULL); + rc = filter_destroy(NULL, exp, &doa, NULL, NULL, NULL, NULL); if (rc && rc != -ENOENT) /* this is pretty fatal... */ CEMERG("error destroying precreate objid "LPU64": %d\n", id, rc); @@ -3554,32 +3628,39 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa, /* update last_id on disk periodically so that if we restart * we don't need to re-scan all of the just-deleted objects. */ if ((id & 511) == 0 && !skip_orphan) { - filter_set_last_id(filter, id - 1, doa.o_gr); - filter_update_last_objid(exp->exp_obd, doa.o_gr, 0); + filter_set_last_id(filter, id - 1, doa.o_seq); + filter_update_last_objid(exp->exp_obd, doa.o_seq, 0); } } CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n", - exp->exp_obd->obd_name, doa.o_gr, oa->o_id); + exp->exp_obd->obd_name, doa.o_seq, oa->o_id); if (!skip_orphan) { - filter_set_last_id(filter, id, doa.o_gr); - rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1); + filter_set_last_id(filter, id, doa.o_seq); + rc = filter_update_last_objid(exp->exp_obd, doa.o_seq, 1); } else { - /* don't reuse orphan object, return last used objid */ + /* + * We have destroyed orphan objects, but don't want to reuse + * them. Therefore we don't reset last_id to the last created + * objects. Instead, we report back to the MDS the object id + * of the last orphan, so that the MDS can restart allocating + * objects from this id + 1 and thus skip the whole orphan + * object id range + */ oa->o_id = last; rc = 0; } - cfs_clear_bit(doa.o_gr, &filter->fo_destroys_in_progress); + cfs_clear_bit(doa.o_seq, &filter->fo_destroys_in_progress); RETURN(rc); } static int filter_precreate(struct obd_device *obd, struct obdo *oa, - obd_gr group, int *num); + obd_seq group, int *num); /* returns a negative error or a nonnegative number of files to create */ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa, - obd_gr group, struct obd_trans_info *oti) + obd_seq group, struct obd_trans_info *oti) { struct obd_device *obd = exp->exp_obd; struct filter_obd *filter = &obd->u.filter; @@ -3597,18 +3678,18 @@ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa, } /* This causes inflight precreates to abort and drop lock */ cfs_set_bit(group, &filter->fo_destroys_in_progress); - cfs_down(&filter->fo_create_locks[group]); + cfs_mutex_lock(&filter->fo_create_locks[group]); if (!cfs_test_bit(group, &filter->fo_destroys_in_progress)) { CERROR("%s:["LPU64"] destroys_in_progress already cleared\n", exp->exp_obd->obd_name, group); - cfs_up(&filter->fo_create_locks[group]); + cfs_mutex_unlock(&filter->fo_create_locks[group]); RETURN(0); } diff = oa->o_id - last; CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n", last, diff); - if (-diff > OST_MAX_PRECREATE) { + if (-diff > (OST_MAX_PRECREATE * 3) / 2) { CERROR("%s: ignoring bogus orphan destroy request: " "obdid "LPU64" last_id "LPU64"\n", obd->obd_name, oa->o_id, last); @@ -3626,14 +3707,14 @@ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa, cfs_clear_bit(group, &filter->fo_destroys_in_progress); } } else { - cfs_down(&filter->fo_create_locks[group]); + cfs_mutex_lock(&filter->fo_create_locks[group]); if (oti->oti_conn_cnt < exp->exp_conn_cnt) { CERROR("%s: dropping old precreate request\n", obd->obd_name); GOTO(out, rc = 0); } /* only precreate if group == 0 and o_id is specfied */ - if (!filter_group_is_mds(group) || oa->o_id == 0) + if (!fid_seq_is_mdt(group) || oa->o_id == 0) diff = 1; else diff = oa->o_id - filter_last_id(filter, group); @@ -3648,22 +3729,24 @@ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa, oa->o_id = filter_last_id(&obd->u.filter, group); rc = filter_precreate(obd, oa, group, &diff); oa->o_id = filter_last_id(&obd->u.filter, group); - oa->o_gr = group; - oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP; + oa->o_seq = group; + oa->o_valid |= (OBD_MD_FLID | OBD_MD_FLGROUP); GOTO(out, rc); } /* else diff == 0 */ GOTO(out, rc = 0); out: - cfs_up(&filter->fo_create_locks[group]); + cfs_mutex_unlock(&filter->fo_create_locks[group]); return rc; } -static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs, - __u64 max_age, __u32 flags) +static int filter_statfs(const struct lu_env *env, struct obd_export *exp, + struct obd_statfs *osfs, __u64 max_age, __u32 flags) { + struct obd_device *obd = class_exp2obd(exp); struct filter_obd *filter = &obd->u.filter; int blockbits = obd->u.obt.obt_sb->s_blocksize_bits; + struct lr_server_data *lsd = class_server_data(obd); int rc; ENTRY; @@ -3681,22 +3764,19 @@ static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs, filter->fo_tot_pending, osfs->os_bfree << blockbits, osfs->os_bavail << blockbits); - filter_grant_sanity_check(obd, __FUNCTION__); + filter_grant_sanity_check(obd, __func__); osfs->os_bavail -= min(osfs->os_bavail, GRANT_FOR_LLOG(obd) + ((filter->fo_tot_dirty + filter->fo_tot_pending + osfs->os_bsize - 1) >> blockbits)); - if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC)) { - struct lr_server_data *lsd = class_server_data(obd); - int index = le32_to_cpu(lsd->lsd_ost_index); + if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOSPC, + le32_to_cpu(lsd->lsd_ost_index))) + osfs->os_bfree = osfs->os_bavail = 2; - if (obd_fail_val == -1 || - index == obd_fail_val) - osfs->os_bfree = osfs->os_bavail = 2; - else if (obd_fail_loc & OBD_FAIL_ONCE) - obd_fail_loc &= ~OBD_FAILED; /* reset flag */ - } + if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOINO, + le32_to_cpu(lsd->lsd_ost_index))) + osfs->os_ffree = 0; /* set EROFS to state field if FS is mounted as RDONLY. The goal is to * stop creating files on MDS if OST is not good shape to create @@ -3761,11 +3841,12 @@ static __u64 filter_calc_free_inodes(struct obd_device *obd) * Caller must hold fo_create_locks[group] */ static int filter_precreate(struct obd_device *obd, struct obdo *oa, - obd_gr group, int *num) + obd_seq group, int *num) { struct dentry *dchild = NULL, *dparent = NULL; struct filter_obd *filter; struct obd_statfs *osfs; + struct iattr iattr; int err = 0, rc = 0, recreate_obj = 0, i; cfs_time_t enough_time = cfs_time_shift(DISK_TIMEOUT/2); __u64 os_ffree; @@ -3775,7 +3856,7 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, filter = &obd->u.filter; - LASSERT(down_trylock(&filter->fo_create_locks[group]) != 0); + LASSERT_MUTEX_LOCKED(&filter->fo_create_locks[group]); OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_PRECREATE, obd_timeout / 2); @@ -3786,13 +3867,21 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, OBD_ALLOC(osfs, sizeof(*osfs)); if (osfs == NULL) RETURN(-ENOMEM); - rc = filter_statfs(obd, osfs, cfs_time_current_64() - CFS_HZ, + rc = filter_statfs(NULL, obd->obd_self_export, osfs, + cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS), 0); if (rc == 0 && osfs->os_bavail < (osfs->os_blocks >> 10)) { CDEBUG(D_RPCTRACE,"%s: not enough space for create " LPU64"\n", obd->obd_name, osfs->os_bavail << - filter->fo_vfsmnt->mnt_sb->s_blocksize_bits); + obd->u.obt.obt_vfsmnt->mnt_sb->s_blocksize_bits); *num = 0; + if (oa->o_valid & OBD_MD_FLFLAGS) + oa->o_flags |= OBD_FL_NOSPC_BLK; + else { + oa->o_valid |= OBD_MD_FLFLAGS; + oa->o_flags = OBD_FL_NOSPC_BLK; + } + rc = -ENOSPC; } OBD_FREE(osfs, sizeof(*osfs)); @@ -3826,6 +3915,19 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, } else next_id = filter_last_id(filter, group) + 1; + /* Don't create objects beyond the valid range for this SEQ */ + if (unlikely(fid_seq_is_mdt0(group) && + next_id >= IDIF_MAX_OID)) { + CERROR("%s:"POSTID" hit the IDIF_MAX_OID (1<<48)!\n", + obd->obd_name, next_id, group); + GOTO(cleanup, rc = -ENOSPC); + } else if (unlikely(!fid_seq_is_mdt0(group) && + next_id >= OBIF_MAX_OID)) { + CERROR("%s:"POSTID" hit the OBIF_MAX_OID (1<<32)!\n", + obd->obd_name, next_id, group); + GOTO(cleanup, rc = -ENOSPC); + } + dparent = filter_parent_lock(obd, group, next_id); if (IS_ERR(dparent)) GOTO(cleanup, rc = PTR_ERR(dparent)); @@ -3847,7 +3949,7 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, } else { /* Use these existing objects if they are * zero length. */ - if (dchild->d_inode->i_size == 0) { + if (i_size_read(dchild->d_inode) == 0) { rc = filter_use_existing_obj(obd,dchild, &handle, &cleanup_phase); if (rc == 0) @@ -3871,7 +3973,7 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, GOTO(cleanup, rc = PTR_ERR(handle)); cleanup_phase = 3; - CDEBUG(D_INODE, "%s: filter_precreate(od->o_gr="LPU64 + CDEBUG(D_INODE, "%s: filter_precreate(od->o_seq="LPU64 ",od->o_id="LPU64")\n", obd->obd_name, group, next_id); @@ -3881,12 +3983,24 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, rc = ll_vfs_create(dparent->d_inode, dchild, S_IFREG | S_ISUID | S_ISGID | 0666, NULL); if (rc) { - CERROR("create failed rc = %d\n", rc); + CWARN("%s: create failed: rc = %d\n", obd->obd_name,rc); if (rc == -ENOSPC) { os_ffree = filter_calc_free_inodes(obd); - if (os_ffree != -1) - CERROR("%s: free inode "LPU64"\n", - obd->obd_name, os_ffree); + if (os_ffree == -1) + GOTO(cleanup, rc); + + if (obd->obd_osfs.os_bavail < + (obd->obd_osfs.os_blocks >> 10)) { + if (oa->o_valid & OBD_MD_FLFLAGS) { + oa->o_flags |= OBD_FL_NOSPC_BLK; + } else { + oa->o_valid |= OBD_MD_FLFLAGS; + oa->o_flags = OBD_FL_NOSPC_BLK; + } + + CWARN("%s: free inode "LPU64"\n", + obd->obd_name, os_ffree); + } } GOTO(cleanup, rc); } @@ -3896,12 +4010,27 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, dchild->d_inode->i_ino); set_last_id: + /* Initialize a/c/m time so any client timestamp will always + * be newer and update the inode. ctime = 0 is also handled + * specially in fsfilt_ext3_setattr(). See LU-221, LU-1042 */ + iattr.ia_valid = ATTR_ATIME | ATTR_MTIME | ATTR_CTIME; + LTIME_S(iattr.ia_atime) = 0; + LTIME_S(iattr.ia_mtime) = 0; + LTIME_S(iattr.ia_ctime) = 0; + err = fsfilt_setattr(obd, dchild, handle, &iattr, 0); + if (err) + CWARN("%s: unable to initialize a/c/m time of newly " + "created object %.*s: rc = %d\n", + obd->obd_name, dchild->d_name.len, + dchild->d_name.name, err); + if (!recreate_obj) { filter_set_last_id(filter, next_id, group); err = filter_update_last_objid(obd, group, 0); if (err) - CERROR("unable to write lastobjid " - "but file created\n"); + CERROR("%s: unable to write lastobjid " + "but file created: rc = %d\n", + obd->obd_name, err); } cleanup: @@ -3924,6 +4053,7 @@ set_last_id: if (rc) break; if (cfs_time_after(jiffies, enough_time)) { + i++; CDEBUG(D_RPCTRACE, "%s: precreate slow - want %d got %d \n", obd->obd_name, *num, i); @@ -3933,41 +4063,46 @@ set_last_id: *num = i; CDEBUG(D_RPCTRACE, - "%s: created %d objects for group "LPU64": "LPU64" rc %d\n", - obd->obd_name, i, group, filter->fo_last_objids[group], rc); + "%s: created %d objects for group "POSTID" rc %d\n", + obd->obd_name, i, filter->fo_last_objids[group], group, rc); RETURN(rc); } -static int filter_create(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md **ea, struct obd_trans_info *oti) +int filter_create(const struct lu_env *env, struct obd_export *exp, + struct obdo *oa, struct lov_stripe_md **ea, + struct obd_trans_info *oti) { struct obd_device *obd = exp->exp_obd; struct filter_export_data *fed; struct filter_obd *filter; struct lvfs_run_ctxt saved; struct lov_stripe_md *lsm = NULL; - int rc = 0, diff, group = oa->o_gr; + int rc = 0, diff; ENTRY; - CDEBUG(D_INODE, "%s: filter_create(od->o_gr="LPU64",od->o_id=" - LPU64")\n", obd->obd_name, oa->o_gr, oa->o_id); - - if (!(oa->o_valid & OBD_MD_FLGROUP)) { - CERROR("!!! nid %s sent invalid object group %d\n", - obd_export_nid2str(exp), group); - RETURN(-EINVAL); - } + CDEBUG(D_INODE, "%s: filter_create(group="LPU64",id=" + LPU64")\n", obd->obd_name, oa->o_seq, oa->o_id); fed = &exp->exp_filter_data; filter = &obd->u.filter; - if (fed->fed_group != group) { - CERROR("!!! this export (nid %s) used object group %d " - "earlier; now it's trying to use group %d! This could " - "be a bug in the MDS. Please report to " - "http://bugzilla.lustre.org/\n", - obd_export_nid2str(exp), fed->fed_group, group); + /* 1.8 client doesn't carry the ocd_group with connect request, + * so the fed_group will always be zero for 1.8 client. */ + if (!(exp->exp_connect_flags & OBD_CONNECT_FULL20)) { + if (oa->o_seq != FID_SEQ_OST_MDT0 && + oa->o_seq != FID_SEQ_LLOG && + oa->o_seq != FID_SEQ_ECHO) { + CERROR("The request from older client has invalid" + " group "LPU64"!\n", oa->o_seq); + RETURN(-EINVAL); + } + } else if (fed->fed_group != oa->o_seq) { + CERROR("%s: this export (nid %s) used object group %d " + "earlier; now it's trying to use group "LPU64"!" + " This could be a bug in the MDS. Please report to " + "http://bugzilla.lustre.org/\n", obd->obd_name, + obd_export_nid2str(exp), fed->fed_group, oa->o_seq); RETURN(-ENOTUNIQ); } @@ -3985,19 +4120,19 @@ static int filter_create(struct obd_export *exp, struct obdo *oa, if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_RECREATE_OBJS)) { - if (oa->o_id > filter_last_id(filter, oa->o_gr)) { + if (!obd->obd_recovering || + oa->o_id > filter_last_id(filter, oa->o_seq)) { CERROR("recreate objid "LPU64" > last id "LPU64"\n", - oa->o_id, filter_last_id(filter, - oa->o_gr)); + oa->o_id, filter_last_id(filter, oa->o_seq)); rc = -EINVAL; } else { diff = 1; - cfs_down(&filter->fo_create_locks[oa->o_gr]); - rc = filter_precreate(obd, oa, oa->o_gr, &diff); - cfs_up(&filter->fo_create_locks[oa->o_gr]); + cfs_mutex_lock(&filter->fo_create_locks[oa->o_seq]); + rc = filter_precreate(obd, oa, oa->o_seq, &diff); + cfs_mutex_unlock(&filter->fo_create_locks[oa->o_seq]); } } else { - rc = filter_handle_precreate(exp, oa, oa->o_gr, oti); + rc = filter_handle_precreate(exp, oa, oa->o_seq, oti); } pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); @@ -4015,9 +4150,10 @@ static int filter_create(struct obd_export *exp, struct obdo *oa, RETURN(rc); } -int filter_destroy(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *md, struct obd_trans_info *oti, - struct obd_export *md_exp, void *capa) +int filter_destroy(const struct lu_env *env, struct obd_export *exp, + struct obdo *oa, struct lov_stripe_md *md, + struct obd_trans_info *oti, struct obd_export *md_exp, + void *capa) { unsigned int qcids[MAXQUOTAS] = {0, 0}; struct obd_device *obd; @@ -4029,10 +4165,10 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, struct llog_cookie *fcc = NULL; int rc, rc2, cleanup_phase = 0, sync = 0; struct iattr iattr; + unsigned long now; ENTRY; - LASSERT(oa->o_valid & OBD_MD_FLGROUP); - rc = filter_auth_capa(exp, NULL, oa->o_gr, + rc = filter_auth_capa(exp, NULL, oa->o_seq, (struct lustre_capa *)capa, CAPA_OPC_OSS_DESTROY); if (rc) RETURN(rc); @@ -4043,26 +4179,26 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); cleanup_phase = 1; - CDEBUG(D_INODE, "%s: filter_destroy(od->o_gr="LPU64",od->o_id=" - LPU64")\n", obd->obd_name, oa->o_gr, oa->o_id); + CDEBUG(D_INODE, "%s: filter_destroy(group="LPU64",oid=" + LPU64")\n", obd->obd_name, oa->o_seq, oa->o_id); - dchild = filter_fid2dentry(obd, NULL, oa->o_gr, oa->o_id); + dchild = filter_fid2dentry(obd, NULL, oa->o_seq, oa->o_id); if (IS_ERR(dchild)) GOTO(cleanup, rc = PTR_ERR(dchild)); cleanup_phase = 2; if (dchild->d_inode == NULL) { - CDEBUG(D_INODE, "destroying non-existent object "LPU64"\n", - oa->o_id); + CDEBUG(D_INODE, "destroying non-existent object "POSTID"\n", + oa->o_id, oa->o_seq); /* If object already gone, cancel cookie right now */ if (oa->o_valid & OBD_MD_FLCOOKIE) { struct llog_ctxt *ctxt; struct obd_llog_group *olg; - olg = filter_find_olg(obd, oa->o_gr); + olg = filter_find_olg(obd, oa->o_seq); if (!olg) { CERROR(" %s: can not find olg of group %d\n", - obd->obd_name, (int)oa->o_gr); + obd->obd_name, (int)oa->o_seq); GOTO(cleanup, rc = PTR_ERR(olg)); } fcc = &oa->o_lcookie; @@ -4074,7 +4210,7 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, GOTO(cleanup, rc = -ENOENT); } - rc = filter_prepare_destroy(obd, oa->o_id, oa->o_gr, &lockh); + rc = filter_prepare_destroy(obd, oa->o_id, oa->o_seq, &lockh); if (rc) GOTO(cleanup, rc); @@ -4084,7 +4220,6 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, if (fcc != NULL) *fcc = oa->o_lcookie; } - DQUOT_INIT(dchild->d_inode); /* we're gonna truncate it first in order to avoid possible deadlock: * P1 P2 @@ -4098,8 +4233,10 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, * between page lock, i_mutex & starting new journal handle. * (see bug 20321) -johann */ + now = jiffies; down_write(&dchild->d_inode->i_alloc_sem); LOCK_INODE_MUTEX(dchild->d_inode); + fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem and i_mutex"); /* VBR: version recovery check */ rc = filter_version_get_check(exp, oti, dchild->d_inode); @@ -4117,6 +4254,9 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, GOTO(cleanup, rc = PTR_ERR(handle)); } + /* Locking order: i_mutex -> journal_lock -> dqptr_sem. LU-952 */ + ll_vfs_dq_init(dchild->d_inode); + iattr.ia_valid = ATTR_SIZE; iattr.ia_size = 0; rc = fsfilt_setattr(obd, dchild, handle, &iattr, 1); @@ -4132,7 +4272,7 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, * here, and not while truncating above. That avoids holding the * parent lock for a long time during truncate, which can block other * threads from doing anything to objects in that directory. bug 7171 */ - dparent = filter_parent_lock(obd, oa->o_gr, oa->o_id); + dparent = filter_parent_lock(obd, oa->o_seq, oa->o_id); if (IS_ERR(dparent)) GOTO(cleanup, rc = PTR_ERR(dparent)); cleanup_phase = 3; /* filter_parent_unlock */ @@ -4146,12 +4286,12 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa, cleanup_phase = 4; /* fsfilt_commit */ /* Quota release need uid/gid of inode */ - obdo_from_inode(oa, dchild->d_inode, OBD_MD_FLUID|OBD_MD_FLGID); + obdo_from_inode(oa, dchild->d_inode, OBD_MD_FLUID | OBD_MD_FLGID); - filter_fmd_drop(exp, oa->o_id, oa->o_gr); + filter_fmd_drop(exp, oa->o_id, oa->o_seq); /* this drops dchild->d_inode->i_mutex unconditionally */ - rc = filter_destroy_internal(obd, oa->o_id, oa->o_gr, dparent, dchild); + rc = filter_destroy_internal(obd, oa->o_id, oa->o_seq, dparent, dchild); EXIT; cleanup: @@ -4207,8 +4347,8 @@ cleanup: } /* NB start and end are used for punch, but not truncate */ -static int filter_truncate(struct obd_export *exp, struct obd_info *oinfo, - struct obd_trans_info *oti, +static int filter_truncate(const struct lu_env *env, struct obd_export *exp, + struct obd_info *oinfo, struct obd_trans_info *oti, struct ptlrpc_request_set *rqset) { int rc; @@ -4221,43 +4361,41 @@ static int filter_truncate(struct obd_export *exp, struct obd_info *oinfo, } CDEBUG(D_INODE, "calling truncate for object "LPU64", valid = "LPX64 - ", o_size = "LPD64"\n", oinfo->oi_oa->o_id, - oinfo->oi_oa->o_valid, oinfo->oi_policy.l_extent.start); + ", o_size = "LPD64"\n", oinfo->oi_oa->o_id,oinfo->oi_oa->o_valid, + oinfo->oi_policy.l_extent.start); oinfo->oi_oa->o_size = oinfo->oi_policy.l_extent.start; - oinfo->oi_oa->o_valid |= OBD_FL_TRUNC; - rc = filter_setattr(exp, oinfo, oti); - oinfo->oi_oa->o_valid &= ~OBD_FL_TRUNC; + rc = filter_setattr(env, exp, oinfo, oti); RETURN(rc); } -static int filter_sync(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *lsm, obd_off start, obd_off end, - void *capa) +static int filter_sync(const struct lu_env *env, struct obd_export *exp, + struct obd_info *oinfo, obd_off start, obd_off end, + struct ptlrpc_request_set *set) { struct lvfs_run_ctxt saved; - struct filter_obd *filter; + struct obd_device_target *obt; struct dentry *dentry; int rc, rc2; ENTRY; - LASSERT(oa->o_valid & OBD_MD_FLGROUP); - rc = filter_auth_capa(exp, NULL, oa->o_gr, - (struct lustre_capa *)capa, CAPA_OPC_OSS_WRITE); + rc = filter_auth_capa(exp, NULL, oinfo->oi_oa->o_seq, + (struct lustre_capa *)oinfo->oi_capa, + CAPA_OPC_OSS_WRITE); if (rc) RETURN(rc); - filter = &exp->exp_obd->u.filter; + obt = &exp->exp_obd->u.obt; /* An objid of zero is taken to mean "sync whole filesystem" */ - if (!oa || !(oa->o_valid & OBD_MD_FLID)) { - rc = fsfilt_sync(exp->exp_obd, filter->fo_obt.obt_sb); + if (!oinfo->oi_oa || !(oinfo->oi_oa->o_valid & OBD_MD_FLID)) { + rc = fsfilt_sync(exp->exp_obd, obt->obt_sb); /* Flush any remaining cancel messages out to the target */ filter_sync_llogs(exp->exp_obd, exp); RETURN(rc); } - dentry = filter_oa2dentry(exp->exp_obd, oa); + dentry = filter_oa2dentry(exp->exp_obd, &oinfo->oi_oa->o_oi); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); @@ -4268,7 +4406,7 @@ static int filter_sync(struct obd_export *exp, struct obdo *oa, rc = filemap_fdatawrite(dentry->d_inode->i_mapping); if (rc == 0) { /* just any file to grab fsync method - "file" arg unused */ - struct file *file = filter->fo_rcvd_filp; + struct file *file = obt->obt_rcvd_filp; if (file->f_op && file->f_op->fsync) rc = file->f_op->fsync(NULL, dentry, 1); @@ -4279,17 +4417,18 @@ static int filter_sync(struct obd_export *exp, struct obdo *oa, } UNLOCK_INODE_MUTEX(dentry->d_inode); - oa->o_valid = OBD_MD_FLID; - obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS); + oinfo->oi_oa->o_valid = OBD_MD_FLID; + obdo_from_inode(oinfo->oi_oa, dentry->d_inode, FILTER_VALID_FLAGS); pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); + filter_counter_incr(exp, LPROC_FILTER_STATS_SYNC, oinfo->oi_jobid, 1); f_dput(dentry); RETURN(rc); } -static int filter_get_info(struct obd_export *exp, __u32 keylen, - void *key, __u32 *vallen, void *val, +static int filter_get_info(const struct lu_env *env, struct obd_export *exp, + __u32 keylen, void *key, __u32 *vallen, void *val, struct lov_stripe_md *lsm) { struct obd_device *obd; @@ -4349,8 +4488,8 @@ static int filter_get_info(struct obd_export *exp, __u32 keylen, RETURN(0); } - dentry = __filter_oa2dentry(exp->exp_obd, &fm_key->oa, - __FUNCTION__, 1); + dentry = __filter_oa2dentry(exp->exp_obd, &fm_key->oa.o_oi, + __func__, 1); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); @@ -4364,6 +4503,12 @@ static int filter_get_info(struct obd_export *exp, __u32 keylen, RETURN(rc); } + if (KEY_IS(KEY_SYNC_LOCK_CANCEL)) { + *((__u32 *) val) = obd->u.filter.fo_sync_lock_cancel; + *vallen = sizeof(__u32); + RETURN(0); + } + CDEBUG(D_IOCTL, "invalid key\n"); RETURN(-EINVAL); } @@ -4424,14 +4569,14 @@ static int filter_set_mds_conn(struct obd_export *exp, void *val) else group = 0; /* default value */ - LASSERT_MDS_GROUP(group); + LASSERT_SEQ_IS_MDT(group); rc = filter_setup_llog_group(exp, obd, group); if (rc) goto out; - if (group == FILTER_GROUP_MDS0) { + if (group == FID_SEQ_OST_MDT0) { /* setup llog group 1 for interop */ - filter_setup_llog_group(exp, obd, FILTER_GROUP_LLOG); + filter_setup_llog_group(exp, obd, FID_SEQ_LLOG); } lquota_setinfo(filter_quota_interface_ref, obd, exp); @@ -4439,7 +4584,8 @@ out: RETURN(rc); } -static int filter_set_info_async(struct obd_export *exp, __u32 keylen, +static int filter_set_info_async(const struct lu_env *env, + struct obd_export *exp, __u32 keylen, void *key, __u32 vallen, void *val, struct ptlrpc_request_set *set) { @@ -4515,8 +4661,8 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, CDEBUG(D_HA, "syncing ost %s\n", obd->obd_name); rc = fsfilt_sync(obd, obd->u.obt.obt_sb); - lvfs_set_rdonly(obd, obd->u.obt.obt_sb); - RETURN(0); + rc = lvfs_set_rdonly(obd, obd->u.obt.obt_sb); + RETURN(rc); } case OBD_IOC_CATLOGLIST: { @@ -4548,7 +4694,7 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, RETURN(0); } -static int filter_health_check(struct obd_device *obd) +static int filter_health_check(const struct lu_env *env, struct obd_device *obd) { #ifdef USE_HEALTH_CHECK_WRITE struct filter_obd *filter = &obd->u.filter; @@ -4596,6 +4742,24 @@ static int filter_process_config(struct obd_device *obd, obd_count len, return rc; } +static int filter_notify(struct obd_device *obd, + struct obd_device *unused, + enum obd_notify_event ev, void *data) +{ + switch (ev) { + case OBD_NOTIFY_CONFIG: + LASSERT(obd->obd_no_conn); + cfs_spin_lock(&obd->obd_dev_lock); + obd->obd_no_conn = 0; + cfs_spin_unlock(&obd->obd_dev_lock); + break; + default: + CDEBUG(D_INFO, "%s: Unhandled notification %#x\n", + obd->obd_name, ev); + } + return 0; +} + static struct lvfs_callback_ops filter_lvfs_ops = { l_fid2dentry: filter_lvfs_fid2dentry, }; @@ -4619,7 +4783,6 @@ static struct obd_ops filter_obd_ops = { .o_create = filter_create, .o_setattr = filter_setattr, .o_destroy = filter_destroy, - .o_brw = filter_brw, .o_punch = filter_truncate, .o_sync = filter_sync, .o_preprw = filter_preprw, @@ -4630,6 +4793,7 @@ static struct obd_ops filter_obd_ops = { .o_iocontrol = filter_iocontrol, .o_health_check = filter_health_check, .o_process_config = filter_process_config, + .o_notify = filter_notify, }; quota_interface_t *filter_quota_interface_ref; @@ -4641,8 +4805,8 @@ static int __init obdfilter_init(void) int rc, i; /** sanity check for group<->mdsno conversion */ - for (i = 0; i < 32; i++) - LASSERT(objgrp_to_mdsno(mdt_to_obd_objgrp(i)) == i); + for (i = 0; i < MAX_MDT_COUNT; i++) + LASSERT(objseq_to_mdsno(mdt_to_obd_objseq(i)) == i); lprocfs_filter_init_vars(&lvars);