X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fobdfilter%2Ffilter.c;h=f00b738061b7758adee815d674c45b5c966bc26b;hp=4a36717d9626e44cc27d65b9ae2808dfe49e12d2;hb=c1f6b32958c799412c830f35f8d16ed7275407ea;hpb=9bc7e4aaddc53c442e50ca49b565ea689db83eed diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c index 4a36717..f00b738 100644 --- a/lustre/obdfilter/filter.c +++ b/lustre/obdfilter/filter.c @@ -24,53 +24,49 @@ */ /* - * Invariant: Get O/R i_sem for lookup, if needed, before any journal ops + * Invariant: Get O/R i_mutex for lookup, if needed, before any journal ops * (which need to get journal_lock, may block if journal full). * * Invariant: Call filter_start_transno() before any journal ops to avoid the * same deadlock problem. We can (and want) to get rid of the - * transno sem in favour of the dir/inode i_sem to avoid single + * transno sem in favour of the dir/inode i_mutex to avoid single * threaded operation on the OST. */ #define DEBUG_SUBSYSTEM S_FILTER +#ifndef AUTOCONF_INCLUDED #include +#endif #include #include #include #include #include -#include -#include -#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0)) -# include -# include -# include -#endif +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include #include "filter_internal.h" /* Group 0 is no longer a legal group, to catch uninitialized IDs */ -#define FILTER_MIN_GROUPS 3 - +#define FILTER_MIN_GROUPS FILTER_GROUP_MDS0 static struct lvfs_callback_ops filter_lvfs_ops; - -static int filter_destroy(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *ea, struct obd_trans_info *); -struct obd_llogs *filter_grab_llog_for_group(struct obd_device *, - int, struct obd_export *); +cfs_mem_cache_t *ll_fmd_cachep; static void filter_commit_cb(struct obd_device *obd, __u64 transno, void *cb_data, int error) @@ -80,14 +76,14 @@ static void filter_commit_cb(struct obd_device *obd, __u64 transno, /* 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) @@ -99,31 +95,42 @@ int filter_finish_transno(struct obd_export *exp, struct obd_trans_info *oti, /* we don't allocate new transnos for replayed requests */ if (oti->oti_transno == 0) { spin_lock(&filter->fo_translock); - last_rcvd = le64_to_cpu(filter->fo_fsd->fsd_last_transno) + 1; - filter->fo_fsd->fsd_last_transno = cpu_to_le64(last_rcvd); + last_rcvd = le64_to_cpu(filter->fo_fsd->lsd_last_transno) + 1; + filter->fo_fsd->lsd_last_transno = cpu_to_le64(last_rcvd); spin_unlock(&filter->fo_translock); oti->oti_transno = last_rcvd; } else { spin_lock(&filter->fo_translock); last_rcvd = oti->oti_transno; - if (last_rcvd > le64_to_cpu(filter->fo_fsd->fsd_last_transno)) - filter->fo_fsd->fsd_last_transno = + if (last_rcvd > le64_to_cpu(filter->fo_fsd->lsd_last_transno)) + filter->fo_fsd->lsd_last_transno = cpu_to_le64(last_rcvd); spin_unlock(&filter->fo_translock); } - fcd->fcd_last_rcvd = cpu_to_le64(last_rcvd); /* could get xid from oti, if it's ever needed */ fcd->fcd_last_xid = 0; off = fed->fed_lr_off; - - fsfilt_add_journal_cb(exp->exp_obd, filter->fo_sb, last_rcvd, - oti->oti_handle, filter_commit_cb, NULL); - - err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp, fcd, - sizeof(*fcd), &off, 0); + if (off <= 0) { + CERROR("%s: client idx %d is %lld\n", exp->exp_obd->obd_name, + fed->fed_lr_idx, fed->fed_lr_off); + err = -EINVAL; + } else { + 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, + 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; if (rc == 0) @@ -146,38 +153,115 @@ void f_dput(struct dentry *dentry) dput(dentry); } +static void init_brw_stats(struct brw_stats *brw_stats) +{ + int i; + for (i = 0; i < BRW_LAST; i++) + 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, + 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) + RETURN(rc); + + 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); +} + /* Add client data to the FILTER. We use a bitmap to locate a free space * in the last_rcvd file if cl_idx is -1 (i.e. a new client). * Otherwise, we have just read the data from the last_rcvd file and * we know its offset. */ -static int filter_client_add(struct obd_device *obd, struct filter_obd *filter, - struct filter_export_data *fed, int cl_idx) +static int filter_client_add(struct obd_device *obd, struct obd_export *exp, + int cl_idx) { + struct filter_obd *filter = &obd->u.filter; + struct filter_export_data *fed = &exp->exp_filter_data; 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); - /* XXX if fcd_uuid were a real obd_uuid, I could use obd_uuid_equals */ - if (!strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid)) + /* Self-export */ + 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 * there's no need for extra complication here */ if (new_client) { - cl_idx = find_first_zero_bit(bitmap, FILTER_LR_MAX_CLIENTS); + cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS); repeat: - if (cl_idx >= FILTER_LR_MAX_CLIENTS) { - CERROR("no client slots - fix FILTER_LR_MAX_CLIENTS\n"); - RETURN(-ENOMEM); + if (cl_idx >= LR_MAX_CLIENTS) { + CERROR("no room for %u client - fix LR_MAX_CLIENTS\n", + cl_idx); + RETURN(-EOVERFLOW); } if (test_and_set_bit(cl_idx, bitmap)) { - CERROR("FILTER client %d: found bit is set in bitmap\n", - cl_idx); - cl_idx = find_next_zero_bit(bitmap, - FILTER_LR_MAX_CLIENTS, + cl_idx = find_next_zero_bit(bitmap, LR_MAX_CLIENTS, cl_idx); goto repeat; } @@ -190,8 +274,9 @@ static int filter_client_add(struct obd_device *obd, struct filter_obd *filter, } fed->fed_lr_idx = cl_idx; - fed->fed_lr_off = le32_to_cpu(filter->fo_fsd->fsd_client_start) + - cl_idx * le16_to_cpu(filter->fo_fsd->fsd_client_size); + fed->fed_lr_off = le32_to_cpu(filter->fo_fsd->lsd_client_start) + + cl_idx * le16_to_cpu(filter->fo_fsd->lsd_client_size); + LASSERTF(fed->fed_lr_off > 0, "fed_lr_off = %llu\n", fed->fed_lr_off); CDEBUG(D_INFO, "client at index %d (%llu) with UUID '%s' added\n", fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid); @@ -199,7 +284,7 @@ static int filter_client_add(struct obd_device *obd, struct filter_obd *filter, if (new_client) { struct lvfs_run_ctxt saved; loff_t off = fed->fed_lr_off; - int err; + int rc; void *handle; CDEBUG(D_INFO, "writing client fcd at idx %u (%llu) (len %u)\n", @@ -211,29 +296,36 @@ static int filter_client_add(struct obd_device *obd, struct filter_obd *filter, filter->fo_rcvd_filp->f_dentry->d_inode, FSFILT_OP_SETATTR, NULL); if (IS_ERR(handle)) { - err = PTR_ERR(handle); - CERROR("unable to start transaction: rc %d\n", err); + rc = PTR_ERR(handle); + CERROR("unable to start transaction: rc %d\n", rc); } else { - err = fsfilt_write_record(obd, filter->fo_rcvd_filp, - fed->fed_fcd, - sizeof(*fed->fed_fcd), - &off, 1); - fsfilt_commit(obd, filter->fo_sb, + rc = fsfilt_add_journal_cb(obd, 0, handle, + target_client_add_cb, exp); + if (rc == 0) { + spin_lock(&exp->exp_lock); + exp->exp_need_sync = 1; + spin_unlock(&exp->exp_lock); + } + rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, + fed->fed_fcd, + sizeof(*fed->fed_fcd), + &off, rc /* sync if no cb */); + fsfilt_commit(obd, filter->fo_rcvd_filp->f_dentry->d_inode, - handle, 1); + handle, 0); } pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - if (err) { + if (rc) { CERROR("error writing %s client idx %u: rc %d\n", - LAST_RCVD, fed->fed_lr_idx, err); - RETURN(err); + LAST_RCVD, fed->fed_lr_idx, rc); + RETURN(rc); } } RETURN(0); } -static int filter_client_free(struct obd_export *exp, int flags) +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; @@ -247,19 +339,24 @@ static int filter_client_free(struct obd_export *exp, int flags) if (fed->fed_fcd == NULL) RETURN(0); - if (flags & OBD_OPT_FAILOVER) - GOTO(free, 0); - /* XXX if fcd_uuid were a real obd_uuid, I could use obd_uuid_equals */ if (strcmp(fed->fed_fcd->fcd_uuid, obd->obd_uuid.uuid ) == 0) GOTO(free, 0); + CDEBUG(D_INFO, "freeing client at idx %u, offset %lld with UUID '%s'\n", + fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid); + LASSERT(filter->fo_last_rcvd_slots != NULL); off = fed->fed_lr_off; - CDEBUG(D_INFO, "freeing client at idx %u (%lld) with UUID '%s'\n", - fed->fed_lr_idx, fed->fed_lr_off, fed->fed_fcd->fcd_uuid); + /* Don't clear fed_lr_idx here as it is likely also unset. At worst + * we leak a client slot that will be cleaned on the next recovery. */ + if (off <= 0) { + CERROR("%s: client idx %d has med_off %lld\n", + obd->obd_name, fed->fed_lr_idx, off); + GOTO(free, rc = -EINVAL); + } /* Clear the bit _after_ zeroing out the client so we don't race with filter_client_add and zero out new clients.*/ @@ -269,20 +366,25 @@ static int filter_client_free(struct obd_export *exp, int flags) LBUG(); } - memset(&zero_fcd, 0, sizeof zero_fcd); - push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - rc = fsfilt_write_record(obd, filter->fo_rcvd_filp, &zero_fcd, - sizeof(zero_fcd), &off, 1); - if (rc == 0) - /* update server's transno */ - filter_update_server_data(obd, filter->fo_rcvd_filp, - filter->fo_fsd, 1); - pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); + if (!(exp->exp_flags & OBD_OPT_FAILOVER)) { + memset(&zero_fcd, 0, sizeof zero_fcd); + 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)); + if (rc == 0) + /* update server's transno */ + filter_update_server_data(obd, filter->fo_rcvd_filp, + filter->fo_fsd, + !exp->exp_libclient); + pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - CDEBUG(rc == 0 ? D_INFO : D_ERROR, - "zeroing disconnecting client %s at idx %u (%llu) in %s rc %d\n", - fed->fed_fcd->fcd_uuid, fed->fed_lr_idx, fed->fed_lr_off, - LAST_RCVD, rc); + CDEBUG(rc == 0 ? D_INFO : D_ERROR, + "zeroing out client %s at idx %u (%llu) in %s rc %d\n", + fed->fed_fcd->fcd_uuid, fed->fed_lr_idx, fed->fed_lr_off, + LAST_RCVD, rc); + } if (!test_and_clear_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) { CERROR("FILTER client %u: bit already clear in bitmap!!\n", @@ -290,39 +392,224 @@ static int filter_client_free(struct obd_export *exp, int flags) LBUG(); } + EXIT; free: OBD_FREE(fed->fed_fcd, sizeof(*fed->fed_fcd)); + fed->fed_fcd = NULL; - RETURN(0); + return 0; +} + +/* drop fmd reference, free it if last ref. must be called with fed_lock held.*/ +static inline void filter_fmd_put_nolock(struct filter_export_data *fed, + struct filter_mod_data *fmd) +{ + LASSERT_SPIN_LOCKED(&fed->fed_lock); + if (--fmd->fmd_refcount == 0) { + /* XXX when we have persistent reservations and the handle + * is stored herein we need to drop it here. */ + fed->fed_mod_count--; + list_del(&fmd->fmd_list); + OBD_SLAB_FREE(fmd, ll_fmd_cachep, sizeof(*fmd)); + } +} + +/* drop fmd reference, free it if last ref */ +void filter_fmd_put(struct obd_export *exp, struct filter_mod_data *fmd) +{ + struct filter_export_data *fed; + + if (fmd == NULL) + return; + + fed = &exp->exp_filter_data; + spin_lock(&fed->fed_lock); + filter_fmd_put_nolock(fed, fmd); /* caller reference */ + spin_unlock(&fed->fed_lock); +} + +/* expire entries from the end of the list if there are too many + * or they are too old */ +static void filter_fmd_expire_nolock(struct filter_obd *filter, + struct filter_export_data *fed, + struct filter_mod_data *keep) +{ + struct filter_mod_data *fmd, *tmp; + + list_for_each_entry_safe(fmd, tmp, &fed->fed_mod_list, fmd_list) { + if (fmd == keep) + break; + + if (time_before(jiffies, fmd->fmd_expire) && + fed->fed_mod_count < filter->fo_fmd_max_num) + break; + + list_del_init(&fmd->fmd_list); + filter_fmd_put_nolock(fed, fmd); /* list reference */ + } +} + +void filter_fmd_expire(struct obd_export *exp) +{ + spin_lock(&exp->exp_filter_data.fed_lock); + filter_fmd_expire_nolock(&exp->exp_obd->u.filter, + &exp->exp_filter_data, NULL); + spin_unlock(&exp->exp_filter_data.fed_lock); +} + +/* find specified objid, group in export fmd list. + * caller must hold fed_lock and take fmd reference itself */ +static struct filter_mod_data *filter_fmd_find_nolock(struct filter_obd *filter, + struct filter_export_data *fed, + obd_id objid, obd_gr group) +{ + struct filter_mod_data *found = NULL, *fmd; + + LASSERT_SPIN_LOCKED(&fed->fed_lock); + + list_for_each_entry_reverse(fmd, &fed->fed_mod_list, fmd_list) { + if (fmd->fmd_id == objid && fmd->fmd_gr == group) { + found = fmd; + list_del(&fmd->fmd_list); + list_add_tail(&fmd->fmd_list, &fed->fed_mod_list); + fmd->fmd_expire = jiffies + filter->fo_fmd_max_age; + break; + } + } + + filter_fmd_expire_nolock(filter, fed, found); + + return found; +} + +/* Find fmd based on objid and group, or return NULL if not found. */ +struct filter_mod_data *filter_fmd_find(struct obd_export *exp, + obd_id objid, obd_gr group) +{ + struct filter_mod_data *fmd; + + spin_lock(&exp->exp_filter_data.fed_lock); + fmd = filter_fmd_find_nolock(&exp->exp_obd->u.filter, + &exp->exp_filter_data, objid, group); + if (fmd) + fmd->fmd_refcount++; /* caller reference */ + spin_unlock(&exp->exp_filter_data.fed_lock); + + return fmd; +} + +/* Find fmd based on objid and group, or create a new one if none is found. + * It is possible for this function to return NULL under memory pressure, + * or if objid = 0 is passed (which will only cause old entries to expire). + * Currently this is not fatal because any fmd state is transient and + * may also be freed when it gets sufficiently old. */ +struct filter_mod_data *filter_fmd_get(struct obd_export *exp, + obd_id objid, obd_gr group) +{ + 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)); + + spin_lock(&fed->fed_lock); + found = filter_fmd_find_nolock(&exp->exp_obd->u.filter,fed,objid,group); + if (fmd_new) { + if (found == NULL) { + list_add_tail(&fmd_new->fmd_list, &fed->fed_mod_list); + fmd_new->fmd_id = objid; + fmd_new->fmd_gr = group; + fmd_new->fmd_refcount++; /* list reference */ + found = fmd_new; + fed->fed_mod_count++; + } else { + OBD_SLAB_FREE(fmd_new, ll_fmd_cachep, sizeof(*fmd_new)); + } + } + if (found) { + found->fmd_refcount++; /* caller reference */ + found->fmd_expire = jiffies + + exp->exp_obd->u.filter.fo_fmd_max_age; + } + + spin_unlock(&fed->fed_lock); + + return found; +} + +#ifdef DO_FMD_DROP +/* drop fmd list reference so it will disappear when last reference is put. + * This isn't so critical because it would in fact only affect the one client + * that is doing the unlink and at worst we have an stale entry referencing + * an object that should never be used again. */ +static void filter_fmd_drop(struct obd_export *exp, obd_id objid, obd_gr group) +{ + struct filter_mod_data *found = NULL; + + spin_lock(&exp->exp_filter_data.fed_lock); + found = filter_fmd_find_nolock(&exp->exp_filter_data, objid, group); + if (found) { + list_del_init(&found->fmd_list); + filter_fmd_put_nolock(&exp->exp_filter_data, found); + } + spin_unlock(&exp->exp_filter_data.fed_lock); +} +#else +#define filter_fmd_drop(exp, objid, group) +#endif + +/* remove all entries from fmd list */ +static void filter_fmd_cleanup(struct obd_export *exp) +{ + struct filter_export_data *fed = &exp->exp_filter_data; + struct filter_mod_data *fmd = NULL, *tmp; + + spin_lock(&fed->fed_lock); + list_for_each_entry_safe(fmd, tmp, &fed->fed_mod_list, fmd_list) { + list_del_init(&fmd->fmd_list); + filter_fmd_put_nolock(fed, fmd); + } + spin_unlock(&fed->fed_lock); +} + +static int filter_init_export(struct obd_export *exp) +{ + spin_lock_init(&exp->exp_filter_data.fed_lock); + 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); + + return 0; } static int filter_free_server_data(struct filter_obd *filter) { OBD_FREE(filter->fo_fsd, sizeof(*filter->fo_fsd)); filter->fo_fsd = NULL; - OBD_FREE(filter->fo_last_rcvd_slots, - FILTER_LR_MAX_CLIENT_WORDS * sizeof(unsigned long)); + OBD_FREE(filter->fo_last_rcvd_slots, LR_MAX_CLIENTS / 8); filter->fo_last_rcvd_slots = NULL; return 0; } /* assumes caller is already in kernel ctxt */ int filter_update_server_data(struct obd_device *obd, struct file *filp, - struct filter_server_data *fsd, int force_sync) + struct lr_server_data *fsd, int force_sync) { loff_t off = 0; int rc; ENTRY; - CDEBUG(D_INODE, "server uuid : %s\n", fsd->fsd_uuid); + CDEBUG(D_INODE, "server uuid : %s\n", fsd->lsd_uuid); CDEBUG(D_INODE, "server last_rcvd : "LPU64"\n", - le64_to_cpu(fsd->fsd_last_transno)); + le64_to_cpu(fsd->lsd_last_transno)); CDEBUG(D_INODE, "server last_mount: "LPU64"\n", - le64_to_cpu(fsd->fsd_mount_count)); + le64_to_cpu(fsd->lsd_mount_count)); - rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off,force_sync); + fsd->lsd_compat14 = fsd->lsd_last_transno; + rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off, force_sync); if (rc) - CERROR("error writing filter_server_data: rc = %d\n", rc); + CERROR("error writing lr_server_data: rc = %d\n", rc); RETURN(rc); } @@ -339,11 +626,11 @@ int filter_update_last_objid(struct obd_device *obd, obd_gr group, if (filter->fo_last_objid_files[group] == NULL) { CERROR("Object group "LPU64" not fully setup; not updating " "last_objid\n", group); - RETURN(0); + RETURN(-EINVAL); } - CDEBUG(D_INODE, "server last_objid for group "LPU64": "LPU64"\n", - group, filter->fo_last_objids[group]); + CDEBUG(D_INODE, "%s: server last_objid for group "LPU64": "LPU64"\n", + obd->obd_name, group, filter->fo_last_objids[group]); tmp = cpu_to_le64(filter->fo_last_objids[group]); rc = fsfilt_write_record(obd, filter->fo_last_objid_files[group], @@ -353,49 +640,49 @@ int filter_update_last_objid(struct obd_device *obd, obd_gr group, 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 filter_obd *filter = &obd->u.filter; - struct filter_server_data *fsd; + 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; int rc; /* ensure padding in the struct is the correct size */ - LASSERT (offsetof(struct filter_server_data, fsd_padding) + - sizeof(fsd->fsd_padding) == FILTER_LR_SERVER_SIZE); - LASSERT (offsetof(struct filter_client_data, fcd_padding) + - sizeof(fcd->fcd_padding) == FILTER_LR_CLIENT_SIZE); + CLASSERT (offsetof(struct lr_server_data, lsd_padding) + + sizeof(fsd->lsd_padding) == LR_SERVER_SIZE); + CLASSERT (offsetof(struct filter_client_data, fcd_padding) + + sizeof(fcd->fcd_padding) == LR_CLIENT_SIZE); OBD_ALLOC(fsd, sizeof(*fsd)); if (!fsd) RETURN(-ENOMEM); filter->fo_fsd = fsd; - OBD_ALLOC(filter->fo_last_rcvd_slots, - FILTER_LR_MAX_CLIENT_WORDS * sizeof(unsigned long)); + OBD_ALLOC(filter->fo_last_rcvd_slots, LR_MAX_CLIENTS / 8); if (filter->fo_last_rcvd_slots == NULL) { OBD_FREE(fsd, sizeof(*fsd)); RETURN(-ENOMEM); } if (last_rcvd_size == 0) { - CWARN("%s: initializing new %s\n", obd->obd_name, LAST_RCVD); - - memcpy(fsd->fsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->fsd_uuid)); - fsd->fsd_last_transno = 0; - mount_count = fsd->fsd_mount_count = 0; - fsd->fsd_server_size = cpu_to_le32(FILTER_LR_SERVER_SIZE); - fsd->fsd_client_start = cpu_to_le32(FILTER_LR_CLIENT_START); - fsd->fsd_client_size = cpu_to_le16(FILTER_LR_CLIENT_SIZE); - fsd->fsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT); + LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name); + + memcpy(fsd->lsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->lsd_uuid)); + fsd->lsd_last_transno = 0; + mount_count = fsd->lsd_mount_count = 0; + fsd->lsd_server_size = cpu_to_le32(LR_SERVER_SIZE); + fsd->lsd_client_start = cpu_to_le32(LR_CLIENT_START); + fsd->lsd_client_size = cpu_to_le16(LR_CLIENT_SIZE); + fsd->lsd_subdir_count = cpu_to_le16(FILTER_SUBDIR_COUNT); filter->fo_subdir_count = FILTER_SUBDIR_COUNT; + fsd->lsd_feature_incompat = cpu_to_le32(OBD_INCOMPAT_OST); } else { rc = fsfilt_read_record(obd, filp, fsd, sizeof(*fsd), &off); if (rc) { @@ -403,52 +690,60 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) LAST_RCVD, rc); GOTO(err_fsd, rc); } - if (strcmp(fsd->fsd_uuid, obd->obd_uuid.uuid) != 0) { - CERROR("OBD UUID %s does not match last_rcvd UUID %s\n", - obd->obd_uuid.uuid, fsd->fsd_uuid); + if (strcmp(fsd->lsd_uuid, obd->obd_uuid.uuid) != 0) { + LCONSOLE_ERROR_MSG(0x134, "Trying to start OBD %s " + "using the wrong disk %s. Were the " + "/dev/ assignments rearranged?\n", + obd->obd_uuid.uuid, fsd->lsd_uuid); GOTO(err_fsd, rc = -EINVAL); } - mount_count = le64_to_cpu(fsd->fsd_mount_count); - filter->fo_subdir_count = le16_to_cpu(fsd->fsd_subdir_count); + 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 & + cpu_to_le32(OBD_INCOMPAT_COMMON_LR))) + fsd->lsd_last_transno = fsd->lsd_compat14; + /* end COMPAT_146 */ } - if (fsd->fsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) { - CERROR("unsupported feature %x\n", - le32_to_cpu(fsd->fsd_feature_incompat) & + if (fsd->lsd_feature_incompat & ~cpu_to_le32(FILTER_INCOMPAT_SUPP)) { + CERROR("%s: unsupported incompat filesystem feature(s) %x\n", + obd->obd_name, le32_to_cpu(fsd->lsd_feature_incompat) & ~FILTER_INCOMPAT_SUPP); GOTO(err_fsd, rc = -EINVAL); } - if (fsd->fsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) { - CERROR("read-only feature %x\n", - le32_to_cpu(fsd->fsd_feature_rocompat) & + if (fsd->lsd_feature_rocompat & ~cpu_to_le32(FILTER_ROCOMPAT_SUPP)) { + CERROR("%s: unsupported read-only filesystem feature(s) %x\n", + obd->obd_name, le32_to_cpu(fsd->lsd_feature_rocompat) & ~FILTER_ROCOMPAT_SUPP); /* Do something like remount filesystem read-only */ GOTO(err_fsd, rc = -EINVAL); } - CDEBUG(D_INODE, "%s: server last_rcvd : "LPU64"\n", - obd->obd_name, le64_to_cpu(fsd->fsd_last_transno)); + CDEBUG(D_INODE, "%s: server last_transno : "LPU64"\n", + obd->obd_name, le64_to_cpu(fsd->lsd_last_transno)); CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n", obd->obd_name, mount_count + 1); CDEBUG(D_INODE, "%s: server data size: %u\n", - obd->obd_name, le32_to_cpu(fsd->fsd_server_size)); + obd->obd_name, le32_to_cpu(fsd->lsd_server_size)); CDEBUG(D_INODE, "%s: per-client data start: %u\n", - obd->obd_name, le32_to_cpu(fsd->fsd_client_start)); + obd->obd_name, le32_to_cpu(fsd->lsd_client_start)); CDEBUG(D_INODE, "%s: per-client data size: %u\n", - obd->obd_name, le32_to_cpu(fsd->fsd_client_size)); + obd->obd_name, le32_to_cpu(fsd->lsd_client_size)); CDEBUG(D_INODE, "%s: server subdir_count: %u\n", - obd->obd_name, le16_to_cpu(fsd->fsd_subdir_count)); + obd->obd_name, le16_to_cpu(fsd->lsd_subdir_count)); CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name, - last_rcvd_size <= le32_to_cpu(fsd->fsd_client_start) ? 0 : - (last_rcvd_size - le32_to_cpu(fsd->fsd_client_start)) / - le16_to_cpu(fsd->fsd_client_size)); + last_rcvd_size <= le32_to_cpu(fsd->lsd_client_start) ? 0 : + (last_rcvd_size - le32_to_cpu(fsd->lsd_client_start)) / + le16_to_cpu(fsd->lsd_client_size)); if (!obd->obd_replayable) { CWARN("%s: recovery support OFF\n", obd->obd_name); GOTO(out, rc = 0); } - for (cl_idx = 0, off = le32_to_cpu(fsd->fsd_client_start); + for (cl_idx = 0, off = le32_to_cpu(fsd->lsd_client_start); off < last_rcvd_size; cl_idx++) { __u64 last_rcvd; struct obd_export *exp; @@ -462,9 +757,9 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) /* Don't assume off is incremented properly by * fsfilt_read_record(), in case sizeof(*fcd) - * isn't the same as fsd->fsd_client_size. */ - off = le32_to_cpu(fsd->fsd_client_start) + - cl_idx * le16_to_cpu(fsd->fsd_client_size); + * isn't the same as fsd->lsd_client_size. */ + off = le32_to_cpu(fsd->lsd_client_start) + + cl_idx * le16_to_cpu(fsd->lsd_client_size); rc = fsfilt_read_record(obd, filp, fcd, sizeof(*fcd), &off); if (rc) { CERROR("error reading FILT %s idx %d off %llu: rc %d\n", @@ -483,63 +778,67 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp) /* These exports are cleaned up by filter_disconnect(), so they * need to be set up like real exports as filter_connect() does. */ - exp = class_new_export(obd); - CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64" " - "srv lr: "LPU64" fcd_group %d \n", fcd->fcd_uuid, cl_idx, - last_rcvd, le64_to_cpu(fsd->fsd_last_transno), - le32_to_cpu(fcd->fcd_group)); - if (exp == NULL) - GOTO(err_client, rc = -ENOMEM); + exp = class_new_export(obd, (struct obd_uuid *)fcd->fcd_uuid); - memcpy(&exp->exp_client_uuid.uuid, fcd->fcd_uuid, - sizeof exp->exp_client_uuid.uuid); - fed = &exp->exp_filter_data; - fed->fed_fcd = fcd; - fed->fed_group = le32_to_cpu(fcd->fcd_group); - filter_client_add(obd, filter, fed, cl_idx); - /* create helper if export init gets more complex */ - spin_lock_init(&fed->fed_lock); - - fcd = NULL; - exp->exp_replay_needed = 1; - obd->obd_recoverable_clients++; - obd->obd_max_recoverable_clients++; - class_export_put(exp); + CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64 + " 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 */ + CERROR("Zeroing out duplicate export due to " + "bug 10479.\n"); + fcd->fcd_uuid[0] = '\0'; + } else { + GOTO(err_client, rc = PTR_ERR(exp)); + } + } else { + fed = &exp->exp_filter_data; + fed->fed_fcd = fcd; + fed->fed_group = le32_to_cpu(fcd->fcd_group); + filter_export_stats_init(obd, exp, NULL); + rc = filter_client_add(obd, exp, cl_idx); + /* can't fail for existing client */ + LASSERTF(rc == 0, "rc = %d\n", rc); + + fcd = NULL; + spin_lock(&exp->exp_lock); + exp->exp_connecting = 0; + exp->exp_in_recovery = 0; + spin_unlock(&exp->exp_lock); + obd->obd_max_recoverable_clients++; + class_export_put(exp); + } + /* Need to check last_rcvd even for duplicated exports. */ CDEBUG(D_OTHER, "client at idx %d has last_rcvd = "LPU64"\n", cl_idx, last_rcvd); - if (last_rcvd > le64_to_cpu(fsd->fsd_last_transno)) - fsd->fsd_last_transno = cpu_to_le64(last_rcvd); - - } - - obd->obd_last_committed = le64_to_cpu(fsd->fsd_last_transno); - - if (obd->obd_recoverable_clients) { - CWARN("RECOVERY: service %s, %d recoverable clients, " - "last_transno "LPU64"\n", obd->obd_name, - obd->obd_recoverable_clients, - le64_to_cpu(fsd->fsd_last_transno)); - obd->obd_next_recovery_transno = obd->obd_last_committed + 1; - target_start_recovery_thread(obd, ost_handle); + if (last_rcvd > le64_to_cpu(fsd->lsd_last_transno)) + fsd->lsd_last_transno = cpu_to_le64(last_rcvd); } if (fcd) OBD_FREE(fcd, sizeof(*fcd)); + obd->obd_last_committed = le64_to_cpu(fsd->lsd_last_transno); + + target_recovery_init(obd, ost_handle); + out: filter->fo_mount_count = mount_count + 1; - fsd->fsd_mount_count = cpu_to_le64(filter->fo_mount_count); + fsd->lsd_mount_count = cpu_to_le64(filter->fo_mount_count); /* save it, so mount count and last_transno is current */ rc = filter_update_server_data(obd, filp, filter->fo_fsd, 1); if (rc) GOTO(err_client, rc); + RETURN(0); err_client: - class_disconnect_exports(obd, 0); + target_recovery_fini(obd); err_fsd: filter_free_server_data(filter); RETURN(rc); @@ -548,42 +847,56 @@ err_fsd: static int filter_cleanup_groups(struct obd_device *obd) { struct filter_obd *filter = &obd->u.filter; + struct file *filp; struct dentry *dentry; - int i, k; + int i, j; ENTRY; - for (i = 0; i < filter->fo_group_count; i++) { - if (filter->fo_subdirs != NULL) { - for (k = 0; k < filter->fo_subdir_count; k++) { - dentry = filter->fo_subdirs[i].dentry[k]; - if (dentry == NULL) - continue; + if (filter->fo_dentry_O_groups != NULL) { + for (i = 0; i < filter->fo_group_count; i++) { + dentry = filter->fo_dentry_O_groups[i]; + if (dentry != NULL) f_dput(dentry); - filter->fo_subdirs[i].dentry[k] = NULL; - } } - if (filter->fo_last_objid_files[i] != NULL) { - filp_close(filter->fo_last_objid_files[i], 0); - filter->fo_last_objid_files[i] = NULL; + OBD_FREE(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->fo_group_count; i++) { + filp = filter->fo_last_objid_files[i]; + if (filp != NULL) + filp_close(filp, 0); } - if (filter->fo_groups[i] != NULL) { - dput(filter->fo_groups[i]); - filter->fo_groups[i] = NULL; + OBD_FREE(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_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_group_count * + sizeof(*filter->fo_dentry_O_sub)); + filter->fo_dentry_O_sub = NULL; } - if (filter->fo_subdirs != NULL) - OBD_FREE(filter->fo_subdirs, - filter->fo_group_count * sizeof(*filter->fo_subdirs)); - if (filter->fo_groups != NULL) - OBD_FREE(filter->fo_groups, - filter->fo_group_count * sizeof(*filter->fo_groups)); - if (filter->fo_last_objids != NULL) + if (filter->fo_last_objids != NULL) { OBD_FREE(filter->fo_last_objids, - filter->fo_group_count * sizeof(__u64)); - if (filter->fo_last_objid_files != NULL) - OBD_FREE(filter->fo_last_objid_files, - filter->fo_group_count * sizeof(struct file *)); - f_dput(filter->fo_dentry_O); + filter->fo_group_count * + sizeof(*filter->fo_last_objids)); + filter->fo_last_objids = NULL; + } + if (filter->fo_dentry_O != NULL) { + f_dput(filter->fo_dentry_O); + filter->fo_dentry_O = NULL; + } RETURN(0); } @@ -645,6 +958,7 @@ static int filter_read_group_internal(struct obd_device *obd, int group, char name[25]; __u64 last_objid; loff_t off = 0; + int len = group + 1; snprintf(name, 24, "%d", group); name[24] = '\0'; @@ -692,14 +1006,14 @@ static int filter_read_group_internal(struct obd_device *obd, int group, char dir[20]; snprintf(dir, sizeof(dir), "d%u", i); - tmp_subdirs->dentry[i] = - simple_mkdir(dentry, dir, 0700, 1); + 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]); } @@ -707,7 +1021,6 @@ static int filter_read_group_internal(struct obd_device *obd, int group, /* 'group' is an index; we need an array of length 'group + 1' */ if (group + 1 > old_count) { - int len = group + 1; OBD_ALLOC(new_objids, len * sizeof(*new_objids)); OBD_ALLOC(new_subdirs, len * sizeof(*new_subdirs)); OBD_ALLOC(new_groups, len * sizeof(*new_groups)); @@ -717,44 +1030,45 @@ static int filter_read_group_internal(struct obd_device *obd, int group, new_groups == NULL || new_files == NULL) GOTO(cleanup, rc = -ENOMEM); - memcpy(new_objids, filter->fo_last_objids, - old_count * sizeof(*new_objids)); - memcpy(new_subdirs, filter->fo_subdirs, - old_count * sizeof(*new_subdirs)); - memcpy(new_groups, filter->fo_groups, - old_count * sizeof(*new_groups)); - memcpy(new_files, filter->fo_last_objid_files, - old_count * sizeof(*new_files)); - 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_subdirs, + OBD_FREE(filter->fo_dentry_O_sub, old_count * sizeof(*new_subdirs)); - OBD_FREE(filter->fo_groups, + 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_subdirs = new_subdirs; - filter->fo_groups = new_groups; + 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_groups[group] = dentry; + filter->fo_dentry_O_groups[group] = dentry; filter->fo_last_objid_files[group] = filp; if (filter->fo_subdir_count) { - filter->fo_subdirs[group] = *tmp_subdirs; + filter->fo_dentry_O_sub[group] = *tmp_subdirs; OBD_FREE(tmp_subdirs, sizeof(*tmp_subdirs)); } filter_update_last_group(obd, group); - - if (filp->f_dentry->d_inode->i_size == 0) { + + if (i_size_read(filp->f_dentry->d_inode) == 0) { filter->fo_last_objids[group] = FILTER_INIT_OBJID; - RETURN(0); + rc = filter_update_last_objid(obd, group, 1); + RETURN(rc); } filter->fo_last_objids[group] = le64_to_cpu(last_objid); @@ -765,13 +1079,13 @@ static int filter_read_group_internal(struct obd_device *obd, int group, switch (stage) { case 4: if (new_objids != NULL) - OBD_FREE(new_objids, group * sizeof(*new_objids)); + OBD_FREE(new_objids, len * sizeof(*new_objids)); if (new_subdirs != NULL) - OBD_FREE(new_subdirs, group * sizeof(*new_subdirs)); + OBD_FREE(new_subdirs, len * sizeof(*new_subdirs)); if (new_groups != NULL) - OBD_FREE(new_groups, group * sizeof(*new_groups)); + OBD_FREE(new_groups, len * sizeof(*new_groups)); if (new_files != NULL) - OBD_FREE(new_files, group * sizeof(*new_files)); + OBD_FREE(new_files, len * sizeof(*new_files)); case 3: if (filter->fo_subdir_count) { for (i = 0; i < filter->fo_subdir_count; i++) { @@ -808,13 +1122,13 @@ static int filter_read_groups(struct obd_device *obd, int last_group, 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; - int rc = 0, cleanup_phase = 0; - struct file *filp = NULL; - int last_group; + struct file *filp; + int last_group, rc = 0, cleanup_phase = 0; loff_t off = 0; ENTRY; @@ -851,17 +1165,17 @@ static int filter_prep_groups(struct obd_device *obd) GOTO(cleanup_O0, rc = -EEXIST); } - down(&O_dentry->d_inode->i_sem); + LOCK_INODE_MUTEX(O_dentry->d_inode); rc = vfs_rename(O_dentry->d_inode, dentry, O_dentry->d_inode, O0_dentry); - up(&O_dentry->d_inode->i_sem); + UNLOCK_INODE_MUTEX(O_dentry->d_inode); if (rc) { CERROR("error renaming O/R to O/0: rc %d\n", rc); GOTO(cleanup_O0, rc); } - filter->fo_fsd->fsd_feature_incompat |= - cpu_to_le32(FILTER_INCOMPAT_GROUPS); + 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); @@ -904,7 +1218,7 @@ static int filter_prep_groups(struct obd_device *obd) rc = filter_read_groups(obd, last_group, 1); if (rc) GOTO(cleanup, rc); - + filp_close(filp, 0); RETURN(0); @@ -921,6 +1235,7 @@ static int filter_prep_groups(struct obd_device *obd) break; } return rc; + } /* setup the object store with correct subdirectories */ @@ -941,40 +1256,61 @@ static int filter_prep(struct obd_device *obd) LAST_RCVD, rc); GOTO(out, rc); } - + filter->fo_rcvd_filp = file; if (!S_ISREG(file->f_dentry->d_inode->i_mode)) { CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD, file->f_dentry->d_inode->i_mode); GOTO(err_filp, rc = -ENOENT); } - /* steal operations */ - inode = file->f_dentry->d_inode; - filter->fo_fop = file->f_op; - filter->fo_iop = inode->i_op; - filter->fo_aops = inode->i_mapping->a_ops; + inode = file->f_dentry->d_parent->d_inode; + /* We use i_op->unlink directly in filter_vfs_unlink() */ + if (!inode->i_op || !inode->i_op->create || !inode->i_op->unlink) { + CERROR("%s: filesystem does not support create/unlink ops\n", + obd->obd_name); + GOTO(err_filp, rc = -EOPNOTSUPP); + } rc = filter_init_server_data(obd, file); if (rc) { CERROR("cannot read %s: rc = %d\n", LAST_RCVD, rc); GOTO(err_filp, rc); } - filter->fo_rcvd_filp = file; + /* open and create health check io file*/ + file = filp_open(HEALTH_CHECK, O_RDWR | O_CREAT, 0644); + if (IS_ERR(file)) { + rc = PTR_ERR(file); + CERROR("OBD filter: cannot open/create %s rc = %d\n", + HEALTH_CHECK, rc); + GOTO(err_server_data, rc); + } + filter->fo_health_check_filp = file; + if (!S_ISREG(file->f_dentry->d_inode->i_mode)) { + CERROR("%s is not a regular file!: mode = %o\n", HEALTH_CHECK, + file->f_dentry->d_inode->i_mode); + GOTO(err_health_check, rc = -ENOENT); + } + rc = lvfs_check_io_health(obd, file); + if (rc) + GOTO(err_health_check, rc); 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); +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_server_data: + target_recovery_fini(obd); filter_free_server_data(filter); - err_filp: - if (filp_close(file, 0)) +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; goto out; @@ -999,7 +1335,7 @@ static void filter_post(struct obd_device *obd) for (i = 1; i < filter->fo_group_count; i++) { rc = filter_update_last_objid(obd, i, - (i == filter->fo_group_count - 1)); + (i == filter->fo_group_count - 1)); if (rc) CERROR("error writing group %d lastobjid: rc = %d\n", i, rc); @@ -1010,29 +1346,37 @@ static void filter_post(struct obd_device *obd) if (rc) CERROR("error closing %s: rc = %d\n", LAST_RCVD, rc); + rc = filp_close(filter->fo_health_check_filp, 0); + filter->fo_health_check_filp = NULL; + if (rc) + CERROR("error closing %s: rc = %d\n", HEALTH_CHECK, 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, int group, obd_id id) +static void filter_set_last_id(struct filter_obd *filter, + obd_id id, obd_gr group) { LASSERT(filter->fo_fsd != NULL); - LASSERT(group > 0); - LASSERT(group < filter->fo_group_count); + LASSERT(group <= filter->fo_group_count); spin_lock(&filter->fo_objidlock); filter->fo_last_objids[group] = id; spin_unlock(&filter->fo_objidlock); } -__u64 filter_last_id(struct filter_obd *filter, int group) +obd_id filter_last_id(struct filter_obd *filter, obd_gr group) { obd_id id; LASSERT(filter->fo_fsd != NULL); - LASSERT(group > 0); - LASSERT(group < filter->fo_group_count); + LASSERT(group <= filter->fo_group_count); + /* FIXME: object groups */ spin_lock(&filter->fo_objidlock); id = filter->fo_last_objids[group]; spin_unlock(&filter->fo_objidlock); @@ -1040,110 +1384,48 @@ __u64 filter_last_id(struct filter_obd *filter, int group) return id; } -/* direct cut-n-paste of mds_blocking_ast() */ -static int filter_blocking_ast(struct ldlm_lock *lock, - struct ldlm_lock_desc *desc, - void *data, int flag) -{ - int do_ast; - ENTRY; - - if (flag == LDLM_CB_CANCELING) { - /* Don't need to do anything here. */ - RETURN(0); - } - - /* XXX layering violation! -phil */ - l_lock(&lock->l_resource->lr_namespace->ns_lock); - /* Get this: if filter_blocking_ast is racing with ldlm_intent_policy, - * such that filter_blocking_ast is called just before l_i_p takes the - * ns_lock, then by the time we get the lock, we might not be the - * correct blocking function anymore. So check, and return early, if - * so. */ - if (lock->l_blocking_ast != filter_blocking_ast) { - l_unlock(&lock->l_resource->lr_namespace->ns_lock); - RETURN(0); - } - - lock->l_flags |= LDLM_FL_CBPENDING; - do_ast = (!lock->l_readers && !lock->l_writers); - l_unlock(&lock->l_resource->lr_namespace->ns_lock); - - if (do_ast) { - struct lustre_handle lockh; - int rc; - - LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel"); - ldlm_lock2handle(lock, &lockh); - rc = ldlm_cli_cancel(&lockh); - if (rc < 0) - CERROR("ldlm_cli_cancel: %d\n", rc); - } else { - LDLM_DEBUG(lock, "Lock still has references, will be " - "cancelled later"); - } - RETURN(0); -} - -extern void *lock_dir(struct inode *dir, struct qstr *name); -extern void unlock_dir(struct inode *dir, void *lock); - -static void * filter_lock_dentry(struct obd_device *obd, - struct dentry *dparent, obd_id id) +static int filter_lock_dentry(struct obd_device *obd, struct dentry *dparent) { -#ifdef S_PDIROPS - struct qstr qstr; - char name[32]; - qstr.name = name; - qstr.len = sprintf(name, LPU64, id); - return lock_dir(dparent->d_inode, &qstr); -#else - down(&dparent->d_inode->i_sem); -#endif + LOCK_INODE_MUTEX(dparent->d_inode); return 0; } /* We never dget the object parent, so DON'T dput it either */ -static void filter_parent_unlock(struct dentry *dparent, void *lock) -{ -#ifdef S_PDIROPS - LASSERT(lock != NULL); - unlock_dir(dparent->d_inode, lock); -#else - up(&dparent->d_inode->i_sem); -#endif -} - -/* We never dget the object parent, so DON'T dput it either */ struct dentry *filter_parent(struct obd_device *obd, obd_gr group, obd_id objid) { struct filter_obd *filter = &obd->u.filter; - LASSERT(group < filter->fo_group_count); - LASSERT(group > 0); + struct filter_subdirs *subdirs; + LASSERT(group < filter->fo_group_count); /* FIXME: object groups */ - if (filter->fo_subdir_count == 0) - return filter->fo_groups[group]; + if (group > 0 || filter->fo_subdir_count == 0) + return filter->fo_dentry_O_groups[group]; - return filter->fo_subdirs[group].dentry[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 */ struct dentry *filter_parent_lock(struct obd_device *obd, obd_gr group, - obd_id objid, void **lock) + obd_id objid) { unsigned long now = jiffies; struct dentry *dparent = filter_parent(obd, group, objid); + int rc; if (IS_ERR(dparent)) return dparent; + if (dparent == NULL) + return ERR_PTR(-ENOENT); - LASSERT(dparent); - LASSERT(dparent->d_inode); + rc = filter_lock_dentry(obd, dparent); + fsfilt_check_slow(obd, now, obd_timeout, "parent lock"); + return rc ? ERR_PTR(rc) : dparent; +} - *lock = filter_lock_dentry(obd, dparent, objid); - if (time_after(jiffies, now + 15 * HZ)) - CERROR("slow parent lock %lus\n", (jiffies - now) / HZ); - return dparent; +/* We never dget the object parent, so DON'T dput it either */ +static void filter_parent_unlock(struct dentry *dparent) +{ + UNLOCK_INODE_MUTEX(dparent->d_inode); } /* How to get files, dentries, inodes from object id's. @@ -1153,17 +1435,23 @@ struct dentry *filter_parent_lock(struct obd_device *obd, obd_gr group, * dir_dentry is NULL, we do a read lock while we do the lookup to * avoid races with create/destroy and such changing the directory * internal to the filesystem code. */ -struct dentry *filter_id2dentry(struct obd_device *obd, - struct dentry *dir_dentry, - obd_gr group, obd_id id) +struct dentry *filter_fid2dentry(struct obd_device *obd, + struct dentry *dir_dentry, + obd_gr group, obd_id id) { struct dentry *dparent = dir_dentry; struct dentry *dchild; - void *lock = NULL; char name[32]; int len; ENTRY; + if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT) && + 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)); @@ -1171,15 +1459,19 @@ struct dentry *filter_id2dentry(struct obd_device *obd, len = sprintf(name, LPU64, id); if (dir_dentry == NULL) { - dparent = filter_parent_lock(obd, group, id, &lock); - if (IS_ERR(dparent)) + dparent = filter_parent_lock(obd, group, id); + if (IS_ERR(dparent)) { + CERROR("%s: error getting object "LPU64":"LPU64 + " parent: rc %ld\n", obd->obd_name, + id, group, PTR_ERR(dparent)); RETURN(dparent); + } } - CDEBUG(D_INODE, "looking up object O/%*s/%s\n", + CDEBUG(D_INODE, "looking up object O/%.*s/%s\n", dparent->d_name.len, dparent->d_name.name, name); dchild = /*ll_*/lookup_one_len(name, dparent, len); if (dir_dentry == NULL) - filter_parent_unlock(dparent, lock); + filter_parent_unlock(dparent); if (IS_ERR(dchild)) { CERROR("%s: child lookup error %ld\n", obd->obd_name, PTR_ERR(dchild)); @@ -1187,7 +1479,8 @@ struct dentry *filter_id2dentry(struct obd_device *obd, } if (dchild->d_inode != NULL && is_bad_inode(dchild->d_inode)) { - CERROR("%s: got bad inode "LPU64"\n", obd->obd_name, id); + CERROR("%s: got bad object "LPU64" inode %lu\n", + obd->obd_name, id, dchild->d_inode->i_ino); f_dput(dchild); RETURN(ERR_PTR(-ENOENT)); } @@ -1205,16 +1498,16 @@ static int filter_prepare_destroy(struct obd_device *obd, obd_id objid, { struct lustre_handle lockh; int flags = LDLM_AST_DISCARD_DATA, rc; - struct ldlm_res_id res_id = { .name = { objid, 0, group, 0 } }; + 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(NULL, NULL, obd->obd_namespace, res_id, - LDLM_EXTENT, &policy, LCK_PW, - &flags, filter_blocking_ast, ldlm_completion_ast, - NULL, NULL, NULL, 0, NULL, &lockh); + 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); /* We only care about the side-effects, just drop the lock. */ if (rc == ELDLM_OK) @@ -1223,131 +1516,264 @@ static int filter_prepare_destroy(struct obd_device *obd, obd_id objid, RETURN(rc); } -/* Caller must hold LCK_PW on parent and push us into kernel context. - * Caller is also required to ensure that dchild->d_inode exists. */ -static int filter_destroy_internal(struct obd_device *obd, obd_id objid, - struct dentry *dparent, - struct dentry *dchild) +/* This is vfs_unlink() without down(i_sem). If we call regular vfs_unlink() + * we have 2.6 lock ordering issues with filter_commitrw_write() as it takes + * i_sem before starting a handle, while filter_destroy() + vfs_unlink do the + * reverse. Caller must take i_sem before starting the transaction and we + * drop it here before the inode is removed from the dentry. bug 4180/6984 */ +int filter_vfs_unlink(struct inode *dir, struct dentry *dentry) { - struct inode *inode = dchild->d_inode; int rc; ENTRY; - if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) { - CERROR("destroying objid %*s nlink = %lu, count = %d\n", - dchild->d_name.len, dchild->d_name.name, - (unsigned long)inode->i_nlink, - atomic_read(&inode->i_count)); - } + /* don't need dir->i_zombie for 2.4, it is for rename/unlink of dir + * itself we already hold dir->i_mutex for child create/unlink ops */ + LASSERT(dentry->d_inode != NULL); + LASSERT(TRYLOCK_INODE_MUTEX(dir) == 0); + LASSERT(TRYLOCK_INODE_MUTEX(dentry->d_inode) == 0); - rc = vfs_unlink(dparent->d_inode, dchild); + /* may_delete() */ + if (/*!dentry->d_inode ||*/dentry->d_parent->d_inode != dir) + GOTO(out, rc = -ENOENT); + + rc = ll_permission(dir, MAY_WRITE | MAY_EXEC, NULL); if (rc) - CERROR("error unlinking objid %*s: rc %d\n", - dchild->d_name.len, dchild->d_name.name, rc); + GOTO(out, rc); + + if (IS_APPEND(dir)) + GOTO(out, rc = -EPERM); + + /* check_sticky() */ + if ((dentry->d_inode->i_uid != current->fsuid && !capable(CAP_FOWNER))|| + IS_APPEND(dentry->d_inode) || IS_IMMUTABLE(dentry->d_inode)) + GOTO(out, rc = -EPERM); + + /* NOTE: This might need to go outside i_mutex, though it isn't clear if + * that was done because of journal_start (which is already done + * here) or some other ordering issue. */ + DQUOT_INIT(dir); + + rc = security_inode_unlink(dir, dentry); + if (rc) + GOTO(out, rc); + + rc = dir->i_op->unlink(dir, dentry); +out: + /* need to drop i_mutex before we lose inode reference */ + UNLOCK_INODE_MUTEX(dentry->d_inode); + if (rc == 0) + d_delete(dentry); RETURN(rc); } +/* Caller must hold LCK_PW on parent and push us into kernel context. + * Caller must hold child i_mutex, we drop it always. + * Caller is also required to ensure that dchild->d_inode exists. */ +static int filter_destroy_internal(struct obd_device *obd, obd_id objid, + obd_gr group, struct dentry *dparent, + struct dentry *dchild) +{ + struct inode *inode = dchild->d_inode; + int rc; + + if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) { + CERROR("destroying objid %.*s ino %lu nlink %lu count %d\n", + dchild->d_name.len, dchild->d_name.name, inode->i_ino, + (unsigned long)inode->i_nlink, + atomic_read(&inode->i_count)); + } + + rc = filter_vfs_unlink(dparent->d_inode, dchild); + if (rc) + CERROR("error unlinking objid %.*s: rc %d\n", + dchild->d_name.len, dchild->d_name.name, rc); + 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 tmpflags = 0, rc, repsize[2] = {sizeof(*rep), sizeof(*reply_lvb)}; + 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) }; ENTRY; policy = ldlm_get_processing_policy(res); LASSERT(policy != NULL); LASSERT(req != NULL); - rc = lustre_pack_reply(req, 2, repsize, NULL); + rc = lustre_pack_reply(req, 3, repsize, NULL); if (rc) RETURN(req->rq_status = rc); - rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*rep)); + rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF, sizeof(*rep)); LASSERT(rep != NULL); - reply_lvb = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*reply_lvb)); + reply_lvb = lustre_msg_buf(req->rq_repmsg, DLM_REPLY_REC_OFF, + sizeof(*reply_lvb)); LASSERT(reply_lvb != NULL); //fixup_handle_for_resent_req(req, lock, &lockh); /* 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; - l_lock(&res->lr_namespace->ns_lock); - - res->lr_tmp = &rpc_list; - rc = policy(lock, &tmpflags, 0, &err); - res->lr_tmp = NULL; + LASSERT(ns == res->lr_namespace); + lock_res(res); + rc = policy(lock, &tmpflags, 0, &err, &rpc_list); + check_res_locked(res); /* FIXME: we should change the policy function slightly, to not make * this list at all, since we just turn around and free it */ while (!list_empty(&rpc_list)) { - struct ldlm_ast_work *w = - list_entry(rpc_list.next, struct ldlm_ast_work, w_list); - list_del(&w->w_list); - LDLM_LOCK_PUT(w->w_lock); - OBD_FREE(w, sizeof(*w)); + struct ldlm_lock *wlock = + list_entry(rpc_list.next, struct ldlm_lock, l_cp_ast); + LASSERT((lock->l_flags & LDLM_FL_AST_SENT) == 0); + LASSERT(lock->l_flags & LDLM_FL_CP_REQD); + lock->l_flags &= ~LDLM_FL_CP_REQD; + list_del_init(&wlock->l_cp_ast); + LDLM_LOCK_PUT(wlock); } + /* The lock met with no resistance; we're finished. */ if (rc == LDLM_ITER_CONTINUE) { - /* The lock met with no resistance; we're finished. */ - l_unlock(&res->lr_namespace->ns_lock); - RETURN(ELDLM_LOCK_REPLACED); + /* do not grant locks to the liblustre clients: they cannot + * handle ASTs robustly. We need to do this while still + * holding ns_lock to avoid the lock remaining on the res_link + * 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_TIMEOUT(OBD_FAIL_LDLM_GLIMPSE, 2)) { + ldlm_resource_unlink_lock(lock); + err = ELDLM_LOCK_ABORTED; + } else { + err = ELDLM_LOCK_REPLACED; + } + unlock_res(res); + RETURN(err); } /* Do not grant any lock, but instead send GL callbacks. The extent * policy nicely created a list of all PW locks for us. We will choose * the highest of those which are larger than the size in the LVB, if * any, and perform a glimpse callback. */ - down(&res->lr_lvb_sem); res_lvb = res->lr_lvb_data; LASSERT(res_lvb != NULL); *reply_lvb = *res_lvb; - up(&res->lr_lvb_sem); - 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; - - if (tmplock->l_policy_data.l_extent.end <= reply_lvb->lvb_size) - 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); } - l_unlock(&res->lr_namespace->ns_lock); + unlock_res(res); /* There were no PW locks beyond the size in the LVB; finished. */ - if (l == NULL) + if (l == NULL) { + if (only_liblustre) { + /* If we discovered a liblustre client with a PW lock, + * however, the LVB may be out of date! The LVB is + * updated only on glimpse (which we don't do for + * liblustre clients) and cancel (which the client + * obviously has not yet done). So if it has written + * data but kept the lock, the LVB is stale and needs + * to be updated from disk. + * + * Of course, this will all disappear when we switch to + * taking liblustre locks on the OST. */ + ldlm_res_lvbo_update(res, NULL, 0, 1); + } RETURN(ELDLM_LOCK_ABORTED); + } + /* + * This check is for lock taken in filter_prepare_destroy() that does + * not have l_glimpse_ast set. So the logic is: if there is a lock + * with no l_glimpse_ast set, this object is being destroyed already. + * + * Hence, if you are grabbing DLM locks on the server, always set + * non-NULL glimpse_ast (e.g., ldlm_request.c:ldlm_glimpse_ast()). + */ if (l->l_glimpse_ast == NULL) { /* We are racing with unlink(); just return -ENOENT */ rep->lock_policy_res1 = -ENOENT; @@ -1355,137 +1781,206 @@ static int filter_intent_policy(struct ldlm_namespace *ns, } LASSERTF(l->l_glimpse_ast != NULL, "l == %p", l); - rc = l->l_glimpse_ast(l, NULL); /* this will update the LVB */ - if (rc != 0 && res->lr_namespace->ns_lvbo && - res->lr_namespace->ns_lvbo->lvbo_update) { - res->lr_namespace->ns_lvbo->lvbo_update(res, NULL, 0, 1); - } + /* Update the LVB from disk if the AST failed (this is a legal race) */ + /* + * XXX nikita: situation when ldlm_server_glimpse_ast() failed before + * sending ast is not handled. This can result in lost client writes. + */ + if (rc != 0) + ldlm_res_lvbo_update(res, NULL, 0, 1); - down(&res->lr_lvb_sem); + lock_res(res); *reply_lvb = *res_lvb; - up(&res->lr_lvb_sem); -out: + unlock_res(res); + + out: LDLM_LOCK_PUT(l); RETURN(ELDLM_LOCK_ABORTED); } -static int filter_post_fs_cleanup(struct obd_device *obd) +/* + * per-obd_device iobuf pool. + * + * To avoid memory deadlocks in low-memory setups, amount of dynamic + * allocations in write-path has to be minimized (see bug 5137). + * + * Pages, niobuf_local's and niobuf_remote's are pre-allocated and attached to + * OST threads (see ost_thread_{init,done}()). + * + * "iobuf's" used by filter cannot be attached to OST thread, however, because + * at the OST layer there are only (potentially) multiple obd_device of type + * unknown at the time of OST thread creation. + * + * Instead array of iobuf's is attached to struct filter_obd (->fo_iobuf_pool + * field). This array has size OST_MAX_THREADS, so that each OST thread uses + * it's very own iobuf. + * + * Functions below + * + * filter_kiobuf_pool_init() + * + * filter_kiobuf_pool_done() + * + * filter_iobuf_get() + * + * operate on this array. They are "generic" in a sense that they don't depend + * on actual type of iobuf's (the latter depending on Linux kernel version). + */ + +/* + * destroy pool created by filter_iobuf_pool_init + */ +static void filter_iobuf_pool_done(struct filter_obd *filter) { - int rc = 0; + struct filter_iobuf **pool; + int i; - rc = fsfilt_post_cleanup(obd); + ENTRY; - RETURN(rc); + pool = filter->fo_iobuf_pool; + if (pool != NULL) { + for (i = 0; i < filter->fo_iobuf_count; ++ i) { + if (pool[i] != NULL) + filter_free_iobuf(pool[i]); + } + OBD_FREE(pool, filter->fo_iobuf_count * sizeof pool[0]); + filter->fo_iobuf_pool = NULL; + } + EXIT; } -static int filter_group_set_fs_flags(struct obd_device *obd, int group) +/* + * pre-allocate pool of iobuf's to be used by filter_{prep,commit}rw_write(). + */ +static int filter_iobuf_pool_init(struct filter_obd *filter) { - struct filter_obd *filter = &obd->u.filter; - int rc = 0, i = 0; - ENTRY; - - /* zero group is not longer valid. */ - if (group== 0) - RETURN(rc); - for (i = 0; i < filter->fo_subdir_count; i++) { - struct dentry *dentry; - dentry = (filter->fo_subdirs + group)->dentry[i]; - rc = fsfilt_set_fs_flags(obd, dentry->d_inode, - SM_DO_REC | SM_DO_COW); - if (rc) - RETURN(rc); - } - RETURN(rc); + void **pool; + + ENTRY; + + + OBD_ALLOC_GFP(filter->fo_iobuf_pool, OSS_THREADS_MAX * sizeof(*pool), + GFP_KERNEL); + if (filter->fo_iobuf_pool == NULL) + RETURN(-ENOMEM); + + filter->fo_iobuf_count = OSS_THREADS_MAX; + + RETURN(0); } -static int filter_post_fs_setup(struct obd_device *obd) + +/* Return iobuf allocated for @thread_id. We don't know in advance how + * many threads there will be so we allocate a large empty array and only + * fill in those slots that are actually in use. + * 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) { - struct filter_obd *filter = &obd->u.filter; - int rc = 0, j = 0; - struct llog_ctxt *ctxt = NULL; + int thread_id = oti ? oti->oti_thread_id : -1; + struct filter_iobuf *pool = NULL; + struct filter_iobuf **pool_place = NULL; - rc = fsfilt_post_setup(obd, filter->fo_dentry_O); - if (rc) - RETURN(rc); - - for (j = 0; j < filter->fo_group_count; j++) { - rc = filter_group_set_fs_flags(obd, j); - if (rc) - return rc; - } + if (thread_id >= 0) { + LASSERT(thread_id < filter->fo_iobuf_count); + pool = *(pool_place = &filter->fo_iobuf_pool[thread_id]); + } - fsfilt_get_reint_log_ctxt(obd, filter->fo_sb, &ctxt); - if (ctxt) { - ctxt->loc_obd = obd; - ctxt->loc_idx = LLOG_REINT_ORIG_CTXT; - obd->obd_llog_ctxt[LLOG_REINT_ORIG_CTXT] = ctxt; + if (unlikely(pool == NULL)) { + pool = filter_alloc_iobuf(filter, OBD_BRW_WRITE, + PTLRPC_MAX_BRW_PAGES); + if (pool_place != NULL) + *pool_place = pool; } - fsfilt_set_ost_flags(obd, filter->fo_sb); - return rc; + + return pool; } -/* mount the file system (secretly) */ -int filter_common_setup(struct obd_device *obd, obd_count len, void *buf, - char *option) +/* mount the file system (secretly). lustre_cfg parameters are: + * 1 = device + * 2 = fstype + * 3 = flags: failover=f, failout=n + * 4 = mount options + */ +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; + struct obd_uuid uuid; + __u8 *uuid_ptr; + char *str, *label; char ns_name[48]; - int rc = 0, i; + int rc, i; ENTRY; - dev_clear_rdonly(2); - - if (!lcfg->lcfg_inlbuf1 || !lcfg->lcfg_inlbuf2) + if (lcfg->lcfg_bufcount < 3 || + LUSTRE_CFG_BUFLEN(lcfg, 1) < 1 || + LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) RETURN(-EINVAL); - obd->obd_fsops = fsfilt_get_ops(lcfg->lcfg_inlbuf2); + lmi = server_get_mount(obd->obd_name); + if (lmi) { + /* We already mounted in lustre_fill_super. + lcfg bufs 1, 2, 4 (device, fstype, mount opts) are ignored.*/ + struct lustre_sb_info *lsi = s2lsi(lmi->lmi_sb); + mnt = lmi->lmi_mnt; + obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd)); + } else { + /* old path - used by lctl */ + CERROR("Using old MDS mount method\n"); + mnt = ll_kern_mount(lustre_cfg_string(lcfg, 2), + MS_NOATIME|MS_NODIRATIME, + lustre_cfg_string(lcfg, 1), option); + if (IS_ERR(mnt)) { + rc = PTR_ERR(mnt); + LCONSOLE_ERROR_MSG(0x135, "Can't mount disk %s (%d)\n", + lustre_cfg_string(lcfg, 1), rc); + RETURN(rc); + } + + obd->obd_fsops = fsfilt_get_ops(lustre_cfg_string(lcfg, 2)); + } if (IS_ERR(obd->obd_fsops)) - RETURN(PTR_ERR(obd->obd_fsops)); + GOTO(err_mntput, rc = PTR_ERR(obd->obd_fsops)); - mnt = do_kern_mount(lcfg->lcfg_inlbuf2, MS_NOATIME | MS_NODIRATIME, - lcfg->lcfg_inlbuf1, option); - rc = PTR_ERR(mnt); - if (IS_ERR(mnt)) + rc = filter_iobuf_pool_init(filter); + if (rc != 0) GOTO(err_ops, rc); - if (lcfg->lcfg_inllen3 > 0 && lcfg->lcfg_inlbuf3) { - if (*lcfg->lcfg_inlbuf3 == 'f') { - obd->obd_replayable = 1; - obd_sync_filter = 1; - CWARN("%s: recovery enabled\n", obd->obd_name); - } else { - if (*lcfg->lcfg_inlbuf3 != 'n') { - CERROR("unrecognised flag '%c'\n", - *lcfg->lcfg_inlbuf3); - } - // XXX Robert? Why do we get errors here - // GOTO(err_mntput, rc = -EINVAL); + LASSERT(!lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb))); + + /* failover is the default */ + obd->obd_replayable = 1; + + if (lcfg->lcfg_bufcount > 3 && LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) { + str = lustre_cfg_string(lcfg, 3); + if (strchr(str, 'n')) { + CWARN("%s: recovery disabled\n", obd->obd_name); + obd->obd_replayable = 0; } } filter->fo_vfsmnt = mnt; - filter->fo_sb = mnt->mnt_sb; + obd->u.obt.obt_sb = mnt->mnt_sb; filter->fo_fstype = mnt->mnt_sb->s_type->name; CDEBUG(D_SUPER, "%s: mnt = %p\n", filter->fo_fstype, mnt); + fsfilt_setup(obd, obd->u.obt.obt_sb); + 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; - rc = fsfilt_setup(obd, mnt->mnt_sb); - if (rc) - GOTO(err_mntput, rc); - sema_init(&filter->fo_init_lock, 1); filter->fo_committed_group = 0; + rc = filter_prep(obd); if (rc) - GOTO(err_mntput, rc); + GOTO(err_ops, rc); filter->fo_destroys_in_progress = 0; for (i = 0; i < 32; i++) @@ -1493,22 +1988,28 @@ int filter_common_setup(struct obd_device *obd, obd_count len, void *buf, 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); - spin_lock_init(&filter->fo_r_pages.oh_lock); - spin_lock_init(&filter->fo_w_pages.oh_lock); - spin_lock_init(&filter->fo_r_discont_pages.oh_lock); - spin_lock_init(&filter->fo_w_discont_pages.oh_lock); - spin_lock_init(&filter->fo_r_discont_blocks.oh_lock); - spin_lock_init(&filter->fo_w_discont_blocks.oh_lock); + 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; - INIT_LIST_HEAD(&filter->fo_llog_list); + CFS_INIT_LIST_HEAD(&filter->fo_llog_list); spin_lock_init(&filter->fo_llog_list_lock); - sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid); - obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER); + 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, + LDLM_NAMESPACE_GREEDY); if (obd->obd_namespace == NULL) GOTO(err_post, rc = -ENOMEM); obd->obd_namespace->ns_lvbp = obd; @@ -1518,158 +2019,545 @@ int filter_common_setup(struct obd_device *obd, obd_count len, void *buf, ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL, "filter_ldlm_cb_client", &obd->obd_ldlm_client); - rc = obd_llog_cat_initialize(obd, &obd->obd_llogs, 1, CATLIST); + rc = llog_cat_initialize(obd, &obd->obd_olg, 1, NULL); if (rc) { CERROR("failed to setup llogging subsystems\n"); GOTO(err_post, rc); } + + rc = lquota_setup(filter_quota_interface_ref, obd); + if (rc) + GOTO(err_post, rc); + + uuid_ptr = fsfilt_uuid(obd, obd->u.obt.obt_sb); + if (uuid_ptr != NULL) { + class_uuid_unparse(uuid_ptr, &uuid); + str = uuid.uuid; + } else { + str = "no UUID"; + } + + label = fsfilt_get_label(obd, obd->u.obt.obt_sb); + + if (obd->obd_recovering) { + LCONSOLE_WARN("OST %s now serving %s (%s%s%s), but will be in " + "recovery until %d %s reconnect, or if no clients" + " reconnect for %d:%.02d; during that time new " + "clients will not be allowed to connect. " + "Recovery progress can be monitored by watching " + "/proc/fs/lustre/obdfilter/%s/recovery_status.\n", + obd->obd_name, lustre_cfg_string(lcfg, 1), + label ?: "", label ? "/" : "", str, + obd->obd_max_recoverable_clients, + (obd->obd_max_recoverable_clients == 1) + ? "client" : "clients", + (int)(OBD_RECOVERY_TIMEOUT) / 60, + (int)(OBD_RECOVERY_TIMEOUT) % 60, + obd->obd_name); + } else { + LCONSOLE_INFO("OST %s now serving %s (%s%s%s) with recovery " + "%s\n", obd->obd_name, lustre_cfg_string(lcfg, 1), + label ?: "", label ? "/" : "", str, + obd->obd_replayable ? "enabled" : "disabled"); + } + RETURN(0); err_post: filter_post(obd); -err_mntput: - unlock_kernel(); - mntput(mnt); - filter->fo_sb = 0; - lock_kernel(); err_ops: fsfilt_put_ops(obd->obd_fsops); + filter_iobuf_pool_done(filter); +err_mntput: + server_put_mount(obd->obd_name, mnt); + obd->u.obt.obt_sb = 0; return rc; } -static int filter_attach(struct obd_device *obd, obd_count len, void *data) +static int filter_setup(struct obd_device *obd, struct lustre_cfg* lcfg) { struct lprocfs_static_vars lvars; + unsigned long addr; + struct page *page; int rc; - lprocfs_init_vars(filter, &lvars); - rc = lprocfs_obd_attach(obd, lvars.obd_vars); - if (rc != 0) - return rc; + CLASSERT(offsetof(struct obd_device, u.obt) == + offsetof(struct obd_device, u.filter.fo_obt)); - rc = lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST); - if (rc != 0) - return rc; + if (!LUSTRE_CFG_BUFLEN(lcfg, 1) || !LUSTRE_CFG_BUFLEN(lcfg, 2)) + RETURN(-EINVAL); - /* Init obdfilter private stats here */ - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES, - LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes"); - lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES, - LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes"); + /* 2.6.9 selinux wants a full option page for do_kern_mount (bug6471) */ + OBD_PAGE_ALLOC(page, CFS_ALLOC_STD); + if (!page) + RETURN(-ENOMEM); + addr = (unsigned long)cfs_page_address(page); + clear_page((void *)addr); + + /* lprocfs must be setup before the filter so state can be safely added + * to /proc incrementally as the filter is setup */ + lprocfs_filter_init_vars(&lvars); + if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0 && + lprocfs_alloc_obd_stats(obd, LPROC_FILTER_LAST) == 0) { + /* Init obdfilter private stats here */ + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_READ_BYTES, + LPROCFS_CNTR_AVGMINMAX, + "read_bytes", "bytes"); + lprocfs_counter_init(obd->obd_stats, LPROC_FILTER_WRITE_BYTES, + LPROCFS_CNTR_AVGMINMAX, + "write_bytes", "bytes"); + + lproc_filter_attach_seqstat(obd); + obd->obd_proc_exports_entry = lprocfs_register("exports", + obd->obd_proc_entry, + NULL, NULL); + if (IS_ERR(obd->obd_proc_exports_entry)) { + rc = PTR_ERR(obd->obd_proc_exports_entry); + CERROR("error %d setting up lprocfs for %s\n", + rc, "exports"); + obd->obd_proc_exports_entry = NULL; + } + } + 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; +} + +static struct llog_operations filter_mds_ost_repl_logops /* initialized below*/; +static struct llog_operations filter_size_orig_logops = { + lop_setup: llog_obd_origin_setup, + lop_cleanup: llog_obd_origin_cleanup, + lop_add: llog_obd_origin_add +}; + +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; + } 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) + GOTO(cleanup, rc); + + /* FIXME - assign unlink_cb for filter's recovery */ + 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, olg, LLOG_SIZE_ORIG_CTXT, tgt, 0, NULL, + &filter_size_orig_logops); - return lproc_filter_attach_seqstat(obd); +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_detach(struct obd_device *dev) +static int filter_group_llog_finish(struct obd_llog_group *olg) { - lprocfs_free_obd_stats(dev); - return lprocfs_obd_detach(dev); + struct llog_ctxt *ctxt; + int rc = 0, rc2 = 0; + ENTRY; + + ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT); + if (ctxt) + rc = llog_cleanup(ctxt); + + ctxt = llog_group_get_ctxt(olg, LLOG_SIZE_ORIG_CTXT); + if (ctxt) + rc2 = llog_cleanup(ctxt); + if (!rc) + rc = rc2; + + RETURN(rc); } -static int filter_setup(struct obd_device *obd, obd_count len, void *buf) +static int filter_llog_finish(struct obd_device *obd, int count) { - struct lustre_cfg* lcfg = buf; int rc; ENTRY; - /* all mount options including errors=remount-ro and asyncdel are passed - * using 4th lcfg param. And it is good, finally we have got rid of - * hardcoded fs types in the code. */ - rc = filter_common_setup(obd, len, buf, lcfg->lcfg_inlbuf4); + + 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) - RETURN(rc); - rc = filter_post_fs_setup(obd); + 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) +{ + int rc = 0; + ENTRY; + + switch(stage) { + case OBD_CLEANUP_EARLY: + break; + case OBD_CLEANUP_EXPORTS: + target_cleanup_recovery(obd); + rc = filter_llog_preclean(obd); + break; + case OBD_CLEANUP_SELF_EXP: + break; + case OBD_CLEANUP_OBD: + break; + } RETURN(rc); } -static int filter_cleanup(struct obd_device *obd, int flags) +static int filter_cleanup(struct obd_device *obd) { struct filter_obd *filter = &obd->u.filter; ENTRY; - if (flags & OBD_OPT_FAILOVER) - CERROR("%s: shutting down for failover; client state will" - " be preserved.\n", obd->obd_name); + if (obd->obd_fail) + LCONSOLE_WARN("%s: shutting down for failover; client state " + "will be preserved.\n", obd->obd_name); if (!list_empty(&obd->obd_exports)) { CERROR("%s: still has clients!\n", obd->obd_name); - class_disconnect_exports(obd, flags); + class_disconnect_exports(obd); if (!list_empty(&obd->obd_exports)) { CERROR("still has exports after forced cleanup?\n"); RETURN(-EBUSY); } } - target_cleanup_recovery(obd); - - ldlm_namespace_free(obd->obd_namespace, flags & OBD_OPT_FORCE); + 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); + + 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; + + fsfilt_put_ops(obd->obd_fsops); + + filter_iobuf_pool_done(filter); + + LCONSOLE_INFO("OST %s has stopped.\n", obd->obd_name); + + RETURN(0); +} - if (filter->fo_sb == NULL) +static int filter_connect_internal(struct obd_export *exp, + struct obd_connect_data *data) +{ + if (!data) RETURN(0); - filter_post_fs_cleanup(obd); - filter_post(obd); + CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64 + " ocd_version: %x ocd_grant: %d ocd_index: %u\n", + exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp, + data->ocd_connect_flags, data->ocd_version, + data->ocd_grant, data->ocd_index); + + data->ocd_connect_flags &= OST_CONNECT_SUPPORTED; + 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; + 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: " + LPU64" left: "LPU64"\n", exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, exp, + data->ocd_grant, want, left); + } - shrink_dcache_parent(filter->fo_sb->s_root); - filter->fo_sb = 0; + if (data->ocd_connect_flags & OBD_CONNECT_INDEX) { + 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 */ + lsd->lsd_ost_index = cpu_to_le32(data->ocd_index); + lsd->lsd_feature_compat |= cpu_to_le32(OBD_COMPAT_OST); + filter_update_server_data(exp->exp_obd, + filter->fo_rcvd_filp, lsd, 1); + } else if (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); + } + } - if (atomic_read(&filter->fo_vfsmnt->mnt_count) > 1) - CERROR("%s: mount point %p busy, mnt_count: %d\n", - obd->obd_name, filter->fo_vfsmnt, - atomic_read(&filter->fo_vfsmnt->mnt_count)); + if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) { + 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); + } - unlock_kernel(); - mntput(filter->fo_vfsmnt); - //destroy_buffers(filter->fo_sb->s_dev); - filter->fo_sb = NULL; - fsfilt_put_ops(obd->obd_fsops); - lock_kernel(); + 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)); + } - dev_clear_rdonly(2); + /* FIXME: Do the same with the MDS UUID and fsd_peeruuid. + * FIXME: We don't strictly need the COMPAT flag for that, + * FIXME: as fsd_peeruuid[0] will tell us if that is set. + * FIXME: We needed it for the index, as index 0 is valid. */ RETURN(0); } -static int filter_connect_post(struct obd_export *exp, unsigned long connect_flags) +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_device *obd = exp->exp_obd; - struct filter_export_data *fed; - char str[PTL_NALFMT_SIZE]; - struct obd_llogs *llog; - struct llog_ctxt *ctxt; int rc; ENTRY; - fed = &exp->exp_filter_data; - if (fed->fed_group < FILTER_MIN_GROUPS) - RETURN(0); - - /* initialize llogs for connections from MDS */ - llog = filter_grab_llog_for_group(obd, fed->fed_group, exp); - LASSERT(llog != NULL); - - ctxt = llog_get_context(llog, LLOG_UNLINK_REPL_CTXT); - LASSERT(ctxt != NULL); + if (exp == NULL || obd == NULL || cluuid == NULL) + RETURN(-EINVAL); - rc = llog_receptor_accept(ctxt, exp->exp_imp_reverse); - portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number, - exp->exp_connection->c_peer.peer_id.nid, str); - CDEBUG(D_OTHER, "%s: init llog ctxt for export "LPX64"/%s, group %d\n", - obd->obd_name, exp->exp_connection->c_peer.peer_id.nid, - str, fed->fed_group); + rc = filter_connect_internal(exp, data); RETURN(rc); } /* nearly identical to mds_connect */ -static int filter_connect(struct lustre_handle *conn, struct obd_device *obd, - struct obd_uuid *cluuid, unsigned long connect_flags) +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, void *localdata) { + struct lvfs_run_ctxt saved; struct obd_export *exp; struct filter_export_data *fed; struct filter_client_data *fcd = NULL; - struct filter_obd *filter = &obd->u.filter; - struct lvfs_run_ctxt saved; - int rc, group; + __u32 group; + int rc; ENTRY; if (conn == NULL || obd == NULL || cluuid == NULL) @@ -1683,10 +2571,12 @@ static int filter_connect(struct lustre_handle *conn, struct obd_device *obd, fed = &exp->exp_filter_data; - spin_lock_init(&fed->fed_lock); + rc = filter_connect_internal(exp, data); + if (rc) + GOTO(cleanup, rc); - /* connection from MDS */ - group = connect_flags; + filter_export_stats_init(obd, exp, localdata); + group = data->ocd_group; if (obd->obd_replayable) { OBD_ALLOC(fcd, sizeof(*fcd)); if (!fcd) { @@ -1697,25 +2587,20 @@ static int filter_connect(struct lustre_handle *conn, struct obd_device *obd, memcpy(fcd->fcd_uuid, cluuid, sizeof(fcd->fcd_uuid)); fed->fed_fcd = fcd; fed->fed_fcd->fcd_group = group; - rc = filter_client_add(obd, filter, fed, -1); + 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); - + obd->obd_name, exp->exp_handle.h_cookie, group); if (group == 0) GOTO(cleanup, rc); - + if (fed->fed_group != 0 && fed->fed_group != group) { - char str[PTL_NALFMT_SIZE]; - portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number, - exp->exp_connection->c_peer.peer_id.nid, str); - CERROR("!!! This export (nid "LPX64"/%s) used object group %d " + 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", - exp->exp_connection->c_peer.peer_id.nid, str, - fed->fed_group, group); + obd_export_nid2str(exp), fed->fed_group, group); GOTO(cleanup, rc = -EPROTO); } fed->fed_group = group; @@ -1727,52 +2612,19 @@ static int filter_connect(struct lustre_handle *conn, struct obd_device *obd, CERROR("can't read group %u\n", group); GOTO(cleanup, rc); } - rc = filter_group_set_fs_flags(obd, group); - if (rc != 0) { - CERROR("can't set kml flags %u\n", group); - GOTO(cleanup, rc); - } + + GOTO(cleanup, rc); cleanup: if (rc) { - if (fcd) + if (fcd) { OBD_FREE(fcd, sizeof(*fcd)); - class_disconnect(exp, 0); + fed->fed_fcd = NULL; + } + class_disconnect(exp); } else { class_export_put(exp); } - return rc; -} - -static int filter_precleanup(struct obd_device *obd, int flags) -{ - struct filter_group_llog *log; - struct filter_obd *filter; - int rc = 0; - ENTRY; - - filter = &obd->u.filter; - - spin_lock(&filter->fo_llog_list_lock); - while (!list_empty(&filter->fo_llog_list)) { - log = list_entry(filter->fo_llog_list.next, - struct filter_group_llog, list); - list_del(&log->list); - spin_unlock(&filter->fo_llog_list_lock); - - rc = obd_llog_finish(obd, log->llogs, 0); - if (rc) - CERROR("failed to cleanup llogging subsystem for %u\n", - log->group); - OBD_FREE(log->llogs, sizeof(*(log->llogs))); - OBD_FREE(log, sizeof(*log)); - spin_lock(&filter->fo_llog_list_lock); - } - spin_unlock(&filter->fo_llog_list_lock); - - rc = obd_llog_finish(obd, &obd->obd_llogs, 0); - if (rc) - CERROR("failed to cleanup llogging subsystem\n"); RETURN(rc); } @@ -1791,19 +2643,37 @@ static void filter_grant_sanity_check(struct obd_device *obd, const char *func) 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) { + int error = 0; fed = &exp->exp_filter_data; - LASSERTF(fed->fed_grant + fed->fed_pending <= maxsize, - "cli %s/%p %lu+%lu > "LPU64"\n", - exp->exp_client_uuid.uuid, exp, - fed->fed_grant, fed->fed_pending, maxsize); - LASSERTF(fed->fed_dirty <= maxsize, "cli %s/%p %lu > "LPU64"\n", - exp->exp_client_uuid.uuid, exp,fed->fed_dirty,maxsize); - CDEBUG(D_CACHE,"%s: cli %s/%p dirty %lu pend %lu grant %lu\n", - obd->obd_name, exp->exp_client_uuid.uuid, exp, - fed->fed_dirty, fed->fed_pending, fed->fed_grant); + if (fed->fed_grant < 0 || fed->fed_pending < 0 || + fed->fed_dirty < 0) + error = 1; + if (maxsize > 0) { /* we may not have done a statfs yet */ + LASSERTF(fed->fed_grant + fed->fed_pending <= maxsize, + "%s: cli %s/%p %ld+%ld > "LPU64"\n", func, + exp->exp_client_uuid.uuid, exp, + fed->fed_grant, fed->fed_pending, maxsize); + LASSERTF(fed->fed_dirty <= maxsize, + "%s: cli %s/%p %ld > "LPU64"\n", func, + exp->exp_client_uuid.uuid, exp, + fed->fed_dirty, maxsize); + } + if (error) + CERROR("%s: cli %s/%p dirty %ld pend %ld grant %ld\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, + fed->fed_dirty, fed->fed_pending,fed->fed_grant); + else + CDEBUG(D_CACHE, "%s: cli %s/%p dirty %ld pend %ld grant %ld\n", + obd->obd_name, exp->exp_client_uuid.uuid, exp, + fed->fed_dirty, fed->fed_pending,fed->fed_grant); tot_granted += fed->fed_grant + fed->fed_pending; tot_pending += fed->fed_pending; tot_dirty += fed->fed_dirty; @@ -1846,25 +2716,22 @@ static void filter_grant_discard(struct obd_export *exp) struct filter_export_data *fed = &exp->exp_filter_data; spin_lock(&obd->obd_osfs_lock); - spin_lock(&exp->exp_obd->obd_dev_lock); + spin_lock(&obd->obd_dev_lock); list_del_init(&exp->exp_obd_chain); - spin_unlock(&exp->exp_obd->obd_dev_lock); - - CDEBUG(D_CACHE, "%s: cli %s/%p dirty %lu pend %lu grant %lu\n", - obd->obd_name, exp->exp_client_uuid.uuid, exp, - fed->fed_dirty, fed->fed_pending, fed->fed_grant); + spin_unlock(&obd->obd_dev_lock); LASSERTF(filter->fo_tot_granted >= fed->fed_grant, - "%s: tot_granted "LPU64" cli %s/%p fed_grant %lu\n", + "%s: tot_granted "LPU64" cli %s/%p fed_grant %ld\n", obd->obd_name, filter->fo_tot_granted, exp->exp_client_uuid.uuid, exp, fed->fed_grant); filter->fo_tot_granted -= fed->fed_grant; - LASSERTF(exp->exp_obd->u.filter.fo_tot_pending >= fed->fed_pending, - "%s: tot_pending "LPU64" cli %s/%p fed_pending %lu\n", + LASSERTF(filter->fo_tot_pending >= fed->fed_pending, + "%s: tot_pending "LPU64" cli %s/%p fed_pending %ld\n", obd->obd_name, filter->fo_tot_pending, exp->exp_client_uuid.uuid, exp, fed->fed_pending); + /* fo_tot_pending is handled in filter_grant_commit as bulk finishes */ LASSERTF(filter->fo_tot_dirty >= fed->fed_dirty, - "%s: tot_dirty "LPU64" cli %s/%p fed_dirty %lu\n", + "%s: tot_dirty "LPU64" cli %s/%p fed_dirty %ld\n", obd->obd_name, filter->fo_tot_dirty, exp->exp_client_uuid.uuid, exp, fed->fed_dirty); filter->fo_tot_dirty -= fed->fed_dirty; @@ -1883,12 +2750,25 @@ static int filter_destroy_export(struct obd_export *exp) 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); + + lprocfs_exp_cleanup(exp); + if (exp->exp_obd->obd_replayable) - filter_client_free(exp, exp->exp_flags); + filter_client_free(exp); + else + fsfilt_sync(exp->exp_obd, exp->exp_obd->u.obt.obt_sb); filter_grant_discard(exp); + filter_fmd_cleanup(exp); + if (!(exp->exp_flags & OBD_OPT_FORCE)) filter_grant_sanity_check(exp->exp_obd, __FUNCTION__); @@ -1897,7 +2777,7 @@ static int filter_destroy_export(struct obd_export *exp) static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp) { - struct filter_group_llog *fglog, *nlog; + struct obd_llog_group *olg_min, *olg; struct filter_obd *filter; int worked = 0, group; struct llog_ctxt *ctxt; @@ -1910,75 +2790,93 @@ static void filter_sync_llogs(struct obd_device *obd, struct obd_export *dexp) * group order and skip already synced llogs -bzzz */ do { /* look for group with min. number, but > worked */ - fglog = NULL; + olg_min = NULL; group = 1 << 30; spin_lock(&filter->fo_llog_list_lock); - list_for_each_entry(nlog, &filter->fo_llog_list, list) { - - if (nlog->group <= worked) { + list_for_each_entry(olg, &filter->fo_llog_list, olg_list) { + if (olg->olg_group <= worked) { /* this group is already synced */ continue; } - - if (group < nlog->group) { + if (group < olg->olg_group) { /* we have group with smaller number to sync */ continue; } - /* store current minimal group */ - fglog = nlog; - group = nlog->group; + olg_min = olg; + group = olg->olg_group; } spin_unlock(&filter->fo_llog_list_lock); - if (fglog == NULL) + if (olg_min == NULL) break; - worked = fglog->group; - if (fglog->exp && (dexp == fglog->exp || dexp == NULL)) { - ctxt = llog_get_context(fglog->llogs, - LLOG_UNLINK_REPL_CTXT); + 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); - llog_sync(ctxt, fglog->exp); + 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 (fglog != NULL); + } while (olg_min != NULL); } /* also incredibly similar to mds_disconnect */ -static int filter_disconnect(struct obd_export *exp, unsigned long flags) +static int filter_disconnect(struct obd_export *exp) { struct obd_device *obd = exp->exp_obd; - unsigned long irqflags; int rc; ENTRY; LASSERT(exp); class_export_get(exp); - spin_lock_irqsave(&exp->exp_lock, irqflags); - exp->exp_flags = flags; - spin_unlock_irqrestore(&exp->exp_lock, irqflags); - - if (!(flags & OBD_OPT_FORCE)) + if (!(exp->exp_flags & OBD_OPT_FORCE)) filter_grant_sanity_check(obd, __FUNCTION__); filter_grant_discard(exp); /* Disconnect early so that clients can't keep using export */ - rc = class_disconnect(exp, flags); - - ldlm_cancel_locks_for_export(exp); + rc = class_disconnect(exp); + if (exp->exp_obd->obd_namespace != NULL) + ldlm_cancel_locks_for_export(exp); - fsfilt_sync(obd, obd->u.filter.fo_sb); + 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); } -struct dentry *__filter_oa2dentry(struct obd_device *obd, - struct obdo *oa, const char *what) +/* 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); + + return 0; +} + +struct dentry *__filter_oa2dentry(struct obd_device *obd, struct obdo *oa, + const char *what, int quiet) { struct dentry *dchild = NULL; obd_gr group = 0; @@ -1986,16 +2884,18 @@ struct dentry *__filter_oa2dentry(struct obd_device *obd, if (oa->o_valid & OBD_MD_FLGROUP) group = oa->o_gr; - dchild = filter_id2dentry(obd, NULL, group, 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); } if (dchild->d_inode == NULL) { - CERROR("%s: %s on non-existent object: "LPU64"\n", - obd->obd_name, what, oa->o_id); + if (!quiet) + CERROR("%s: %s on non-existent object: "LPU64"\n", + obd->obd_name, what, oa->o_id); f_dput(dchild); RETURN(ERR_PTR(-ENOENT)); } @@ -2003,109 +2903,299 @@ struct dentry *__filter_oa2dentry(struct obd_device *obd, return dchild; } -static int filter_getattr(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *md) +static int filter_getattr(struct obd_export *exp, struct obd_info *oinfo) { struct dentry *dentry = NULL; struct obd_device *obd; 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 cookie "LPX64"\n", - exp->exp_handle.h_cookie); + CDEBUG(D_IOCTL, "invalid client export %p\n", exp); RETURN(-EINVAL); } - dentry = filter_oa2dentry(obd, oa); + dentry = filter_oa2dentry(obd, oinfo->oi_oa); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); /* Limit the valid bits in the return data to what we actually use */ - oa->o_valid = OBD_MD_FLID; - obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS); + oinfo->oi_oa->o_valid = OBD_MD_FLID; + obdo_from_inode(oinfo->oi_oa, dentry->d_inode, FILTER_VALID_FLAGS); f_dput(dentry); RETURN(rc); } +/* this should be enabled/disabled in condition to enabled/disabled large + * inodes (fast EAs) in backing store FS. */ +int filter_update_fidea(struct obd_export *exp, struct inode *inode, + void *handle, struct obdo *oa) +{ + struct obd_device *obd = exp->exp_obd; + int rc = 0; + ENTRY; + + if (oa->o_valid & OBD_MD_FLFID) { + struct filter_fid ff; + + 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. */ + ff.ff_fid.id = cpu_to_le64(oa->o_fid); + ff.ff_fid.f_type = cpu_to_le32(oa->o_stripe_idx); + ff.ff_fid.generation = cpu_to_le32(oa->o_generation); + ff.ff_objid = cpu_to_le64(oa->o_id); + ff.ff_group = cpu_to_le64(oa->o_gr); + + CDEBUG(D_INODE, "storing filter fid EA ("LPU64"/%u/%u" + LPU64"/"LPU64")\n", oa->o_fid, oa->o_stripe_idx, + oa->o_generation, oa->o_id, oa->o_gr); + + rc = fsfilt_set_md(obd, inode, handle, &ff, sizeof(ff), "fid"); + if (rc) + CERROR("store fid in object failed! rc: %d\n", rc); + } else { + CDEBUG(D_HA, "OSS object without fid info!\n"); + } + + RETURN(rc); +} + /* this is called from filter_truncate() until we have filter_punch() */ -static int filter_setattr(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *md, struct obd_trans_info *oti) +int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry, + struct obdo *oa, struct obd_trans_info *oti) { - struct lvfs_run_ctxt saved; + unsigned int orig_ids[MAXQUOTAS] = {0, 0}; + struct llog_cookie *fcc = NULL; struct filter_obd *filter; - struct dentry *dentry; + int rc, err, locked = 0, sync = 0; + unsigned int ia_valid; + struct inode *inode; struct iattr iattr; - struct ldlm_res_id res_id = { .name = { oa->o_id, 0, oa->o_gr, 0 } }; - struct ldlm_resource *res; void *handle; - int rc, rc2; ENTRY; - LASSERT(oti != NULL); + LASSERT(dentry != NULL); + LASSERT(!IS_ERR(dentry)); - dentry = filter_oa2dentry(exp->exp_obd, oa); - if (IS_ERR(dentry)) - RETURN(PTR_ERR(dentry)); + inode = dentry->d_inode; + LASSERT(inode != NULL); filter = &exp->exp_obd->u.filter; - iattr_from_obdo(&iattr, oa, oa->o_valid); + ia_valid = iattr.ia_valid; - push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); - lock_kernel(); + if (oa->o_valid & OBD_MD_FLCOOKIE) { + OBD_ALLOC(fcc, sizeof(*fcc)); + if (fcc != NULL) + memcpy(fcc, obdo_logcookie(oa), sizeof(*fcc)); + } - if (iattr.ia_valid & ATTR_SIZE) - down(&dentry->d_inode->i_sem); - handle = fsfilt_start(exp->exp_obd, dentry->d_inode, FSFILT_OP_SETATTR, - oti); - if (IS_ERR(handle)) - GOTO(out_unlock, rc = PTR_ERR(handle)); - - /* XXX this could be a rwsem instead, if filter_preprw played along */ - if (iattr.ia_valid & ATTR_ATTR_FLAG) - rc = fsfilt_iocontrol(exp->exp_obd, dentry->d_inode, NULL, - EXT3_IOC_SETFLAGS, - (long)&iattr.ia_attr_flags); - else - rc = fsfilt_setattr(exp->exp_obd, dentry, handle, &iattr, 1); - rc = filter_finish_transno(exp, oti, rc); - rc2 = fsfilt_commit(exp->exp_obd, filter->fo_sb, dentry->d_inode, handle, 0); - if (rc2) { - CERROR("error on commit, err = %d\n", rc2); - if (!rc) - rc = rc2; + if (ia_valid & ATTR_SIZE || ia_valid & (ATTR_UID | ATTR_GID)) { + DQUOT_INIT(inode); + LOCK_INODE_MUTEX(inode); + locked = 1; } - if (iattr.ia_valid & ATTR_SIZE) { - res = ldlm_resource_get(exp->exp_obd->obd_namespace, NULL, - res_id, LDLM_EXTENT, 0); - if (res == NULL) { - CERROR("!!! resource_get failed for object "LPU64" -- " - "filter_setattr with no lock?\n", oa->o_id); - } else { - if (res->lr_namespace->ns_lvbo && - res->lr_namespace->ns_lvbo->lvbo_update) { - rc = res->lr_namespace->ns_lvbo->lvbo_update - (res, NULL, 0, 0); + /* If the inode still has SUID+SGID bits set (see filter_precreate()) + * then we will accept the UID+GID sent by the client during write for + * initializing the ownership of this inode. We only allow this to + * happen once so clear these bits in setattr. In 2.6 kernels it is + * possible to get ATTR_UID and ATTR_GID separately, so we only clear + * the flags that are actually being set. */ + if (ia_valid & (ATTR_UID | ATTR_GID)) { + CDEBUG(D_INODE, "update UID/GID to %lu/%lu\n", + (unsigned long)oa->o_uid, (unsigned long)oa->o_gid); + + if ((inode->i_mode & S_ISUID) && (ia_valid & ATTR_UID)) { + if (!(ia_valid & ATTR_MODE)) { + iattr.ia_mode = inode->i_mode; + iattr.ia_valid |= ATTR_MODE; } - ldlm_resource_putref(res); + iattr.ia_mode &= ~S_ISUID; } + if ((inode->i_mode & S_ISGID) && (ia_valid & ATTR_GID)) { + if (!(iattr.ia_valid & ATTR_MODE)) { + iattr.ia_mode = inode->i_mode; + iattr.ia_valid |= ATTR_MODE; + } + iattr.ia_mode &= ~S_ISGID; + } + + orig_ids[USRQUOTA] = inode->i_uid; + 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)); + + /* update inode EA only once when inode is suid bit marked. As + * on 2.6.x UID and GID may be set separately, we check here + * only one of them to avoid double setting. */ + if (inode->i_mode & S_ISUID) + filter_update_fidea(exp, inode, handle, oa); + } 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) + /* set cancel cookie callback function */ + sync = fsfilt_add_journal_cb(exp->exp_obd, 0, handle, + filter_cancel_cookies_cb, + fcc); } - oa->o_valid = OBD_MD_FLID; - obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS); + if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_CREDITS)) + fsfilt_extend(exp->exp_obd, inode, 0, handle); + + /* The truncate might have used up our transaction credits. Make + * sure we have one left for the last_rcvd update. */ + err = fsfilt_extend(exp->exp_obd, inode, 1, handle); + + 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) { + /* 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; + } + EXIT; out_unlock: - if (iattr.ia_valid & ATTR_SIZE) - up(&dentry->d_inode->i_sem); - unlock_kernel(); - pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); + if (locked) + UNLOCK_INODE_MUTEX(inode); + + if (fcc) + OBD_FREE(fcc, sizeof(*fcc)); + + /* 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, + orig_ids, rc, FSFILT_OP_SETATTR); + CDEBUG(rc2 ? D_ERROR : D_QUOTA, + "filter adjust qunit. (rc:%d)\n", rc2); + } + return rc; +} + +/* this is called from filter_truncate() until we have filter_punch() */ +int filter_setattr(struct obd_export *exp, struct obd_info *oinfo, + struct obd_trans_info *oti) +{ + 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; + struct ldlm_resource *res; + struct dentry *dentry; + 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)) + RETURN(PTR_ERR(dentry)); + + filter = &exp->exp_obd->u.filter; + push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); + lock_kernel(); + + if (oinfo->oi_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); + 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); + if (rc) + GOTO(out_unlock, rc); + + res = ldlm_resource_get(exp->exp_obd->obd_namespace, NULL, + &res_id, LDLM_EXTENT, 0); + if (res != NULL) { + rc = ldlm_res_lvbo_update(res, NULL, 0, 0); + ldlm_resource_putref(res); + } + + oinfo->oi_oa->o_valid = OBD_MD_FLID; + + /* Quota release need uid/gid info */ + obdo_from_inode(oinfo->oi_oa, dentry->d_inode, + FILTER_VALID_FLAGS | OBD_MD_FLUID | OBD_MD_FLGID); + + EXIT; +out_unlock: + unlock_kernel(); f_dput(dentry); - RETURN(rc); + pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); + return rc; } /* XXX identical to osc_unpackmd */ @@ -2134,6 +3224,7 @@ static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp, RETURN(lsm_size); if (*lsmp != NULL && lmm == NULL) { + OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo)); OBD_FREE(*lsmp, lsm_size); *lsmp = NULL; RETURN(0); @@ -2144,7 +3235,12 @@ static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp, if (*lsmp == NULL) RETURN(-ENOMEM); - loi_init((*lsmp)->lsm_oinfo); + OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo)); + if ((*lsmp)->lsm_oinfo[0] == NULL) { + OBD_FREE(*lsmp, lsm_size); + RETURN(-ENOMEM); + } + loi_init((*lsmp)->lsm_oinfo[0]); } if (lmm != NULL) { @@ -2158,111 +3254,202 @@ static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp, RETURN(lsm_size); } -static void filter_destroy_precreated(struct obd_export *exp, struct obdo *oa, +/* 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 obdo *doa = NULL; - __u64 last, id; + struct obdo doa; /* XXX obdo on stack */ + obd_id last, id; + int rc; ENTRY; - + LASSERT(oa); LASSERT(oa->o_gr != 0); LASSERT(oa->o_valid & OBD_MD_FLGROUP); + LASSERT(down_trylock(&filter->fo_create_locks[oa->o_gr]) != 0); - OBD_ALLOC(doa, sizeof(*doa)); - doa = obdo_alloc(); - if (doa == NULL) { - CERROR("cannot allocate doa, error %d\n", - -ENOMEM); - EXIT; - return; - } + memset(&doa, 0, sizeof(doa)); - doa->o_mode = S_IFREG; - doa->o_gr = oa->o_gr; - doa->o_valid = oa->o_valid & (OBD_MD_FLGROUP | OBD_MD_FLID); + doa.o_valid |= OBD_MD_FLGROUP; + doa.o_gr = oa->o_gr; + doa.o_mode = S_IFREG; - set_bit(doa->o_gr, &filter->fo_destroys_in_progress); - down(&filter->fo_create_locks[doa->o_gr]); - if (!test_bit(doa->o_gr, &filter->fo_destroys_in_progress)) { - CERROR("%s:["LPU64"] destroy_in_progress already cleared\n", - exp->exp_obd->obd_name, doa->o_gr); - up(&filter->fo_create_locks[doa->o_gr]); - GOTO(out_free_doa, 0); + 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:["LPU64"] deleting orphan objects from "LPU64" to "LPU64"\n", - exp->exp_obd->obd_name, doa->o_gr, oa->o_id + 1, last); - for (id = oa->o_id + 1; id <= last; id++) { - doa->o_id = id; - filter_destroy(exp, doa, NULL, NULL); + 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); + 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) + filter_update_last_objid(exp->exp_obd, doa.o_gr, 0); } - CDEBUG(D_HA, "%s:["LPU64"] after destroy: set last_objids = "LPU64"\n", - exp->exp_obd->obd_name, doa->o_gr, oa->o_id); - - filter_set_last_id(filter, doa->o_gr, oa->o_id); + CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n", + exp->exp_obd->obd_name, doa.o_gr, oa->o_id); - clear_bit(doa->o_gr, &filter->fo_destroys_in_progress); - up(&filter->fo_create_locks[doa->o_gr]); + rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1); + clear_bit(doa.o_gr, &filter->fo_destroys_in_progress); - EXIT; -out_free_doa: - obdo_free(doa); + RETURN(rc); } +static int filter_precreate(struct obd_device *obd, struct obdo *oa, + obd_gr group, int *num); /* returns a negative error or a nonnegative number of files to create */ -static int filter_should_precreate(struct obd_export *exp, struct obdo *oa, - obd_gr group) +static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa, + obd_gr group, struct obd_trans_info *oti) { struct obd_device *obd = exp->exp_obd; struct filter_obd *filter = &obd->u.filter; int diff, rc; ENTRY; - diff = oa->o_id - filter_last_id(filter, oa->o_gr); - CDEBUG(D_INFO, "filter_last_id() = "LPU64" -> diff = %d\n", - filter_last_id(filter, oa->o_gr), diff); - /* delete orphans request */ - if ((oa->o_valid & OBD_MD_FLFLAGS) && - (oa->o_flags & OBD_FL_DELORPHAN)) { - if (diff >= 0) - RETURN(diff); - if (-diff > 10000) { /* XXX make this smarter */ - CERROR("ignoring bogus orphan destroy request: obdid " - LPU64" last_id "LPU64"\n", - oa->o_id, filter_last_id(filter, oa->o_gr)); - RETURN(-EINVAL); + if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_DELORPHAN)){ + if (oti->oti_conn_cnt < exp->exp_conn_cnt) { + CERROR("%s: dropping old orphan cleanup request\n", + obd->obd_name); + RETURN(0); + } + /* This causes inflight precreates to abort and drop 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); + + 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)); + /* FIXME: should reset precreate_next_id on MDS */ + GOTO(out, rc = -EINVAL); + } + if (diff < 0) { + rc = filter_destroy_precreated(exp, oa, filter); + if (rc) + CERROR("%s: unable to write lastobjid, but " + "orphans were deleted\n", obd->obd_name); + GOTO(out, rc); + } else { + /* XXX: Used by MDS for the first time! */ + clear_bit(group, &filter->fo_destroys_in_progress); } - filter_destroy_precreated(exp, oa, filter); - rc = filter_update_last_objid(obd, group, 0); - if (rc) - CERROR("unable to write lastobjid, but orphans" - "were deleted\n"); - RETURN(0); } else { + 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 (!(oa->o_valid & OBD_FL_DELORPHAN) && - (/*group != 0 ||*/ oa->o_id == 0)) - RETURN(1); + if (group < FILTER_GROUP_MDS0 || oa->o_id == 0) + diff = 1; + else + diff = oa->o_id - filter_last_id(filter, group); + CDEBUG(D_RPCTRACE, "filter_last_id() = "LPU64" -> diff = %d\n", + filter_last_id(filter, group), diff); - LASSERT(diff >= 0); - RETURN(diff); + LASSERTF(diff >= 0,"%s: "LPU64" - "LPU64" = %d\n",obd->obd_name, + oa->o_id, filter_last_id(filter, group), diff); } + + if (diff > 0) { + oa->o_id = filter_last_id(&obd->u.filter, group); + rc = filter_precreate(obd, oa, group, &diff); + oa->o_id = filter_last_id(&obd->u.filter, group); + oa->o_gr = group; + oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP; + GOTO(out, rc); + } + /* else diff == 0 */ + GOTO(out, rc = 0); +out: + up(&filter->fo_create_locks[group]); + return rc; +} + +static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs, + __u64 max_age, __u32 flags) +{ + struct filter_obd *filter = &obd->u.filter; + int blockbits = obd->u.obt.obt_sb->s_blocksize_bits; + int rc; + ENTRY; + + /* at least try to account for cached pages. its still racey and + * might be under-reporting if clients haven't announced their + * caches with brw recently */ + spin_lock(&obd->obd_osfs_lock); + rc = fsfilt_statfs(obd, obd->u.obt.obt_sb, max_age); + memcpy(osfs, &obd->obd_osfs, sizeof(*osfs)); + spin_unlock(&obd->obd_osfs_lock); + + CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64 + " pending "LPU64" free "LPU64" avail "LPU64"\n", + filter->fo_tot_dirty, filter->fo_tot_granted, + filter->fo_tot_pending, + osfs->os_bfree << blockbits, osfs->os_bavail << blockbits); + + filter_grant_sanity_check(obd, __FUNCTION__); + + osfs->os_bavail -= min(osfs->os_bavail, GRANT_FOR_LLOG(obd) + + ((filter->fo_tot_dirty + filter->fo_tot_pending + + osfs->os_bsize - 1) >> blockbits)); + + /* set EROFS to state field if FS is mounted as RDONLY. The goal is to + * stop creating files on MDS if OST is not good shape to create + * objects.*/ + osfs->os_state = (filter->fo_obt.obt_sb->s_flags & MS_RDONLY) ? + EROFS : 0; + RETURN(rc); } -static int filter_precreate_rec(struct obd_device *obd, struct dentry *dentry, - int *number, struct obdo *oa) + +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; - ENTRY; - rc = fsfilt_precreate_rec(obd, dentry, number, oa); + if ((inode->i_mode & (S_ISUID | S_ISGID)) == (S_ISUID|S_ISGID)) + return 0; - RETURN(rc); + *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: @@ -2271,38 +3458,60 @@ static int filter_precreate_rec(struct obd_device *obd, struct dentry *dentry, * thread 2: gets id (x + 1) from filter_next_id * thread 2: creates object (x + 1) * thread 1: tries to create object x, gets -ENOSPC + * + * 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; - int err = 0, rc = 0, recreate_obj = 0, i; 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; void *handle = NULL; - void *lock = NULL; - __u64 next_id; ENTRY; filter = &obd->u.filter; + 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 { + OBD_ALLOC(osfs, sizeof(*osfs)); + if (osfs == NULL) + RETURN(-ENOMEM); + rc = filter_statfs(obd, osfs, cfs_time_current_64() - HZ, 0); + if (rc == 0 && osfs->os_bavail < (osfs->os_blocks >> 10)) { + CDEBUG(D_RPCTRACE,"%s: not enough space for create " + LPU64"\n", obd->obd_name, osfs->os_bavail << + filter->fo_vfsmnt->mnt_sb->s_blocksize_bits); + *num = 0; + rc = -ENOSPC; + } + OBD_FREE(osfs, sizeof(*osfs)); + if (rc) + RETURN(rc); } - CDEBUG(D_HA, "%s: precreating %d objects\n", obd->obd_name, *num); - - down(&filter->fo_create_locks[group]); + 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 (test_bit(group, &filter->fo_destroys_in_progress)) { - CWARN("%s: precreate aborted by destroy\n", + CWARN("%s: create aborted by destroy\n", obd->obd_name); + rc = -EAGAIN; break; } - + if (recreate_obj) { __u64 last_id; next_id = oa->o_id; @@ -2313,38 +3522,42 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, next_id, last_id); GOTO(cleanup, rc = -EINVAL); } - } else { + } 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, &lock); + dparent = filter_parent_lock(obd, group, next_id); if (IS_ERR(dparent)) GOTO(cleanup, rc = PTR_ERR(dparent)); - cleanup_phase = 1; - - /*only do precreate rec record. so clean kml flags here*/ - fsfilt_clear_fs_flags(obd, dparent->d_inode, SM_DO_REC); + cleanup_phase = 1; /* filter_parent_unlock(dparent) */ - dchild = filter_id2dentry(obd, dparent, group, next_id); + dchild = filter_fid2dentry(obd, dparent, group, next_id); if (IS_ERR(dchild)) GOTO(cleanup, rc = PTR_ERR(dchild)); - cleanup_phase = 2; + 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: Serious error: recreating obj %*s " - "but obj already exists \n", + CERROR("%s: recreating existing object %.*s?\n", obd->obd_name, dchild->d_name.len, dchild->d_name.name); - LBUG(); } else { - CERROR("%s: Serious error: objid %*s already " + /* 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, dchild->d_name.name); @@ -2359,26 +3572,29 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, GOTO(cleanup, rc = PTR_ERR(handle)); cleanup_phase = 3; - rc = ll_vfs_create(dparent->d_inode, dchild, S_IFREG, NULL); + /* 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. */ + rc = ll_vfs_create(dparent->d_inode, dchild, + S_IFREG | S_ISUID | S_ISGID | 0666, NULL); if (rc) { CERROR("create failed rc = %d\n", rc); GOTO(cleanup, rc); } +set_last_id: if (!recreate_obj) { - filter_set_last_id(filter, group, next_id); + filter_set_last_id(filter, next_id, group); err = filter_update_last_objid(obd, group, 0); if (err) CERROR("unable to write lastobjid " "but file created\n"); } - fsfilt_set_fs_flags(obd, dparent->d_inode, SM_DO_REC); - + cleanup: switch(cleanup_phase) { case 3: - err = fsfilt_commit(obd, filter->fo_sb, - dparent->d_inode, handle, 0); + err = fsfilt_commit(obd, dparent->d_inode, handle, 0); if (err) { CERROR("error on commit, err = %d\n", err); if (!rc) @@ -2387,29 +3603,25 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, case 2: f_dput(dchild); case 1: - filter_parent_unlock(dparent, lock); + filter_parent_unlock(dparent); case 0: break; } if (rc) break; + if (time_after(jiffies, enough_time)) { + CDEBUG(D_RPCTRACE, + "%s: precreate slow - want %d got %d \n", + obd->obd_name, *num, i); + break; + } } - *num = i; - /* check if we have an error after ll_vfs_create(). It is possible that - * there will be say -ENOSPC and we will leak it. */ - if (rc == 0) - rc = filter_precreate_rec(obd, dparent, num, oa); - - up(&filter->fo_create_locks[group]); - - CDEBUG(D_HA, "%s: server last_objid for group "LPU64": "LPU64"\n", - obd->obd_name, group, filter->fo_last_objids[group]); - - CDEBUG(D_HA, "%s: filter_precreate() created %d objects\n", - obd->obd_name, i); + CDEBUG(D_RPCTRACE, + "%s: created %d objects for group "LPU64": "LPU64" rc %d\n", + obd->obd_name, i, group, filter->fo_last_objids[group], rc); RETURN(rc); } @@ -2417,46 +3629,34 @@ static int filter_precreate(struct obd_device *obd, struct obdo *oa, 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; - struct filter_export_data *fed; - char str[PTL_NALFMT_SIZE]; - int group = oa->o_gr, rc = 0, diff, recreate_objs = 0; + int rc = 0, diff, group = oa->o_gr; ENTRY; if (!(oa->o_valid & OBD_MD_FLGROUP) || group == 0) { - portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number, - exp->exp_connection->c_peer.peer_id.nid, str); - CERROR("!!! nid "LPX64"/%s sent invalid object group %d\n", - exp->exp_connection->c_peer.peer_id.nid, str, group); + CERROR("!!! nid %s sent invalid object group %d\n", + obd_export_nid2str(exp), group); RETURN(-EINVAL); } - if ((oa->o_valid & OBD_MD_FLFLAGS) && - (oa->o_flags & OBD_FL_RECREATE_OBJS)) { - recreate_objs = 1; - } - obd = exp->exp_obd; fed = &exp->exp_filter_data; filter = &obd->u.filter; - if (fed->fed_group != group && !recreate_objs && - !(oa->o_valid & OBD_MD_REINT)) { - portals_nid2str(exp->exp_connection->c_peer.peer_ni->pni_number, - exp->exp_connection->c_peer.peer_id.nid, str); - CERROR("!!! This export (nid "LPX64"/%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", - exp->exp_connection->c_peer.peer_id.nid, str, - fed->fed_group, group); + 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=%d,od->o_id="LPU64")\n", - group, oa->o_id); + 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) { @@ -2469,26 +3669,21 @@ static int filter_create(struct obd_export *exp, struct obdo *oa, obd = exp->exp_obd; push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - if (oa->o_valid & OBD_MD_REINT) { - int num = *((int*)oa->o_inline); - rc = filter_precreate(obd, oa, oa->o_gr, &num); - } else if (recreate_objs) { - if (oa->o_id > filter_last_id(&obd->u.filter, group)) { + if ((oa->o_valid & OBD_MD_FLFLAGS) && + (oa->o_flags & OBD_FL_RECREATE_OBJS)) { + if (oa->o_id > filter_last_id(filter, oa->o_gr)) { CERROR("recreate objid "LPU64" > last id "LPU64"\n", - oa->o_id, filter_last_id(&obd->u.filter, group)); + oa->o_id, filter_last_id(filter, + oa->o_gr)); rc = -EINVAL; } else { diff = 1; - rc = filter_precreate(obd, oa, group, &diff); + down(&filter->fo_create_locks[oa->o_gr]); + rc = filter_precreate(obd, oa, oa->o_gr, &diff); + up(&filter->fo_create_locks[oa->o_gr]); } } else { - diff = filter_should_precreate(exp, oa, group); - if (diff > 0) { - oa->o_id = filter_last_id(&obd->u.filter, group); - rc = filter_precreate(obd, oa, group, &diff); - oa->o_id += diff; - oa->o_valid = OBD_MD_FLID; - } + rc = filter_handle_precreate(exp, oa, oa->o_gr, oti); } pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); @@ -2500,24 +3695,24 @@ static int filter_create(struct obd_export *exp, struct obdo *oa, * don't go touching that. This needs to be fixed in a * big way. */ lsm->lsm_object_id = oa->o_id; - lsm->lsm_object_gr = oa->o_gr; *ea = lsm; } RETURN(rc); } -static int filter_destroy(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *ea, struct obd_trans_info *oti) +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) { - int rc, rc2, cleanup_phase = 0, have_prepared = 0; + unsigned int qcids[MAXQUOTAS] = {0, 0}; + struct obd_device *obd; + struct filter_obd *filter; struct dentry *dchild = NULL, *dparent = NULL; - struct llog_cookie *fcc = NULL; struct lvfs_run_ctxt saved; - struct filter_obd *filter; - struct obd_device *obd; void *handle = NULL; - void *lock = NULL; + struct llog_cookie *fcc = NULL; + int rc, rc2, cleanup_phase = 0, sync = 0; struct iattr iattr; ENTRY; @@ -2527,44 +3722,35 @@ static int filter_destroy(struct obd_export *exp, struct obdo *oa, filter = &obd->u.filter; push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - - acquire_locks: - dparent = filter_parent_lock(obd, oa->o_gr, oa->o_id, &lock); - if (IS_ERR(dparent)) - GOTO(cleanup, rc = PTR_ERR(dparent)); cleanup_phase = 1; - dchild = filter_id2dentry(obd, dparent, oa->o_gr, oa->o_id); + dchild = filter_fid2dentry(obd, NULL, oa->o_gr, oa->o_id); if (IS_ERR(dchild)) - GOTO(cleanup, rc = -ENOENT); + GOTO(cleanup, rc = PTR_ERR(dchild)); cleanup_phase = 2; if (dchild->d_inode == NULL) { CDEBUG(D_INODE, "destroying non-existent object "LPU64"\n", oa->o_id); + /* 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); + 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); } - if (!have_prepared) { - /* If we're really going to destroy the object, get ready - * by getting the clients to discard their cached data. - * - * We have to drop the parent lock, because - * filter_prepare_destroy will acquire a PW on the object, and - * we don't want to deadlock with an incoming write to the - * object, which has the extent PW and then wants to get the - * parent dentry to do the lookup. - * - * We dput the child because it's not worth the extra - * complication of condition the above code to skip it on the - * second time through. */ - f_dput(dchild); - filter_parent_unlock(dparent, lock); - - filter_prepare_destroy(obd, oa->o_id, oa->o_gr); - have_prepared = 1; - goto acquire_locks; - } + 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) { @@ -2572,69 +3758,92 @@ static int filter_destroy(struct obd_export *exp, struct obdo *oa, if (fcc != NULL) memcpy(fcc, obdo_logcookie(oa), sizeof(*fcc)); } + DQUOT_INIT(dchild->d_inode); - /* we're gonna truncate it first in order to avoid possible - * deadlock: + /* we're gonna truncate it first in order to avoid possible deadlock: * P1 P2 * open trasaction open transaction * down(i_zombie) down(i_zombie) * restart transaction * (see BUG 4180) -bzzz */ - down(&dchild->d_inode->i_sem); - handle = fsfilt_start_log(obd, dparent->d_inode,FSFILT_OP_SETATTR,NULL,1); + LOCK_INODE_MUTEX(dchild->d_inode); + handle = fsfilt_start_log(obd, dchild->d_inode, FSFILT_OP_SETATTR, + NULL, 1); if (IS_ERR(handle)) { - up(&dchild->d_inode->i_sem); + UNLOCK_INODE_MUTEX(dchild->d_inode); GOTO(cleanup, rc = PTR_ERR(handle)); } iattr.ia_valid = ATTR_SIZE; iattr.ia_size = 0; rc = fsfilt_setattr(obd, dchild, handle, &iattr, 1); - - rc2 = fsfilt_commit(obd, filter->fo_sb, dparent->d_inode, handle, 0); - up(&dchild->d_inode->i_sem); + rc2 = fsfilt_commit(obd, dchild->d_inode, handle, 0); + UNLOCK_INODE_MUTEX(dchild->d_inode); if (rc) GOTO(cleanup, rc); if (rc2) GOTO(cleanup, rc = rc2); + /* We don't actually need to lock the parent until we are unlinking + * here, and not while truncating above. That avoids holding the + * parent lock for a long time during truncate, which can block other + * threads from doing anything to objects in that directory. bug 7171 */ + dparent = filter_parent_lock(obd, oa->o_gr, oa->o_id); + if (IS_ERR(dparent)) + GOTO(cleanup, rc = PTR_ERR(dparent)); + cleanup_phase = 3; /* filter_parent_unlock */ + + LOCK_INODE_MUTEX(dchild->d_inode); handle = fsfilt_start_log(obd, dparent->d_inode,FSFILT_OP_UNLINK,oti,1); - if (IS_ERR(handle)) + if (IS_ERR(handle)) { + UNLOCK_INODE_MUTEX(dchild->d_inode); GOTO(cleanup, rc = PTR_ERR(handle)); + } + cleanup_phase = 4; /* fsfilt_commit */ - cleanup_phase = 3; + /* Quota release need uid/gid of inode */ + obdo_from_inode(oa, dchild->d_inode, OBD_MD_FLUID|OBD_MD_FLGID); - rc = filter_destroy_internal(obd, oa->o_id, dparent, dchild); + filter_fmd_drop(exp, oa->o_id, oa->o_gr); + /* this drops dchild->d_inode->i_mutex unconditionally */ + rc = filter_destroy_internal(obd, oa->o_id, oa->o_gr, dparent, dchild); + + EXIT; cleanup: switch(cleanup_phase) { - case 3: - if (fcc != NULL) { - if (oti != NULL) - fsfilt_add_journal_cb(obd, filter->fo_sb, 0, - oti->oti_handle, - filter_cancel_cookies_cb, - fcc); - else - fsfilt_add_journal_cb(obd, filter->fo_sb, 0, - handle, - filter_cancel_cookies_cb, - fcc); - } - rc = filter_finish_transno(exp, oti, rc); - rc2 = fsfilt_commit(obd, filter->fo_sb, dparent->d_inode, - handle, 0); + case 4: + 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); case 2: f_dput(dchild); + if (fcc != NULL) + OBD_FREE(fcc, sizeof(*fcc)); case 1: - filter_parent_unlock(dparent, lock); - case 0: pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); break; default: @@ -2642,57 +3851,79 @@ cleanup: LBUG(); } - RETURN(rc); + /* trigger quota release */ + qcids[USRQUOTA] = oa->o_uid; + 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); + return rc; } /* NB start and end are used for punch, but not truncate */ -static int filter_truncate(struct obd_export *exp, struct obdo *oa, - struct lov_stripe_md *lsm, - obd_off start, obd_off end, - struct obd_trans_info *oti) +static int filter_truncate(struct obd_export *exp, struct obd_info *oinfo, + struct obd_trans_info *oti, + struct ptlrpc_request_set *rqset) { - int error; + int rc; ENTRY; - if (end != OBD_OBJECT_EOF) + if (oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) { CERROR("PUNCH not supported, only truncate: end = "LPX64"\n", - end); + oinfo->oi_policy.l_extent.end); + RETURN(-EFAULT); + } + + CDEBUG(D_INODE, "calling truncate for object "LPU64", valid = "LPX64 + ", o_size = "LPD64"\n", oinfo->oi_oa->o_id, + oinfo->oi_oa->o_valid, oinfo->oi_policy.l_extent.start); + + rc = filter_auth_capa(exp, NULL, oinfo_mdsno(oinfo), + oinfo_capa(oinfo), CAPA_OPC_OSS_TRUNC); + if (rc) + RETURN(rc); - CDEBUG(D_INODE, "calling truncate for object "LPU64", valid = "LPU64", " - "o_size = "LPD64"\n", oa->o_id, oa->o_valid, start); - oa->o_size = start; - error = filter_setattr(exp, oa, NULL, oti); - RETURN(error); + 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 obd_device *obd = exp->exp_obd; struct lvfs_run_ctxt saved; struct filter_obd *filter; struct dentry *dentry; int rc, rc2; ENTRY; - filter = &obd->u.filter; + 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(obd, filter->fo_sb); + rc = fsfilt_sync(exp->exp_obd, filter->fo_obt.obt_sb); /* flush any remaining cancel messages out to the target */ - filter_sync_llogs(obd, NULL); + filter_sync_llogs(exp->exp_obd, exp); RETURN(rc); } - dentry = filter_oa2dentry(obd, oa); + dentry = filter_oa2dentry(exp->exp_obd, oa); if (IS_ERR(dentry)) RETURN(PTR_ERR(dentry)); push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL); - down(&dentry->d_inode->i_sem); - rc = filemap_fdatasync(dentry->d_inode->i_mapping); + LOCK_INODE_MUTEX(dentry->d_inode); + + rc = filemap_fdatawrite(dentry->d_inode->i_mapping); if (rc == 0) { /* just any file to grab fsync method - "file" arg unused */ struct file *file = filter->fo_rcvd_filp; @@ -2704,7 +3935,7 @@ static int filter_sync(struct obd_export *exp, struct obdo *oa, if (!rc) rc = rc2; } - up(&dentry->d_inode->i_sem); + UNLOCK_INODE_MUTEX(dentry->d_inode); oa->o_valid = OBD_MD_FLID; obdo_from_inode(oa, dentry->d_inode, FILTER_VALID_FLAGS); @@ -2715,84 +3946,50 @@ static int filter_sync(struct obd_export *exp, struct obdo *oa, RETURN(rc); } -static int filter_statfs(struct obd_device *obd, struct obd_statfs *osfs, - unsigned long max_age) -{ - struct filter_obd *filter = &obd->u.filter; - int blockbits = filter->fo_sb->s_blocksize_bits; - int rc; - ENTRY; - - /* at least try to account for cached pages. its still racey and - * might be under-reporting if clients haven't announced their - * caches with brw recently */ - spin_lock(&obd->obd_osfs_lock); - rc = fsfilt_statfs(obd, filter->fo_sb, max_age); - memcpy(osfs, &obd->obd_osfs, sizeof(*osfs)); - spin_unlock(&obd->obd_osfs_lock); - - CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64 - " pending "LPU64" free "LPU64" avail "LPU64"\n", - filter->fo_tot_dirty, filter->fo_tot_granted, - filter->fo_tot_pending, - osfs->os_bfree << blockbits, osfs->os_bavail << blockbits); - - filter_grant_sanity_check(obd, __FUNCTION__); - - osfs->os_bavail -= min(osfs->os_bavail, - (filter->fo_tot_dirty + filter->fo_tot_pending + - osfs->os_bsize -1) >> blockbits); - - RETURN(rc); -} - static int filter_get_info(struct obd_export *exp, __u32 keylen, void *key, __u32 *vallen, void *val) { - struct filter_export_data *fed = &exp->exp_filter_data; struct obd_device *obd; ENTRY; obd = class_exp2obd(exp); if (obd == NULL) { - CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n", - exp->exp_handle.h_cookie); + CDEBUG(D_IOCTL, "invalid client export %p\n", exp); 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.filter.fo_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.filter.fo_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; - *last_id = filter_last_id(&obd->u.filter, fed->fed_group); - RETURN(0); - } - if (keylen >= strlen("reint_log") && memcmp(key, "reint_log", 9) == 0) { - /*Get log_context handle*/ - unsigned long *llh_handle = val; - *vallen = sizeof(unsigned long); - *llh_handle = (unsigned long)obd->obd_llog_ctxt[LLOG_REINT_ORIG_CTXT]; - RETURN(0); - } - if (keylen >= strlen("cache_sb") && memcmp(key, "cache_sb", 8) == 0) { - /*Get log_context handle*/ - unsigned long *sb = val; - *vallen = sizeof(unsigned long); - *sb = (unsigned long)obd->u.filter.fo_sb; + /* FIXME: object groups */ + 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); } @@ -2800,72 +3997,60 @@ static int filter_get_info(struct obd_export *exp, __u32 keylen, RETURN(-EINVAL); } -struct obd_llogs *filter_grab_llog_for_group(struct obd_device *obd, int group, - struct obd_export *export) +static int filter_set_info_async(struct obd_export *exp, __u32 keylen, + void *key, __u32 vallen, void *val, + struct ptlrpc_request_set *set) { - struct filter_group_llog *fglog, *nlog; - char name[32] = "CATLIST"; - struct filter_obd *filter; + struct obd_device *obd; + struct obd_llog_group *olg; struct llog_ctxt *ctxt; - struct list_head *cur; - int rc; + int rc = 0, group; + ENTRY; - filter = &obd->u.filter; + obd = exp->exp_obd; + if (obd == NULL) { + CDEBUG(D_IOCTL, "invalid export %p\n", exp); + RETURN(-EINVAL); + } - spin_lock(&filter->fo_llog_list_lock); - list_for_each(cur, &filter->fo_llog_list) { - fglog = list_entry(cur, struct filter_group_llog, list); - if (fglog->group == group) { - if (!(fglog->exp == NULL || fglog->exp == export || export == NULL)) - CWARN("%s: export for group %d changes: 0x%p -> 0x%p\n", - obd->obd_name, group, fglog->exp, export); - spin_unlock(&filter->fo_llog_list_lock); - goto init; - } + 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); } - spin_unlock(&filter->fo_llog_list_lock); - if (export == NULL) - RETURN(NULL); + if (!KEY_IS(KEY_MDS_CONN)) + RETURN(-EINVAL); - OBD_ALLOC(fglog, sizeof(*fglog)); - if (fglog == NULL) - RETURN(NULL); - fglog->group = group; + LCONSOLE_WARN("%s: received MDS connection from %s\n", obd->obd_name, + obd_export_nid2str(exp)); + obd->u.filter.fo_mdc_conn.cookie = exp->exp_handle.h_cookie; - OBD_ALLOC(fglog->llogs, sizeof(struct obd_llogs)); - if (fglog->llogs == NULL) { - OBD_FREE(fglog, sizeof(*fglog)); - RETURN(NULL); - } + /* setup llog imports */ + LASSERT(val != NULL); + group = (int)(*(__u32 *)val); + LASSERT(group >= FILTER_GROUP_MDS0); - spin_lock(&filter->fo_llog_list_lock); - list_for_each(cur, &filter->fo_llog_list) { - nlog = list_entry(cur, struct filter_group_llog, list); - LASSERT(nlog->group != group); - } - list_add(&fglog->list, &filter->fo_llog_list); - spin_unlock(&filter->fo_llog_list_lock); + olg = filter_find_olg(obd, group); + if (IS_ERR(olg)) + RETURN(PTR_ERR(olg)); + llog_group_set_export(olg, exp); - rc = obd_llog_cat_initialize(obd, fglog->llogs, 1, name); - if (rc) { - OBD_FREE(fglog->llogs, sizeof(*(fglog->llogs))); - OBD_FREE(fglog, sizeof(*fglog)); - RETURN(NULL); - } + ctxt = llog_group_get_ctxt(olg, LLOG_MDS_OST_REPL_CTXT); + LASSERTF(ctxt != NULL, "ctxt is null\n"), -init: - if (export) { - fglog->exp = export; - ctxt = llog_get_context(fglog->llogs, LLOG_UNLINK_REPL_CTXT); - LASSERT(ctxt != NULL); + rc = llog_receptor_accept(ctxt, exp->exp_imp_reverse); + llog_ctxt_put(ctxt); - llog_receptor_accept(ctxt, export->exp_imp_reverse); - } - CDEBUG(D_OTHER, "%s: new llog 0x%p for group %u\n", - obd->obd_name, fglog->llogs, group); + lquota_setinfo(filter_quota_interface_ref, exp, obd); - RETURN(fglog->llogs); + RETURN(rc); } int filter_iocontrol(unsigned int cmd, struct obd_export *exp, @@ -2876,23 +4061,34 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, int rc = 0; switch (cmd) { - case OBD_IOC_ABORT_RECOVERY: + case OBD_IOC_ABORT_RECOVERY: { + CERROR("aborting recovery for device %s\n", obd->obd_name); target_stop_recovery_thread(obd); RETURN(0); + } + + case OBD_IOC_SYNC: { + CDEBUG(D_RPCTRACE, "syncing ost %s\n", obd->obd_name); + rc = fsfilt_sync(obd, obd->u.obt.obt_sb); + RETURN(rc); + } case OBD_IOC_SET_READONLY: { void *handle; - struct super_block *sb = obd->u.filter.fo_sb; + struct super_block *sb = obd->u.obt.obt_sb; struct inode *inode = sb->s_root->d_inode; BDEVNAME_DECLARE_STORAGE(tmp); - CERROR("setting device %s read-only\n", + CERROR("*** setting device %s read-only ***\n", ll_bdevname(sb, tmp)); handle = fsfilt_start(obd, inode, FSFILT_OP_MKNOD, NULL); - LASSERT(handle); - (void)fsfilt_commit(obd, sb, inode, handle, 1); + if (!IS_ERR(handle)) + rc = fsfilt_commit(obd, inode, handle, 1); - dev_set_rdonly(ll_sbdev(obd->u.filter.fo_sb), 2); + CDEBUG(D_HA, "syncing ost %s\n", obd->obd_name); + rc = fsfilt_sync(obd, obd->u.obt.obt_sb); + + lvfs_set_rdonly(obd, obd->u.obt.obt_sb); RETURN(0); } @@ -2900,6 +4096,7 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, rc = llog_catalog_list(obd, 1, data); RETURN(rc); } + case OBD_IOC_LLOG_CANCEL: case OBD_IOC_LLOG_REMOVE: case OBD_IOC_LLOG_INFO: @@ -2909,9 +4106,9 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, /* struct llog_ctxt *ctxt = NULL; - push_ctxt(&saved, &ctxt->loc_ctxt, NULL); + push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL); rc = llog_ioctl(ctxt, cmd, data); - pop_ctxt(&saved, &ctxt->loc_ctxt, NULL); + pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL); RETURN(rc); */ @@ -2924,130 +4121,97 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp, RETURN(0); } -static struct llog_operations filter_unlink_repl_logops; -static struct llog_operations filter_size_orig_logops = { - lop_setup: llog_obd_origin_setup, - lop_cleanup: llog_catalog_cleanup, - lop_add: llog_catalog_add, -}; - -static int filter_llog_init(struct obd_device *obd, struct obd_llogs *llogs, - struct obd_device *tgt, int count, - struct llog_catid *catid) +static int filter_health_check(struct obd_device *obd) { - struct llog_ctxt *ctxt; - int rc; - ENTRY; +#ifdef USE_HEALTH_CHECK_WRITE + struct filter_obd *filter = &obd->u.filter; +#endif + int rc = 0; - filter_unlink_repl_logops = llog_client_ops; - filter_unlink_repl_logops.lop_cancel = llog_obd_repl_cancel; - filter_unlink_repl_logops.lop_connect = llog_repl_connect; - filter_unlink_repl_logops.lop_sync = llog_obd_repl_sync; + /* + * health_check to return 0 on healthy + * and 1 on unhealthy. + */ + if (obd->u.obt.obt_sb->s_flags & MS_RDONLY) + rc = 1; - rc = obd_llog_setup(obd, llogs, LLOG_UNLINK_REPL_CTXT, tgt, 0, NULL, - &filter_unlink_repl_logops); - if (rc) - RETURN(rc); - /* FIXME - assign unlink_cb for filter's recovery */ - ctxt = llog_get_context(llogs, LLOG_UNLINK_REPL_CTXT); - ctxt->llog_proc_cb = filter_recov_log_unlink_cb; +#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; +} - /* FIXME - count should be 1 to setup size log */ - rc = obd_llog_setup(obd, llogs, LLOG_SIZE_ORIG_CTXT, tgt, 0, - &catid->lci_logid, &filter_size_orig_logops); - RETURN(rc); +static struct dentry *filter_lvfs_fid2dentry(__u64 id, __u32 gen, __u64 gr, + void *data) +{ + return filter_fid2dentry(data, NULL, gr, id); } -static int filter_llog_finish(struct obd_device *obd, - struct obd_llogs *llogs, int count) +static int filter_process_config(struct obd_device *obd, obd_count len, + void *buf) { - int rc; - ENTRY; + struct lustre_cfg *lcfg = buf; + struct lprocfs_static_vars lvars; + int rc = 0; - rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_UNLINK_REPL_CTXT)); - if (rc) - RETURN(rc); + 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; - rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_SIZE_ORIG_CTXT)); - RETURN(rc); -} + log = sptlrpc_conf_log_extract(lcfg); + if (IS_ERR(log)) { + rc = PTR_ERR(log); + break; + } -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_llogs *llog; - int rc; - ENTRY; + sptlrpc_rule_set_init(&tmp_rset); - 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); - llog = filter_grab_llog_for_group(obd, body->lgdc_logid.lgl_ogr, exp); - LASSERT(llog != NULL); - ctxt = llog_get_context(llog, body->lgdc_ctxt_idx); - rc = llog_connect(ctxt, 1, &body->lgdc_logid, - &body->lgdc_gen, NULL); - if (rc != 0) - CERROR("failed to connect\n"); + 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; + } - RETURN(rc); -} + 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); -static struct dentry *filter_lvfs_id2dentry(__u64 id, __u32 gen, - __u64 gr, void *data) -{ - return filter_id2dentry(data, NULL, gr, id); + 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 = { - l_id2dentry: filter_lvfs_id2dentry, + l_fid2dentry: filter_lvfs_fid2dentry, }; static struct obd_ops filter_obd_ops = { .o_owner = THIS_MODULE, - .o_attach = filter_attach, - .o_detach = filter_detach, .o_get_info = filter_get_info, + .o_set_info_async = filter_set_info_async, .o_setup = filter_setup, .o_precleanup = filter_precleanup, .o_cleanup = filter_cleanup, .o_connect = filter_connect, - .o_connect_post = filter_connect_post, + .o_reconnect = filter_reconnect, .o_disconnect = filter_disconnect, - .o_statfs = filter_statfs, - .o_getattr = filter_getattr, - .o_unpackmd = filter_unpackmd, - .o_create = filter_create, - .o_setattr = filter_setattr, - .o_destroy = filter_destroy, - .o_brw = filter_brw, - .o_punch = filter_truncate, - .o_sync = filter_sync, - .o_preprw = filter_preprw, - .o_commitrw = filter_commitrw, - .o_do_cow = filter_do_cow, - .o_write_extents = filter_write_extents, + .o_ping = filter_ping, + .o_init_export = filter_init_export, .o_destroy_export = filter_destroy_export, - .o_llog_init = filter_llog_init, - .o_llog_finish = filter_llog_finish, - .o_llog_connect = filter_llog_connect, - .o_iocontrol = filter_iocontrol, -}; - -static struct obd_ops filter_sanobd_ops = { - .o_owner = THIS_MODULE, - .o_attach = filter_attach, - .o_detach = filter_detach, - .o_get_info = filter_get_info, - .o_setup = filter_san_setup, - .o_precleanup = filter_precleanup, - .o_cleanup = filter_cleanup, - .o_connect = filter_connect, - .o_connect_post = filter_connect_post, - .o_disconnect = filter_disconnect, .o_statfs = filter_statfs, .o_getattr = filter_getattr, .o_unpackmd = filter_unpackmd, @@ -3059,41 +4223,75 @@ static struct obd_ops filter_sanobd_ops = { .o_sync = filter_sync, .o_preprw = filter_preprw, .o_commitrw = filter_commitrw, - .o_do_cow = filter_do_cow, - .o_write_extents = filter_write_extents, - .o_san_preprw = filter_san_preprw, - .o_destroy_export = filter_destroy_export, .o_llog_init = filter_llog_init, - .o_llog_finish = filter_llog_finish, .o_llog_connect = filter_llog_connect, + .o_llog_finish = filter_llog_finish, .o_iocontrol = filter_iocontrol, + .o_health_check = filter_health_check, + .o_process_config = filter_process_config, }; +quota_interface_t *filter_quota_interface_ref; +extern quota_interface_t filter_quota_interface; + static int __init obdfilter_init(void) { struct lprocfs_static_vars lvars; int rc; - printk(KERN_INFO "Lustre: Filtering OBD driver; info@clusterfs.com\n"); + lprocfs_filter_init_vars(&lvars); + + request_module("lquota"); + OBD_ALLOC(obdfilter_created_scratchpad, + OBDFILTER_CREATED_SCRATCHPAD_ENTRIES * + sizeof(*obdfilter_created_scratchpad)); + if (obdfilter_created_scratchpad == NULL) + return -ENOMEM; - lprocfs_init_vars(filter, &lvars); + 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, NULL, lvars.module_vars, - OBD_FILTER_DEVICENAME); - if (rc) - return rc; + LUSTRE_OST_NAME, NULL); + if (rc) { + int err; + + 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: + if (filter_quota_interface_ref) + PORTAL_SYMBOL_PUT(filter_quota_interface); + + OBD_FREE(obdfilter_created_scratchpad, + OBDFILTER_CREATED_SCRATCHPAD_ENTRIES * + sizeof(*obdfilter_created_scratchpad)); + } - rc = class_register_type(&filter_sanobd_ops, NULL, lvars.module_vars, - OBD_FILTER_SAN_DEVICENAME); - if (rc) - class_unregister_type(OBD_FILTER_DEVICENAME); return rc; } static void __exit obdfilter_exit(void) { - class_unregister_type(OBD_FILTER_SAN_DEVICENAME); - class_unregister_type(OBD_FILTER_DEVICENAME); + if (filter_quota_interface_ref) + PORTAL_SYMBOL_PUT(filter_quota_interface); + + if (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; + } + + class_unregister_type(LUSTRE_OST_NAME); + OBD_FREE(obdfilter_created_scratchpad, + OBDFILTER_CREATED_SCRATCHPAD_ENTRIES * + sizeof(*obdfilter_created_scratchpad)); } MODULE_AUTHOR("Cluster File Systems, Inc. ");