Make blockstore list operation return consistent snapshots
parent
6982fe1255
commit
46f9bd2a69
|
@ -145,7 +145,7 @@ void blockstore_impl_t::loop()
|
||||||
}
|
}
|
||||||
unsigned ring_space = ringloop->space_left();
|
unsigned ring_space = ringloop->space_left();
|
||||||
unsigned prev_sqe_pos = ringloop->save();
|
unsigned prev_sqe_pos = ringloop->save();
|
||||||
int dequeue_op = 0;
|
bool dequeue_op = false;
|
||||||
if (op->opcode == BS_OP_READ)
|
if (op->opcode == BS_OP_READ)
|
||||||
{
|
{
|
||||||
dequeue_op = dequeue_read(op);
|
dequeue_op = dequeue_read(op);
|
||||||
|
@ -175,17 +175,34 @@ void blockstore_impl_t::loop()
|
||||||
}
|
}
|
||||||
else if (op->opcode == BS_OP_STABLE)
|
else if (op->opcode == BS_OP_STABLE)
|
||||||
{
|
{
|
||||||
|
if (has_writes == 2)
|
||||||
|
{
|
||||||
|
// Don't submit additional flushes before completing previous LISTs
|
||||||
|
break;
|
||||||
|
}
|
||||||
dequeue_op = dequeue_stable(op);
|
dequeue_op = dequeue_stable(op);
|
||||||
}
|
}
|
||||||
else if (op->opcode == BS_OP_ROLLBACK)
|
else if (op->opcode == BS_OP_ROLLBACK)
|
||||||
{
|
{
|
||||||
|
if (has_writes == 2)
|
||||||
|
{
|
||||||
|
// Don't submit additional flushes before completing previous LISTs
|
||||||
|
break;
|
||||||
|
}
|
||||||
dequeue_op = dequeue_rollback(op);
|
dequeue_op = dequeue_rollback(op);
|
||||||
}
|
}
|
||||||
else if (op->opcode == BS_OP_LIST)
|
else if (op->opcode == BS_OP_LIST)
|
||||||
|
{
|
||||||
|
// Block LIST operation by previous modifications,
|
||||||
|
// so it always returns a consistent state snapshot
|
||||||
|
if (has_writes == 2 || inflight_writes > 0)
|
||||||
|
has_writes = 2;
|
||||||
|
else
|
||||||
{
|
{
|
||||||
process_list(op);
|
process_list(op);
|
||||||
dequeue_op = true;
|
dequeue_op = true;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if (dequeue_op)
|
if (dequeue_op)
|
||||||
{
|
{
|
||||||
submit_queue.erase(op_ptr);
|
submit_queue.erase(op_ptr);
|
||||||
|
|
|
@ -230,6 +230,7 @@ class blockstore_impl_t
|
||||||
|
|
||||||
bool live = false, queue_stall = false;
|
bool live = false, queue_stall = false;
|
||||||
ring_loop_t *ringloop;
|
ring_loop_t *ringloop;
|
||||||
|
int inflight_writes = 0;
|
||||||
|
|
||||||
bool stop_sync_submitted;
|
bool stop_sync_submitted;
|
||||||
|
|
||||||
|
|
|
@ -18,6 +18,10 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
|
||||||
});
|
});
|
||||||
if (dirty_it == dirty_db.begin())
|
if (dirty_it == dirty_db.begin())
|
||||||
{
|
{
|
||||||
|
if (v->version == 0)
|
||||||
|
{
|
||||||
|
// Already rolled back
|
||||||
|
}
|
||||||
bad_op:
|
bad_op:
|
||||||
op->retval = -EINVAL;
|
op->retval = -EINVAL;
|
||||||
FINISH_OP(op);
|
FINISH_OP(op);
|
||||||
|
@ -115,6 +119,7 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
|
||||||
PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
||||||
PRIV(op)->pending_ops = s;
|
PRIV(op)->pending_ops = s;
|
||||||
PRIV(op)->op_state = 1;
|
PRIV(op)->op_state = 1;
|
||||||
|
inflight_writes++;
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -175,6 +180,7 @@ resume_5:
|
||||||
erase_dirty(rm_start, rm_end, UINT64_MAX);
|
erase_dirty(rm_start, rm_end, UINT64_MAX);
|
||||||
}
|
}
|
||||||
journal.trim();
|
journal.trim();
|
||||||
|
inflight_writes--;
|
||||||
// Acknowledge op
|
// Acknowledge op
|
||||||
op->retval = 0;
|
op->retval = 0;
|
||||||
FINISH_OP(op);
|
FINISH_OP(op);
|
||||||
|
@ -186,6 +192,7 @@ void blockstore_impl_t::handle_rollback_event(ring_data_t *data, blockstore_op_t
|
||||||
live = true;
|
live = true;
|
||||||
if (data->res != data->iov.iov_len)
|
if (data->res != data->iov.iov_len)
|
||||||
{
|
{
|
||||||
|
inflight_writes--;
|
||||||
throw std::runtime_error(
|
throw std::runtime_error(
|
||||||
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
||||||
"). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
|
"). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
|
||||||
|
|
|
@ -108,6 +108,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
|
||||||
}
|
}
|
||||||
for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
|
for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
|
||||||
{
|
{
|
||||||
|
// FIXME: Only stabilize versions that aren't stable yet
|
||||||
auto unstab_it = unstable_writes.find(v->oid);
|
auto unstab_it = unstable_writes.find(v->oid);
|
||||||
if (unstab_it != unstable_writes.end() &&
|
if (unstab_it != unstable_writes.end() &&
|
||||||
unstab_it->second <= v->version)
|
unstab_it->second <= v->version)
|
||||||
|
@ -132,6 +133,7 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
|
||||||
PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
||||||
PRIV(op)->pending_ops = s;
|
PRIV(op)->pending_ops = s;
|
||||||
PRIV(op)->op_state = 1;
|
PRIV(op)->op_state = 1;
|
||||||
|
inflight_writes++;
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -209,6 +211,7 @@ resume_5:
|
||||||
flusher->enqueue_flush(*v);
|
flusher->enqueue_flush(*v);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
inflight_writes--;
|
||||||
// Acknowledge op
|
// Acknowledge op
|
||||||
op->retval = 0;
|
op->retval = 0;
|
||||||
FINISH_OP(op);
|
FINISH_OP(op);
|
||||||
|
@ -220,6 +223,7 @@ void blockstore_impl_t::handle_stable_event(ring_data_t *data, blockstore_op_t *
|
||||||
live = true;
|
live = true;
|
||||||
if (data->res != data->iov.iov_len)
|
if (data->res != data->iov.iov_len)
|
||||||
{
|
{
|
||||||
|
inflight_writes--;
|
||||||
throw std::runtime_error(
|
throw std::runtime_error(
|
||||||
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
||||||
"). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
|
"). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
|
||||||
|
|
|
@ -276,6 +276,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
|
||||||
PRIV(op)->op_state = 3;
|
PRIV(op)->op_state = 3;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
inflight_writes++;
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -357,6 +358,7 @@ resume_4:
|
||||||
dirty_it++;
|
dirty_it++;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
inflight_writes--;
|
||||||
// Acknowledge write
|
// Acknowledge write
|
||||||
op->retval = op->len;
|
op->retval = op->len;
|
||||||
FINISH_OP(op);
|
FINISH_OP(op);
|
||||||
|
@ -368,6 +370,7 @@ void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *o
|
||||||
live = true;
|
live = true;
|
||||||
if (data->res != data->iov.iov_len)
|
if (data->res != data->iov.iov_len)
|
||||||
{
|
{
|
||||||
|
inflight_writes--;
|
||||||
// FIXME: our state becomes corrupted after a write error. maybe do something better than just die
|
// FIXME: our state becomes corrupted after a write error. maybe do something better than just die
|
||||||
throw std::runtime_error(
|
throw std::runtime_error(
|
||||||
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
||||||
|
|
Loading…
Reference in New Issue