#define DEBUG_SUBSYSTEM S_FILTER
+#ifndef AUTOCONF_INCLUDED
#include <linux/config.h>
+#endif
#include <linux/module.h>
#include <linux/fs.h>
#include <linux/dcache.h>
#include <linux/init.h>
#include <linux/version.h>
#include <linux/sched.h>
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
-# include <linux/mount.h>
-# include <linux/buffer_head.h>
-#endif
+#include <linux/mount.h>
+#include <linux/buffer_head.h>
+#include <obd_cksum.h>
#include <obd_class.h>
#include <obd_lov.h>
#include <lustre_dlm.h>
#include "filter_internal.h"
+/* Group 0 is no longer a legal group, to catch uninitialized IDs */
+#define FILTER_MIN_GROUPS FILTER_GROUP_MDS0
static struct lvfs_callback_ops filter_lvfs_ops;
-kmem_cache_t *ll_fmd_cachep;
+cfs_mem_cache_t *ll_fmd_cachep;
static void filter_commit_cb(struct obd_device *obd, __u64 transno,
void *cb_data, int error)
/* Assumes caller has already pushed us into the kernel context. */
int filter_finish_transno(struct obd_export *exp, struct obd_trans_info *oti,
- int rc)
+ int rc, int force_sync)
{
struct filter_obd *filter = &exp->exp_obd->u.filter;
struct filter_export_data *fed = &exp->exp_filter_data;
struct filter_client_data *fcd = fed->fed_fcd;
__u64 last_rcvd;
loff_t off;
- int err, log_pri = D_HA;
+ int err, log_pri = D_RPCTRACE;
/* Propagate error code. */
if (rc)
fed->fed_lr_idx, fed->fed_lr_off);
err = -EINVAL;
} else {
- fsfilt_add_journal_cb(exp->exp_obd, last_rcvd, oti->oti_handle,
- filter_commit_cb, NULL);
+ if (!force_sync)
+ force_sync = fsfilt_add_journal_cb(exp->exp_obd,
+ last_rcvd,
+ oti->oti_handle,
+ filter_commit_cb,
+ NULL);
+
err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp,
fcd, sizeof(*fcd), &off,
- exp->exp_need_sync);
+ force_sync | exp->exp_need_sync);
+ if (force_sync)
+ filter_commit_cb(exp->exp_obd, last_rcvd, NULL, err);
}
if (err) {
log_pri = D_ERROR;
spin_lock_init(&brw_stats->hist[i].oh_lock);
}
+static int lprocfs_init_rw_stats(struct obd_device *obd,
+ struct lprocfs_stats **stats)
+{
+ int num_stats;
+
+ num_stats = (sizeof(*obd->obd_type->typ_dt_ops) / sizeof(void *)) +
+ LPROC_FILTER_LAST - 1;
+ *stats = lprocfs_alloc_stats(num_stats, 0);
+ if (*stats == NULL)
+ return -ENOMEM;
+
+ lprocfs_init_ops_stats(LPROC_FILTER_LAST, *stats);
+ lprocfs_counter_init(*stats, LPROC_FILTER_READ_BYTES,
+ LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
+ lprocfs_counter_init(*stats, LPROC_FILTER_WRITE_BYTES,
+ LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
+
+ return(0);
+}
+
/* brw_stats are 2128, ops are 3916, ldlm are 204, so 6248 bytes per client,
plus the procfs overhead :( */
static int filter_export_stats_init(struct obd_device *obd,
- struct obd_export *exp)
+ struct obd_export *exp,
+ void *client_nid)
{
struct filter_export_data *fed = &exp->exp_filter_data;
- struct proc_dir_entry *brw_entry;
- int rc, num_stats;
+ 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);
- if (rc)
+
+ rc = lprocfs_exp_setup(exp, client_nid, &newnid);
+ if (rc)
RETURN(rc);
- /* Create a per export proc entry for brw_stats */
- brw_entry = create_proc_entry("brw_stats", 0644, exp->exp_proc);
- if (brw_entry == NULL)
- RETURN(-ENOMEM);
- brw_entry->proc_fops = &filter_per_export_stats_fops;
- brw_entry->data = fed;
-
- /* Create a per export proc entry for ops stats */
- num_stats = (sizeof(*obd->obd_type->typ_ops) / sizeof(void *)) +
- LPROC_FILTER_LAST - 1;
- exp->exp_ops_stats = lprocfs_alloc_stats(num_stats);
- if (exp->exp_ops_stats == NULL)
- RETURN(-ENOMEM);
- lprocfs_init_ops_stats(LPROC_FILTER_LAST, exp->exp_ops_stats);
- lprocfs_counter_init(exp->exp_ops_stats, LPROC_FILTER_READ_BYTES,
- LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
- lprocfs_counter_init(exp->exp_ops_stats, LPROC_FILTER_WRITE_BYTES,
- LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
- lprocfs_register_stats(exp->exp_proc, "stats", exp->exp_ops_stats);
+ if (newnid) {
+ struct nid_stat *tmp = exp->exp_nid_stats;
+ LASSERT(tmp != NULL);
+
+ OBD_ALLOC(tmp->nid_brw_stats, sizeof(struct brw_stats));
+ if (tmp->nid_brw_stats == NULL)
+ RETURN(-ENOMEM);
+
+ init_brw_stats(tmp->nid_brw_stats);
+ rc = lprocfs_seq_create(exp->exp_nid_stats->nid_proc, "brw_stats",
+ 0644, &filter_per_nid_stats_fops,
+ exp->exp_nid_stats);
+ if (rc)
+ CWARN("Error adding the brw_stats file\n");
+
+ rc = lprocfs_init_rw_stats(obd, &exp->exp_nid_stats->nid_stats);
+ if (rc)
+ RETURN(rc);
+
+ rc = lprocfs_register_stats(tmp->nid_proc, "stats",
+ tmp->nid_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);
LASSERTF(cl_idx > -2, "%d\n", cl_idx);
/* Self-export */
- if (strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid) == 0)
+ if (strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid) == 0)
RETURN(0);
/* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
repeat:
if (cl_idx >= LR_MAX_CLIENTS) {
- CERROR("no room for %u clients - fix LR_MAX_CLIENTS\n",
+ CERROR("no room for %u client - fix LR_MAX_CLIENTS\n",
cl_idx);
RETURN(-EOVERFLOW);
}
if (new_client) {
struct lvfs_run_ctxt saved;
loff_t off = fed->fed_lr_off;
- void *handle;
int rc;
+ void *handle;
CDEBUG(D_INFO, "writing client fcd at idx %u (%llu) (len %u)\n",
fed->fed_lr_idx,off,(unsigned int)sizeof(*fed->fed_fcd));
GOTO(free, 0);
CDEBUG(D_INFO, "freeing client at idx %u, offset %lld with UUID '%s'\n",
- fed->fed_lr_idx, off, fed->fed_fcd->fcd_uuid);
+ fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid);
LASSERT(filter->fo_last_rcvd_slots != NULL);
push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, &zero_fcd,
sizeof(zero_fcd), &off,
- (!exp->exp_libclient || exp->exp_need_sync));
-
+ (!exp->exp_libclient ||
+ exp->exp_need_sync));
if (rc == 0)
/* update server's transno */
filter_update_server_data(obd, filter->fo_rcvd_filp,
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, SLAB_NOFS, sizeof(*fmd_new));
+ OBD_SLAB_ALLOC(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO, sizeof(*fmd_new));
spin_lock(&fed->fed_lock);
found = filter_fmd_find_nolock(&exp->exp_obd->u.filter,fed,objid,group);
static int filter_init_export(struct obd_export *exp)
{
spin_lock_init(&exp->exp_filter_data.fed_lock);
- INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
-
+ CFS_INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
+
spin_lock(&exp->exp_lock);
exp->exp_connecting = 1;
spin_unlock(&exp->exp_lock);
le64_to_cpu(fsd->lsd_mount_count));
fsd->lsd_compat14 = fsd->lsd_last_transno;
- rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off,force_sync);
+ rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off, force_sync);
if (rc)
CERROR("error writing lr_server_data: rc = %d\n", rc);
group, rc);
RETURN(rc);
}
-
+extern int ost_handle(struct ptlrpc_request *req);
/* assumes caller has already in kernel ctxt */
static int filter_init_server_data(struct obd_device *obd, struct file * filp)
{
struct lr_server_data *fsd;
struct filter_client_data *fcd = NULL;
struct inode *inode = filp->f_dentry->d_inode;
- unsigned long last_rcvd_size = inode->i_size;
+ unsigned long last_rcvd_size = i_size_read(inode);
__u64 mount_count;
int cl_idx;
loff_t off = 0;
GOTO(err_fsd, rc);
}
if (strcmp(fsd->lsd_uuid, obd->obd_uuid.uuid) != 0) {
- LCONSOLE_ERROR("Trying to start OBD %s using the wrong"
- " disk %s. Were the /dev/ assignments "
- "rearranged?\n",
- obd->obd_uuid.uuid, fsd->lsd_uuid);
+ 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);
}
mount_count = le64_to_cpu(fsd->lsd_mount_count);
filter->fo_subdir_count = le16_to_cpu(fsd->lsd_subdir_count);
/* COMPAT_146 */
/* Assume old last_rcvd format unless I_C_LR is set */
- if (!(fsd->lsd_feature_incompat &
+ if (!(fsd->lsd_feature_incompat &
cpu_to_le32(OBD_INCOMPAT_COMMON_LR)))
fsd->lsd_last_transno = fsd->lsd_compat14;
/* end COMPAT_146 */
* need to be set up like real exports as filter_connect() does.
*/
exp = class_new_export(obd, (struct obd_uuid *)fcd->fcd_uuid);
+
CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64
- " srv lr: "LPU64"\n", fcd->fcd_uuid, cl_idx,
- last_rcvd, le64_to_cpu(fsd->lsd_last_transno));
+ " srv lr: "LPU64" fcd_group %d\n", fcd->fcd_uuid, cl_idx,
+ last_rcvd, le64_to_cpu(fsd->lsd_last_transno),
+ le32_to_cpu(fcd->fcd_group));
if (IS_ERR(exp)) {
if (PTR_ERR(exp) == -EALREADY) {
/* export already exists, zero out this one */
} else {
fed = &exp->exp_filter_data;
fed->fed_fcd = fcd;
- filter_export_stats_init(obd, exp);
+ fed->fed_group = le32_to_cpu(fcd->fcd_group);
+ filter_export_stats_init(obd, exp, NULL);
rc = filter_client_add(obd, exp, cl_idx);
- LASSERTF(rc == 0, "rc = %d\n", rc); /* can't fail existing */
+ /* can't fail for existing client */
+ LASSERTF(rc == 0, "rc = %d\n", rc);
fcd = NULL;
-
spin_lock(&exp->exp_lock);
- exp->exp_replay_needed = 1;
exp->exp_connecting = 0;
+ exp->exp_in_recovery = 0;
spin_unlock(&exp->exp_lock);
-
- obd->obd_recoverable_clients++;
obd->obd_max_recoverable_clients++;
class_export_put(exp);
}
obd->obd_last_committed = le64_to_cpu(fsd->lsd_last_transno);
- if (obd->obd_recoverable_clients) {
- CWARN("RECOVERY: service %s, %d recoverable clients, "
- "last_rcvd "LPU64"\n", obd->obd_name,
- obd->obd_recoverable_clients,
- le64_to_cpu(fsd->lsd_last_transno));
- obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
- obd->obd_recovering = 1;
- obd->obd_recovery_start = CURRENT_SECONDS;
- /* Only used for lprocfs_status */
- obd->obd_recovery_end = obd->obd_recovery_start +
- OBD_RECOVERY_TIMEOUT;
- }
+ target_recovery_init(obd, ost_handle);
out:
filter->fo_mount_count = mount_count + 1;
RETURN(0);
err_client:
- class_disconnect_exports(obd);
+ target_recovery_fini(obd);
err_fsd:
filter_free_server_data(filter);
RETURN(rc);
struct filter_obd *filter = &obd->u.filter;
struct file *filp;
struct dentry *dentry;
- int i;
+ int i, j;
ENTRY;
if (filter->fo_dentry_O_groups != NULL) {
- for (i = 0; i < FILTER_GROUPS; i++) {
+ for (i = 0; i < filter->fo_group_count; i++) {
dentry = filter->fo_dentry_O_groups[i];
if (dentry != NULL)
f_dput(dentry);
}
OBD_FREE(filter->fo_dentry_O_groups,
- FILTER_GROUPS * sizeof(*filter->fo_dentry_O_groups));
+ filter->fo_group_count *
+ sizeof(*filter->fo_dentry_O_groups));
filter->fo_dentry_O_groups = NULL;
}
if (filter->fo_last_objid_files != NULL) {
- for (i = 0; i < FILTER_GROUPS; i++) {
+ for (i = 0; i < filter->fo_group_count; i++) {
filp = filter->fo_last_objid_files[i];
if (filp != NULL)
filp_close(filp, 0);
}
OBD_FREE(filter->fo_last_objid_files,
- FILTER_GROUPS * sizeof(*filter->fo_last_objid_files));
+ filter->fo_group_count *
+ sizeof(*filter->fo_last_objid_files));
filter->fo_last_objid_files = NULL;
}
if (filter->fo_dentry_O_sub != NULL) {
- for (i = 0; i < filter->fo_subdir_count; i++) {
- dentry = filter->fo_dentry_O_sub[i];
- if (dentry != NULL)
- f_dput(dentry);
+ for (i = 0; i < filter->fo_group_count; i++) {
+ for (j = 0; j < filter->fo_subdir_count; j++) {
+ dentry = filter->fo_dentry_O_sub[i].dentry[j];
+ if (dentry != NULL)
+ f_dput(dentry);
+ }
}
OBD_FREE(filter->fo_dentry_O_sub,
- filter->fo_subdir_count *
+ filter->fo_group_count *
sizeof(*filter->fo_dentry_O_sub));
filter->fo_dentry_O_sub = NULL;
}
if (filter->fo_last_objids != NULL) {
OBD_FREE(filter->fo_last_objids,
- FILTER_GROUPS * sizeof(*filter->fo_last_objids));
+ filter->fo_group_count *
+ sizeof(*filter->fo_last_objids));
filter->fo_last_objids = NULL;
}
if (filter->fo_dentry_O != NULL) {
RETURN(0);
}
+static int filter_update_last_group(struct obd_device *obd, int group)
+{
+ struct filter_obd *filter = &obd->u.filter;
+ struct file *filp = NULL;
+ int last_group = 0, rc;
+ loff_t off = 0;
+ ENTRY;
+
+ if (group <= filter->fo_committed_group)
+ RETURN(0);
+
+ filp = filp_open("LAST_GROUP", O_RDWR, 0700);
+ if (IS_ERR(filp)) {
+ rc = PTR_ERR(filp);
+ filp = NULL;
+ CERROR("cannot open LAST_GROUP: rc = %d\n", rc);
+ GOTO(cleanup, rc);
+ }
+
+ rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
+ if (rc) {
+ CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n",rc);
+ GOTO(cleanup, rc);
+ }
+ LASSERT(off == 0 || last_group >= FILTER_MIN_GROUPS);
+ CDEBUG(D_INODE, "%s: previous %d, new %d\n",
+ obd->obd_name, last_group, group);
+
+ off = 0;
+ last_group = group;
+ /* must be sync: bXXXX */
+ rc = fsfilt_write_record(obd, filp, &last_group, sizeof(__u32), &off, 1);
+ if (rc) {
+ CDEBUG(D_INODE, "error updating LAST_GROUP: rc %d\n", rc);
+ GOTO(cleanup, rc);
+ }
+
+ filter->fo_committed_group = group;
+cleanup:
+ if (filp)
+ filp_close(filp, 0);
+ RETURN(rc);
+}
+
+static int filter_read_group_internal(struct obd_device *obd, int group,
+ int create)
+{
+ struct filter_obd *filter = &obd->u.filter;
+ __u64 *new_objids = NULL;
+ struct filter_subdirs *new_subdirs = NULL, *tmp_subdirs = NULL;
+ struct dentry **new_groups = NULL;
+ struct file **new_files = NULL;
+ struct dentry *dentry;
+ struct file *filp;
+ int old_count = filter->fo_group_count, rc, stage = 0, i;
+ char name[25];
+ __u64 last_objid;
+ loff_t off = 0;
+ int len = group + 1;
+
+ snprintf(name, 24, "%d", group);
+ name[24] = '\0';
+
+ if (!create) {
+ dentry = ll_lookup_one_len(name, filter->fo_dentry_O,
+ strlen(name));
+ if (IS_ERR(dentry)) {
+ CERROR("Cannot lookup expected object group %d: %ld\n",
+ group, PTR_ERR(dentry));
+ RETURN(PTR_ERR(dentry));
+ }
+ } else {
+ dentry = simple_mkdir(filter->fo_dentry_O, name, 0700, 1);
+ if (IS_ERR(dentry)) {
+ CERROR("cannot lookup/create O/%s: rc = %ld\n", name,
+ PTR_ERR(dentry));
+ RETURN(PTR_ERR(dentry));
+ }
+ }
+ stage = 1;
+
+ snprintf(name, 24, "O/%d/LAST_ID", group);
+ name[24] = '\0';
+ filp = filp_open(name, O_CREAT | O_RDWR, 0700);
+ if (IS_ERR(filp)) {
+ CERROR("cannot create %s: rc = %ld\n", name, PTR_ERR(filp));
+ GOTO(cleanup, rc = PTR_ERR(filp));
+ }
+ stage = 2;
+
+ rc = fsfilt_read_record(obd, filp, &last_objid, sizeof(__u64), &off);
+ if (rc) {
+ CDEBUG(D_INODE, "error reading %s: rc %d\n", name, rc);
+ GOTO(cleanup, rc);
+ }
+
+ if (filter->fo_subdir_count) {
+ OBD_ALLOC(tmp_subdirs, sizeof(*tmp_subdirs));
+ if (tmp_subdirs == NULL)
+ GOTO(cleanup, rc = -ENOMEM);
+ stage = 3;
+
+ for (i = 0; i < filter->fo_subdir_count; i++) {
+ char dir[20];
+ snprintf(dir, sizeof(dir), "d%u", i);
+
+ tmp_subdirs->dentry[i] = simple_mkdir(dentry, dir, 0700, 1);
+ if (IS_ERR(tmp_subdirs->dentry[i])) {
+ rc = PTR_ERR(tmp_subdirs->dentry[i]);
+ CERROR("can't lookup/create O/%d/%s: rc = %d\n",
+ group, dir, rc);
+ GOTO(cleanup, rc);
+ }
+
+ CDEBUG(D_INODE, "got/created O/%d/%s: %p\n", group, dir,
+ tmp_subdirs->dentry[i]);
+ }
+ }
+
+ /* 'group' is an index; we need an array of length 'group + 1' */
+ if (group + 1 > old_count) {
+ OBD_ALLOC(new_objids, len * sizeof(*new_objids));
+ OBD_ALLOC(new_subdirs, len * sizeof(*new_subdirs));
+ OBD_ALLOC(new_groups, len * sizeof(*new_groups));
+ OBD_ALLOC(new_files, len * sizeof(*new_files));
+ stage = 4;
+ if (new_objids == NULL || new_subdirs == NULL ||
+ new_groups == NULL || new_files == NULL)
+ GOTO(cleanup, rc = -ENOMEM);
+
+ if (old_count) {
+ memcpy(new_objids, filter->fo_last_objids,
+ old_count * sizeof(*new_objids));
+ memcpy(new_subdirs, filter->fo_dentry_O_sub,
+ old_count * sizeof(*new_subdirs));
+ memcpy(new_groups, filter->fo_dentry_O_groups,
+ old_count * sizeof(*new_groups));
+ memcpy(new_files, filter->fo_last_objid_files,
+ old_count * sizeof(*new_files));
+
+ OBD_FREE(filter->fo_last_objids,
+ old_count * sizeof(*new_objids));
+ OBD_FREE(filter->fo_dentry_O_sub,
+ old_count * sizeof(*new_subdirs));
+ OBD_FREE(filter->fo_dentry_O_groups,
+ old_count * sizeof(*new_groups));
+ OBD_FREE(filter->fo_last_objid_files,
+ old_count * sizeof(*new_files));
+ }
+ filter->fo_last_objids = new_objids;
+ filter->fo_dentry_O_sub = new_subdirs;
+ filter->fo_dentry_O_groups = new_groups;
+ filter->fo_last_objid_files = new_files;
+ filter->fo_group_count = len;
+ }
+
+ filter->fo_dentry_O_groups[group] = dentry;
+ filter->fo_last_objid_files[group] = filp;
+ if (filter->fo_subdir_count) {
+ filter->fo_dentry_O_sub[group] = *tmp_subdirs;
+ OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
+ }
+
+ filter_update_last_group(obd, group);
+
+ if (i_size_read(filp->f_dentry->d_inode) == 0) {
+ filter->fo_last_objids[group] = FILTER_INIT_OBJID;
+ rc = filter_update_last_objid(obd, group, 1);
+ RETURN(rc);
+ }
+
+ filter->fo_last_objids[group] = le64_to_cpu(last_objid);
+ CDEBUG(D_INODE, "%s: server last_objid group %d: "LPU64"\n",
+ obd->obd_name, group, last_objid);
+ RETURN(0);
+ cleanup:
+ switch (stage) {
+ case 4:
+ if (new_objids != NULL)
+ OBD_FREE(new_objids, len * sizeof(*new_objids));
+ if (new_subdirs != NULL)
+ OBD_FREE(new_subdirs, len * sizeof(*new_subdirs));
+ if (new_groups != NULL)
+ OBD_FREE(new_groups, len * sizeof(*new_groups));
+ if (new_files != NULL)
+ OBD_FREE(new_files, len * sizeof(*new_files));
+ case 3:
+ if (filter->fo_subdir_count) {
+ for (i = 0; i < filter->fo_subdir_count; i++) {
+ if (tmp_subdirs->dentry[i] != NULL)
+ dput(tmp_subdirs->dentry[i]);
+ }
+ OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs));
+ }
+ case 2:
+ filp_close(filp, 0);
+ case 1:
+ dput(dentry);
+ }
+ RETURN(rc);
+}
+
+static int filter_read_groups(struct obd_device *obd, int last_group,
+ int create)
+{
+ struct filter_obd *filter = &obd->u.filter;
+ int old_count, group, rc = 0;
+
+ 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)
+ break;
+ }
+ up(&filter->fo_init_lock);
+ return rc;
+}
+
/* FIXME: object groups */
static int filter_prep_groups(struct obd_device *obd)
{
struct filter_obd *filter = &obd->u.filter;
struct dentry *dentry, *O_dentry;
struct file *filp;
- int i, rc = 0, cleanup_phase = 0;
+ int last_group, rc = 0, cleanup_phase = 0;
+ loff_t off = 0;
ENTRY;
O_dentry = simple_mkdir(current->fs->pwd, "O", 0700, 1);
}
filter->fo_dentry_O = O_dentry;
cleanup_phase = 1; /* O_dentry */
-
- OBD_ALLOC(filter->fo_last_objids, FILTER_GROUPS * sizeof(__u64));
- if (filter->fo_last_objids == NULL)
- GOTO(cleanup, rc = -ENOMEM);
- cleanup_phase = 2; /* groups */
-
- OBD_ALLOC(filter->fo_dentry_O_groups, FILTER_GROUPS * sizeof(dentry));
- if (filter->fo_dentry_O_groups == NULL)
- GOTO(cleanup, rc = -ENOMEM);
- OBD_ALLOC(filter->fo_last_objid_files, FILTER_GROUPS * sizeof(filp));
- if (filter->fo_last_objid_files == NULL)
- GOTO(cleanup, rc = -ENOMEM);
-
- for (i = 0; i < FILTER_GROUPS; i++) {
- char name[25];
- loff_t off = 0;
- sprintf(name, "%d", i);
- dentry = simple_mkdir(O_dentry, name, 0700, 1);
- CDEBUG(D_INODE, "got/created O/%s: %p\n", name, dentry);
- if (IS_ERR(dentry)) {
- rc = PTR_ERR(dentry);
- CERROR("cannot lookup/create O/%s: rc = %d\n",
- name, rc);
- GOTO(cleanup, rc);
+ /* Lookup "R" to tell if we're on an old OST FS and need to convert
+ * from O/R/<dir>/<objid> to O/0/<dir>/<objid>. This can be removed
+ * some time post 1.0 when all old-style OSTs have converted along
+ * with the init_objid hack. */
+ dentry = ll_lookup_one_len("R", O_dentry, 1);
+ if (IS_ERR(dentry))
+ GOTO(cleanup, rc = PTR_ERR(dentry));
+ if (dentry->d_inode && S_ISDIR(dentry->d_inode->i_mode)) {
+ struct dentry *O0_dentry = lookup_one_len("0", O_dentry, 1);
+ ENTRY;
+
+ CWARN("converting OST to new object layout\n");
+ if (IS_ERR(O0_dentry)) {
+ rc = PTR_ERR(O0_dentry);
+ CERROR("error looking up O/0: rc %d\n", rc);
+ GOTO(cleanup_R, rc);
}
- filter->fo_dentry_O_groups[i] = dentry;
- sprintf(name, "O/%d/LAST_ID", i);
- filp = filp_open(name, O_CREAT | O_RDWR, 0700);
- if (IS_ERR(filp)) {
- rc = PTR_ERR(filp);
- CERROR("cannot create %s: rc = %d\n", name, rc);
- GOTO(cleanup, rc);
+ if (O0_dentry->d_inode) {
+ CERROR("Both O/R and O/0 exist. Fix manually.\n");
+ GOTO(cleanup_O0, rc = -EEXIST);
}
- filter->fo_last_objid_files[i] = filp;
- if (filp->f_dentry->d_inode->i_size == 0) {
- filter->fo_last_objids[i] = FILTER_INIT_OBJID;
- rc = filter_update_last_objid(obd, i, 1);
- if (rc)
- GOTO(cleanup, rc);
- continue;
- }
+ LOCK_INODE_MUTEX(O_dentry->d_inode);
+ rc = vfs_rename(O_dentry->d_inode, dentry,
+ O_dentry->d_inode, O0_dentry);
+ UNLOCK_INODE_MUTEX(O_dentry->d_inode);
- rc = fsfilt_read_record(obd, filp, &filter->fo_last_objids[i],
- sizeof(__u64), &off);
if (rc) {
- CDEBUG(D_INODE,"OBD filter: error reading %s: rc %d\n",
- name, rc);
- GOTO(cleanup, rc);
+ CERROR("error renaming O/R to O/0: rc %d\n", rc);
+ GOTO(cleanup_O0, rc);
}
- filter->fo_last_objids[i] =
- le64_to_cpu(filter->fo_last_objids[i]);
- CDEBUG(D_HA, "%s: server last_objid group %d: "LPU64"\n",
- obd->obd_name, i, filter->fo_last_objids[i]);
+ filter->fo_fsd->lsd_feature_incompat |=
+ cpu_to_le32(OBD_INCOMPAT_GROUPS);
+ rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
+ filter->fo_fsd, 1);
+ GOTO(cleanup_O0, rc);
+
+ cleanup_O0:
+ f_dput(O0_dentry);
+ cleanup_R:
+ f_dput(dentry);
+ if (rc)
+ GOTO(cleanup, rc);
+ } else {
+ f_dput(dentry);
}
- if (filter->fo_subdir_count) {
- O_dentry = filter->fo_dentry_O_groups[0];
- OBD_ALLOC(filter->fo_dentry_O_sub,
- filter->fo_subdir_count * sizeof(dentry));
- if (filter->fo_dentry_O_sub == NULL)
- GOTO(cleanup, rc = -ENOMEM);
+ cleanup_phase = 2; /* groups */
- for (i = 0; i < filter->fo_subdir_count; i++) {
- char dir[20];
- snprintf(dir, sizeof(dir), "d%u", i);
+ /* we have to initialize all groups before first connections from
+ * 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));
+ GOTO(cleanup, rc = PTR_ERR(filp));
+ }
+ cleanup_phase = 3; /* filp */
- dentry = simple_mkdir(O_dentry, dir, 0700, 1);
- CDEBUG(D_INODE, "got/created O/0/%s: %p\n", dir,dentry);
- if (IS_ERR(dentry)) {
- rc = PTR_ERR(dentry);
- CERROR("can't lookup/create O/0/%s: rc = %d\n",
- dir, rc);
- GOTO(cleanup, rc);
- }
- filter->fo_dentry_O_sub[i] = dentry;
- }
+ rc = fsfilt_read_record(obd, filp, &last_group, sizeof(__u32), &off);
+ if (rc) {
+ CDEBUG(D_INODE, "error reading LAST_GROUP: rc %d\n", rc);
+ GOTO(cleanup, rc);
+ }
+ if (off == 0) {
+ last_group = FILTER_MIN_GROUPS;
+ } else {
+ LASSERT(last_group >= FILTER_MIN_GROUPS);
}
+
+ CWARN("%s: initialize groups [%d,%d]\n", obd->obd_name,
+ FILTER_MIN_GROUPS, last_group);
+ filter->fo_committed_group = last_group;
+ rc = filter_read_groups(obd, last_group, 1);
+ if (rc)
+ GOTO(cleanup, rc);
+
+ filp_close(filp, 0);
RETURN(0);
cleanup:
- filter_cleanup_groups(obd);
+ switch (cleanup_phase) {
+ case 3:
+ filp_close(filp, 0);
+ case 2:
+ filter_cleanup_groups(obd);
+ case 1:
+ f_dput(filter->fo_dentry_O);
+ filter->fo_dentry_O = NULL;
+ default:
+ break;
+ }
return rc;
+
}
/* setup the object store with correct subdirectories */
rc = PTR_ERR(file);
CERROR("OBD filter: cannot open/create %s rc = %d\n",
HEALTH_CHECK, rc);
- GOTO(err_filp, rc);
+ GOTO(err_server_data, rc);
}
filter->fo_health_check_filp = file;
if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
rc = filter_prep_groups(obd);
if (rc)
- GOTO(err_server_data, rc);
- out:
+ GOTO(err_health_check, rc);
+out:
pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
return(rc);
- err_server_data:
- //class_disconnect_exports(obd, 0);
- filter_free_server_data(filter);
- err_health_check:
+err_health_check:
if (filp_close(filter->fo_health_check_filp, 0))
CERROR("can't close %s after error\n", HEALTH_CHECK);
filter->fo_health_check_filp = NULL;
- err_filp:
+err_server_data:
+ target_recovery_fini(obd);
+ filter_free_server_data(filter);
+err_filp:
if (filp_close(filter->fo_rcvd_filp, 0))
CERROR("can't close %s after error\n", LAST_RCVD);
filter->fo_rcvd_filp = NULL;
if (rc)
CERROR("error writing server data: rc = %d\n", rc);
- for (i = 0; i < FILTER_GROUPS; i++) {
- rc = filter_update_last_objid(obd, i, (i == FILTER_GROUPS - 1));
+ for (i = 1; i < filter->fo_group_count; i++) {
+ rc = filter_update_last_objid(obd, i,
+ (i == filter->fo_group_count - 1));
if (rc)
CERROR("error writing group %d lastobjid: rc = %d\n",
i, rc);
filter_cleanup_groups(obd);
filter_free_server_data(filter);
pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+
+ filter_free_capa_keys(filter);
+ cleanup_capa_hash(filter->fo_capa_hash);
}
static void filter_set_last_id(struct filter_obd *filter,
obd_id id, obd_gr group)
{
LASSERT(filter->fo_fsd != NULL);
- LASSERT(group <= FILTER_GROUPS);
+ LASSERT(group <= filter->fo_group_count);
spin_lock(&filter->fo_objidlock);
filter->fo_last_objids[group] = id;
{
obd_id id;
LASSERT(filter->fo_fsd != NULL);
- LASSERT(group <= FILTER_GROUPS);
+ LASSERT(group <= filter->fo_group_count);
/* FIXME: object groups */
spin_lock(&filter->fo_objidlock);
struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid)
{
struct filter_obd *filter = &obd->u.filter;
- LASSERT(group < FILTER_GROUPS); /* FIXME: object groups */
+ struct filter_subdirs *subdirs;
+ LASSERT(group < filter->fo_group_count); /* FIXME: object groups */
if (group > 0 || filter->fo_subdir_count == 0)
return filter->fo_dentry_O_groups[group];
- return filter->fo_dentry_O_sub[objid & (filter->fo_subdir_count - 1)];
+ subdirs = &filter->fo_dentry_O_sub[group];
+ return subdirs->dentry[objid & (filter->fo_subdir_count - 1)];
}
/* We never dget the object parent, so DON'T dput it either */
if (IS_ERR(dparent))
return dparent;
+ if (dparent == NULL)
+ return ERR_PTR(-ENOENT);
rc = filter_lock_dentry(obd, dparent);
fsfilt_check_slow(obd, now, obd_timeout, "parent lock");
ENTRY;
if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT) &&
- !obd->u.filter.fo_destroy_in_progress) {
+ 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);
RETURN(ERR_PTR(-ENOENT));
}
-
if (id == 0) {
CERROR("fatal: invalid object id 0\n");
RETURN(ERR_PTR(-ESTALE));
RETURN(dchild);
}
-static int filter_prepare_destroy(struct obd_device *obd, obd_id objid)
+static int filter_prepare_destroy(struct obd_device *obd, obd_id objid,
+ obd_id group)
{
struct lustre_handle lockh;
int flags = LDLM_AST_DISCARD_DATA, rc;
- struct ldlm_res_id res_id = { .name = { objid } };
+ struct ldlm_res_id res_id = { .name = { objid, 0, group, 0} };
ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
ENTRY;
/* Tell the clients that the object is gone now and that they should
* throw away any cached pages. */
- rc = ldlm_cli_enqueue_local(obd->obd_namespace, res_id, LDLM_EXTENT,
+ 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);
* here) or some other ordering issue. */
DQUOT_INIT(dir);
-#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
rc = security_inode_unlink(dir, dentry);
if (rc)
GOTO(out, rc);
-#endif
rc = dir->i_op->unlink(dir, dentry);
out:
return(rc);
}
+struct filter_intent_args {
+ struct ldlm_lock **victim;
+ __u64 size;
+ int *liblustre;
+};
+
+static enum interval_iter filter_intent_cb(struct interval_node *n,
+ void *args)
+{
+ struct ldlm_interval *node = (struct ldlm_interval *)n;
+ struct filter_intent_args *arg = (struct filter_intent_args*)args;
+ __u64 size = arg->size;
+ struct ldlm_lock **v = arg->victim;
+ struct ldlm_lock *lck;
+
+ /* If the interval is lower than the current file size,
+ * just break. */
+ if (interval_high(n) <= size)
+ return INTERVAL_ITER_STOP;
+
+ list_for_each_entry(lck, &node->li_group, l_sl_policy) {
+ /* Don't send glimpse ASTs to liblustre clients.
+ * They aren't listening for them, and they do
+ * entirely synchronous I/O anyways. */
+ if (lck->l_export == NULL ||
+ lck->l_export->exp_libclient == 1)
+ continue;
+
+ if (*arg->liblustre)
+ *arg->liblustre = 0;
+
+ if (*v == NULL) {
+ *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);
+ *v = LDLM_LOCK_GET(lck);
+ }
+
+ /* the same policy group - every lock has the
+ * same extent, so needn't do it any more */
+ break;
+ }
+
+ return INTERVAL_ITER_CONT;
+}
+
static int filter_intent_policy(struct ldlm_namespace *ns,
struct ldlm_lock **lockp, void *req_cookie,
ldlm_mode_t mode, int flags, void *data)
{
- struct list_head rpc_list = LIST_HEAD_INIT(rpc_list);
+ 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;
ldlm_processing_policy policy;
struct ost_lvb *res_lvb, *reply_lvb;
struct ldlm_reply *rep;
- struct list_head *tmp;
ldlm_error_t err;
- int rc, tmpflags = 0, only_liblustre = 0;
+ int idx, rc, tmpflags = 0, only_liblustre = 1;
+ struct ldlm_interval_tree *tree;
+ struct filter_intent_args arg;
int repsize[3] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
[DLM_LOCKREPLY_OFF] = sizeof(*rep),
[DLM_REPLY_REC_OFF] = sizeof(*reply_lvb) };
/* If we grant any lock at all, it will be a whole-file read lock.
* Call the extent policy function to see if our request can be
- * granted, or is blocked. */
+ * granted, or is blocked.
+ * If the OST lock has LDLM_FL_HAS_INTENT set, it means a glimpse lock
+ */
lock->l_policy_data.l_extent.start = 0;
lock->l_policy_data.l_extent.end = OBD_OBJECT_EOF;
lock->l_req_mode = LCK_PR;
* 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 ||
- OBD_FAIL_CHECK(OBD_FAIL_LDLM_GLIMPSE)) {
- OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_GLIMPSE, 2);
+ OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_GLIMPSE, 2)) {
ldlm_resource_unlink_lock(lock);
err = ELDLM_LOCK_ABORTED;
} else {
LASSERT(res_lvb != NULL);
*reply_lvb = *res_lvb;
- list_for_each(tmp, &res->lr_granted) {
- struct ldlm_lock *tmplock =
- list_entry(tmp, struct ldlm_lock, l_res_link);
-
- if (tmplock->l_granted_mode == LCK_PR)
- continue;
- /*
- * ->ns_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.
- */
- if (tmplock->l_policy_data.l_extent.end <= reply_lvb->lvb_size)
- continue;
-
- /* Don't send glimpse ASTs to liblustre clients. They aren't
- * listening for them, and they do entirely synchronous I/O
- * anyways. */
- if (tmplock->l_export == NULL ||
- tmplock->l_export->exp_libclient == 1) {
- only_liblustre = 1;
- continue;
- }
-
- if (l == NULL) {
- l = LDLM_LOCK_GET(tmplock);
- continue;
- }
-
- if (l->l_policy_data.l_extent.start >
- tmplock->l_policy_data.l_extent.start)
+ /*
+ * ->ns_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.
+ */
+ arg.size = reply_lvb->lvb_size;
+ arg.victim = &l;
+ arg.liblustre = &only_liblustre;
+ for (idx = 0; idx < LCK_MODE_NUM; idx++) {
+ tree = &res->lr_itree[idx];
+ if (tree->lit_mode == LCK_PR)
continue;
- LDLM_LOCK_PUT(l);
- l = LDLM_LOCK_GET(tmplock);
+ interval_iterate_reverse(tree->lit_root,
+ filter_intent_cb, &arg);
}
unlock_res(res);
*
* Of course, this will all disappear when we switch to
* taking liblustre locks on the OST. */
- if (ns->ns_lvbo && ns->ns_lvbo->lvbo_update)
- ns->ns_lvbo->lvbo_update(res, NULL, 0, 1);
+ ldlm_res_lvbo_update(res, NULL, 0, 1);
}
RETURN(ELDLM_LOCK_ABORTED);
}
* 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 && ns->ns_lvbo && ns->ns_lvbo->lvbo_update)
- ns->ns_lvbo->lvbo_update(res, NULL, 0, 1);
+ if (rc != 0)
+ ldlm_res_lvbo_update(res, NULL, 0, 1);
lock_res(res);
*reply_lvb = *res_lvb;
ENTRY;
+
OBD_ALLOC_GFP(filter->fo_iobuf_pool, OSS_THREADS_MAX * sizeof(*pool),
GFP_KERNEL);
if (filter->fo_iobuf_pool == NULL)
* 3 = flags: failover=f, failout=n
* 4 = mount options
*/
-int filter_common_setup(struct obd_device *obd, obd_count len, void *buf,
+int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg,
void *option)
{
- struct lustre_cfg* lcfg = buf;
struct filter_obd *filter = &obd->u.filter;
struct vfsmount *mnt;
struct lustre_mount_info *lmi;
__u8 *uuid_ptr;
char *str, *label;
char ns_name[48];
- int rc;
+ int rc, i;
ENTRY;
if (lcfg->lcfg_bufcount < 3 ||
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);
+ lustre_cfg_string(lcfg, 1), option);
if (IS_ERR(mnt)) {
rc = PTR_ERR(mnt);
- LCONSOLE_ERROR("Can't mount disk %s (%d)\n",
- lustre_cfg_string(lcfg, 1), rc);
+ LCONSOLE_ERROR_MSG(0x135, "Can't mount disk %s (%d)\n",
+ lustre_cfg_string(lcfg, 1), rc);
RETURN(rc);
}
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);
+
OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
obd->obd_lvfs_ctxt.pwdmnt = mnt;
obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
obd->obd_lvfs_ctxt.fs = get_ds();
obd->obd_lvfs_ctxt.cb_ops = filter_lvfs_ops;
+ sema_init(&filter->fo_init_lock, 1);
+ filter->fo_committed_group = 0;
+
rc = filter_prep(obd);
if (rc)
GOTO(err_ops, rc);
- filter->fo_destroy_in_progress = 0;
- sema_init(&filter->fo_create_lock, 1);
+ filter->fo_destroys_in_progress = 0;
+ for (i = 0; i < 32; i++)
+ sema_init(&filter->fo_create_locks[i], 1);
+
spin_lock_init(&filter->fo_translock);
spin_lock_init(&filter->fo_objidlock);
- INIT_LIST_HEAD(&filter->fo_export_list);
+ CFS_INIT_LIST_HEAD(&filter->fo_export_list);
sema_init(&filter->fo_alloc_lock, 1);
init_brw_stats(&filter->fo_filter_stats);
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;
+ CFS_INIT_LIST_HEAD(&filter->fo_capa_keys);
+ filter->fo_capa_hash = init_capa_hash();
+ if (filter->fo_capa_hash == NULL)
+ GOTO(err_ops, rc = -ENOMEM);
+
sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid);
- obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER);
+ obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER,
+ LDLM_NAMESPACE_GREEDY);
if (obd->obd_namespace == NULL)
GOTO(err_post, rc = -ENOMEM);
obd->obd_namespace->ns_lvbp = obd;
ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
"filter_ldlm_cb_client", &obd->obd_ldlm_client);
- rc = llog_cat_initialize(obd, 1, NULL);
+ rc = llog_cat_initialize(obd, &obd->obd_olg, 1, NULL);
if (rc) {
CERROR("failed to setup llogging subsystems\n");
GOTO(err_post, rc);
} else {
str = "no UUID";
}
-
+
label = fsfilt_get_label(obd, obd->u.obt.obt_sb);
if (obd->obd_recovering) {
"/proc/fs/lustre/obdfilter/%s/recovery_status.\n",
obd->obd_name, lustre_cfg_string(lcfg, 1),
label ?: "", label ? "/" : "", str,
- obd->obd_recoverable_clients,
- (obd->obd_recoverable_clients == 1)
+ obd->obd_max_recoverable_clients,
+ (obd->obd_max_recoverable_clients == 1)
? "client" : "clients",
(int)(OBD_RECOVERY_TIMEOUT) / 60,
(int)(OBD_RECOVERY_TIMEOUT) % 60,
return rc;
}
-static int filter_setup(struct obd_device *obd, obd_count len, void *buf)
+static int filter_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
{
struct lprocfs_static_vars lvars;
- struct lustre_cfg* lcfg = buf;
- unsigned long page;
+ unsigned long addr;
+ struct page *page;
int rc;
CLASSERT(offsetof(struct obd_device, u.obt) ==
RETURN(-EINVAL);
/* 2.6.9 selinux wants a full option page for do_kern_mount (bug6471) */
- page = get_zeroed_page(GFP_KERNEL);
+ OBD_PAGE_ALLOC(page, CFS_ALLOC_STD);
if (!page)
RETURN(-ENOMEM);
+ addr = (unsigned long)cfs_page_address(page);
+ clear_page((void *)addr);
- memcpy((void *)page, lustre_cfg_buf(lcfg, 4),
- LUSTRE_CFG_BUFLEN(lcfg, 4));
- rc = filter_common_setup(obd, len, buf, (void *)page);
- free_page(page);
-
- lprocfs_init_vars(filter, &lvars);
- if (rc == 0 && lprocfs_obd_setup(obd, lvars.obd_vars) == 0 &&
+ /* 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_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES,
LPROCFS_CNTR_AVGMINMAX,
"write_bytes", "bytes");
+
lproc_filter_attach_seqstat(obd);
- obd->obd_proc_exports = proc_mkdir("exports",
- obd->obd_proc_entry);
+ 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;
+ }
+ }
+ 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);
+
+ 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);
}
return rc;
lop_add: llog_obd_origin_add
};
-static int filter_llog_init(struct obd_device *obd, struct obd_device *tgt,
- int count, struct llog_catid *catid,
+static int filter_llog_init(struct obd_device *obd, int group,
+ struct obd_device *tgt, int count,
+ struct llog_catid *catid,
struct obd_uuid *uuid)
{
+ struct filter_obd *filter = &obd->u.filter;
+ struct obd_llog_group *olg;
struct llog_ctxt *ctxt;
int rc;
ENTRY;
+ olg = filter_find_olg(obd, group);
+ if (IS_ERR(olg))
+ RETURN(PTR_ERR(olg));
+
+ if (group == OBD_LLOG_GROUP) {
+ LASSERT(filter->fo_lcm == NULL);
+ OBD_ALLOC(filter->fo_lcm, sizeof(struct llog_commit_master));
+ if (!filter->fo_lcm)
+ RETURN(-ENOMEM);
+
+ rc = llog_init_commit_master((struct llog_commit_master *)
+ filter->fo_lcm);
+ if (rc)
+ GOTO(cleanup, rc);
+
filter_mds_ost_repl_logops = llog_client_ops;
filter_mds_ost_repl_logops.lop_cancel = llog_obd_repl_cancel;
filter_mds_ost_repl_logops.lop_connect = llog_repl_connect;
filter_mds_ost_repl_logops.lop_sync = llog_obd_repl_sync;
-
- rc = llog_setup(obd, LLOG_MDS_OST_REPL_CTXT, tgt, 0, NULL,
+ } else {
+ LASSERT(filter->fo_lcm != NULL);
+ }
+ rc = llog_setup(obd, olg, LLOG_MDS_OST_REPL_CTXT, tgt, 0, NULL,
&filter_mds_ost_repl_logops);
if (rc)
- RETURN(rc);
+ GOTO(cleanup, rc);
/* FIXME - assign unlink_cb for filter's recovery */
- ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
+ LASSERT(olg);
+ ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
+
+ LASSERT(ctxt != NULL);
ctxt->llog_proc_cb = filter_recov_log_mds_ost_cb;
+ ctxt->loc_lcm = obd->u.filter.fo_lcm;
+ rc = llog_start_commit_thread(ctxt->loc_lcm);
+ llog_ctxt_put(ctxt);
+ if (rc)
+ GOTO(cleanup, rc);
- rc = llog_setup(obd, LLOG_SIZE_ORIG_CTXT, tgt, 0, NULL,
+ rc = llog_setup(obd, olg, LLOG_SIZE_ORIG_CTXT, tgt, 0, NULL,
&filter_size_orig_logops);
+
+cleanup:
+ if (rc) {
+ llog_cleanup_commit_master(filter->fo_lcm, 0);
+ OBD_FREE(filter->fo_lcm, sizeof(struct llog_commit_master));
+ filter->fo_lcm = NULL;
+ }
RETURN(rc);
}
-static int filter_llog_finish(struct obd_device *obd, int count)
+static int filter_group_llog_finish(struct obd_llog_group *olg)
{
struct llog_ctxt *ctxt;
int rc = 0, rc2 = 0;
ENTRY;
- ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
+ ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT);
if (ctxt)
rc = llog_cleanup(ctxt);
- ctxt = llog_get_context(obd, LLOG_SIZE_ORIG_CTXT);
+ ctxt = llog_group_get_ctxt(olg, LLOG_SIZE_ORIG_CTXT);
if (ctxt)
rc2 = llog_cleanup(ctxt);
if (!rc)
RETURN(rc);
}
+static int filter_llog_finish(struct obd_device *obd, int count)
+{
+ int rc;
+ ENTRY;
+
+ if (obd->u.filter.fo_lcm) {
+ llog_cleanup_commit_master((struct llog_commit_master *)
+ obd->u.filter.fo_lcm, 0);
+ OBD_FREE(obd->u.filter.fo_lcm,
+ sizeof(struct llog_commit_master));
+ obd->u.filter.fo_lcm = NULL;
+ }
+ /* finish obd llog group */
+ rc = filter_group_llog_finish(&obd->obd_olg);
+
+ RETURN(rc);
+}
+
+struct obd_llog_group *filter_find_olg(struct obd_device *obd, int group)
+{
+ struct obd_llog_group *olg, *nolg;
+ struct filter_obd *filter;
+ int rc;
+
+ filter = &obd->u.filter;
+
+ if (group == OBD_LLOG_GROUP)
+ RETURN(&obd->obd_olg);
+
+ spin_lock(&filter->fo_llog_list_lock);
+ list_for_each_entry(olg, &filter->fo_llog_list, olg_list) {
+ if (olg->olg_group == group) {
+ spin_unlock(&filter->fo_llog_list_lock);
+ RETURN(olg);
+ }
+ }
+ spin_unlock(&filter->fo_llog_list_lock);
+
+ OBD_ALLOC_PTR(olg);
+ if (olg == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+
+ llog_group_init(olg, group);
+ spin_lock(&filter->fo_llog_list_lock);
+ list_for_each_entry(nolg, &filter->fo_llog_list, olg_list) {
+ LASSERT(nolg->olg_group != group);
+ }
+ list_add(&olg->olg_list, &filter->fo_llog_list);
+ spin_unlock(&filter->fo_llog_list_lock);
+
+ rc = llog_cat_initialize(obd, olg, 1, NULL);
+ if (rc) {
+ spin_lock(&filter->fo_llog_list_lock);
+ list_del(&olg->olg_list);
+ spin_unlock(&filter->fo_llog_list_lock);
+ OBD_FREE_PTR(olg);
+ RETURN(ERR_PTR(rc));
+ }
+ CDEBUG(D_OTHER, "%s: new llog group %u (0x%p)\n",
+ obd->obd_name, group, olg);
+
+ RETURN(olg);
+}
+
+static int filter_llog_connect(struct obd_export *exp,
+ struct llogd_conn_body *body)
+{
+ struct obd_device *obd = exp->exp_obd;
+ struct llog_ctxt *ctxt;
+ struct obd_llog_group *olg;
+ 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);
+
+ olg = filter_find_olg(obd, body->lgdc_logid.lgl_ogr);
+ if (IS_ERR(olg))
+ RETURN(PTR_ERR(olg));
+ 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,
+ &body->lgdc_gen, NULL);
+ llog_ctxt_put(ctxt);
+ if (rc != 0)
+ CERROR("failed to connect rc %d idx %d\n", rc,
+ body->lgdc_ctxt_idx);
+
+ RETURN(rc);
+}
+
+static int filter_llog_preclean (struct obd_device *obd)
+{
+ struct obd_llog_group *olg;
+ struct filter_obd *filter;
+ int rc = 0;
+ ENTRY;
+
+ rc = obd_llog_finish(obd, 0);
+ if (rc)
+ CERROR("failed to cleanup llogging subsystem\n");
+
+ filter = &obd->u.filter;
+ spin_lock(&filter->fo_llog_list_lock);
+ while (!list_empty(&filter->fo_llog_list)) {
+ olg = list_entry(filter->fo_llog_list.next,
+ struct obd_llog_group, olg_list);
+ list_del(&olg->olg_list);
+ spin_unlock(&filter->fo_llog_list_lock);
+
+ rc = filter_group_llog_finish(olg);
+ if (rc)
+ CERROR("failed to cleanup llogging subsystem for %u\n",
+ olg->olg_group);
+ OBD_FREE_PTR(olg);
+ spin_lock(&filter->fo_llog_list_lock);
+ }
+ spin_unlock(&filter->fo_llog_list_lock);
+
+ RETURN(rc);
+}
+
static int filter_precleanup(struct obd_device *obd,
enum obd_cleanup_stage stage)
{
break;
case OBD_CLEANUP_EXPORTS:
target_cleanup_recovery(obd);
+ rc = filter_llog_preclean(obd);
break;
case OBD_CLEANUP_SELF_EXP:
- rc = filter_llog_finish(obd, 0);
break;
case OBD_CLEANUP_OBD:
break;
}
}
- lprocfs_obd_cleanup(obd);
+ 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);
+ /* Stop recovery before namespace cleanup. */
+ target_stop_recovery_thread(obd);
+ target_cleanup_recovery(obd);
+
ldlm_namespace_free(obd->obd_namespace, obd->obd_force);
+ sptlrpc_rule_set_free(&filter->fo_sptlrpc_rset);
+
if (obd->u.obt.obt_sb == NULL)
RETURN(0);
filter_post(obd);
- shrink_dcache_parent(obd->u.obt.obt_sb->s_root);
-
LL_DQUOT_OFF(obd->u.obt.obt_sb);
+ shrink_dcache_sb(obd->u.obt.obt_sb);
server_put_mount(obd->obd_name, filter->fo_vfsmnt);
obd->u.obt.obt_sb = NULL;
exp->exp_connect_flags = data->ocd_connect_flags;
data->ocd_version = LUSTRE_VERSION_CODE;
+ if ((exp->exp_connect_flags & OBD_CONNECT_FID) == 0) {
+ CWARN("%s: OST requires FID support (flag="LPX64
+ "), but client not\n",
+ exp->exp_obd->obd_name,
+ exp->exp_connect_flags);
+ RETURN(-EBADF);
+ }
+
if (exp->exp_connect_flags & OBD_CONNECT_GRANT) {
+ struct filter_export_data *fed = &exp->exp_filter_data;
obd_size left, want;
spin_lock(&exp->exp_obd->obd_osfs_lock);
left = filter_grant_space_left(exp);
want = data->ocd_grant;
- data->ocd_grant = filter_grant(exp, 0, want, left);
+ filter_grant(exp, fed->fed_grant, want, left);
+ data->ocd_grant = fed->fed_grant;
spin_unlock(&exp->exp_obd->obd_osfs_lock);
CDEBUG(D_CACHE, "%s: cli %s/%p ocd_grant: %d want: "
struct filter_obd *filter = &exp->exp_obd->u.filter;
struct lr_server_data *lsd = filter->fo_fsd;
int index = le32_to_cpu(lsd->lsd_ost_index);
-
+
if (!(lsd->lsd_feature_compat &
cpu_to_le32(OBD_COMPAT_OST))) {
/* this will only happen on the first connect */
filter_update_server_data(exp->exp_obd,
filter->fo_rcvd_filp, lsd, 1);
} else if (index != data->ocd_index) {
- LCONSOLE_ERROR("Connection from %s to index %u doesn't "
- "match actual OST index %u in last_rcvd "
- "file, bad configuration?\n",
- obd_export_nid2str(exp), index,
- data->ocd_index);
+ LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
+ " %u doesn't match actual OST index"
+ " %u in last_rcvd file, bad "
+ "configuration?\n",
+ obd_export_nid2str(exp), index,
+ data->ocd_index);
RETURN(-EBADF);
}
}
data->ocd_brw_size = 65536;
} 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);
+ (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT));
+ LASSERT(data->ocd_brw_size);
+ }
+
+ if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) {
+ __u32 cksum_types = data->ocd_cksum_types;
+
+ /* 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 = OBD_CKSUM_CRC32;
+
+ CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return "
+ "%x\n", exp->exp_obd->obd_name,
+ obd_export_nid2str(exp), cksum_types,
+ data->ocd_cksum_types);
+ } else {
+ /* This client does not support OBD_CONNECT_CKSUM
+ * fall back to CRC32 */
+ CDEBUG(D_RPCTRACE, "%s: cli %s does not support "
+ "OBD_CONNECT_CKSUM, CRC32 will be used\n",
+ exp->exp_obd->obd_name,
+ obd_export_nid2str(exp));
}
/* FIXME: Do the same with the MDS UUID and fsd_peeruuid.
RETURN(0);
}
-static int filter_reconnect(struct obd_export *exp, struct obd_device *obd,
+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)
{
}
/* nearly identical to mds_connect */
-static int filter_connect(struct lustre_handle *conn, struct obd_device *obd,
+static int filter_connect(const struct lu_env *env,
+ struct lustre_handle *conn, struct obd_device *obd,
struct obd_uuid *cluuid,
- struct obd_connect_data *data)
+ struct obd_connect_data *data, void *localdata)
{
+ struct lvfs_run_ctxt saved;
struct obd_export *exp;
struct filter_export_data *fed;
struct filter_client_data *fcd = NULL;
+ __u32 group;
int rc;
ENTRY;
if (rc)
GOTO(cleanup, rc);
- filter_export_stats_init(obd, exp);
-
- if (!obd->obd_replayable)
- GOTO(cleanup, rc = 0);
+ filter_export_stats_init(obd, exp, localdata);
+ group = data->ocd_group;
+ if (obd->obd_replayable) {
+ OBD_ALLOC(fcd, sizeof(*fcd));
+ if (!fcd) {
+ CERROR("filter: out of memory for client data\n");
+ GOTO(cleanup, rc = -ENOMEM);
+ }
- OBD_ALLOC(fcd, sizeof(*fcd));
- if (!fcd) {
- CERROR("filter: out of memory for client data\n");
- GOTO(cleanup, rc = -ENOMEM);
+ memcpy(fcd->fcd_uuid, cluuid, sizeof(fcd->fcd_uuid));
+ fed->fed_fcd = fcd;
+ fed->fed_fcd->fcd_group = group;
+ rc = filter_client_add(obd, exp, -1);
+ if (rc)
+ GOTO(cleanup, rc);
}
+ CWARN("%s: Received MDS connection ("LPX64"); group %d\n",
+ obd->obd_name, exp->exp_handle.h_cookie, group);
+ if (group == 0)
+ GOTO(cleanup, rc);
- memcpy(fcd->fcd_uuid, cluuid, sizeof(fcd->fcd_uuid));
- fed->fed_fcd = fcd;
+ if (fed->fed_group != 0 && 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. Tell CFS.\n",
+ obd_export_nid2str(exp), fed->fed_group, group);
+ GOTO(cleanup, rc = -EPROTO);
+ }
+ fed->fed_group = group;
- rc = filter_client_add(obd, exp, -1);
+ push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+ rc = filter_read_groups(obd, group, 1);
+ pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+ if (rc != 0) {
+ CERROR("can't read group %u\n", group);
+ GOTO(cleanup, rc);
+ }
GOTO(cleanup, rc);
if (list_empty(&obd->obd_exports))
return;
+ /* We don't want to do this for large machines that do lots of
+ mounts or unmounts. It burns... */
+ if (obd->obd_num_exports > 100)
+ return;
+
spin_lock(&obd->obd_osfs_lock);
spin_lock(&obd->obd_dev_lock);
list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
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);
+ */
+
target_destroy_export(exp);
if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid))
RETURN(0);
}
+static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp)
+{
+ struct obd_llog_group *olg_min, *olg;
+ struct filter_obd *filter;
+ int worked = 0, group;
+ struct llog_ctxt *ctxt;
+ ENTRY;
+
+ filter = &obd->u.filter;
+
+ /* we can't sync log holding spinlock. also, we do not want to get
+ * into livelock. so we do following: loop over MDS's exports in
+ * group order and skip already synced llogs -bzzz */
+ do {
+ /* look for group with min. number, but > worked */
+ olg_min = NULL;
+ group = 1 << 30;
+ spin_lock(&filter->fo_llog_list_lock);
+ list_for_each_entry(olg, &filter->fo_llog_list, olg_list) {
+ if (olg->olg_group <= worked) {
+ /* this group is already synced */
+ continue;
+ }
+ if (group < olg->olg_group) {
+ /* we have group with smaller number to sync */
+ continue;
+ }
+ /* store current minimal group */
+ olg_min = olg;
+ group = olg->olg_group;
+ }
+ spin_unlock(&filter->fo_llog_list_lock);
+
+ if (olg_min == NULL)
+ break;
+
+ worked = olg_min->olg_group;
+ 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);
+ 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);
+ }
+ } while (olg_min != NULL);
+}
+
/* also incredibly similar to mds_disconnect */
static int filter_disconnect(struct obd_export *exp)
{
struct obd_device *obd = exp->exp_obd;
- struct llog_ctxt *ctxt;
- int rc, err;
+ int rc;
ENTRY;
LASSERT(exp);
if (exp->exp_obd->obd_namespace != NULL)
ldlm_cancel_locks_for_export(exp);
- /* flush any remaining cancel messages out to the target */
- ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
- err = llog_sync(ctxt, exp);
- if (err)
- CERROR("error flushing logs to MDS: rc %d\n", err);
+ fsfilt_sync(obd, obd->u.obt.obt_sb);
+ /* flush any remaining cancel messages out to the target */
+ filter_sync_llogs(obd, exp);
class_export_put(exp);
RETURN(rc);
}
+/* reverse import is changed, sync all cancels */
+static void filter_revimp_update(struct obd_export *exp)
+{
+ ENTRY;
+
+ LASSERT(exp);
+ class_export_get(exp);
+
+ /* flush any remaining cancel messages out to the target */
+ filter_sync_llogs(exp->exp_obd, exp);
+ class_export_put(exp);
+ EXIT;
+}
+
static int filter_ping(struct obd_export *exp)
{
filter_fmd_expire(exp);
const char *what, int quiet)
{
struct dentry *dchild = NULL;
+ obd_gr group = 0;
- if (!(oa->o_valid & OBD_MD_FLGROUP))
- oa->o_gr = 0;
+ if (oa->o_valid & OBD_MD_FLGROUP)
+ group = oa->o_gr;
- dchild = filter_fid2dentry(obd, NULL, oa->o_gr, oa->o_id);
+ dchild = filter_fid2dentry(obd, NULL, group, oa->o_id);
if (IS_ERR(dchild)) {
- CERROR("%s error looking up object: "LPU64"\n",
- what, oa->o_id);
+ CERROR("%s error looking up object: "LPU64":"LPU64"\n",
+ what, group, oa->o_id);
RETURN(dchild);
}
int rc = 0;
ENTRY;
+ rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
+ oinfo_capa(oinfo), CAPA_OPC_META_READ);
+ if (rc)
+ RETURN(rc);
+
obd = class_exp2obd(exp);
if (obd == NULL) {
CDEBUG(D_IOCTL, "invalid client export %p\n", exp);
if (!(oa->o_valid & OBD_MD_FLGROUP))
oa->o_gr = 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. */
unsigned int orig_ids[MAXQUOTAS] = {0, 0};
struct llog_cookie *fcc = NULL;
struct filter_obd *filter;
- int rc, err, locked = 0;
+ int rc, err, locked = 0, sync = 0;
unsigned int ia_valid;
struct inode *inode;
struct iattr iattr;
orig_ids[GRPQUOTA] = inode->i_gid;
handle = fsfilt_start_log(exp->exp_obd, inode,
FSFILT_OP_SETATTR, oti, 1);
-
if (IS_ERR(handle))
GOTO(out_unlock, rc = PTR_ERR(handle));
} else {
handle = fsfilt_start(exp->exp_obd, inode,
FSFILT_OP_SETATTR, oti);
-
if (IS_ERR(handle))
GOTO(out_unlock, rc = PTR_ERR(handle));
}
-
if (oa->o_valid & OBD_MD_FLFLAGS) {
rc = fsfilt_iocontrol(exp->exp_obd, inode, NULL,
EXT3_IOC_SETFLAGS, (long)&oa->o_flags);
} else {
rc = fsfilt_setattr(exp->exp_obd, dentry, handle, &iattr, 1);
- if (fcc != NULL) {
+ if (fcc != NULL)
/* set cancel cookie callback function */
- if (fsfilt_add_journal_cb(exp->exp_obd, 0, handle,
- filter_cancel_cookies_cb,
- fcc)) {
- spin_lock(&exp->exp_lock);
- exp->exp_need_sync = 1;
- spin_unlock(&exp->exp_lock);
- } else {
- fcc = NULL;
- }
- }
+ sync = fsfilt_add_journal_cb(exp->exp_obd, 0, handle,
+ filter_cancel_cookies_cb,
+ fcc);
}
if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_CREDITS))
/* 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);
- rc = filter_finish_transno(exp, oti, rc);
+
+ rc = filter_finish_transno(exp, oti, rc, sync);
+ if (sync) {
+ filter_cancel_cookies_cb(exp->exp_obd, 0, fcc, rc);
+ fcc = NULL;
+ }
err = fsfilt_commit(exp->exp_obd, inode, handle, 0);
if (err) {
CERROR("error on commit, err = %d\n", err);
if (!rc)
rc = err;
+ } else {
+ fcc = NULL;
}
if (locked) {
/* trigger quota release */
if (ia_valid & (ATTR_SIZE | ATTR_UID | ATTR_GID)) {
unsigned int cur_ids[MAXQUOTAS] = {oa->o_uid, oa->o_gid};
- int rc2 = lquota_adjust(filter_quota_interface_ref, exp->exp_obd, cur_ids,
+ int rc2 = lquota_adjust(filter_quota_interface_ref,
+ exp->exp_obd, cur_ids,
orig_ids, rc, FSFILT_OP_SETATTR);
CDEBUG(rc2 ? D_ERROR : D_QUOTA,
"filter adjust qunit. (rc:%d)\n", rc2);
int filter_setattr(struct obd_export *exp, struct obd_info *oinfo,
struct obd_trans_info *oti)
{
- struct ldlm_res_id res_id = { .name = { oinfo->oi_oa->o_id } };
- struct ldlm_valblock_ops *ns_lvbo;
+ struct ldlm_res_id res_id = { .name = { oinfo->oi_oa->o_id, 0,
+ oinfo->oi_oa->o_gr, 0 } };
struct filter_mod_data *fmd;
struct lvfs_run_ctxt saved;
struct filter_obd *filter;
int rc;
ENTRY;
+ rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo),
+ oinfo_capa(oinfo), CAPA_OPC_META_WRITE);
+ if (rc)
+ RETURN(rc);
+
+ /* 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 &&
+ oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) {
+ static char mdsinum[48];
+
+ if (oinfo->oi_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);
+ 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);
+ RETURN(-EPERM);
+ }
+
dentry = __filter_oa2dentry(exp->exp_obd, oinfo->oi_oa,
__FUNCTION__, 1);
if (IS_ERR(dentry))
GOTO(out_unlock, rc);
res = ldlm_resource_get(exp->exp_obd->obd_namespace, NULL,
- res_id, LDLM_EXTENT, 0);
+ &res_id, LDLM_EXTENT, 0);
if (res != NULL) {
- ns_lvbo = res->lr_namespace->ns_lvbo;
- if (ns_lvbo && ns_lvbo->lvbo_update)
- rc = ns_lvbo->lvbo_update(res, NULL, 0, 0);
+ rc = ldlm_res_lvbo_update(res, NULL, 0, 0);
ldlm_resource_putref(res);
}
OBD_ALLOC(*lsmp, lsm_size);
if (*lsmp == NULL)
RETURN(-ENOMEM);
+
OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
if ((*lsmp)->lsm_oinfo[0] == NULL) {
OBD_FREE(*lsmp, lsm_size);
RETURN(lsm_size);
}
-/* caller must hold fo_create_lock */
+/* 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)
{
ENTRY;
LASSERT(oa);
- LASSERT(down_trylock(&filter->fo_create_lock) != 0);
+ LASSERT(oa->o_gr != 0);
+ LASSERT(oa->o_valid & OBD_MD_FLGROUP);
+ LASSERT(down_trylock(&filter->fo_create_locks[oa->o_gr]) != 0);
memset(&doa, 0, sizeof(doa));
- if (oa->o_valid & OBD_MD_FLGROUP) {
- doa.o_valid |= OBD_MD_FLGROUP;
- doa.o_gr = oa->o_gr;
- } else {
- doa.o_gr = 0;
- }
+
+ doa.o_valid |= OBD_MD_FLGROUP;
+ doa.o_gr = oa->o_gr;
doa.o_mode = S_IFREG;
- if (!filter->fo_destroy_in_progress) {
- CERROR("%s: destroy_in_progress already cleared\n",
- exp->exp_obd->obd_name);
+ if (!test_bit(doa.o_gr, &filter->fo_destroys_in_progress)) {
+ CERROR("%s:["LPU64"] destroys_in_progress already cleared\n",
+ exp->exp_obd->obd_name, doa.o_gr);
RETURN(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);
+
for (id = last; id > oa->o_id; id--) {
doa.o_id = id;
rc = filter_destroy(exp, &doa, NULL, NULL, NULL);
exp->exp_obd->obd_name, doa.o_gr, oa->o_id);
rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
- filter->fo_destroy_in_progress = 0;
+ clear_bit(doa.o_gr, &filter->fo_destroys_in_progress);
RETURN(rc);
}
obd->obd_name);
RETURN(0);
}
-
/* This causes inflight precreates to abort and drop lock */
- filter->fo_destroy_in_progress = 1;
- down(&filter->fo_create_lock);
+ set_bit(group, &filter->fo_destroys_in_progress);
+ down(&filter->fo_create_locks[group]);
+ if (!test_bit(group, &filter->fo_destroys_in_progress)) {
+ CERROR("%s:["LPU64"] destroys_in_progress already cleared\n",
+ exp->exp_obd->obd_name, group);
+ up(&filter->fo_create_locks[group]);
+ RETURN(0);
+ }
diff = oa->o_id - filter_last_id(filter, group);
CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n",
filter_last_id(filter, group), diff);
"orphans were deleted\n", obd->obd_name);
GOTO(out, rc);
} else {
- /*XXX used by MDS for the first time! */
- filter->fo_destroy_in_progress = 0;
+ /* XXX: Used by MDS for the first time! */
+ clear_bit(group, &filter->fo_destroys_in_progress);
}
} else {
- down(&filter->fo_create_lock);
+ down(&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 (group != 0 || oa->o_id == 0)
+ if (group < FILTER_GROUP_MDS0 || oa->o_id == 0)
diff = 1;
else
diff = oa->o_id - filter_last_id(filter, group);
- CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n",
+ CDEBUG(D_RPCTRACE, "filter_last_id() = "LPU64" -> diff = %d\n",
filter_last_id(filter, group), diff);
LASSERTF(diff >= 0,"%s: "LPU64" - "LPU64" = %d\n",obd->obd_name,
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_valid = OBD_MD_FLID;
+ oa->o_gr = group;
+ oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
GOTO(out, rc);
}
/* else diff == 0 */
GOTO(out, rc = 0);
out:
- up(&filter->fo_create_lock);
+ up(&filter->fo_create_locks[group]);
return rc;
}
static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs,
- __u64 max_age)
+ __u64 max_age, __u32 flags)
{
struct filter_obd *filter = &obd->u.filter;
int blockbits = obd->u.obt.obt_sb->s_blocksize_bits;
RETURN(rc);
}
+static int filter_use_existing_obj(struct obd_device *obd,
+ struct dentry *dchild, void **handle,
+ int *cleanup_phase)
+{
+ struct inode *inode = dchild->d_inode;
+ struct iattr iattr;
+ int rc;
+
+ if ((inode->i_mode & (S_ISUID | S_ISGID)) == (S_ISUID|S_ISGID))
+ return 0;
+
+ *handle = fsfilt_start_log(obd, inode, FSFILT_OP_SETATTR, NULL, 1);
+ if (IS_ERR(*handle))
+ return PTR_ERR(*handle);
+
+ iattr.ia_valid = ATTR_MODE;
+ iattr.ia_mode = S_ISUID | S_ISGID |0666;
+ rc = fsfilt_setattr(obd, dchild, *handle, &iattr, 1);
+ if (rc == 0)
+ *cleanup_phase = 3;
+
+ return rc;
+}
+
+
/* We rely on the fact that only one thread will be creating files in a given
* group at a time, which is why we don't need an atomic filter_get_new_id.
* Even if we had that atomic function, the following race would exist:
* thread 2: creates object (x + 1)
* thread 1: tries to create object x, gets -ENOSPC
*
- * Caller must hold fo_create_lock
+ * Caller must hold fo_create_locks[group]
*/
static int filter_precreate(struct obd_device *obd, struct obdo *oa,
obd_gr group, int *num)
{
struct dentry *dchild = NULL, *dparent = NULL;
struct filter_obd *filter;
+ struct obd_statfs *osfs;
int err = 0, rc = 0, recreate_obj = 0, i;
unsigned long enough_time = jiffies + min(obd_timeout * HZ / 4, 10U*HZ);
obd_id next_id;
filter = &obd->u.filter;
- LASSERT(down_trylock(&filter->fo_create_lock) != 0);
+ LASSERT(down_trylock(&filter->fo_create_locks[group]) != 0);
+
+ OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_PRECREATE, obd_timeout / 2);
if ((oa->o_valid & OBD_MD_FLFLAGS) &&
(oa->o_flags & OBD_FL_RECREATE_OBJS)) {
recreate_obj = 1;
} else {
- struct obd_statfs *osfs;
-
OBD_ALLOC(osfs, sizeof(*osfs));
if (osfs == NULL)
RETURN(-ENOMEM);
- rc = filter_statfs(obd, osfs, cfs_time_current_64() - HZ);
+ rc = filter_statfs(obd, osfs, cfs_time_current_64() - HZ, 0);
if (rc == 0 && osfs->os_bavail < (osfs->os_blocks >> 10)) {
- CDEBUG(D_HA,"%s: not enough space for create "LPU64"\n",
- obd->obd_name, osfs->os_bavail <<
+ 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);
*num = 0;
rc = -ENOSPC;
RETURN(rc);
}
- CDEBUG(D_HA, "%s: precreating %d objects in group "LPU64" at "LPU64"\n",
- obd->obd_name, *num, group, oa->o_id);
+ CDEBUG(D_RPCTRACE, "%s: precreating %d objects in group "LPU64
+ " at "LPU64"\n", obd->obd_name, *num, group, oa->o_id);
for (i = 0; i < *num && err == 0; i++) {
int cleanup_phase = 0;
- if (filter->fo_destroy_in_progress) {
- CWARN("%s: precreate aborted by destroy\n",
+ if (test_bit(group, &filter->fo_destroys_in_progress)) {
+ CWARN("%s: create aborted by destroy\n",
obd->obd_name);
rc = -EAGAIN;
break;
}
-
+
if (recreate_obj) {
__u64 last_id;
next_id = oa->o_id;
dparent = filter_parent_lock(obd, group, next_id);
if (IS_ERR(dparent))
GOTO(cleanup, rc = PTR_ERR(dparent));
- cleanup_phase = 1; /* filter_parent_unlock(dparent) */
+ cleanup_phase = 1; /* filter_parent_unlock(dparent) */
dchild = filter_fid2dentry(obd, dparent, group, next_id);
if (IS_ERR(dchild))
GOTO(cleanup, rc = PTR_ERR(dchild));
- cleanup_phase = 2; /* f_dput(dchild) */
+ cleanup_phase = 2; /* f_dput(dchild) */
if (dchild->d_inode != NULL) {
/* This would only happen if lastobjid was bad on disk*/
- /* Could also happen if recreating missing obj but
- * already exists
- */
+ /* Could also happen if recreating missing obj but it
+ * already exists. */
if (recreate_obj) {
CERROR("%s: recreating existing object %.*s?\n",
obd->obd_name, dchild->d_name.len,
dchild->d_name.name);
} else {
+ /* Use these existing objects if they are
+ * zero length. */
+ if (dchild->d_inode->i_size == 0) {
+ rc = filter_use_existing_obj(obd,dchild,
+ &handle, &cleanup_phase);
+ if (rc == 0)
+ goto set_last_id;
+ else
+ GOTO(cleanup, rc);
+ }
+
CERROR("%s: Serious error: objid %.*s already "
"exists; is this filesystem corrupt?\n",
obd->obd_name, dchild->d_name.len,
GOTO(cleanup, rc);
}
+set_last_id:
if (!recreate_obj) {
filter_set_last_id(filter, next_id, group);
err = filter_update_last_objid(obd, group, 0);
if (rc)
break;
if (time_after(jiffies, enough_time)) {
- CDEBUG(D_HA, "%s: precreate slow - want %d got %d \n",
+ CDEBUG(D_RPCTRACE,
+ "%s: precreate slow - want %d got %d \n",
obd->obd_name, *num, i);
break;
}
}
*num = i;
- CDEBUG(D_HA, "%s: created %d objects for group "LPU64": "LPU64" rc %d\n",
+ 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);
RETURN(rc);
static int filter_create(struct obd_export *exp, struct obdo *oa,
struct lov_stripe_md **ea, struct obd_trans_info *oti)
{
+ 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;
+ int rc = 0, diff, group = oa->o_gr;
ENTRY;
- if (!(oa->o_valid & OBD_MD_FLGROUP))
- oa->o_gr = 0;
+ if (!(oa->o_valid & OBD_MD_FLGROUP) || group == 0) {
+ 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;
+
+ 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. Tell CFS.\n",
+ obd_export_nid2str(exp), fed->fed_group, group);
+ RETURN(-ENOTUNIQ);
+ }
CDEBUG(D_INFO, "filter_create(od->o_gr="LPU64",od->o_id="LPU64")\n",
oa->o_gr, oa->o_id);
if ((oa->o_valid & OBD_MD_FLFLAGS) &&
(oa->o_flags & OBD_FL_RECREATE_OBJS)) {
- if (oa->o_id > filter_last_id(&obd->u.filter, oa->o_gr)) {
+ if (oa->o_id > filter_last_id(filter, oa->o_gr)) {
CERROR("recreate objid "LPU64" > last id "LPU64"\n",
- oa->o_id, filter_last_id(&obd->u.filter,
+ oa->o_id, filter_last_id(filter,
oa->o_gr));
rc = -EINVAL;
} else {
- struct filter_obd *filter = &obd->u.filter;
- int diff = 1;
-
- down(&filter->fo_create_lock);
+ diff = 1;
+ down(&filter->fo_create_locks[oa->o_gr]);
rc = filter_precreate(obd, oa, oa->o_gr, &diff);
- up(&filter->fo_create_lock);
+ up(&filter->fo_create_locks[oa->o_gr]);
}
} else {
rc = filter_handle_precreate(exp, oa, oa->o_gr, oti);
struct lvfs_run_ctxt saved;
void *handle = NULL;
struct llog_cookie *fcc = NULL;
- int rc, rc2, cleanup_phase = 0;
+ int rc, rc2, cleanup_phase = 0, sync = 0;
struct iattr iattr;
ENTRY;
- if (!(oa->o_valid & OBD_MD_FLGROUP))
- oa->o_gr = 0;
+ LASSERT(oa->o_valid & OBD_MD_FLGROUP);
obd = exp->exp_obd;
filter = &obd->u.filter;
oa->o_id);
/* If object already gone, cancel cookie right now */
if (oa->o_valid & OBD_MD_FLCOOKIE) {
+ struct llog_ctxt *ctxt;
+ struct obd_llog_group *olg;
fcc = obdo_logcookie(oa);
- llog_cancel(llog_get_context(obd, fcc->lgc_subsys + 1),
- NULL, 1, fcc, 0);
+ olg = filter_find_olg(obd, oa->o_gr);
+ if (IS_ERR(olg))
+ GOTO(cleanup, rc = PTR_ERR(olg));
+ llog_group_set_export(olg, exp);
+
+ ctxt = llog_group_get_ctxt(olg, fcc->lgc_subsys + 1);
+ llog_cancel(ctxt, NULL, 1, fcc, 0);
+ llog_ctxt_put(ctxt);
fcc = NULL; /* we didn't allocate fcc, don't free it */
}
GOTO(cleanup, rc = -ENOENT);
}
- filter_prepare_destroy(obd, oa->o_id);
+ filter_prepare_destroy(obd, oa->o_id, oa->o_gr);
/* Our MDC connection is established by the MDS to us */
if (oa->o_valid & OBD_MD_FLCOOKIE) {
cleanup:
switch(cleanup_phase) {
case 4:
- if (fcc != NULL) {
- if (fsfilt_add_journal_cb(obd, 0, oti ?
- oti->oti_handle : handle,
- filter_cancel_cookies_cb,
- fcc) == 0)
- fcc = NULL;
- }
- rc = filter_finish_transno(exp, oti, rc);
+ if (fcc != NULL)
+ sync = fsfilt_add_journal_cb(obd, 0, oti ?
+ oti->oti_handle : handle,
+ 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.
+ */
+ rc = filter_finish_transno(exp, oti, rc, sync);
+ if (sync) {
+ filter_cancel_cookies_cb(obd, 0, fcc, rc);
+ fcc = NULL;
+ }
rc2 = fsfilt_commit(obd, dparent->d_inode, handle, 0);
if (rc2) {
CERROR("error on commit, err = %d\n", rc2);
if (!rc)
rc = rc2;
+ } else {
+ fcc = NULL;
}
case 3:
filter_parent_unlock(dparent);
qcids[GRPQUOTA] = oa->o_gid;
rc2 = lquota_adjust(filter_quota_interface_ref, obd, qcids, NULL, rc,
FSFILT_OP_UNLINK);
+
if (rc2)
- CERROR("filter adjust qunit! (rc:%d)\n", rc2);
+ CDEBUG(D_QUOTA, "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;
rc = filter_setattr(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)
+ struct lov_stripe_md *lsm, obd_off start, obd_off end,
+ void *capa)
{
struct lvfs_run_ctxt saved;
struct filter_obd *filter;
struct dentry *dentry;
- struct llog_ctxt *ctxt;
int rc, rc2;
ENTRY;
+ rc = filter_auth_capa(exp, NULL, obdo_mdsno(oa),
+ (struct lustre_capa *)capa, CAPA_OPC_OSS_WRITE);
+ if (rc)
+ RETURN(rc);
+
filter = &exp->exp_obd->u.filter;
/* 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 */
- ctxt = llog_get_context(exp->exp_obd, LLOG_MDS_OST_REPL_CTXT);
- llog_sync(ctxt, exp);
+ filter_sync_llogs(exp->exp_obd, exp);
RETURN(rc);
}
RETURN(-EINVAL);
}
- if (keylen == strlen("blocksize") &&
- memcmp(key, "blocksize", keylen) == 0) {
+ if (KEY_IS("blocksize")) {
__u32 *blocksize = val;
+ if (blocksize) {
+ if (*vallen < sizeof(*blocksize))
+ RETURN(-EOVERFLOW);
+ *blocksize = obd->u.obt.obt_sb->s_blocksize;
+ }
*vallen = sizeof(*blocksize);
- *blocksize = obd->u.obt.obt_sb->s_blocksize;
RETURN(0);
}
- if (keylen == strlen("blocksize_bits") &&
- memcmp(key, "blocksize_bits", keylen) == 0) {
+ if (KEY_IS("blocksize_bits")) {
__u32 *blocksize_bits = val;
+ if (blocksize_bits) {
+ if (*vallen < sizeof(*blocksize_bits))
+ RETURN(-EOVERFLOW);
+ *blocksize_bits = obd->u.obt.obt_sb->s_blocksize_bits;
+ }
*vallen = sizeof(*blocksize_bits);
- *blocksize_bits = obd->u.obt.obt_sb->s_blocksize_bits;
RETURN(0);
}
- if (keylen >= strlen("last_id") && memcmp(key, "last_id", 7) == 0) {
+ if (KEY_IS("last_id")) {
obd_id *last_id = val;
/* FIXME: object groups */
- *last_id = filter_last_id(&obd->u.filter, 0);
+ if (last_id) {
+ if (*vallen < sizeof(*last_id))
+ RETURN(-EOVERFLOW);
+ *last_id = filter_last_id(&obd->u.filter,
+ exp->exp_filter_data.fed_group);
+ }
+ *vallen = sizeof(*last_id);
RETURN(0);
}
+
CDEBUG(D_IOCTL, "invalid key\n");
RETURN(-EINVAL);
}
struct ptlrpc_request_set *set)
{
struct obd_device *obd;
+ struct obd_llog_group *olg;
struct llog_ctxt *ctxt;
- int rc = 0;
+ int rc = 0, group;
ENTRY;
obd = exp->exp_obd;
RETURN(-EINVAL);
}
- if (keylen < strlen(KEY_MDS_CONN) ||
- memcmp(key, KEY_MDS_CONN, keylen) != 0)
+ if (KEY_IS(KEY_CAPA_KEY)) {
+ rc = filter_update_capa_key(obd, (struct lustre_capa_key *)val);
+ if (rc)
+ CERROR("filter update capability key failed: %d\n", rc);
+ RETURN(rc);
+ }
+
+ if (KEY_IS(KEY_REVIMP_UPD)) {
+ filter_revimp_update(exp);
+ RETURN(0);
+ }
+
+ if (!KEY_IS(KEY_MDS_CONN))
RETURN(-EINVAL);
LCONSOLE_WARN("%s: received MDS connection from %s\n", obd->obd_name,
obd->u.filter.fo_mdc_conn.cookie = exp->exp_handle.h_cookie;
/* setup llog imports */
- ctxt = llog_get_context(obd, LLOG_MDS_OST_REPL_CTXT);
+ LASSERT(val != NULL);
+ group = (int)(*(__u32 *)val);
+ LASSERT(group >= FILTER_GROUP_MDS0);
+
+ olg = filter_find_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);
lquota_setinfo(filter_quota_interface_ref, exp, obd);
switch (cmd) {
case OBD_IOC_ABORT_RECOVERY: {
CERROR("aborting recovery for device %s\n", obd->obd_name);
- target_abort_recovery(obd);
+ target_stop_recovery_thread(obd);
RETURN(0);
}
case OBD_IOC_SYNC: {
- CDEBUG(D_HA, "syncing ost %s\n", obd->obd_name);
+ CDEBUG(D_RPCTRACE, "syncing ost %s\n", obd->obd_name);
rc = fsfilt_sync(obd, obd->u.obt.obt_sb);
RETURN(rc);
}
static int filter_health_check(struct obd_device *obd)
{
+#ifdef USE_HEALTH_CHECK_WRITE
struct filter_obd *filter = &obd->u.filter;
+#endif
int rc = 0;
/*
if (obd->u.obt.obt_sb->s_flags & MS_RDONLY)
rc = 1;
+#ifdef USE_HEALTH_CHECK_WRITE
LASSERT(filter->fo_health_check_filp != NULL);
rc |= !!lvfs_check_io_health(obd, filter->fo_health_check_filp);
-
+#endif
return rc;
}
return filter_fid2dentry(data, NULL, gr, id);
}
-static int filter_process_config(struct obd_device *obd, obd_count len, void *buf)
+static int filter_process_config(struct obd_device *obd, obd_count len,
+ void *buf)
{
struct lustre_cfg *lcfg = buf;
struct lprocfs_static_vars lvars;
int rc = 0;
- lprocfs_init_vars(filter, &lvars);
-
- rc = class_process_proc_param(PARAM_OST, lvars.obd_vars, lcfg, obd);
+ 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;
- return(rc);
+ 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);
+ break;
+ }
+
+ return rc;
}
static struct lvfs_callback_ops filter_lvfs_ops = {
.o_preprw = filter_preprw,
.o_commitrw = filter_commitrw,
.o_llog_init = filter_llog_init,
+ .o_llog_connect = filter_llog_connect,
.o_llog_finish = filter_llog_finish,
.o_iocontrol = filter_iocontrol,
.o_health_check = filter_health_check,
struct lprocfs_static_vars lvars;
int rc;
- printk(KERN_INFO "Lustre: Filtering OBD driver; info@clusterfs.com\n");
-
- lprocfs_init_vars(filter, &lvars);
+ lprocfs_filter_init_vars(&lvars);
request_module("lquota");
OBD_ALLOC(obdfilter_created_scratchpad,
if (obdfilter_created_scratchpad == NULL)
return -ENOMEM;
- ll_fmd_cachep = kmem_cache_create("ll_fmd_cache",
- sizeof(struct filter_mod_data),
- 0, 0, NULL, NULL);
+ ll_fmd_cachep = cfs_mem_cache_create("ll_fmd_cache",
+ sizeof(struct filter_mod_data),
+ 0, 0);
if (!ll_fmd_cachep)
GOTO(out, rc = -ENOMEM);
filter_quota_interface_ref = PORTAL_SYMBOL_GET(filter_quota_interface);
init_obd_quota_ops(filter_quota_interface_ref, &filter_obd_ops);
- rc = class_register_type(&filter_obd_ops, lvars.module_vars,
- LUSTRE_OST_NAME);
+ rc = class_register_type(&filter_obd_ops, NULL, lvars.module_vars,
+ LUSTRE_OST_NAME, NULL);
if (rc) {
int err;
- err = kmem_cache_destroy(ll_fmd_cachep);
+ err = cfs_mem_cache_destroy(ll_fmd_cachep);
LASSERTF(err == 0, "Cannot destroy ll_fmd_cachep: rc %d\n",err);
ll_fmd_cachep = NULL;
out:
PORTAL_SYMBOL_PUT(filter_quota_interface);
if (ll_fmd_cachep) {
- int rc = kmem_cache_destroy(ll_fmd_cachep);
+ int rc = cfs_mem_cache_destroy(ll_fmd_cachep);
LASSERTF(rc == 0, "Cannot destroy ll_fmd_cachep: rc %d\n", rc);
ll_fmd_cachep = NULL;
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff