Whamcloud - gitweb
git://git.whamcloud.com
/
fs
/
lustre-release.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
| inline |
side by side
LU-5324 lfsck: invalidly memory access in lfsck_del_target
[fs/lustre-release.git]
/
lustre
/
osp
/
osp_object.c
diff --git
a/lustre/osp/osp_object.c
b/lustre/osp/osp_object.c
index
4658f48
..
690e1ce
100644
(file)
--- a/
lustre/osp/osp_object.c
+++ b/
lustre/osp/osp_object.c
@@
-281,7
+281,6
@@
static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
{
struct osp_object *obj = dt2osp_obj(dt);
struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
- struct dt_update_request *update;
int rc = 0;
if (obj->opo_ooa == NULL) {
@@
-291,14
+290,9
@@
static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
}
mutex_lock(&osp->opd_async_requests_mutex);
- update = osp_find_or_create_async_update_request(osp);
- if (IS_ERR(update))
- rc = PTR_ERR(update);
- else
- rc = osp_insert_async_update(env, update, OUT_ATTR_GET, obj, 0,
- NULL, NULL,
- &obj->opo_ooa->ooa_attr,
- osp_attr_get_interpterer);
+ rc = osp_insert_async_request(env, OUT_ATTR_GET, obj, 0, NULL, NULL,
+ &obj->opo_ooa->ooa_attr,
+ osp_attr_get_interpterer);
mutex_unlock(&osp->opd_async_requests_mutex);
return rc;
@@
-370,13
+364,6
@@
int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
if (rc != 0)
GOTO(out, rc);
- if (!is_ost_obj(&dt->do_lu)) {
- if (attr->la_flags == 1)
- obj->opo_empty = 0;
- else
- obj->opo_empty = 1;
- }
-
GOTO(out, rc = 0);
out:
@@
-557,7
+544,6
@@
static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
{
struct osp_object *obj = dt2osp_obj(dt);
struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
- struct dt_update_request *update;
struct osp_xattr_entry *oxe;
int namelen = strlen(name);
int rc = 0;
@@
-580,33
+566,28
@@
static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
return -ENOMEM;
mutex_lock(&osp->opd_async_requests_mutex);
- update = osp_find_or_create_async_update_request(osp);
- if (IS_ERR(update)) {
- rc = PTR_ERR(update);
+ rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
+ &namelen, &name, oxe,
+ osp_xattr_get_interpterer);
+ if (rc != 0) {
mutex_unlock(&osp->opd_async_requests_mutex);
osp_oac_xattr_put(oxe);
} else {
- rc = osp_insert_async_update(env, update, OUT_XATTR_GET, obj,
- 1, &namelen, &name, oxe,
- osp_xattr_get_interpterer);
- if (rc != 0) {
+ struct dt_update_request *update;
+
+ /* XXX: Currently, we trigger the batched async OUT
+ * RPC via dt_declare_xattr_get(). It is not
+ * perfect solution, but works well now.
+ *
+ * We will improve it in the future. */
+ update = osp->opd_async_requests;
+ if (update != NULL && update->dur_req != NULL &&
+ update->dur_req->ourq_count > 0) {
+ osp->opd_async_requests = NULL;
mutex_unlock(&osp->opd_async_requests_mutex);
-
osp_oac_xattr_put(ox
e);
+
rc = osp_unplug_async_request(env, osp, updat
e);
} else {
- /* XXX: Currently, we trigger the batched async OUT
- * RPC via dt_declare_xattr_get(). It is not
- * perfect solution, but works well now.
- *
- * We will improve it in the future. */
- update = osp->opd_async_requests;
- if (update != NULL && update->dur_req != NULL &&
- update->dur_req->ourq_count > 0) {
- osp->opd_async_requests = NULL;
- mutex_unlock(&osp->opd_async_requests_mutex);
- rc = osp_unplug_async_update(env, osp, update);
- } else {
- mutex_unlock(&osp->opd_async_requests_mutex);
- }
+ mutex_unlock(&osp->opd_async_requests_mutex);
}
}
@@
-753,7
+734,7
@@
unlock:
"cache for "DFID": rc = %d\n",
dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
spin_lock(&obj->opo_lock);
- o
xe
->oxe_ready = 0;
+ o
ld
->oxe_ready = 0;
spin_unlock(&obj->opo_lock);
GOTO(out, rc);
@@
-840,7
+821,7
@@
int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
name, PFID(lu_object_fid(&dt->do_lu)), rc);
spin_lock(&o->opo_lock);
- o
xe
->oxe_ready = 0;
+ o
ld
->oxe_ready = 0;
spin_unlock(&o->opo_lock);
return 0;
@@
-871,6
+852,36
@@
int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
return 0;
}
+int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
+ const char *name, struct thandle *th)
+{
+ struct dt_update_request *update;
+ const struct lu_fid *fid;
+ int size = strlen(name);
+ int rc;
+
+ update = out_find_create_update_loc(th, dt);
+ if (IS_ERR(update))
+ return PTR_ERR(update);
+
+ fid = lu_object_fid(&dt->do_lu);
+
+ rc = out_insert_update(env, update, OUT_XATTR_DEL, fid, 1, &size,
+ (const char **)&name);
+
+ return rc;
+}
+
+int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
+ const char *name, struct thandle *th,
+ struct lustre_capa *capa)
+{
+ CDEBUG(D_INFO, "xattr %s del object "DFID"\n", name,
+ PFID(&dt->do_lu.lo_header->loh_fid));
+
+ return 0;
+}
+
static int osp_declare_object_create(const struct lu_env *env,
struct dt_object *dt,
struct lu_attr *attr,
@@
-1084,21
+1095,6
@@
int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
RETURN(rc);
}
-struct osp_orphan_it {
- int ooi_pos0;
- int ooi_pos1;
- int ooi_pos2;
- int ooi_total_npages;
- int ooi_valid_npages;
- unsigned int ooi_swab:1;
- __u64 ooi_next;
- struct dt_object *ooi_obj;
- struct lu_orphan_ent *ooi_ent;
- struct page *ooi_cur_page;
- struct lu_idxpage *ooi_cur_idxpage;
- struct page **ooi_pages;
-};
-
static int osp_orphan_index_lookup(const struct lu_env *env,
struct dt_object *dt,
struct dt_rec *rec,
@@
-1145,30
+1141,27
@@
static int osp_orphan_index_delete(const struct lu_env *env,
return -EOPNOTSUPP;
}
-static struct dt_it *osp_orphan_it_init(const struct lu_env *env,
- struct dt_object *dt,
- __u32 attr,
- struct lustre_capa *capa)
+struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
+ __u32 attr, struct lustre_capa *capa)
{
- struct osp_
orphan_
it *it;
+ struct osp_it *it;
OBD_ALLOC_PTR(it);
if (it == NULL)
return ERR_PTR(-ENOMEM);
- it->ooi_pos
2
= -1;
+ it->ooi_pos
_ent
= -1;
it->ooi_obj = dt;
return (struct dt_it *)it;
}
-static void osp_orphan_it_fini(const struct lu_env *env,
- struct dt_it *di)
+void osp_it_fini(const struct lu_env *env, struct dt_it *di)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
- struct page
**pages
= it->ooi_pages;
- int
npages
= it->ooi_total_npages;
- int
i;
+ struct osp_
it *it = (struct osp
_it *)di;
+ struct page
**pages
= it->ooi_pages;
+ int
npages
= it->ooi_total_npages;
+ int i;
if (pages != NULL) {
for (i = 0; i < npages; i++) {
@@
-1185,8
+1178,7
@@
static void osp_orphan_it_fini(const struct lu_env *env,
OBD_FREE_PTR(it);
}
-static int osp_orphan_it_fetch(const struct lu_env *env,
- struct osp_orphan_it *it)
+static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
{
struct lu_device *dev = it->ooi_obj->do_lu.lo_dev;
struct osp_device *osp = lu2osp_dev(dev);
@@
-1227,6
+1219,27
@@
static int osp_orphan_it_fetch(const struct lu_env *env,
}
req->rq_request_portal = OUT_PORTAL;
+ ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
+ memset(ii, 0, sizeof(*ii));
+ if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
+ /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
+ * ost_index, 0] with LAST_ID FID, so it needs to replace
+ * the FID with orphan FID here */
+ ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
+ ii->ii_fid.f_oid = osp->opd_index;
+ ii->ii_fid.f_ver = 0;
+ ii->ii_flags = II_FL_NOHASH;
+ } else {
+ ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
+ ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
+ II_FL_VARREC;
+ }
+ ii->ii_magic = IDX_INFO_MAGIC;
+ ii->ii_count = npages * LU_PAGE_COUNT;
+ ii->ii_hash_start = it->ooi_next;
+ ii->ii_attrs =
+ osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
+
ptlrpc_at_set_req_timeout(req);
desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
@@
-1239,18
+1252,6
@@
static int osp_orphan_it_fetch(const struct lu_env *env,
for (i = 0; i < npages; i++)
ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
- ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
- memset(ii, 0, sizeof(*ii));
- ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
- ii->ii_fid.f_oid = osp->opd_index;
- ii->ii_fid.f_ver = 0;
- ii->ii_magic = IDX_INFO_MAGIC;
- ii->ii_flags = II_FL_NOHASH;
- ii->ii_count = npages * LU_PAGE_COUNT;
- ii->ii_hash_start = it->ooi_next;
- ii->ii_attrs =
- osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
-
ptlrpc_request_set_replen(req);
rc = ptlrpc_queue_wait(req);
if (rc != 0)
@@
-1260,6
+1261,7
@@
static int osp_orphan_it_fetch(const struct lu_env *env,
req->rq_bulk->bd_nob_transferred);
if (rc < 0)
GOTO(out, rc);
+ rc = 0;
ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
if (ii->ii_magic != IDX_INFO_MAGIC)
@@
-1279,22
+1281,19
@@
static int osp_orphan_it_fetch(const struct lu_env *env,
it->ooi_next = ii->ii_hash_end;
- GOTO(out, rc = 0);
-
out:
ptlrpc_req_finished(req);
return rc;
}
-static int osp_orphan_it_next(const struct lu_env *env,
- struct dt_it *di)
+int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
- struct lu_idxpage
*idxpage;
+ struct osp_
it *it = (struct osp
_it *)di;
+ struct lu_idxpage *idxpage;
struct page **pages;
- int
rc;
- int
i;
+ int rc;
+ int i;
ENTRY;
again2:
@@
-1303,23
+1302,17
@@
again2:
if (idxpage->lip_nr == 0)
RETURN(1);
- it->ooi_pos2++;
- if (it->ooi_pos2 < idxpage->lip_nr) {
- it->ooi_ent =
- (struct lu_orphan_ent *)idxpage->lip_entries +
- it->ooi_pos2;
- if (it->ooi_swab)
- lustre_swab_orphan_ent(it->ooi_ent);
+ if (it->ooi_pos_ent < idxpage->lip_nr) {
+ CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
+ (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
RETURN(0);
}
-
it->ooi_cur_idxpage = NULL;
- it->ooi_pos1++;
-
+ it->ooi_pos_lu_page++;
again1:
- if (it->ooi_pos
1
< LU_PAGE_COUNT) {
+ if (it->ooi_pos
_lu_page
< LU_PAGE_COUNT) {
it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
-
LU_PAGE_SIZE * it->ooi_pos1
;
+
LU_PAGE_SIZE * it->ooi_pos_lu_page
;
if (it->ooi_swab)
lustre_swab_lip_header(it->ooi_cur_idxpage);
if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
@@
-1330,24
+1323,25
@@
again1:
"%d/%d while read layout orphan index\n",
osp->opd_obd->obd_name,
it->ooi_cur_idxpage->lip_magic,
- LIP_MAGIC, it->ooi_pos0, it->ooi_pos1);
+ LIP_MAGIC, it->ooi_pos_page,
+ it->ooi_pos_lu_page);
/* Skip this lu_page next time. */
- it->ooi_pos
2
= idxpage->lip_nr - 1;
+ it->ooi_pos
_ent
= idxpage->lip_nr - 1;
RETURN(-EINVAL);
}
- it->ooi_pos
2
= -1;
+ it->ooi_pos
_ent
= -1;
goto again2;
}
kunmap(it->ooi_cur_page);
it->ooi_cur_page = NULL;
- it->ooi_pos
0
++;
+ it->ooi_pos
_page
++;
again0:
pages = it->ooi_pages;
- if (it->ooi_pos
0
< it->ooi_valid_npages) {
- it->ooi_cur_page = kmap(pages[it->ooi_pos
0
]);
- it->ooi_pos
1
= 0;
+ if (it->ooi_pos
_page
< it->ooi_valid_npages) {
+ it->ooi_cur_page = kmap(pages[it->ooi_pos
_page
]);
+ it->ooi_pos
_lu_page
= 0;
goto again1;
}
@@
-1357,7
+1351,7
@@
again0:
}
OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
- it->ooi_pos
0
= 0;
+ it->ooi_pos
_page
= 0;
it->ooi_total_npages = 0;
it->ooi_valid_npages = 0;
it->ooi_swab = 0;
@@
-1370,30
+1364,59
@@
again0:
if (it->ooi_next == II_END_OFF)
RETURN(1);
- rc = osp_
orphan_
it_fetch(env, it);
+ rc = osp_it_fetch(env, it);
if (rc == 0)
goto again0;
RETURN(rc);
}
-static int osp_orphan_it_get(const struct lu_env *env,
- struct dt_it *di,
- const struct dt_key *key)
+int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
+{
+ struct osp_it *it = (struct osp_it *)di;
+ struct lu_idxpage *idxpage;
+ int rc;
+ ENTRY;
+
+again:
+ idxpage = it->ooi_cur_idxpage;
+ if (idxpage != NULL) {
+ if (idxpage->lip_nr == 0)
+ RETURN(1);
+
+ it->ooi_pos_ent++;
+ if (it->ooi_pos_ent < idxpage->lip_nr) {
+ it->ooi_ent =
+ (struct lu_orphan_ent *)idxpage->lip_entries +
+ it->ooi_pos_ent;
+ if (it->ooi_swab)
+ lustre_swab_orphan_ent(it->ooi_ent);
+ RETURN(0);
+ }
+ }
+
+ rc = osp_it_next_page(env, di);
+ if (rc == 0)
+ goto again;
+
+ RETURN(rc);
+}
+
+int osp_it_get(const struct lu_env *env, struct dt_it *di,
+ const struct dt_key *key)
{
- return
-ENOSYS
;
+ return
1
;
}
-static void osp_orphan_it_put(const struct lu_env *env,
- struct dt_it *di)
+void osp_it_put(const struct lu_env *env, struct dt_it *di)
{
}
-st
atic st
ruct dt_key *osp_orphan_it_key(const struct lu_env *env,
-
const struct dt_it *di)
+struct dt_key *osp_orphan_it_key(const struct lu_env *env,
+ const struct dt_it *di)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
- struct lu_orphan_ent *ent = it->ooi_ent;
+ struct osp_
it *it = (struct osp
_it *)di;
+ struct lu_orphan_ent *ent =
(struct lu_orphan_ent *)
it->ooi_ent;
if (likely(ent != NULL))
return (struct dt_key *)(&ent->loe_key);
@@
-1401,19
+1424,16
@@
static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
return NULL;
}
-static int osp_orphan_it_key_size(const struct lu_env *env,
- const struct dt_it *di)
+int osp_orphan_it_key_size(const struct lu_env *env, const struct dt_it *di)
{
return sizeof(struct lu_fid);
}
-static int osp_orphan_it_rec(const struct lu_env *env,
- const struct dt_it *di,
- struct dt_rec *rec,
- __u32 attr)
+int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
+ struct dt_rec *rec, __u32 attr)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
- struct lu_orphan_ent *ent = it->ooi_ent;
+ struct osp_
it *it = (struct osp
_it *)di;
+ struct lu_orphan_ent *ent =
(struct lu_orphan_ent *)
it->ooi_ent;
if (likely(ent != NULL)) {
*(struct lu_orphan_rec *)rec = ent->loe_rec;
@@
-1423,10
+1443,9
@@
static int osp_orphan_it_rec(const struct lu_env *env,
return -EINVAL;
}
-static __u64 osp_orphan_it_store(const struct lu_env *env,
- const struct dt_it *di)
+__u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
+ struct osp_
it *it = (struct osp
_it *)di;
return it->ooi_next;
}
@@
-1437,11
+1456,10
@@
static __u64 osp_orphan_it_store(const struct lu_env *env,
* call next() to move to a valid position.
* \retval -ve: on error
*/
-static int osp_orphan_it_load(const struct lu_env *env,
- const struct dt_it *di,
- __u64 hash)
+int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
+ __u64 hash)
{
- struct osp_
orphan_it *it = (struct osp_orphan
_it *)di;
+ struct osp_
it *it = (struct osp
_it *)di;
int rc;
it->ooi_next = hash;
@@
-1455,9
+1473,8
@@
static int osp_orphan_it_load(const struct lu_env *env,
return rc;
}
-static int osp_orphan_it_key_rec(const struct lu_env *env,
- const struct dt_it *di,
- void *key_rec)
+int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
+ void *key_rec)
{
return 0;
}
@@
-1469,17
+1486,17
@@
static const struct dt_index_operations osp_orphan_index_ops = {
.dio_declare_delete = osp_orphan_index_declare_delete,
.dio_delete = osp_orphan_index_delete,
.dio_it = {
- .init = osp_
orphan_
it_init,
- .fini = osp_
orphan_
it_fini,
+ .init = osp_it_init,
+ .fini = osp_it_fini,
.next = osp_orphan_it_next,
- .get = osp_
orphan_
it_get,
- .put = osp_
orphan_
it_put,
+ .get = osp_it_get,
+ .put = osp_it_put,
.key = osp_orphan_it_key,
.key_size = osp_orphan_it_key_size,
.rec = osp_orphan_it_rec,
- .store = osp_
orphan_
it_store,
+ .store = osp_it_store,
.load = osp_orphan_it_load,
- .key_rec = osp_
orphan_
it_key_rec,
+ .key_rec = osp_it_key_rec,
}
};
@@
-1489,13
+1506,11
@@
static int osp_index_try(const struct lu_env *env,
{
const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
- if (fid_is_last_id(fid) && fid_is_idif(fid))
{
+ if (fid_is_last_id(fid) && fid_is_idif(fid))
dt->do_index_ops = &osp_orphan_index_ops;
-
- return 0;
- }
-
- return -EINVAL;
+ else
+ dt->do_index_ops = &osp_md_index_ops;
+ return 0;
}
struct dt_object_operations osp_obj_ops = {