#include <linux/obd_lmv.h>
#include "lmv_internal.h"
-LIST_HEAD(lmv_obj_list);
-spinlock_t lmv_obj_list_lock = SPIN_LOCK_UNLOCKED;
+static LIST_HEAD(lmv_obj_list);
+static spinlock_t lmv_obj_list_lock = SPIN_LOCK_UNLOCKED;
/* creates new obj on passed @fid and @mea. */
-static struct lmv_obj *
-__lmv_alloc_obj(struct obd_device *obd, struct ll_fid *fid,
- struct mea *mea)
+struct lmv_obj *
+lmv_alloc_obj(struct obd_device *obd, struct ll_fid *fid,
+ struct mea *mea)
{
int i;
struct lmv_obj *obj;
unsigned int obj_size;
struct lmv_obd *lmv = &obd->u.lmv;
+ LASSERT(mea->mea_magic == MEA_MAGIC_LAST_CHAR
+ || mea->mea_magic == MEA_MAGIC_ALL_CHARS);
+
OBD_ALLOC(obj, sizeof(*obj));
if (!obj)
return NULL;
obj->obd = obd;
+ obj->state = 0;
obj->fid = *fid;
+ obj->hashtype = mea->mea_magic;
+ init_MUTEX(&obj->guard);
atomic_set(&obj->count, 0);
obj->objcount = mea->mea_count;
}
/* destroys passed @obj. */
-static void
-__lmv_free_obj(struct lmv_obj *obj)
+void
+lmv_free_obj(struct lmv_obj *obj)
{
unsigned int obj_size;
struct lmv_obd *lmv = &obj->obd->u.lmv;
OBD_FREE(obj, sizeof(*obj));
}
-struct lmv_obj *
-lmv_get_obj(struct lmv_obj *obj)
+static void
+__add_obj(struct lmv_obj *obj)
+{
+ atomic_inc(&obj->count);
+ list_add(&obj->list, &lmv_obj_list);
+}
+
+void
+lmv_add_obj(struct lmv_obj *obj)
+{
+ spin_lock(&lmv_obj_list_lock);
+ __add_obj(obj);
+ spin_unlock(&lmv_obj_list_lock);
+}
+
+static void
+__del_obj(struct lmv_obj *obj)
+{
+ if (!(obj->state & O_FREEING))
+ LBUG();
+
+ list_del(&obj->list);
+ lmv_free_obj(obj);
+}
+
+void
+lmv_del_obj(struct lmv_obj *obj)
+{
+ spin_lock(&lmv_obj_list_lock);
+ __del_obj(obj);
+ spin_unlock(&lmv_obj_list_lock);
+}
+
+static struct lmv_obj *
+__get_obj(struct lmv_obj *obj)
{
LASSERT(obj);
atomic_inc(&obj->count);
return obj;
}
-void
-lmv_put_obj(struct lmv_obj *obj)
+struct lmv_obj *
+lmv_get_obj(struct lmv_obj *obj)
+{
+ spin_lock(&lmv_obj_list_lock);
+ __get_obj(obj);
+ spin_unlock(&lmv_obj_list_lock);
+
+ return obj;
+}
+
+static void
+__put_obj(struct lmv_obj *obj)
{
LASSERT(obj);
CDEBUG(D_OTHER, "last reference to %lu/%lu/%lu - destroying\n",
(unsigned long)fid->mds, (unsigned long)fid->id,
(unsigned long)fid->generation);
- __lmv_free_obj(obj);
+ __del_obj(obj);
}
}
+void
+lmv_put_obj(struct lmv_obj *obj)
+{
+ spin_lock(&lmv_obj_list_lock);
+ __put_obj(obj);
+ spin_unlock(&lmv_obj_list_lock);
+}
+
static struct lmv_obj *
-__lmv_grab_obj(struct obd_device *obd, struct ll_fid *fid)
+__grab_obj(struct obd_device *obd, struct ll_fid *fid)
{
struct lmv_obj *obj;
struct list_head *cur;
list_for_each(cur, &lmv_obj_list) {
obj = list_entry(cur, struct lmv_obj, list);
+
+ /* check if object is in progress of destroying. If so - skip
+ * it. */
+ if (obj->state & O_FREEING)
+ continue;
+
+ /* check if this is waht we're looking for. */
if (fid_equal(&obj->fid, fid))
- return lmv_get_obj(obj);
+ return __get_obj(obj);
}
+
return NULL;
}
ENTRY;
spin_lock(&lmv_obj_list_lock);
- obj = __lmv_grab_obj(obd, fid);
+ obj = __grab_obj(obd, fid);
spin_unlock(&lmv_obj_list_lock);
RETURN(obj);
}
/* looks in objects list for an object that matches passed @fid. If it is not
- * found -- creates it using passed @mea and puts to list. */
+ * found -- creates it using passed @mea and puts onto list. */
static struct lmv_obj *
-__lmv_create_obj(struct obd_device *obd, struct ll_fid *fid,
- struct mea *mea)
+__create_obj(struct obd_device *obd, struct ll_fid *fid, struct mea *mea)
{
- struct lmv_obj *obj, *cobj;
+ struct lmv_obj *new, *obj;
ENTRY;
obj = lmv_grab_obj(obd, fid);
if (obj)
RETURN(obj);
- /* no such object yet, allocate and initialize them. */
- obj = __lmv_alloc_obj(obd, fid, mea);
- if (!obj)
+ /* no such object yet, allocate and initialize it. */
+ new = lmv_alloc_obj(obd, fid, mea);
+ if (!new)
RETURN(NULL);
/* check if someone create it already while we were dealing with
* allocating @obj. */
spin_lock(&lmv_obj_list_lock);
- cobj = __lmv_grab_obj(obd, fid);
- if (cobj) {
+ obj = __grab_obj(obd, fid);
+ if (obj) {
/* someone created it already - put @obj and getting out. */
- __lmv_free_obj(obj);
+ lmv_free_obj(new);
spin_unlock(&lmv_obj_list_lock);
- RETURN(cobj);
+ RETURN(obj);
}
- /* object is referenced by list and thus should have additional
- * reference counted. */
- lmv_get_obj(obj);
- list_add(&obj->list, &lmv_obj_list);
+ __add_obj(new);
+ __get_obj(new);
+
spin_unlock(&lmv_obj_list_lock);
CDEBUG(D_OTHER, "new obj in lmv cache: %lu/%lu/%lu\n",
(unsigned long)fid->mds, (unsigned long)fid->id,
(unsigned long)fid->generation);
- RETURN(lmv_get_obj(obj));
+ RETURN(new);
}
-int
-lmv_create_obj(struct obd_export *exp,
- struct ll_fid *fid, struct mea *mea)
+/* creates object from passed @fid and @mea. If @mea is NULL, it will be
+ * obtained from correct MDT and used for constructing the object. */
+struct lmv_obj *
+lmv_create_obj(struct obd_export *exp, struct ll_fid *fid, struct mea *mea)
{
struct obd_device *obd = exp->exp_obd;
struct lmv_obd *lmv = &obd->u.lmv;
struct ptlrpc_request *req = NULL;
struct lmv_obj *obj;
struct lustre_md md;
- int mealen, i, rc = 0;
+ int mealen, i, rc;
ENTRY;
CDEBUG(D_OTHER, "get mea for %lu/%lu/%lu and create lmv obj\n",
valid, mealen, &req);
if (rc) {
CERROR("md_getattr() failed, error %d\n", rc);
- GOTO(cleanup, rc);
+ GOTO(cleanup, obj = ERR_PTR(rc));
}
rc = mdc_req2lustre_md(exp, req, 0, NULL, &md);
if (rc) {
CERROR("mdc_req2lustre_md() failed, error %d\n", rc);
- GOTO(cleanup, rc);
+ GOTO(cleanup, obj = ERR_PTR(rc));
}
if (!md.mea)
- GOTO(cleanup, rc = -ENODATA);
+ GOTO(cleanup, obj = ERR_PTR(-ENODATA));
mea = md.mea;
}
/* got mea, now create obj for it. */
- obj = __lmv_create_obj(obd, fid, mea);
+ obj = __create_obj(obd, fid, mea);
if (!obj) {
CERROR("Can't create new object %lu/%lu/%lu\n",
(unsigned long)fid->mds, (unsigned long)fid->id,
(unsigned long)fid->generation);
- GOTO(cleanup, rc = -ENOMEM);
- } else
- lmv_put_obj(obj);
+ GOTO(cleanup, obj = ERR_PTR(-ENOMEM));
+ }
cleanup:
if (req)
ptlrpc_req_finished(req);
- RETURN(rc);
+ RETURN(obj);
+}
+
+/* looks for object with @fid and orders to destroy it. It possible the object
+ * will not be destroyed right now, because it is still using by someone. In
+ * this case it will be marked as "freeing" and will not be accessible anymore
+ * for subsequent callers of lmv_grab_obj(). */
+int
+lmv_delete_obj(struct obd_export *exp, struct ll_fid *fid)
+{
+ struct obd_device *obd = exp->exp_obd;
+ struct lmv_obj *obj;
+ int rc = 0;
+ ENTRY;
+
+ spin_lock(&lmv_obj_list_lock);
+
+ obj = __grab_obj(obd, fid);
+ if (obj) {
+ obj->state |= O_FREEING;
+
+ if (atomic_read(&obj->count) > 1)
+ CERROR("obj %lu/%lu/%lu has count > 2 (%d)\n",
+ (unsigned long) obj->fid.mds,
+ (unsigned long) obj->fid.id,
+ (unsigned long) obj->fid.generation,
+ atomic_read(&obj->count));
+ __put_obj(obj);
+ __put_obj(obj);
+ rc = 1;
+ }
+
+ spin_unlock(&lmv_obj_list_lock);
+ RETURN(rc);
}
int
lmv_setup_mgr(struct obd_device *obd)
{
- CWARN("LMV object manager setup\n");
+ CDEBUG(D_INFO, "LMV object manager setup (%s)\n", obd->obd_uuid.uuid);
return 0;
}
struct lmv_obj *obj;
struct list_head *cur, *tmp;
- CWARN("LMV object manager cleanup\n");
+ CDEBUG(D_INFO, "LMV object manager cleanup (%s)\n", obd->obd_uuid.uuid);
spin_lock(&lmv_obj_list_lock);
list_for_each_safe(cur, tmp, &lmv_obj_list) {
if (obj->obd != obd)
continue;
- list_del(&obj->list);
-
- if (atomic_read(&obj->count) > 1) {
- struct ll_fid *fid = &obj->fid;
-
- CERROR("Object %lu/%lu/%lu has invalid ref count %d\n",
- (unsigned long)fid->mds, (unsigned long)fid->id,
- (unsigned long)fid->generation,
+ obj->state |= O_FREEING;
+ if (atomic_read(&obj->count) > 1)
+ CERROR("obj %lu/%lu/%lu has count > 1 (%d)\n",
+ (unsigned long) obj->fid.mds,
+ (unsigned long) obj->fid.id,
+ (unsigned long) obj->fid.generation,
atomic_read(&obj->count));
- }
-
- /* list does not use object anymore, ref counter should be
- * descreased. */
- lmv_put_obj(obj);
+ __put_obj(obj);
}
spin_unlock(&lmv_obj_list_lock);
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff