filter.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,698 行 · 第 1/5 页

                RETURN(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, off, fed->fed_fcd->fcd_uuid);

        LASSERT(filter->fo_last_rcvd_slots != NULL);

        off = fed->fed_lr_off;

        /* 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.*/
        if (!test_bit(fed->fed_lr_idx, filter->fo_last_rcvd_slots)) {
                CERROR("FILTER client %u: bit already clear in bitmap!!\n",
                       fed->fed_lr_idx);
                LBUG();
        }

        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 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",
                       fed->fed_lr_idx);
                LBUG();
        }

        EXIT;
free:
        OBD_FREE(fed->fed_fcd, sizeof(*fed->fed_fcd));
        fed->fed_fcd = NULL;

        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);
        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, 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 lr_server_data *fsd, int force_sync)
{
        loff_t off = 0;
        int rc;
        ENTRY;

        CDEBUG(D_INODE, "server uuid      : %s\n", fsd->lsd_uuid);
        CDEBUG(D_INODE, "server last_rcvd : "LPU64"\n",
               le64_to_cpu(fsd->lsd_last_transno));
        CDEBUG(D_INODE, "server last_mount: "LPU64"\n",
               le64_to_cpu(fsd->lsd_mount_count));

        fsd->lsd_compat14 = fsd->lsd_last_transno;
        rc = fsfilt_write_record(obd, filp, fsd, sizeof(*fsd), &off,force_sync);
        if (rc)
                CERROR("error writing lr_server_data: rc = %d\n", rc);

        RETURN(rc);
}

int filter_update_last_objid(struct obd_device *obd, obd_gr group,
                             int force_sync)
{
        struct filter_obd *filter = &obd->u.filter;
        __u64 tmp;
        loff_t off = 0;
        int rc;
        ENTRY;

        if (filter->fo_last_objid_files[group] == NULL) {
                CERROR("Object group "LPU64" not fully setup; not updating "
                       "last_objid\n", group);
                RETURN(-EINVAL);
        }

        CDEBUG(D_INODE, "%s: server last_objid for group "LPU64": "LPU64"\n",
               obd->obd_name, group, filter->fo_last_objids[group]);

        tmp = cpu_to_le64(filter->fo_last_objids[group]);
        rc = fsfilt_write_record(obd, filter->fo_last_objid_files[group],
                                 &tmp, sizeof(tmp), &off, force_sync);
        if (rc)
                CERROR("error writing group "LPU64" last objid: rc = %d\n",
                       group, rc);
        RETURN(rc);
}

/* 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 lr_server_data *fsd;
        struct filter_client_data *fcd = NULL;
        struct inode *inode = filp->f_dentry->d_inode;
        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 */
        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, LR_MAX_CLIENTS / 8);
        if (filter->fo_last_rcvd_slots == NULL) {
                OBD_FREE(fsd, sizeof(*fsd));
                RETURN(-ENOMEM);
        }

        if (last_rcvd_size == 0) {
                LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name);

                memcpy(fsd->lsd_uuid, obd->obd_uuid.uuid,sizeof(fsd->lsd_uuid));
                fsd->lsd_last_transno = 0;