RepoMirrors/mars
1
0
Fork 0
mirror of https://github.com/schoebel/mars synced 2025-03-01 19:10:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

trans_logger: fix potential race on mref reuse

Browse Source
This commit is contained in:
Thomas Schoebel-Theuer 2012-12-07 11:30:33 +01:00 committed by Thomas Schoebel-Theuer
parent 6ec548251d
commit f1af13b52b
1 changed files with 13 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
mars_trans_logger.c
View File

@ -659,31 +659,35 @@ int _write_ref_get(struct trans_logger_output *output, struct trans_logger_mref_
static noinline static noinline
int trans_logger_ref_get(struct trans_logger_output *output, struct mref_object *mref) int trans_logger_ref_get(struct trans_logger_output *output, struct mref_object *mref)
{ {
struct trans_logger_brick *brick = output->brick; struct trans_logger_brick *brick;
struct trans_logger_mref_aspect *mref_a; struct trans_logger_mref_aspect *mref_a;
loff_t base_offset; loff_t base_offset;
CHECK_PTR(output, err); CHECK_PTR(output, err);
brick = output->brick;
CHECK_PTR(brick, err);
CHECK_PTR(mref, err);
MARS_IO("pos = %lld len = %d\n", mref->ref_pos, mref->ref_len); MARS_IO("pos = %lld len = %d\n", mref->ref_pos, mref->ref_len);
if (mref->ref_len > brick->max_mref_size && brick->max_mref_size > 0) mref_a = trans_logger_mref_get_aspect(brick, mref);
mref->ref_len = brick->max_mref_size; CHECK_PTR(mref_a, err);
CHECK_ASPECT(mref_a, mref, err);
atomic_inc(&brick->outer_balance_count); atomic_inc(&brick->outer_balance_count);
if (atomic_read(&mref->ref_count) > 0) { // setup already performed if (mref_a->stamp.tv_sec) { // setup already performed
MARS_IO("again %d\n", atomic_read(&mref->ref_count)); MARS_IO("again %d\n", atomic_read(&mref->ref_count));
atomic_inc(&mref->ref_count); CHECK_ATOMIC(&mref->ref_count, 1);
atomic_inc(&mref->ref_count); // must be paired with __trans_logger_ref_put()
return mref->ref_len; return mref->ref_len;
} }
mref_a = trans_logger_mref_get_aspect(brick, mref);
CHECK_PTR(mref_a, err);
CHECK_PTR(mref_a->object, err);
get_lamport(&mref_a->stamp); get_lamport(&mref_a->stamp);
if (mref->ref_len > brick->max_mref_size && brick->max_mref_size > 0)
mref->ref_len = brick->max_mref_size;
// ensure that REGION_SIZE boundaries are obeyed by hashing // ensure that REGION_SIZE boundaries are obeyed by hashing
base_offset = mref->ref_pos & (loff_t)(REGION_SIZE - 1); base_offset = mref->ref_pos & (loff_t)(REGION_SIZE - 1);
if (mref->ref_len > REGION_SIZE - base_offset) { if (mref->ref_len > REGION_SIZE - base_offset) {