Vitaliy Filippov
571be0f380
"2-phase" (write->stabilize) process is pointless for deletions because it doesn't protect us from incomplete objects. This happens because it removes the version information from metadata after stabilization. Deletions require "3-phase" process with a potentially very long 3rd phase. So, deletions will be allowed to generate degraded and incomplete objects, and for it to not affect users' ability to delete something, the cluster will allow to delete whole inodes while storing a list of them in etcd. Proper TRIM will be impossible until the implementation of the aforementioned "3-phase" process, though. By the way, this change also fixes a possible write stall after rebalancing which was caused by the lack of "stabilize delete" operations.
513 lines
18 KiB
C++
513 lines
18 KiB
C++
#include "blockstore_impl.h"
|
|
|
|
bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
|
|
{
|
|
// Check or assign version number
|
|
bool found = false, deleted = false, is_del = (op->opcode == BS_OP_DELETE);
|
|
bool is_inflight_big = false;
|
|
uint64_t version = 1;
|
|
if (dirty_db.size() > 0)
|
|
{
|
|
auto dirty_it = dirty_db.upper_bound((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = UINT64_MAX,
|
|
});
|
|
dirty_it--; // segfaults when dirty_db is empty
|
|
if (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
|
|
{
|
|
found = true;
|
|
version = dirty_it->first.version + 1;
|
|
deleted = IS_DELETE(dirty_it->second.state);
|
|
is_inflight_big = dirty_it->second.state >= ST_D_IN_FLIGHT &&
|
|
dirty_it->second.state < ST_D_SYNCED ||
|
|
dirty_it->second.state == ST_J_WAIT_BIG;
|
|
}
|
|
}
|
|
if (!found)
|
|
{
|
|
auto clean_it = clean_db.find(op->oid);
|
|
if (clean_it != clean_db.end())
|
|
{
|
|
version = clean_it->second.version + 1;
|
|
}
|
|
else
|
|
{
|
|
deleted = true;
|
|
}
|
|
}
|
|
if (op->version == 0)
|
|
{
|
|
op->version = version;
|
|
}
|
|
else if (op->version < version)
|
|
{
|
|
// Invalid version requested
|
|
op->retval = -EEXIST;
|
|
return false;
|
|
}
|
|
if (deleted && is_del)
|
|
{
|
|
// Already deleted
|
|
op->retval = 0;
|
|
return false;
|
|
}
|
|
if (is_inflight_big && !is_del && !deleted && op->len < block_size &&
|
|
immediate_commit != IMMEDIATE_ALL)
|
|
{
|
|
// Issue an additional sync so that the previous big write can reach the journal
|
|
blockstore_op_t *sync_op = new blockstore_op_t;
|
|
sync_op->opcode = BS_OP_SYNC;
|
|
sync_op->callback = [this, op](blockstore_op_t *sync_op)
|
|
{
|
|
delete sync_op;
|
|
};
|
|
enqueue_op(sync_op);
|
|
}
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
if (is_del)
|
|
printf("Delete %lu:%lu v%lu\n", op->oid.inode, op->oid.stripe, op->version);
|
|
else
|
|
printf("Write %lu:%lu v%lu offset=%u len=%u\n", op->oid.inode, op->oid.stripe, op->version, op->offset, op->len);
|
|
#endif
|
|
// No strict need to add it into dirty_db here, it's just left
|
|
// from the previous implementation where reads waited for writes
|
|
dirty_db.emplace((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
}, (dirty_entry){
|
|
.state = (uint32_t)(
|
|
is_del
|
|
? ST_DEL_IN_FLIGHT
|
|
: (op->len == block_size || deleted ? ST_D_IN_FLIGHT : (is_inflight_big ? ST_J_WAIT_BIG : ST_J_IN_FLIGHT))
|
|
),
|
|
.flags = 0,
|
|
.location = 0,
|
|
.offset = is_del ? 0 : op->offset,
|
|
.len = is_del ? 0 : op->len,
|
|
.journal_sector = 0,
|
|
});
|
|
return true;
|
|
}
|
|
|
|
// First step of the write algorithm: dequeue operation and submit initial write(s)
|
|
int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
|
|
{
|
|
if (PRIV(op)->op_state)
|
|
{
|
|
return continue_write(op);
|
|
}
|
|
auto dirty_it = dirty_db.find((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
assert(dirty_it != dirty_db.end());
|
|
if (dirty_it->second.state == ST_J_WAIT_BIG)
|
|
{
|
|
return 0;
|
|
}
|
|
else if (dirty_it->second.state == ST_D_IN_FLIGHT)
|
|
{
|
|
blockstore_journal_check_t space_check(this);
|
|
if (!space_check.check_available(op, unsynced_big_writes.size() + 1, sizeof(journal_entry_big_write), JOURNAL_STABILIZE_RESERVATION))
|
|
{
|
|
return 0;
|
|
}
|
|
// Big (redirect) write
|
|
uint64_t loc = data_alloc->find_free();
|
|
if (loc == UINT64_MAX)
|
|
{
|
|
// no space
|
|
if (flusher->is_active())
|
|
{
|
|
// hope that some space will be available after flush
|
|
PRIV(op)->wait_for = WAIT_FREE;
|
|
return 0;
|
|
}
|
|
op->retval = -ENOSPC;
|
|
FINISH_OP(op);
|
|
return 1;
|
|
}
|
|
BS_SUBMIT_GET_SQE(sqe, data);
|
|
dirty_it->second.location = loc << block_order;
|
|
dirty_it->second.state = ST_D_SUBMITTED;
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
printf("Allocate block %lu\n", loc);
|
|
#endif
|
|
data_alloc->set(loc, true);
|
|
uint64_t stripe_offset = (op->offset % bitmap_granularity);
|
|
uint64_t stripe_end = (op->offset + op->len) % bitmap_granularity;
|
|
// Zero fill up to bitmap_granularity
|
|
int vcnt = 0;
|
|
if (stripe_offset)
|
|
{
|
|
PRIV(op)->iov_zerofill[vcnt++] = (struct iovec){ zero_object, stripe_offset };
|
|
}
|
|
PRIV(op)->iov_zerofill[vcnt++] = (struct iovec){ op->buf, op->len };
|
|
if (stripe_end)
|
|
{
|
|
stripe_end = bitmap_granularity - stripe_end;
|
|
PRIV(op)->iov_zerofill[vcnt++] = (struct iovec){ zero_object, stripe_end };
|
|
}
|
|
data->iov.iov_len = op->len + stripe_offset + stripe_end; // to check it in the callback
|
|
data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
|
|
my_uring_prep_writev(
|
|
sqe, data_fd, PRIV(op)->iov_zerofill, vcnt, data_offset + (loc << block_order) + op->offset - stripe_offset
|
|
);
|
|
PRIV(op)->pending_ops = 1;
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
if (immediate_commit != IMMEDIATE_ALL)
|
|
{
|
|
// Remember big write as unsynced
|
|
unsynced_big_writes.push_back((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
PRIV(op)->op_state = 3;
|
|
}
|
|
else
|
|
{
|
|
PRIV(op)->op_state = 1;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
// Small (journaled) write
|
|
// First check if the journal has sufficient space
|
|
blockstore_journal_check_t space_check(this);
|
|
if (unsynced_big_writes.size() && !space_check.check_available(op, unsynced_big_writes.size(), sizeof(journal_entry_big_write), 0)
|
|
|| !space_check.check_available(op, 1, sizeof(journal_entry_small_write), op->len + JOURNAL_STABILIZE_RESERVATION))
|
|
{
|
|
return 0;
|
|
}
|
|
// There is sufficient space. Get SQE(s)
|
|
struct io_uring_sqe *sqe1 = NULL;
|
|
if (immediate_commit != IMMEDIATE_NONE ||
|
|
(journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_small_write) &&
|
|
journal.sector_info[journal.cur_sector].dirty)
|
|
{
|
|
// Write current journal sector only if it's dirty and full, or in the immediate_commit mode
|
|
BS_SUBMIT_GET_SQE_DECL(sqe1);
|
|
}
|
|
struct io_uring_sqe *sqe2 = NULL;
|
|
if (op->len > 0)
|
|
{
|
|
BS_SUBMIT_GET_SQE_DECL(sqe2);
|
|
}
|
|
// Got SQEs. Prepare previous journal sector write if required
|
|
auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
|
|
if (immediate_commit == IMMEDIATE_NONE)
|
|
{
|
|
if (sqe1)
|
|
{
|
|
prepare_journal_sector_write(journal, journal.cur_sector, sqe1, cb);
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
|
PRIV(op)->pending_ops++;
|
|
}
|
|
else
|
|
{
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
}
|
|
}
|
|
// Then pre-fill journal entry
|
|
journal_entry_small_write *je = (journal_entry_small_write*)
|
|
prefill_single_journal_entry(journal, JE_SMALL_WRITE, sizeof(journal_entry_small_write));
|
|
dirty_it->second.journal_sector = journal.sector_info[journal.cur_sector].offset;
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]++;
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
printf(
|
|
"journal offset %08lx is used by %lu:%lu v%lu (%lu refs)\n",
|
|
dirty_it->second.journal_sector, dirty_it->first.oid.inode, dirty_it->first.oid.stripe, dirty_it->first.version,
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]
|
|
);
|
|
#endif
|
|
// Figure out where data will be
|
|
journal.next_free = (journal.next_free + op->len) <= journal.len ? journal.next_free : journal_block_size;
|
|
je->oid = op->oid;
|
|
je->version = op->version;
|
|
je->offset = op->offset;
|
|
je->len = op->len;
|
|
je->data_offset = journal.next_free;
|
|
je->crc32_data = crc32c(0, op->buf, op->len);
|
|
je->crc32 = je_crc32((journal_entry*)je);
|
|
journal.crc32_last = je->crc32;
|
|
if (immediate_commit != IMMEDIATE_NONE)
|
|
{
|
|
prepare_journal_sector_write(journal, journal.cur_sector, sqe1, cb);
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
|
PRIV(op)->pending_ops++;
|
|
}
|
|
if (op->len > 0)
|
|
{
|
|
// Prepare journal data write
|
|
if (journal.inmemory)
|
|
{
|
|
// Copy data
|
|
memcpy(journal.buffer + journal.next_free, op->buf, op->len);
|
|
}
|
|
ring_data_t *data2 = ((ring_data_t*)sqe2->user_data);
|
|
data2->iov = (struct iovec){ op->buf, op->len };
|
|
data2->callback = cb;
|
|
my_uring_prep_writev(
|
|
sqe2, journal.fd, &data2->iov, 1, journal.offset + journal.next_free
|
|
);
|
|
PRIV(op)->pending_ops++;
|
|
}
|
|
else
|
|
{
|
|
// Zero-length overwrite. Allowed to bump object version in EC placement groups without actually writing data
|
|
}
|
|
dirty_it->second.location = journal.next_free;
|
|
dirty_it->second.state = ST_J_SUBMITTED;
|
|
journal.next_free += op->len;
|
|
if (journal.next_free >= journal.len)
|
|
{
|
|
journal.next_free = journal_block_size;
|
|
}
|
|
if (immediate_commit == IMMEDIATE_NONE)
|
|
{
|
|
// Remember small write as unsynced
|
|
unsynced_small_writes.push_back((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
}
|
|
if (!PRIV(op)->pending_ops)
|
|
{
|
|
PRIV(op)->op_state = 4;
|
|
continue_write(op);
|
|
}
|
|
else
|
|
{
|
|
PRIV(op)->op_state = 3;
|
|
}
|
|
}
|
|
inflight_writes++;
|
|
return 1;
|
|
}
|
|
|
|
int blockstore_impl_t::continue_write(blockstore_op_t *op)
|
|
{
|
|
io_uring_sqe *sqe = NULL;
|
|
journal_entry_big_write *je;
|
|
auto dirty_it = dirty_db.find((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
assert(dirty_it != dirty_db.end());
|
|
if (PRIV(op)->op_state == 2)
|
|
goto resume_2;
|
|
else if (PRIV(op)->op_state == 4)
|
|
goto resume_4;
|
|
else
|
|
return 1;
|
|
resume_2:
|
|
// Only for the immediate_commit mode: prepare and submit big_write journal entry
|
|
sqe = get_sqe();
|
|
if (!sqe)
|
|
{
|
|
return 0;
|
|
}
|
|
je = (journal_entry_big_write*)prefill_single_journal_entry(journal, JE_BIG_WRITE, sizeof(journal_entry_big_write));
|
|
dirty_it->second.journal_sector = journal.sector_info[journal.cur_sector].offset;
|
|
journal.sector_info[journal.cur_sector].dirty = false;
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]++;
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
printf(
|
|
"journal offset %08lx is used by %lu:%lu v%lu (%lu refs)\n",
|
|
journal.sector_info[journal.cur_sector].offset, op->oid.inode, op->oid.stripe, op->version,
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]
|
|
);
|
|
#endif
|
|
je->oid = op->oid;
|
|
je->version = op->version;
|
|
je->offset = op->offset;
|
|
je->len = op->len;
|
|
je->location = dirty_it->second.location;
|
|
je->crc32 = je_crc32((journal_entry*)je);
|
|
journal.crc32_last = je->crc32;
|
|
prepare_journal_sector_write(journal, journal.cur_sector, sqe,
|
|
[this, op](ring_data_t *data) { handle_write_event(data, op); });
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
|
PRIV(op)->pending_ops = 1;
|
|
PRIV(op)->op_state = 3;
|
|
return 1;
|
|
resume_4:
|
|
// Switch object state
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
printf("Ack write %lu:%lu v%lu = %d\n", op->oid.inode, op->oid.stripe, op->version, dirty_it->second.state);
|
|
#endif
|
|
bool imm = dirty_it->second.state == ST_D_SUBMITTED
|
|
? (immediate_commit == IMMEDIATE_ALL)
|
|
: (immediate_commit != IMMEDIATE_NONE);
|
|
if (imm)
|
|
{
|
|
auto & unstab = unstable_writes[op->oid];
|
|
unstab = unstab < op->version ? op->version : unstab;
|
|
}
|
|
if (dirty_it->second.state == ST_J_SUBMITTED)
|
|
{
|
|
dirty_it->second.state = imm ? ST_J_SYNCED : ST_J_WRITTEN;
|
|
}
|
|
else if (dirty_it->second.state == ST_D_SUBMITTED)
|
|
{
|
|
dirty_it->second.state = imm ? ST_D_SYNCED : ST_D_WRITTEN;
|
|
}
|
|
else if (dirty_it->second.state == ST_DEL_SUBMITTED)
|
|
{
|
|
dirty_it->second.state = imm ? ST_DEL_SYNCED : ST_DEL_WRITTEN;
|
|
if (imm)
|
|
{
|
|
// Deletions are treated as immediately stable
|
|
mark_stable(dirty_it->first);
|
|
}
|
|
}
|
|
if (immediate_commit == IMMEDIATE_ALL)
|
|
{
|
|
dirty_it++;
|
|
while (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
|
|
{
|
|
if (dirty_it->second.state == ST_J_WAIT_BIG)
|
|
{
|
|
dirty_it->second.state = ST_J_IN_FLIGHT;
|
|
}
|
|
dirty_it++;
|
|
}
|
|
}
|
|
inflight_writes--;
|
|
// Acknowledge write
|
|
op->retval = op->len;
|
|
FINISH_OP(op);
|
|
return 1;
|
|
}
|
|
|
|
void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op)
|
|
{
|
|
live = true;
|
|
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
|
|
throw std::runtime_error(
|
|
"write operation failed ("+std::to_string(data->res)+" != "+std::to_string(data->iov.iov_len)+
|
|
"). in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111"
|
|
);
|
|
}
|
|
PRIV(op)->pending_ops--;
|
|
if (PRIV(op)->pending_ops == 0)
|
|
{
|
|
release_journal_sectors(op);
|
|
PRIV(op)->op_state++;
|
|
if (!continue_write(op))
|
|
{
|
|
submit_queue.push_front(op);
|
|
}
|
|
}
|
|
}
|
|
|
|
void blockstore_impl_t::release_journal_sectors(blockstore_op_t *op)
|
|
{
|
|
// Release flushed journal sectors
|
|
if (PRIV(op)->min_flushed_journal_sector > 0 &&
|
|
PRIV(op)->max_flushed_journal_sector > 0)
|
|
{
|
|
uint64_t s = PRIV(op)->min_flushed_journal_sector;
|
|
while (1)
|
|
{
|
|
journal.sector_info[s-1].usage_count--;
|
|
if (s != (1+journal.cur_sector) && journal.sector_info[s-1].usage_count == 0)
|
|
{
|
|
// We know for sure that we won't write into this sector anymore
|
|
uint64_t new_ds = journal.sector_info[s-1].offset + journal.block_size;
|
|
if (new_ds >= journal.len)
|
|
{
|
|
new_ds = journal.block_size;
|
|
}
|
|
if ((journal.dirty_start + (journal.dirty_start >= journal.used_start ? 0 : journal.len)) <
|
|
(new_ds + (new_ds >= journal.used_start ? 0 : journal.len)))
|
|
{
|
|
journal.dirty_start = new_ds;
|
|
}
|
|
}
|
|
if (s == PRIV(op)->max_flushed_journal_sector)
|
|
break;
|
|
s = 1 + s % journal.sector_count;
|
|
}
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
}
|
|
}
|
|
|
|
int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
|
|
{
|
|
auto dirty_it = dirty_db.find((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
assert(dirty_it != dirty_db.end());
|
|
blockstore_journal_check_t space_check(this);
|
|
if (!space_check.check_available(op, 1, sizeof(journal_entry_del), 0))
|
|
{
|
|
return 0;
|
|
}
|
|
io_uring_sqe *sqe = NULL;
|
|
if (immediate_commit != IMMEDIATE_NONE ||
|
|
(journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_del) &&
|
|
journal.sector_info[journal.cur_sector].dirty)
|
|
{
|
|
// Write current journal sector only if it's dirty and full, or in the immediate_commit mode
|
|
BS_SUBMIT_GET_SQE_DECL(sqe);
|
|
}
|
|
auto cb = [this, op](ring_data_t *data) { handle_write_event(data, op); };
|
|
// Prepare journal sector write
|
|
if (immediate_commit == IMMEDIATE_NONE)
|
|
{
|
|
if (sqe)
|
|
{
|
|
prepare_journal_sector_write(journal, journal.cur_sector, sqe, cb);
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
|
PRIV(op)->pending_ops++;
|
|
}
|
|
else
|
|
{
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 0;
|
|
}
|
|
}
|
|
// Pre-fill journal entry
|
|
journal_entry_del *je = (journal_entry_del*)
|
|
prefill_single_journal_entry(journal, JE_DELETE, sizeof(struct journal_entry_del));
|
|
dirty_it->second.journal_sector = journal.sector_info[journal.cur_sector].offset;
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]++;
|
|
#ifdef BLOCKSTORE_DEBUG
|
|
printf(
|
|
"journal offset %08lx is used by %lu:%lu v%lu (%lu refs)\n",
|
|
dirty_it->second.journal_sector, dirty_it->first.oid.inode, dirty_it->first.oid.stripe, dirty_it->first.version,
|
|
journal.used_sectors[journal.sector_info[journal.cur_sector].offset]
|
|
);
|
|
#endif
|
|
je->oid = op->oid;
|
|
je->version = op->version;
|
|
je->crc32 = je_crc32((journal_entry*)je);
|
|
journal.crc32_last = je->crc32;
|
|
dirty_it->second.state = ST_DEL_SUBMITTED;
|
|
if (immediate_commit != IMMEDIATE_NONE)
|
|
{
|
|
prepare_journal_sector_write(journal, journal.cur_sector, sqe, cb);
|
|
PRIV(op)->min_flushed_journal_sector = PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
|
|
PRIV(op)->pending_ops++;
|
|
// Remember small write as unsynced
|
|
unsynced_small_writes.push_back((obj_ver_id){
|
|
.oid = op->oid,
|
|
.version = op->version,
|
|
});
|
|
}
|
|
if (!PRIV(op)->pending_ops)
|
|
{
|
|
PRIV(op)->op_state = 4;
|
|
continue_write(op);
|
|
}
|
|
else
|
|
{
|
|
PRIV(op)->op_state = 3;
|
|
}
|
|
return 1;
|
|
}
|