#include "filter_internal.h"
/* Group 0 is no longer a legal group, to catch uninitialized IDs */
-#define FILTER_MIN_GROUPS FILTER_GROUP_MDS0
+#define FILTER_MIN_GROUPS FILTER_GROUP_MDS1_N_BASE
static struct lvfs_callback_ops filter_lvfs_ops;
cfs_mem_cache_t *ll_fmd_cachep;
err = -EINVAL;
} else {
if (!force_sync)
- force_sync = fsfilt_add_journal_cb(exp->exp_obd,
+ force_sync = fsfilt_add_journal_cb(exp->exp_obd,
last_rcvd,
oti->oti_handle,
filter_commit_cb,
struct obd_export *exp,
void *client_nid)
{
- struct filter_export_data *fed = &exp->exp_filter_data;
int rc, newnid = 0;
ENTRY;
- init_brw_stats(&fed->fed_brw_stats);
-
if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
/* Self-export gets no proc entry */
RETURN(0);
rc = lprocfs_exp_setup(exp, client_nid, &newnid);
- if (rc)
+ if (rc) {
+ /* Mask error for already created
+ * /proc entries */
+ if (rc == -EALREADY)
+ rc = 0;
RETURN(rc);
+ }
if (newnid) {
struct nid_stat *tmp = exp->exp_nid_stats;
tmp->nid_stats);
if (rc)
RETURN(rc);
+ /* Always add in ldlm_stats */
+ tmp->nid_ldlm_stats = lprocfs_alloc_stats(LDLM_LAST_OPC -
+ LDLM_FIRST_OPC, 0);
+ if (tmp->nid_ldlm_stats == NULL)
+ return -ENOMEM;
+
+ lprocfs_init_ldlm_stats(tmp->nid_ldlm_stats);
+ rc = lprocfs_register_stats(tmp->nid_proc, "ldlm_stats",
+ tmp->nid_ldlm_stats);
+ if (rc)
+ RETURN(rc);
}
RETURN(0);
struct filter_export_data *fed = &exp->exp_filter_data;
unsigned long *bitmap = filter->fo_last_rcvd_slots;
int new_client = (cl_idx == -1);
-
+
ENTRY;
LASSERT(bitmap != NULL);
RETURN(0);
}
+struct lsd_client_data zero_lcd; /* globals are implicitly zeroed */
+
static int filter_client_free(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 lsd_client_data zero_lcd;
struct lvfs_run_ctxt saved;
int rc;
loff_t off;
}
if (!(exp->exp_flags & OBD_OPT_FAILOVER)) {
- memset(&zero_lcd, 0, sizeof zero_lcd);
+ /* Don't force sync on disconnect if aborting recovery,
+ * or it does num_clients * num_osts. b=17194 */
+ int need_sync = (!exp->exp_libclient || exp->exp_need_sync) &&
+ !(exp->exp_flags&OBD_OPT_ABORT_RECOV);
push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, &zero_lcd,
- sizeof(zero_lcd), &off,
- (!exp->exp_libclient ||
- exp->exp_need_sync));
+ sizeof(zero_lcd), &off, 0);
+
+ /* Make sure the server's last_transno is up to date. Do this
+ * after the client is freed so we know all the client's
+ * transactions have been committed. */
if (rc == 0)
- /* update server's transno */
filter_update_server_data(obd, filter->fo_rcvd_filp,
- filter->fo_fsd,
- !exp->exp_libclient);
+ filter->fo_fsd, need_sync);
pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
CDEBUG(rc == 0 ? D_INFO : D_ERROR,
- "zeroing out client %s at idx %u (%llu) in %s rc %d\n",
+ "zero out client %s at idx %u/%llu in %s %ssync rc %d\n",
fed->fed_lcd->lcd_uuid, fed->fed_lr_idx, fed->fed_lr_off,
- LAST_RCVD, rc);
+ LAST_RCVD, need_sync ? "" : "a", rc);
}
if (!test_and_clear_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
struct filter_export_data *fed = &exp->exp_filter_data;
struct filter_mod_data *found = NULL, *fmd_new = NULL;
- OBD_SLAB_ALLOC(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO, sizeof(*fmd_new));
+ OBD_SLAB_ALLOC_PTR_GFP(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO);
spin_lock(&fed->fed_lock);
found = filter_fmd_find_nolock(&exp->exp_obd->u.filter,fed,objid,group);
exp->exp_connecting = 1;
spin_unlock(&exp->exp_lock);
- return 0;
+ return ldlm_init_export(exp);
}
static int filter_free_server_data(struct filter_obd *filter)
CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n",rc);
GOTO(cleanup, rc);
}
- LASSERT(off == 0 || last_group >= FILTER_MIN_GROUPS);
+ LASSERTF(off == 0 || CHECK_MDS_GROUP(last_group),
+ "off = %llu and last_group = %d\n", off, last_group);
+
CDEBUG(D_INODE, "%s: previous %d, new %d\n",
obd->obd_name, last_group, group);
RETURN(PTR_ERR(dentry));
}
} else {
- dentry = simple_mkdir(filter->fo_dentry_O, filter->fo_vfsmnt,
+ dentry = simple_mkdir(filter->fo_dentry_O, filter->fo_vfsmnt,
name, 0700, 1);
if (IS_ERR(dentry)) {
CERROR("cannot lookup/create O/%s: rc = %ld\n", name,
char dir[20];
snprintf(dir, sizeof(dir), "d%u", i);
- tmp_subdirs->dentry[i] = simple_mkdir(dentry,
+ tmp_subdirs->dentry[i] = simple_mkdir(dentry,
filter->fo_vfsmnt,
dir, 0700, 1);
if (IS_ERR(tmp_subdirs->dentry[i])) {
down(&filter->fo_init_lock);
old_count = filter->fo_group_count;
for (group = old_count; group <= last_group; group++) {
- if (group == 0)
- continue; /* no group zero */
rc = filter_read_group_internal(obd, group, create);
if (rc != 0)
loff_t off = 0;
ENTRY;
- O_dentry = simple_mkdir(current->fs->pwd, filter->fo_vfsmnt,
+ O_dentry = simple_mkdir(current->fs->pwd, filter->fo_vfsmnt,
"O", 0700, 1);
CDEBUG(D_INODE, "got/created O: %p\n", O_dentry);
if (IS_ERR(O_dentry)) {
if (off == 0) {
last_group = FILTER_MIN_GROUPS;
} else {
- LASSERT(last_group >= FILTER_MIN_GROUPS);
+ LASSERT_MDS_GROUP(last_group);
}
CWARN("%s: initialize groups [%d,%d]\n", obd->obd_name,
if (rc)
CERROR("error writing server data: rc = %d\n", rc);
- for (i = 1; i < filter->fo_group_count; i++) {
+ for (i = 0; i < filter->fo_group_count; i++) {
rc = filter_update_last_objid(obd, i,
(i == filter->fo_group_count - 1));
if (rc)
spin_lock(&filter->fo_objidlock);
id = filter->fo_last_objids[group];
spin_unlock(&filter->fo_objidlock);
-
return id;
}
static int filter_lock_dentry(struct obd_device *obd, struct dentry *dparent)
{
- LOCK_INODE_MUTEX(dparent->d_inode);
+ LOCK_INODE_MUTEX_PARENT(dparent->d_inode);
return 0;
}
struct filter_subdirs *subdirs;
LASSERT(group < filter->fo_group_count); /* FIXME: object groups */
- if (group > 0 || filter->fo_subdir_count == 0)
+ if ((group > FILTER_GROUP_MDS0 && group < FILTER_GROUP_MDS1_N_BASE) ||
+ filter->fo_subdir_count == 0)
return filter->fo_dentry_O_groups[group];
subdirs = &filter->fo_dentry_O_sub[group];
rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id, LDLM_EXTENT,
&policy, LCK_PW, &flags, ldlm_blocking_ast,
ldlm_completion_ast, NULL, NULL, 0, NULL,
- &lockh);
+ NULL, &lockh);
/* We only care about the side-effects, just drop the lock. */
if (rc == ELDLM_OK)
*v = LDLM_LOCK_GET(lck);
} else if ((*v)->l_policy_data.l_extent.start <
lck->l_policy_data.l_extent.start) {
- LDLM_LOCK_PUT(*v);
+ LDLM_LOCK_RELEASE(*v);
*v = LDLM_LOCK_GET(lck);
}
LASSERT(lock->l_flags & LDLM_FL_CP_REQD);
lock->l_flags &= ~LDLM_FL_CP_REQD;
list_del_init(&wlock->l_cp_ast);
- LDLM_LOCK_PUT(wlock);
+ LDLM_LOCK_RELEASE(wlock);
}
/* The lock met with no resistance; we're finished. */
if (tree->lit_mode == LCK_PR)
continue;
- interval_iterate_reverse(tree->lit_root,
+ interval_iterate_reverse(tree->lit_root,
filter_intent_cb, &arg);
}
unlock_res(res);
unlock_res(res);
out:
- LDLM_LOCK_PUT(l);
+ LDLM_LOCK_RELEASE(l);
RETURN(ELDLM_LOCK_ABORTED);
}
EXIT;
}
+static int filter_adapt_sptlrpc_conf(struct obd_device *obd, int initial)
+{
+ struct filter_obd *filter = &obd->u.filter;
+ struct sptlrpc_rule_set tmp_rset;
+ int rc;
+
+ sptlrpc_rule_set_init(&tmp_rset);
+ rc = sptlrpc_conf_target_get_rules(obd, &tmp_rset, initial);
+ if (rc) {
+ CERROR("obd %s: failed get sptlrpc rules: %d\n",
+ obd->obd_name, rc);
+ return rc;
+ }
+
+ sptlrpc_target_update_exp_flavor(obd, &tmp_rset);
+
+ write_lock(&filter->fo_sptlrpc_lock);
+ sptlrpc_rule_set_free(&filter->fo_sptlrpc_rset);
+ filter->fo_sptlrpc_rset = tmp_rset;
+ write_unlock(&filter->fo_sptlrpc_lock);
+
+ return 0;
+}
+
/*
* pre-allocate pool of iobuf's to be used by filter_{prep,commit}rw_write().
*/
* If we haven't allocated a pool entry for this thread before, do so now. */
void *filter_iobuf_get(struct filter_obd *filter, struct obd_trans_info *oti)
{
- int thread_id = oti ? oti->oti_thread_id : -1;
+ int thread_id = (oti && oti->oti_thread) ?
+ oti->oti_thread->t_id : -1;
struct filter_iobuf *pool = NULL;
struct filter_iobuf **pool_place = NULL;
if (rc != 0)
GOTO(err_ops, rc);
- LASSERT(!lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb)));
+ if (lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb))) {
+ CERROR("%s: Underlying device is marked as read-only. "
+ "Setup failed\n", obd->obd_name);
+ GOTO(err_ops, rc = -EROFS);
+ }
/* failover is the default */
obd->obd_replayable = 1;
CFS_INIT_LIST_HEAD(&filter->fo_export_list);
sema_init(&filter->fo_alloc_lock, 1);
init_brw_stats(&filter->fo_filter_stats);
+ 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;
CFS_INIT_LIST_HEAD(&filter->fo_llog_list);
spin_lock_init(&filter->fo_llog_list_lock);
- filter->fo_sptlrpc_lock = RW_LOCK_UNLOCKED;
- sptlrpc_rule_set_init(&filter->fo_sptlrpc_rset);
-
- filter->fo_fl_oss_capa = 0;
+ filter->fo_fl_oss_capa = 1;
CFS_INIT_LIST_HEAD(&filter->fo_capa_keys);
filter->fo_capa_hash = init_capa_hash();
if (filter->fo_capa_hash == NULL)
ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
"filter_ldlm_cb_client", &obd->obd_ldlm_client);
- rc = llog_cat_initialize(obd, &obd->obd_olg, 1, NULL);
+ rc = obd_llog_init(obd, &obd->obd_olg, obd, 1, NULL, NULL);
if (rc) {
CERROR("failed to setup llogging subsystems\n");
GOTO(err_post, rc);
}
+ rwlock_init(&filter->fo_sptlrpc_lock);
+ sptlrpc_rule_set_init(&filter->fo_sptlrpc_rset);
+ /* do this after llog being initialized */
+ filter_adapt_sptlrpc_conf(obd, 1);
+
rc = lquota_setup(filter_quota_interface_ref, obd);
if (rc)
GOTO(err_post, rc);
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",
if (obd->obd_proc_exports_entry)
lprocfs_add_simple(obd->obd_proc_exports_entry, "clear",
lprocfs_nid_stats_clear_read,
- lprocfs_nid_stats_clear_write, obd);
+ lprocfs_nid_stats_clear_write, obd, NULL);
memcpy((void *)addr, lustre_cfg_buf(lcfg, 4),
LUSTRE_CFG_BUFLEN(lcfg, 4));
rc = llog_cleanup(ctxt);
ctxt = llog_group_get_ctxt(olg, LLOG_SIZE_ORIG_CTXT);
- if (ctxt)
+ if (ctxt) {
rc2 = llog_cleanup(ctxt);
- if (!rc)
- rc = rc2;
+ if (!rc)
+ rc = rc2;
+ }
+
+ ctxt = llog_group_get_ctxt(olg, LLOG_CONFIG_ORIG_CTXT);
+ if (ctxt) {
+ rc2 = llog_cleanup(ctxt);
+ if (!rc)
+ rc = rc2;
+ }
RETURN(rc);
}
-static int
+static int
filter_olg_init(struct obd_device *obd, struct obd_llog_group *olg,
struct obd_device *tgt)
{
/**
* Init the default olg, which is embeded in the obd_device, for filter.
*/
-static int
+static int
filter_default_olg_init(struct obd_device *obd, struct obd_llog_group *olg,
struct obd_device *tgt)
{
if (rc)
GOTO(cleanup_lcm, rc);
+ rc = llog_setup(obd, olg, LLOG_CONFIG_ORIG_CTXT, tgt, 0, NULL,
+ &llog_lvfs_ops);
+ if (rc)
+ GOTO(cleanup_olg, rc);
+
ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
if (!ctxt) {
- CERROR("Can't get ctxt for %p:%x\n", olg,
+ CERROR("Can't get ctxt for %p:%x\n", olg,
LLOG_MDS_OST_REPL_CTXT);
GOTO(cleanup_olg, rc = -ENODEV);
}
filter->fo_lcm = NULL;
return rc;
}
-
-static int
+
+static int
filter_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
struct obd_device *tgt, int count, struct llog_catid *catid,
struct obd_uuid *uuid)
RETURN(rc);
ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
if (!ctxt) {
- CERROR("Can't get ctxt for %p:%x\n", olg,
+ CERROR("Can't get ctxt for %p:%x\n", olg,
LLOG_MDS_OST_REPL_CTXT);
filter_olg_fini(olg);
RETURN(-ENODEV);
struct filter_obd *filter = &obd->u.filter;
struct llog_ctxt *ctxt;
ENTRY;
-
+
ctxt = llog_group_get_ctxt(&obd->obd_olg, LLOG_MDS_OST_REPL_CTXT);
- LASSERT(ctxt != NULL);
- mutex_down(&ctxt->loc_sem);
- if (ctxt->loc_imp) {
+ if (ctxt) {
+ /*
+ * Make sure that no cached llcds left in recov_thread.
+ * We actually do sync in disconnect time, but disconnect
+ * may not come being marked rq_no_resend = 1.
+ */
+ llog_sync(ctxt, NULL);
+
/*
- * Balance class_import_get() in llog_receptor_accept(). This
- * is safe to do here, as llog is already synchronized and its
- * import may go.
+ * Balance class_import_get() in llog_receptor_accept().
+ * This is safe to do, as llog is already synchronized
+ * and its import may go.
*/
- class_import_put(ctxt->loc_imp);
- ctxt->loc_imp = NULL;
+ mutex_down(&ctxt->loc_sem);
+ if (ctxt->loc_imp) {
+ class_import_put(ctxt->loc_imp);
+ ctxt->loc_imp = NULL;
+ }
+ mutex_up(&ctxt->loc_sem);
+ llog_ctxt_put(ctxt);
}
- mutex_up(&ctxt->loc_sem);
- llog_ctxt_put(ctxt);
if (filter->fo_lcm) {
llog_recov_thread_fini(filter->fo_lcm, obd->obd_force);
filter = &obd->u.filter;
- if (group == OBD_LLOG_GROUP)
+ if (group == FILTER_GROUP_LLOG)
RETURN(&obd->obd_olg);
spin_lock(&filter->fo_llog_list_lock);
olg = filter_find_olg_internal(filter, group);
spin_unlock(&filter->fo_llog_list_lock);
- RETURN(olg);
+ RETURN(olg);
}
/**
* Find the llog_group of the filter according to the group. If it can not
- * find, create the llog_group, which only happens when mds is being synced
+ * find, create the llog_group, which only happens when mds is being synced
* with OST.
*/
struct obd_llog_group *filter_find_create_olg(struct obd_device *obd, int group)
filter = &obd->u.filter;
- if (group == OBD_LLOG_GROUP)
+ if (group == FILTER_GROUP_LLOG)
RETURN(&obd->obd_olg);
spin_lock(&filter->fo_llog_list_lock);
int rc;
ENTRY;
- CDEBUG(D_OTHER, "handle connect for %s: %u/%u/%u\n", obd->obd_name,
- (unsigned) body->lgdc_logid.lgl_ogr,
- (unsigned) body->lgdc_logid.lgl_oid,
- (unsigned) body->lgdc_logid.lgl_ogen);
+ 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);
olg = filter_find_olg(obd, body->lgdc_logid.lgl_ogr);
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_ogr);
RETURN(-ENOENT);
}
llog_group_set_export(olg, exp);
ctxt = llog_group_get_ctxt(olg, body->lgdc_ctxt_idx);
LASSERTF(ctxt != NULL, "ctxt is not null, ctxt idx %d \n",
body->lgdc_ctxt_idx);
- rc = llog_connect(ctxt, 1, &body->lgdc_logid,
+
+ 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);
+
+ rc = llog_connect(ctxt, &body->lgdc_logid,
&body->lgdc_gen, NULL);
llog_ctxt_put(ctxt);
if (rc != 0)
list_add(&olg->olg_list, &remove_list);
}
spin_unlock(&filter->fo_llog_list_lock);
-
- list_for_each_entry_safe(olg, tmp, &remove_list, olg_list) {
+
+ list_for_each_entry_safe(olg, tmp, &remove_list, olg_list) {
list_del_init(&olg->olg_list);
rc = filter_olg_fini(olg);
if (rc)
exp->exp_connect_flags = data->ocd_connect_flags;
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))
+ RETURN(-EPROTO);
+
if (exp->exp_connect_flags & OBD_CONNECT_GRANT) {
+ struct filter_obd *filter = &exp->exp_obd->u.filter;
obd_size left, want;
spin_lock(&exp->exp_obd->obd_osfs_lock);
LPU64" left: "LPU64"\n", exp->exp_obd->obd_name,
exp->exp_client_uuid.uuid, exp,
data->ocd_grant, want, left);
+
+ filter->fo_tot_granted_clients ++;
}
if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
static int filter_reconnect(const struct lu_env *env,
struct obd_export *exp, struct obd_device *obd,
struct obd_uuid *cluuid,
- struct obd_connect_data *data)
+ struct obd_connect_data *data,
+ void *localdata)
{
int rc;
ENTRY;
RETURN(-EINVAL);
rc = filter_connect_internal(exp, data);
+ if (rc == 0)
+ filter_export_stats_init(obd, exp, localdata);
RETURN(rc);
}
/* nearly identical to mds_connect */
static int filter_connect(const struct lu_env *env,
- struct lustre_handle *conn, struct obd_device *obd,
+ struct obd_export **exp, struct obd_device *obd,
struct obd_uuid *cluuid,
struct obd_connect_data *data, void *localdata)
{
struct lvfs_run_ctxt saved;
- struct obd_export *exp;
+ 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;
- if (conn == NULL || obd == NULL || cluuid == NULL)
+ if (exp == NULL || obd == NULL || cluuid == NULL)
RETURN(-EINVAL);
- rc = class_connect(conn, obd, cluuid);
+ rc = class_connect(&conn, obd, cluuid);
if (rc)
RETURN(rc);
- exp = class_conn2export(conn);
- LASSERT(exp != NULL);
+ lexp = class_conn2export(&conn);
+ LASSERT(lexp != NULL);
- fed = &exp->exp_filter_data;
+ fed = &lexp->exp_filter_data;
- rc = filter_connect_internal(exp, data);
+ rc = filter_connect_internal(lexp, data);
if (rc)
GOTO(cleanup, rc);
- filter_export_stats_init(obd, exp, localdata);
+ filter_export_stats_init(obd, lexp, localdata);
if (obd->obd_replayable) {
OBD_ALLOC(lcd, sizeof(*lcd));
if (!lcd) {
memcpy(lcd->lcd_uuid, cluuid, sizeof(lcd->lcd_uuid));
fed->fed_lcd = lcd;
- rc = filter_client_add(obd, exp, -1);
+ rc = filter_client_add(obd, lexp, -1);
if (rc)
GOTO(cleanup, rc);
}
group = data->ocd_group;
- if (group == 0)
- GOTO(cleanup, rc);
CWARN("%s: Received MDS connection ("LPX64"); group %d\n",
- obd->obd_name, exp->exp_handle.h_cookie, group);
+ obd->obd_name, lexp->exp_handle.h_cookie, group);
push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
rc = filter_read_groups(obd, group, 1);
OBD_FREE_PTR(lcd);
fed->fed_lcd = NULL;
}
- class_disconnect(exp);
+ class_disconnect(lexp);
+ *exp = NULL;
} else {
- class_export_put(exp);
+ *exp = lexp;
}
RETURN(rc);
exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
exp, exp->exp_filter_data.fed_pending);
- /* Not ported yet the b1_6 quota functionality
- * lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd);
- */
+ 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);
filter_grant_discard(exp);
filter_fmd_cleanup(exp);
+ if (exp->exp_connect_flags & OBD_CONNECT_GRANT_SHRINK) {
+ struct filter_obd *filter = &exp->exp_obd->u.filter;
+ if (filter->fo_tot_granted_clients > 0)
+ filter->fo_tot_granted_clients --;
+ }
+
if (!(exp->exp_flags & OBD_OPT_FORCE))
filter_grant_sanity_check(exp->exp_obd, __FUNCTION__);
{
struct obd_llog_group *olg_min, *olg;
struct filter_obd *filter;
- int worked = 0, group;
+ int worked = -1, group;
struct llog_ctxt *ctxt;
ENTRY;
(dexp == olg_min->olg_exp || dexp == NULL)) {
int err;
ctxt = llog_group_get_ctxt(olg_min,
- LLOG_MDS_OST_REPL_CTXT);
- LASSERT(ctxt != NULL);
- err = llog_sync(ctxt, olg_min->olg_exp);
- llog_ctxt_put(ctxt);
- if (err)
- CERROR("error flushing logs to MDS: rc %d\n",
- err);
+ LLOG_MDS_OST_REPL_CTXT);
+ if (ctxt) {
+ err = llog_sync(ctxt, olg_min->olg_exp);
+ llog_ctxt_put(ctxt);
+ if (err) {
+ CERROR("error flushing logs to MDS: "
+ "rc %d\n", err);
+ }
+ }
}
} while (olg_min != NULL);
}
-/* also incredibly similar to mds_disconnect */
+/* Also incredibly similar to mds_disconnect */
static int filter_disconnect(struct obd_export *exp)
{
struct obd_device *obd = exp->exp_obd;
filter_grant_sanity_check(obd, __FUNCTION__);
filter_grant_discard(exp);
+ /* Flush any remaining cancel messages out to the target */
+ filter_sync_llogs(obd, exp);
+
+ lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd);
+
/* Disconnect early so that clients can't keep using export */
rc = class_disconnect(exp);
if (exp->exp_obd->obd_namespace != NULL)
lprocfs_exp_cleanup(exp);
- /* flush any remaining cancel messages out to the target */
- filter_sync_llogs(obd, exp);
class_export_put(exp);
RETURN(rc);
}
struct llog_cookie *fcc = NULL;
struct filter_obd *filter;
int rc, err, locked = 0, sync = 0;
+ loff_t old_size = 0;
unsigned int ia_valid;
struct inode *inode;
struct iattr iattr;
if (ia_valid & ATTR_SIZE || ia_valid & (ATTR_UID | ATTR_GID)) {
DQUOT_INIT(inode);
LOCK_INODE_MUTEX(inode);
+ old_size = i_size_read(inode);
locked = 1;
}
fcc = NULL;
}
+ /* For a partial-page truncate flush the page to disk immediately
+ * to avoid data corruption during direct disk write. b=17397 */
+ if (!sync && (iattr.ia_valid & ATTR_SIZE) &&
+ old_size != iattr.ia_size && (iattr.ia_size & ~CFS_PAGE_MASK)) {
+ err = filemap_fdatawrite_range(inode->i_mapping, iattr.ia_size,
+ iattr.ia_size + 1);
+ if (!rc)
+ rc = err;
+ }
+
if (locked) {
- /* Let's flush truncated page on disk immediately, then we can
- * avoid need to search for page aliases before directio writes
- * and this sort of stuff at expense of somewhat slower
- * truncates not on a page boundary. I believe this is the only
- * place in filter code that can lead to pages getting to
- * pagecache so far. */
- filter_clear_truncated_page(inode);
UNLOCK_INODE_MUTEX(inode);
locked = 0;
}
int filter_setattr(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);
struct ldlm_res_id res_id;
struct filter_mod_data *fmd;
struct lvfs_run_ctxt saved;
struct filter_obd *filter;
struct ldlm_resource *res;
struct dentry *dentry;
+ __u64 opc = CAPA_OPC_META_WRITE;
int rc;
ENTRY;
- osc_build_res_name(oinfo->oi_oa->o_id, oinfo->oi_oa->o_gr, &res_id);
- rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
- oinfo_capa(oinfo), CAPA_OPC_META_WRITE);
+ if (oa->o_valid & OBD_FL_TRUNC)
+ opc |= CAPA_OPC_OSS_TRUNC;
+ rc = filter_auth_capa(exp, NULL, obdo_mdsno(oa), capa, opc);
if (rc)
RETURN(rc);
+ if (oa->o_valid & (OBD_MD_FLUID | OBD_MD_FLGID)) {
+ rc = filter_capa_fixoa(exp, oa, obdo_mdsno(oa), capa);
+ if (rc)
+ RETURN(rc);
+ }
+
+ osc_build_res_name(oa->o_id, oa->o_gr, &res_id);
/* This would be very bad - accidentally truncating a file when
* changing the time or similar - bug 12203. */
- if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE &&
+ if (oa->o_valid & OBD_MD_FLSIZE &&
oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) {
static char mdsinum[48];
- if (oinfo->oi_oa->o_valid & OBD_MD_FLFID)
+ if (oa->o_valid & OBD_MD_FLFID)
snprintf(mdsinum, sizeof(mdsinum) - 1,
- " of inode "LPU64"/%u", oinfo->oi_oa->o_fid,
- oinfo->oi_oa->o_generation);
+ " of inode "LPU64"/%u", oa->o_fid,
+ oa->o_generation);
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),
- oinfo->oi_oa->o_id, mdsinum);
+ oa->o_id, mdsinum);
RETURN(-EPERM);
}
- dentry = __filter_oa2dentry(exp->exp_obd, oinfo->oi_oa,
- __FUNCTION__, 1);
+ dentry = __filter_oa2dentry(exp->exp_obd, oa, __FUNCTION__, 1);
if (IS_ERR(dentry))
RETURN(PTR_ERR(dentry));
push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
lock_kernel();
- if (oinfo->oi_oa->o_valid &
+ if (oa->o_valid &
(OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME)) {
- fmd = filter_fmd_get(exp,oinfo->oi_oa->o_id,oinfo->oi_oa->o_gr);
+ fmd = filter_fmd_get(exp, oa->o_id, oa->o_gr);
if (fmd && fmd->fmd_mactime_xid < oti->oti_xid)
fmd->fmd_mactime_xid = oti->oti_xid;
filter_fmd_put(exp, fmd);
}
/* setting objects attributes (including owner/group) */
- rc = filter_setattr_internal(exp, dentry, oinfo->oi_oa, oti);
+ rc = filter_setattr_internal(exp, dentry, oa, oti);
if (rc)
GOTO(out_unlock, rc);
&res_id, LDLM_EXTENT, 0);
if (res != NULL) {
+ LDLM_RESOURCE_ADDREF(res);
rc = ldlm_res_lvbo_update(res, NULL, 0, 0);
+ LDLM_RESOURCE_DELREF(res);
ldlm_resource_putref(res);
}
- oinfo->oi_oa->o_valid = OBD_MD_FLID;
+ oa->o_valid = OBD_MD_FLID;
/* Quota release need uid/gid info */
- obdo_from_inode(oinfo->oi_oa, dentry->d_inode,
+ obdo_from_inode(oa, dentry->d_inode,
FILTER_VALID_FLAGS | OBD_MD_FLUID | OBD_MD_FLGID);
EXIT;
/* caller must hold fo_create_locks[oa->o_gr] */
static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
- struct filter_obd *filter)
+ struct filter_obd *filter)
{
struct obdo doa; /* XXX obdo on stack */
obd_id last, id;
- int rc;
+ int rc = 0;
+ int skip_orphan;
ENTRY;
LASSERT(oa);
- LASSERT(oa->o_gr != 0);
+ LASSERT_MDS_GROUP(oa->o_gr);
LASSERT(oa->o_valid & OBD_MD_FLGROUP);
LASSERT(down_trylock(&filter->fo_create_locks[oa->o_gr]) != 0);
last = filter_last_id(filter, doa.o_gr);
- CWARN("%s: deleting orphan objects from "LPU64" to "LPU64"\n",
- exp->exp_obd->obd_name, oa->o_id + 1, last);
-
+ skip_orphan = !!(exp->exp_connect_flags & OBD_CONNECT_SKIP_ORPHAN);
+
+ CWARN("%s: deleting orphan objects from "LPU64" to "LPU64"%s\n",
+ exp->exp_obd->obd_name, oa->o_id + 1, last,
+ skip_orphan ? ", orphan objids won't be reused any more." : ".");
+
for (id = last; id > oa->o_id; id--) {
doa.o_id = id;
- rc = filter_destroy(exp, &doa, NULL, NULL, NULL);
+ rc = filter_destroy(exp, &doa, NULL, NULL, NULL, NULL);
if (rc && rc != -ENOENT) /* this is pretty fatal... */
CEMERG("error destroying precreate objid "LPU64": %d\n",
id, rc);
- filter_set_last_id(filter, id - 1, doa.o_gr);
+
/* 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)
+ 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);
+ }
}
CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n",
exp->exp_obd->obd_name, doa.o_gr, oa->o_id);
- rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
+ if (!skip_orphan) {
+ filter_set_last_id(filter, id, doa.o_gr);
+ rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
+ } else {
+ /* don't reuse orphan object, return last used objid */
+ oa->o_id = last;
+ rc = 0;
+ }
clear_bit(doa.o_gr, &filter->fo_destroys_in_progress);
RETURN(rc);
/* delete orphans request */
if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_DELORPHAN)){
+ obd_id last = filter_last_id(filter, group);
+
if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
CERROR("%s: dropping old orphan cleanup request\n",
obd->obd_name);
up(&filter->fo_create_locks[group]);
RETURN(0);
}
- diff = oa->o_id - filter_last_id(filter, group);
+ diff = oa->o_id - last;
CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n",
- filter_last_id(filter, group), diff);
+ last, diff);
if (-diff > OST_MAX_PRECREATE) {
CERROR("%s: ignoring bogus orphan destroy request: "
"obdid "LPU64" last_id "LPU64"\n", obd->obd_name,
- oa->o_id, filter_last_id(filter, group));
+ oa->o_id, last);
/* FIXME: should reset precreate_next_id on MDS */
GOTO(out, rc = -EINVAL);
}
GOTO(out, rc = 0);
}
/* only precreate if group == 0 and o_id is specfied */
- if (group < FILTER_GROUP_MDS0 || oa->o_id == 0)
+ if (group == FILTER_GROUP_LLOG || oa->o_id == 0)
diff = 1;
else
diff = oa->o_id - filter_last_id(filter, group);
rc = -EAGAIN;
break;
}
-
+
if (recreate_obj) {
__u64 last_id;
next_id = oa->o_id;
} else
next_id = filter_last_id(filter, group) + 1;
- CDEBUG(D_INFO, "precreate objid "LPU64"\n", next_id);
-
dparent = filter_parent_lock(obd, group, next_id);
if (IS_ERR(dparent))
GOTO(cleanup, rc = PTR_ERR(dparent));
GOTO(cleanup, rc = PTR_ERR(handle));
cleanup_phase = 3;
+ CDEBUG(D_INODE, "%s: filter_precreate(od->o_gr="LPU64
+ ",od->o_id="LPU64")\n", obd->obd_name, group,
+ next_id);
+
/* We mark object SUID+SGID to flag it for accepting UID+GID
* from client on first write. Currently the permission bits
* on the OST are never used, so this is OK. */
static int filter_create(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 obd_device *obd = NULL;
struct filter_obd *filter;
struct lvfs_run_ctxt saved;
struct lov_stripe_md *lsm = NULL;
int rc = 0, diff, group = oa->o_gr;
ENTRY;
- if (!(oa->o_valid & OBD_MD_FLGROUP) || group == 0) {
+ 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);
}
- obd = exp->exp_obd;
fed = &exp->exp_filter_data;
filter = &obd->u.filter;
RETURN(-ENOTUNIQ);
}
- CDEBUG(D_INFO, "filter_create(od->o_gr="LPU64",od->o_id="LPU64")\n",
- oa->o_gr, oa->o_id);
if (ea != NULL) {
lsm = *ea;
if (lsm == NULL) {
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)
+ struct obd_export *md_exp, void *capa)
{
unsigned int qcids[MAXQUOTAS] = {0, 0};
struct obd_device *obd;
LASSERT(oa->o_valid & OBD_MD_FLGROUP);
+ rc = filter_auth_capa(exp, NULL, obdo_mdsno(oa),
+ (struct lustre_capa *)capa, CAPA_OPC_OSS_DESTROY);
+ if (rc)
+ RETURN(rc);
+
obd = exp->exp_obd;
filter = &obd->u.filter;
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);
+
dchild = filter_fid2dentry(obd, NULL, oa->o_gr, oa->o_id);
if (IS_ERR(dchild))
GOTO(cleanup, rc = PTR_ERR(dchild));
filter_cancel_cookies_cb,
fcc);
/* If add_journal_cb failed, then filter_finish_transno
- * will commit the handle and we will do a sync
- * on commit. then we call callback directly to free
- * the fcc.
+ * will commit the handle and we will do a sync
+ * on commit. then we call callback directly to free
+ * the fcc.
*/
rc = filter_finish_transno(exp, oti, rc, sync);
if (sync) {
- filter_cancel_cookies_cb(obd, 0, fcc, rc);
+ filter_cancel_cookies_cb(obd, 0, fcc, rc);
fcc = NULL;
}
rc2 = fsfilt_commit(obd, dparent->d_inode, handle, 0);
qcids[GRPQUOTA] = oa->o_gid;
rc2 = lquota_adjust(filter_quota_interface_ref, obd, qcids, NULL, rc,
FSFILT_OP_UNLINK);
-
if (rc2)
- CDEBUG(D_QUOTA, "filter adjust qunit! (rc:%d)\n", rc2);
+ CERROR("filter adjust qunit! (rc:%d)\n", rc2);
return rc;
}
", o_size = "LPD64"\n", oinfo->oi_oa->o_id,
oinfo->oi_oa->o_valid, oinfo->oi_policy.l_extent.start);
- rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
- oinfo_capa(oinfo), CAPA_OPC_OSS_TRUNC);
- if (rc)
- RETURN(rc);
-
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;
RETURN(rc);
}
filter = &exp->exp_obd->u.filter;
- /* an objid of zero is taken to mean "sync whole filesystem" */
+ /* 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);
- /* flush any remaining cancel messages out to the target */
+ /* Flush any remaining cancel messages out to the target */
filter_sync_llogs(exp->exp_obd, exp);
RETURN(rc);
}
push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
rc = fsfilt_iocontrol(obd, dentry->d_inode, NULL,
EXT3_IOC_FIEMAP, (long)fiemap);
- if (rc) {
- f_dput(dentry);
- RETURN(rc);
- }
pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
f_dput(dentry);
- RETURN(0);
+ RETURN(rc);
}
CDEBUG(D_IOCTL, "invalid key\n");
RETURN(-EINVAL);
}
+static inline int filter_setup_llog_group(struct obd_export *exp,
+ struct obd_device *obd,
+ int group)
+{
+ struct obd_llog_group *olg;
+ struct llog_ctxt *ctxt;
+ int rc;
+
+ olg = filter_find_create_olg(obd, group);
+ if (IS_ERR(olg))
+ RETURN(PTR_ERR(olg));
+
+ llog_group_set_export(olg, exp);
+
+ ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
+ LASSERTF(ctxt != NULL, "ctxt is null\n");
+
+ rc = llog_receptor_accept(ctxt, exp->exp_imp_reverse);
+ llog_ctxt_put(ctxt);
+ return rc;
+}
static int filter_set_info_async(struct obd_export *exp, __u32 keylen,
void *key, __u32 vallen, void *val,
struct ptlrpc_request_set *set)
{
struct obd_device *obd;
- struct obd_llog_group *olg;
- struct llog_ctxt *ctxt;
int rc = 0, group;
ENTRY;
if (KEY_IS(KEY_REVIMP_UPD)) {
filter_revimp_update(exp);
+ lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd);
+ RETURN(0);
+ }
+
+ if (KEY_IS(KEY_SPTLRPC_CONF)) {
+ filter_adapt_sptlrpc_conf(obd, 0);
RETURN(0);
}
+ if (KEY_IS(KEY_GRANT_SHRINK)) {
+ struct ost_body *body = (struct ost_body *)val;
+ /* handle shrink grant */
+ spin_lock(&exp->exp_obd->obd_osfs_lock);
+ filter_grant_incoming(exp, &body->oa);
+ spin_unlock(&exp->exp_obd->obd_osfs_lock);
+ RETURN(rc);
+ }
+
if (!KEY_IS(KEY_MDS_CONN))
RETURN(-EINVAL);
obd->u.filter.fo_mdc_conn.cookie = exp->exp_handle.h_cookie;
/* setup llog imports */
- LASSERT(val != NULL);
- group = (int)(*(__u32 *)val);
- LASSERT(group >= FILTER_GROUP_MDS0);
-
- olg = filter_find_create_olg(obd, group);
- if (IS_ERR(olg))
- RETURN(PTR_ERR(olg));
-
- llog_group_set_export(olg, exp);
-
- ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
- LASSERTF(ctxt != NULL, "ctxt is null\n"),
+ if (val != NULL)
+ group = (int)(*(__u32 *)val);
+ else
+ group = 0; /* default value */
- rc = llog_receptor_accept(ctxt, exp->exp_imp_reverse);
- llog_ctxt_put(ctxt);
+ LASSERT_MDS_GROUP(group);
+ rc = filter_setup_llog_group(exp, obd, group);
+ if (rc)
+ goto out;
- lquota_setinfo(filter_quota_interface_ref, exp, obd);
+ lquota_setinfo(filter_quota_interface_ref, obd, exp);
+ if (group == FILTER_GROUP_MDS0) {
+ /* setup llog group 1 for interop */
+ filter_setup_llog_group(exp, obd, FILTER_GROUP_LLOG);
+ }
+out:
RETURN(rc);
}
int rc = 0;
switch (lcfg->lcfg_command) {
- case LCFG_SPTLRPC_CONF: {
- struct filter_obd *filter = &obd->u.filter;
- struct sptlrpc_conf_log *log;
- struct sptlrpc_rule_set tmp_rset;
-
- log = sptlrpc_conf_log_extract(lcfg);
- if (IS_ERR(log)) {
- rc = PTR_ERR(log);
- break;
- }
-
- sptlrpc_rule_set_init(&tmp_rset);
-
- rc = sptlrpc_rule_set_from_log(&tmp_rset, log);
- if (rc) {
- CERROR("obd %s: failed get sptlrpc rules: %d\n",
- obd->obd_name, rc);
- break;
- }
-
- write_lock(&filter->fo_sptlrpc_lock);
- sptlrpc_rule_set_free(&filter->fo_sptlrpc_rset);
- filter->fo_sptlrpc_rset = tmp_rset;
- write_unlock(&filter->fo_sptlrpc_lock);
-
- sptlrpc_target_update_exp_flavor(obd, &tmp_rset);
- break;
- }
default:
lprocfs_filter_init_vars(&lvars);
rc = class_process_proc_param(PARAM_OST, lvars.obd_vars,
lcfg, obd);
+ if (rc > 0)
+ rc = 0;
break;
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff