filter.c

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

{
        struct lvfs_run_ctxt saved;
        struct filter_obd *filter = &obd->u.filter;
        struct file *file;
        struct inode *inode;
        int rc = 0;
        ENTRY;

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        file = filp_open(LAST_RCVD, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
        if (!file || IS_ERR(file)) {
                rc = PTR_ERR(file);
                CERROR("OBD filter: cannot open/create %s: rc = %d\n",
                       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);
        }

        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);
        }
        /* 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_filp, 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:
        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);

        return(rc);

 err_server_data:
        //class_disconnect_exports(obd, 0);
        filter_free_server_data(filter);
 err_health_check:
        if (filp_close(filter->fo_health_check_filp, 0))
                CERROR("can't close %s after error\n", HEALTH_CHECK);
        filter->fo_health_check_filp = NULL;
 err_filp:
        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;
}

/* cleanup the filter: write last used object id to status file */
static void filter_post(struct obd_device *obd)
{
        struct lvfs_run_ctxt saved;
        struct filter_obd *filter = &obd->u.filter;
        int rc, i;

        /* XXX: filter_update_lastobjid used to call fsync_dev.  It might be
         * best to start a transaction with h_sync, because we removed this
         * from lastobjid */

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        rc = filter_update_server_data(obd, filter->fo_rcvd_filp,
                                       filter->fo_fsd, 0);
        if (rc)
                CERROR("error writing server data: rc = %d\n", rc);

        for (i = 0; i < FILTER_GROUPS; i++) {
                rc = filter_update_last_objid(obd, i, (i == FILTER_GROUPS - 1));
                if (rc)
                        CERROR("error writing group %d lastobjid: rc = %d\n",
                               i, rc);
        }

        rc = filp_close(filter->fo_rcvd_filp, 0);
        filter->fo_rcvd_filp = NULL;
        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);
}

static void filter_set_last_id(struct filter_obd *filter,
                               obd_id id, obd_gr group)
{
        LASSERT(filter->fo_fsd != NULL);
        LASSERT(group <= FILTER_GROUPS);

        spin_lock(&filter->fo_objidlock);
        filter->fo_last_objids[group] = id;
        spin_unlock(&filter->fo_objidlock);
}

obd_id filter_last_id(struct filter_obd *filter, obd_gr group)
{
        obd_id id;
        LASSERT(filter->fo_fsd != NULL);
        LASSERT(group <= FILTER_GROUPS);

        /* FIXME: object groups */
        spin_lock(&filter->fo_objidlock);
        id = filter->fo_last_objids[group];
        spin_unlock(&filter->fo_objidlock);

        return id;
}

static int filter_lock_dentry(struct obd_device *obd, struct dentry *dparent)
{
        LOCK_INODE_MUTEX(dparent->d_inode);
        return 0;
}

/* 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_GROUPS); /* FIXME: object groups */

        if (group > 0 || filter->fo_subdir_count == 0)
                return filter->fo_dentry_O_groups[group];

        return filter->fo_dentry_O_sub[objid & (filter->fo_subdir_count - 1)];
}

/* 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)
{
        unsigned long now = jiffies;
        struct dentry *dparent = filter_parent(obd, group, objid);
        int rc;

        if (IS_ERR(dparent))
                return dparent;

        rc = filter_lock_dentry(obd, dparent);
        fsfilt_check_slow(obd, now, "parent lock");
        return rc ? ERR_PTR(rc) : 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.
 *
 * If dir_dentry is passed, the caller has already locked the parent
 * appropriately for this operation (normally a write lock).  If
 * 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_fid2dentry(struct obd_device *obd,
                                 struct dentry *dir_dentry,
                                 obd_gr group, obd_id id)
{
        struct dentry *dparent = dir_dentry;
        struct dentry *dchild;
        char name[32];
        int len;
        ENTRY;

        if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT) &&
            !obd->u.filter.fo_destroy_in_progress) {
                /* 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));
        }

        len = sprintf(name, LPU64, id);
        if (dir_dentry == NULL) {
                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",
               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);
        if (IS_ERR(dchild)) {
                CERROR("%s: child lookup error %ld\n", obd->obd_name,
                       PTR_ERR(dchild));
                RETURN(dchild);
        }

        if (dchild->d_inode != NULL && is_bad_inode(dchild->d_inode)) {
                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));
        }

        CDEBUG(D_INODE, "got child objid %s: %p, count = %d\n",
               name, dchild, atomic_read(&dchild->d_count));

        LASSERT(atomic_read(&dchild->d_count) > 0);

        RETURN(dchild);
}

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 } };
        ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };

        ENTRY;
        /* Tell the clients that the object is gone now and that they should
         * throw away any cached pages. */
        rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id, LDLM_EXTENT,
                                    &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)
                ldlm_lock_decref(&lockh, LCK_PW);

        RETURN(rc);
}

/* 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)
{
        int rc;
        ENTRY;

        /* 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);


        /* 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)
                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);

#if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
        rc = security_inode_unlink(dir, dentry);
        if (rc)
                GOTO(out, rc);
#endif

        rc = dir->i_op->unlink(dir, dentry);
out:
        /* 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);