Compare commits

...

11 Commits

Author SHA1 Message Date
ba2f65fa8a WIP Use the disk interface wrapper only in the new blockstore 2026-09-26 17:35:39 +03:00
47fbf6fdc3 Implement user-space NVMe wrapper via libvfn 2026-09-26 17:26:16 +03:00
96d0e7cc2e Remove creepy ringloop->restore usage from the new store 2026-09-26 17:25:18 +03:00
Changwei Ye
b51c05200d Store v2: fix compaction skipping stable overwrites and BIG_INTENT type
The original compaction condition only postponed compaction when not under
pressure, and only for BIG_WRITE and DELETE types:
  if (!under_pressure && (BIG_WRITE || DELETE))
This had two bugs:
1. BIG_INTENT overwrites were not included, causing add_compact() to be
   called on BIG_INTENT entries. add_compact() unconditionally rejects
   any stable overwrite with EBUSY, leading to unnecessary retry loops.
2. When under pressure, stable overwrites (BIG_WRITE, BIG_INTENT, DELETE
   with BS_HEAP_STABLE set) were not skipped. add_compact() rejects stable
   overwrites with EBUSY regardless of pressure, so attempting to compact
   them under pressure also always fails.
Fix: change the condition to:
  if ((!under_pressure || stable) && (BIG_WRITE || BIG_INTENT || DELETE))
When under pressure with an unstable overwrite, we now fall through to
`continue`, skipping it and scanning older entries (SMALL_WRITE /
INTENT_WRITE) that can actually be compacted to reclaim space.
2026-09-24 17:06:20 +08:00
6d1b9165a2 Store v2: decouple trim_lsn from force_fsync
Commit df9a2850 introduced force_fsync to resolve the placement contract
conflict (EAGAIN) when a new metadata entry depends on a previous unsynced
entry that is already in-flight. However, it mistakenly grouped the
'force_fsync > 0' condition together with the Superblock Checkpoint logic.

As a result, every time the internal torn-write prevention mechanism
requested a force_fsync, the system inadvertently executed an expensive
trim_lsn() (updating the Superblock with RWF_DSYNC), causing unnecessary
write latency and I/O amplification during congested object updates.

This change minimally refactors the state machine to ensure trim_lsn() is
strictly bound to the intended journal_trim_interval, while force_fsync
only triggers the necessary fsync_buffer().
2026-09-24 11:08:15 +03:00
e58b37b284 Fix CAS merge assert(rwo->todo == 1) 2026-09-23 00:04:01 +03:00
53c7da3e2d %lu -> %ju, peer_fd -> client_id in one log message 2026-09-21 01:10:31 +03:00
75a35c3b36 Unmark previous partial writes on a repeated partial write to a different OSD 2026-09-20 00:32:50 +03:00
a375f43a93 Fix possible "operation sequencing is broken on client XX: expected num 2, got 3, stopping client" during shitstorm :] 2026-09-19 21:29:47 +03:00
22ae66b95f Now fix build with qemu < 8 :D 2026-09-13 12:39:31 +03:00
456fad7fa7 Add opool to docs 2026-09-13 12:13:06 +03:00
47 changed files with 2686 additions and 641 deletions

View File

@@ -196,6 +196,7 @@ Options can be specified in classic dd style (`key=value`) or like usual (`--key
| `iimg=<image>` | Copy from Vitastor image `<image>` |
| `if=<file>` | Copy from file `<file>` |
| `oimg=<image>` | Copy to Vitastor image `<image>` |
| `opool=<pool>` | Pool to create output image if it doesn't exist |
| `of=<file>` | Copy to file `<file>` |
| `bs=1M` | Set copy block size |
| `count=N` | Copy only N input blocks. If N ends in B it counts bytes, not blocks |

View File

@@ -203,6 +203,7 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
| `iimg=<image>` | Копировать из образа Vitastor `<image>` |
| `if=<file>` | Копировать из файла `<file>` |
| `oimg=<image>` | Копировать в образ Vitastor `<image>` |
| `opool=<pool>` | Пул для создания выходного образа, если он не существует |
| `of=<file>` | Копировать в файл `<file>` |
| `bs=1M` | Задать размер блока копирования |
| `count=N` | Копировать не более N блоков. Если N заканчивается на B - то N байт. |

View File

@@ -86,6 +86,11 @@ pkg_check_modules(RDMACM librdmacm)
if (RDMACM_LIBRARIES)
add_definitions(-DWITH_RDMACM)
endif (RDMACM_LIBRARIES)
pkg_check_modules(VFN libvfn)
pkg_check_modules(NVME libnvme)
if (VFN_LIBRARIES AND NVME_LIBRARIES)
add_definitions(-DWITH_VFN)
endif (VFN_LIBRARIES AND NVME_LIBRARIES)
find_package(OpenSSL 3.0 REQUIRED)
add_definitions(-DWITH_OPENSSL)

View File

@@ -1,10 +1,6 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#include <sys/file.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <stdexcept>
#include "blockstore.h"
@@ -355,31 +351,22 @@ void blockstore_disk_t::check_lengths()
}
}
static int bs_openmode(const std::string & mode)
{
if (mode == "directsync")
return O_DIRECT|O_SYNC;
else if (mode == "cached")
return O_SYNC;
else
return O_DIRECT;
}
void blockstore_disk_t::open_data()
{
if (data_fd >= 0)
if (data_disk)
{
throw std::runtime_error("data device is already opened");
}
data_fd = mock_mode ? MOCK_DATA_FD : open(data_device.c_str(), bs_openmode(data_io) | O_RDWR);
if (data_fd == -1)
data_disk = mock_mode
? disk_i::create_mock(MOCK_DATA_FD, data_device_size, data_device_sect)
: disk_i::create_and_open(data_device, data_io, disable_flock);
if (!data_disk)
{
throw std::runtime_error("Failed to open data device "+data_device+": "+std::string(strerror(errno)));
}
if (!mock_mode)
{
check_size(data_fd, &data_device_size, &data_device_sect, "data device");
throw std::runtime_error("Failed to open data device "+data_device);
}
data_off_ring = data_disk->is_off_ring();
data_device_size = data_disk->get_size();
data_device_sect = data_disk->get_block_size();
if (disk_alignment % data_device_sect)
{
throw std::runtime_error(
@@ -391,41 +378,33 @@ void blockstore_disk_t::open_data()
{
throw std::runtime_error("data_offset exceeds device size = "+std::to_string(data_device_size));
}
if (!mock_mode && !disable_flock && flock(data_fd, LOCK_EX|LOCK_NB) != 0)
{
throw std::runtime_error(std::string("Failed to lock data device: ") + strerror(errno));
}
}
void blockstore_disk_t::open_meta()
{
if (meta_fd >= 0)
if (meta_disk)
{
throw std::runtime_error("metadata device is already opened");
}
if (meta_device != data_device || meta_io != data_io)
{
meta_fd = mock_mode ? MOCK_META_FD : open(meta_device.c_str(), bs_openmode(meta_io) | O_RDWR);
if (meta_fd == -1)
meta_disk = mock_mode
? disk_i::create_mock(MOCK_META_FD, meta_device_size, meta_device_sect)
: disk_i::create_and_open(meta_device, meta_io, meta_device == data_device || disable_flock);
if (!meta_disk)
{
throw std::runtime_error("Failed to open metadata device "+meta_device+": "+std::string(strerror(errno)));
}
if (!mock_mode)
{
check_size(meta_fd, &meta_device_size, &meta_device_sect, "metadata device");
throw std::runtime_error("Failed to open metadata device "+meta_device);
}
meta_device_size = meta_disk->get_size();
meta_device_sect = meta_disk->get_block_size();
if (meta_offset >= meta_device_size)
{
throw std::runtime_error("meta_offset exceeds device size = "+std::to_string(meta_device_size));
}
if (!mock_mode && !disable_flock && meta_device != data_device && flock(meta_fd, LOCK_EX|LOCK_NB) != 0)
{
throw std::runtime_error(std::string("Failed to lock metadata device: ") + strerror(errno));
}
}
else
{
meta_fd = data_fd;
meta_disk = data_disk;
meta_device_sect = data_device_sect;
meta_device_size = 0;
if (meta_offset >= data_device_size)
@@ -433,6 +412,7 @@ void blockstore_disk_t::open_meta()
throw std::runtime_error("meta_offset exceeds device size = "+std::to_string(data_device_size));
}
}
meta_off_ring = meta_disk->is_off_ring();
if (meta_block_size % meta_device_sect)
{
throw std::runtime_error(
@@ -444,33 +424,29 @@ void blockstore_disk_t::open_meta()
void blockstore_disk_t::open_journal()
{
if (journal_fd >= 0)
if (journal_disk)
{
throw std::runtime_error("journal device is already opened");
}
if (journal_device != meta_device || journal_io != meta_io)
{
journal_fd = mock_mode ? MOCK_JOURNAL_FD : open(journal_device.c_str(), bs_openmode(journal_io) | O_RDWR);
if (journal_fd == -1)
journal_disk = mock_mode
? disk_i::create_mock(MOCK_JOURNAL_FD, journal_device_size, journal_device_sect)
: disk_i::create_and_open(journal_device, journal_io, journal_device == meta_device || disable_flock);
if (!journal_disk)
{
throw std::runtime_error("Failed to open journal device "+journal_device+": "+std::string(strerror(errno)));
}
if (!mock_mode)
{
check_size(journal_fd, &journal_device_size, &journal_device_sect, "journal device");
throw std::runtime_error("Failed to open journal device "+journal_device);
}
journal_device_size = journal_disk->get_size();
journal_device_sect = journal_disk->get_block_size();
if (journal_offset >= journal_device_size)
{
throw std::runtime_error("journal_offset exceeds device size = "+std::to_string(journal_device_size));
}
if (!mock_mode && !disable_flock && journal_device != meta_device && flock(journal_fd, LOCK_EX|LOCK_NB) != 0)
{
throw std::runtime_error(std::string("Failed to lock journal device: ") + strerror(errno));
}
}
else
{
journal_fd = meta_fd;
journal_disk = meta_disk;
journal_device_sect = meta_device_sect;
journal_device_size = 0;
if (journal_offset >= (meta_device_size > 0 ? meta_device_size : data_device_size))
@@ -478,6 +454,7 @@ void blockstore_disk_t::open_journal()
throw std::runtime_error("journal_offset exceeds device size");
}
}
journal_off_ring = journal_disk->is_off_ring();
if (journal_block_size % journal_device_sect)
{
throw std::runtime_error(
@@ -489,26 +466,19 @@ void blockstore_disk_t::open_journal()
void blockstore_disk_t::close_all()
{
if (!mock_mode)
{
if (data_fd >= 0)
close(data_fd);
if (meta_fd >= 0 && meta_fd != data_fd)
close(meta_fd);
if (journal_fd >= 0 && journal_fd != meta_fd)
close(journal_fd);
}
data_fd = meta_fd = journal_fd = -1;
if (journal_disk && journal_disk != meta_disk && journal_disk != data_disk)
delete journal_disk;
if (meta_disk && meta_disk != data_disk)
delete meta_disk;
if (data_disk)
delete data_disk;
data_disk = meta_disk = journal_disk = NULL;
}
// Sadly DISCARD only works through ioctl(), but it seems to always block the device queue,
// so it's not a big deal that we can only run it synchronously.
int blockstore_disk_t::trim_data(std::function<bool(uint64_t)> is_used)
{
if (mock_mode)
{
return -EINVAL;
}
int r = 0;
uint64_t j = 0, i = 0;
uint64_t discarded = 0;
@@ -533,12 +503,12 @@ int blockstore_disk_t::trim_data(std::function<bool(uint64_t)> is_used)
}
if (range[1] > 0)
{
r = ioctl(data_fd, BLKDISCARD, &range);
r = data_disk->discard_sync(range[0], range[1]);
if (r != 0)
{
fprintf(stderr, "Failed to execute BLKDISCARD %ju+%ju on %s: %s (code %d)\n",
fprintf(stderr, "Failed to discard %ju+%ju bytes on %s: %s (code %d)\n",
range[0], range[1], data_device.c_str(), strerror(-r), r);
return -errno;
return r;
}
discarded += range[1];
}

View File

@@ -24,6 +24,8 @@
class allocator_t;
class disk_i;
struct blockstore_disk_t
{
std::string data_device, meta_device, journal_device;
@@ -66,6 +68,8 @@ struct blockstore_disk_t
uint64_t discard_granularity = 0;
int meta_fd = -1, data_fd = -1, journal_fd = -1;
disk_i *meta_disk = NULL, *data_disk = NULL, *journal_disk = NULL;
bool meta_off_ring = false, data_off_ring = false, journal_off_ring = false;
uint64_t meta_offset = 0, meta_device_sect = 0, meta_device_size = 0, meta_area_size = 0, min_meta_len = 0;
uint64_t data_offset = 0, data_device_sect = 0, data_device_size = 0, data_len = 0;
uint64_t journal_offset = 0, journal_device_sect = 0, journal_device_size = 0, journal_len = 0;

View File

@@ -37,18 +37,18 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs)
journal_flusher_co::journal_flusher_co()
{
wait_state = 0;
simple_callback_r = [this](ring_data_t* data)
simple_callback_r = [this](int res)
{
bs->live = true;
if (data->res != data->iov.iov_len)
bs->disk_error_abort("read operation during flush", data->res, data->iov.iov_len);
if (res < 0)
bs->disk_error_abort("read operation during flush", res);
wait_count--;
};
simple_callback_w = [this](ring_data_t* data)
simple_callback_w = [this](int res)
{
bs->live = true;
if (data->res != data->iov.iov_len)
bs->disk_error_abort("write operation during flush", data->res, data->iov.iov_len);
if (res < 0)
bs->disk_error_abort("write operation during flush", res);
wait_count--;
};
}
@@ -161,19 +161,8 @@ void journal_flusher_t::loop()
bs->ringloop->wakeup();
}
#define await_sqe(label) \
resume_##label:\
sqe = bs->get_sqe();\
if (!sqe)\
{\
wait_state = wait_base+label;\
return false;\
}\
data = ((ring_data_t*)sqe->user_data);
bool journal_flusher_co::loop()
{
int wait_base = 0;
// This is much better than implementing the whole function as an FSM
// Maybe I should consider a coroutine library like https://github.com/hnes/libaco ...
// Or just C++ coroutines, but they require some wrappers
@@ -213,18 +202,25 @@ resume_0:
res = bs->heap->get_next_compact(cur_oid);
if ((bs->intent_write_counter >= bs->journal_trim_interval || flusher->force_fsync > 0) && co_id == 0)
{
// Advance fsynced_lsn every <journal_trim_interval> intent writes
bs->intent_write_counter = 0;
resume_17:
resume_18:
resume_19:
if (!fsync_buffer(17))
{
return false;
}
if (bs->intent_write_counter >= bs->journal_trim_interval)
{
// Advance fsynced_lsn every <journal_trim_interval> intent writes
bs->intent_write_counter = 0;
resume_29:
resume_30:
resume_31:
if (!trim_lsn(29))
return false;
if (!trim_lsn(29))
{
return false;
}
}
}
if (res == ENOENT && flusher->force_start > 0 && co_id == 0 &&
(!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync || !bs->dsk.disable_data_fsync))
@@ -458,13 +454,22 @@ resume_9:
if ((read_vec[i].copy_flags & COPY_BUF_JOURNAL) &&
!(read_vec[i].copy_flags & COPY_BUF_COALESCED))
{
resume_10:
assert(read_vec[i].buf);
await_sqe(10);
data->iov = (struct iovec){ read_vec[i].buf + (read_vec[i].copy_flags & COPY_BUF_PADDED
? read_vec[i].offset - read_vec[i].disk_offset : 0), (size_t)read_vec[i].len };
data->callback = simple_callback_w;
assert(clean_loc + read_vec[i].offset + data->iov.iov_len <= bs->dsk.block_count*bs->dsk.data_block_size);
io_uring_prep_writev(sqe, bs->dsk.data_fd, &data->iov, 1, bs->dsk.data_offset + clean_loc + read_vec[i].offset);
assert(clean_loc + read_vec[i].offset + read_vec[i].len <= bs->dsk.block_count*bs->dsk.data_block_size);
res = bs->dsk.data_disk->write_async(
bs->dsk.data_offset + clean_loc + read_vec[i].offset,
(size_t)read_vec[i].len,
read_vec[i].buf + (read_vec[i].copy_flags & COPY_BUF_PADDED
? read_vec[i].offset - read_vec[i].disk_offset : 0),
simple_callback_w
);
if (res == -EAGAIN)
{
wait_state = 10;
return false;
}
assert(res == 0);
wait_count++;
}
}
@@ -572,7 +577,7 @@ void journal_flusher_co::fill_partial_checksum_blocks()
.disk_loc = clean_loc,
.disk_offset = blk_begin,
.disk_len = blk_end - blk_begin,
.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, blk_end - blk_begin),
.buf = (uint8_t*)bs->dsk.data_disk->alloc_iomem(blk_end - blk_begin),
.wr = compact_info.clean_wr,
});
}
@@ -597,7 +602,8 @@ void journal_flusher_co::free_buffers()
if (it->buf && !(it->copy_flags & COPY_BUF_COALESCED) &&
(!bs->dsk.inmemory_journal || it->buf < bs->buffer_area || it->buf >= (uint8_t*)bs->buffer_area + bs->dsk.journal_len))
{
free(it->buf);
disk_i *di = ((it->copy_flags & COPY_BUF_JOURNAL) ? bs->dsk.journal_disk : bs->dsk.data_disk);
di->free_iomem(it->buf);
}
}
read_vec.clear();
@@ -748,6 +754,8 @@ resume_1:
bool journal_flusher_co::read_buffered(int wait_base)
{
disk_i *di = NULL;
copy_buffer_t *vec = NULL;
if (wait_state == wait_base)
goto resume_0;
else if (wait_state == wait_base+1)
@@ -765,19 +773,24 @@ bool journal_flusher_co::read_buffered(int wait_base)
if (!(read_vec[i].copy_flags & COPY_BUF_COALESCED) &&
((read_vec[i].copy_flags & COPY_BUF_JOURNAL) && !bs->dsk.inmemory_journal || (read_vec[i].copy_flags & COPY_BUF_DATA)))
{
await_sqe(0);
auto & vec = read_vec[i];
if (!vec.buf)
vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len);
data->iov = (struct iovec){ vec.buf, (size_t)vec.disk_len };
wait_count++;
io_uring_prep_readv(
sqe,
(vec.copy_flags & COPY_BUF_JOURNAL) ? bs->dsk.journal_fd : bs->dsk.data_fd,
&data->iov, 1,
((vec.copy_flags & COPY_BUF_JOURNAL) ? bs->dsk.journal_offset : bs->dsk.data_offset) + vec.disk_loc + vec.disk_offset
resume_0:
vec = &read_vec[i];
di = ((vec->copy_flags & COPY_BUF_JOURNAL) ? bs->dsk.journal_disk : bs->dsk.data_disk);
if (!vec->buf)
vec->buf = (uint8_t*)di->alloc_iomem(vec->disk_len);
res = di->read_async(
((vec->copy_flags & COPY_BUF_JOURNAL) ? bs->dsk.journal_offset : bs->dsk.data_offset) + vec->disk_loc + vec->disk_offset,
(size_t)vec->disk_len,
vec->buf,
simple_callback_r
);
data->callback = simple_callback_r;
if (res == -EAGAIN)
{
wait_state = wait_base+0;
return false;
}
assert(!res);
wait_count++;
}
}
// Wait for reads/writes if the journal is not inmemory
@@ -820,18 +833,29 @@ resume_0:
if (!cur_sync->sent)
{
// Sync batch is ready. Do it.
await_sqe(1);
data->iov = { 0 };
data->callback = simple_callback_w;
io_uring_prep_fsync(sqe, bs->dsk.data_fd, IORING_FSYNC_DATASYNC);
cur_sync->sent = true;
wait_count++;
resume_2:
if (wait_count > 0)
resume_1:
res = bs->dsk.data_disk->flush_async(simple_callback_w);
if (res == -EAGAIN)
{
wait_state = wait_base+2;
wait_state = wait_base+1;
return false;
}
else if (res == -EALREADY)
{
cur_sync->sent = true;
}
else
{
assert(res == 0);
cur_sync->sent = true;
wait_count++;
resume_2:
if (wait_count > 0)
{
wait_state = wait_base+2;
return false;
}
}
cur_sync->done = true;
// Wake up other flushers
bs->ringloop->wakeup();
@@ -868,17 +892,27 @@ resume_0:
}
flusher->fsyncing_meta = true;
// Sync batch is ready. Do it.
await_sqe(1);
data->iov = { 0 };
data->callback = simple_callback_w;
io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
wait_count++;
resume_2:
if (wait_count > 0)
resume_1:
res = bs->dsk.meta_disk->flush_async(simple_callback_w);
if (res == -EAGAIN)
{
wait_state = wait_base+2;
wait_state = wait_base+1;
return false;
}
else if (res == -EALREADY)
{
}
else
{
assert(res == 0);
wait_count++;
resume_2:
if (wait_count > 0)
{
wait_state = wait_base+2;
return false;
}
}
// Sync completed. All previous coroutines waiting for it must be resumed
flusher->fsyncing_meta = false;
bs->ringloop->wakeup();
@@ -905,7 +939,7 @@ resume_0:
resume_1:
assert(!wait_count);
fsynced_lsn = bs->heap->get_completed_lsn();
if (!bs->submit_fsyncs(wait_count))
if (bs->submit_fsyncs(wait_count) != 0)
{
wait_state = wait_base+1;
return false;
@@ -944,12 +978,16 @@ resume_0:
flusher->active_flushers++;
((blockstore_meta_header_v3_t*)bs->meta_superblock)->completed_lsn = fsynced_lsn;
((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c();
await_sqe(1);
data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
data->callback = simple_callback_w;
io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
// Update superblock with datasync
sqe->rw_flags = RWF_DSYNC;
// Update superblock with FUA (datasync)
resume_1:
res = bs->dsk.meta_disk->write_async(bs->dsk.meta_offset, bs->dsk.meta_block_size,
bs->meta_superblock, simple_callback_w, true);
if (res == -EAGAIN)
{
wait_state = wait_base+1;
return false;
}
assert(res == 0);
wait_count++;
resume_2:
if (wait_count > 0)

View File

@@ -43,14 +43,12 @@ class journal_flusher_co
journal_flusher_t *flusher;
int co_id;
int wait_state, wait_count;
struct io_uring_sqe *sqe;
struct ring_data_t *data;
uint8_t *new_csums = NULL;
uint8_t *new_bmp = NULL;
uint8_t *punch_bmp = NULL;
uint8_t *new_ext_bmp = NULL;
std::function<void(ring_data_t*)> simple_callback_r, simple_callback_w;
std::function<void(int)> simple_callback_r, simple_callback_w;
object_id cur_oid;
heap_entry_t *cur_obj;

View File

@@ -422,7 +422,7 @@ corrupted_object:
if ((wr->type() == BS_HEAP_BIG_INTENT || wr->type() == BS_HEAP_BIG_WRITE) &&
wr->big().block_num >= dsk->block_count)
{
fprintf(stderr, "Error: big_write or big_intent entry %jx:%jx v%ju block_num is too large: %u > %lu. Metadata is incompatible with current parameters. ",
fprintf(stderr, "Error: big_write or big_intent entry %jx:%jx v%ju block_num is too large: %u > %ju. Metadata is incompatible with current parameters. ",
wr->inode, wr->stripe, wr->version, wr->big_intent().block_num, dsk->block_count);
goto corrupted_object;
}
@@ -2436,9 +2436,11 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
res.do_delete = false;
res.compact_lsn = 0;
res.compact_version = 0;
if (!under_pressure && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_DELETE))
if ((!under_pressure || stable) && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT || wr->type() == BS_HEAP_DELETE))
{
// We may postpone compaction if we have an unstable overwrite when not under pressure
// We MUST postpone if the overwrite is STABLE (even if under pressure),
// because add_compact will reject STABLE overwrites with EBUSY.
return res;
}
continue;

View File

@@ -30,12 +30,15 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
dsk.close_all();
throw;
}
meta_superblock = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
dsk.data_disk->set_ringloop(ringloop);
dsk.meta_disk->set_ringloop(ringloop);
dsk.journal_disk->set_ringloop(ringloop);
meta_superblock = (uint8_t*)dsk.meta_disk->alloc_iomem(dsk.meta_block_size);
memset(meta_superblock, 0, dsk.meta_block_size);
flusher = new journal_flusher_t(this);
if (dsk.inmemory_journal)
{
buffer_area = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.journal_len);
buffer_area = (uint8_t*)dsk.journal_disk->alloc_iomem(dsk.journal_len);
}
heap = new blockstore_heap_t(&dsk, buffer_area, log_level);
ringloop->wakeup();
@@ -66,9 +69,9 @@ blockstore_impl_t::~blockstore_impl_t()
if (heap)
delete heap;
if (buffer_area)
free(buffer_area);
dsk.journal_disk->free_iomem(buffer_area);
if (meta_superblock)
free(meta_superblock);
dsk.meta_disk->free_iomem(meta_superblock);
ringloop->unregister_consumer(&ring_consumer);
dsk.close_all();
}
@@ -127,9 +130,9 @@ void blockstore_impl_t::loop()
if (PRIV(op)->wait_for)
{
check_wait(op);
if (PRIV(op)->wait_for == WAIT_SQE)
if (PRIV(op)->wait_for != WAIT_COMPACTION)
{
// ring is full, stop submission
// disk queue(s) are full, stop submission
break;
}
else if (PRIV(op)->wait_for)
@@ -140,7 +143,6 @@ void blockstore_impl_t::loop()
continue;
}
}
unsigned prev_sqe_pos = ringloop->save();
// 0 = can't submit
// 1 = in progress
// 2 = can be removed from queue
@@ -189,8 +191,7 @@ void blockstore_impl_t::loop()
if (wr_st == 0)
{
PRIV(op)->pending_ops = 0;
ringloop->restore(prev_sqe_pos);
if (PRIV(op)->wait_for == WAIT_SQE)
if (PRIV(op)->wait_for != WAIT_COMPACTION)
{
// ring is full, stop submission
break;
@@ -261,13 +262,34 @@ bool blockstore_impl_t::is_safe_to_stop()
void blockstore_impl_t::check_wait(blockstore_op_t *op)
{
if (PRIV(op)->wait_for == WAIT_SQE)
if (PRIV(op)->wait_for == WAIT_DATA_QUEUE)
{
if (ringloop->space_left() < PRIV(op)->wait_detail)
if (dsk.data_disk->get_queue_space() < PRIV(op)->wait_detail)
{
// stop submission if there's still no free space
#ifdef BLOCKSTORE_DEBUG
printf("Still waiting for %ju SQE(s)\n", PRIV(op)->wait_detail);
printf("Still waiting for %ju free data disk queue slots\n", PRIV(op)->wait_detail);
#endif
return;
}
PRIV(op)->wait_for = 0;
}
else if (PRIV(op)->wait_for == WAIT_META_QUEUE)
{
if (dsk.meta_disk->get_queue_space() < PRIV(op)->wait_detail)
{
#ifdef BLOCKSTORE_DEBUG
printf("Still waiting for %ju free meta disk queue slots\n", PRIV(op)->wait_detail);
#endif
return;
}
PRIV(op)->wait_for = 0;
}
else if (PRIV(op)->wait_for == WAIT_JOURNAL_QUEUE)
{
if (dsk.journal_disk->get_queue_space() < PRIV(op)->wait_detail)
{
#ifdef BLOCKSTORE_DEBUG
printf("Still waiting for %ju free journal disk queue slots\n", PRIV(op)->wait_detail);
#endif
return;
}
@@ -404,7 +426,7 @@ void blockstore_meta_header_v3_t::set_crc32c()
header_csum = calc;
}
void blockstore_impl_t::disk_error_abort(const char *op, int retval, int expected)
void blockstore_impl_t::disk_error_abort(const char *op, int retval)
{
if (retval == -EAGAIN)
{
@@ -413,7 +435,7 @@ void blockstore_impl_t::disk_error_abort(const char *op, int retval, int expecte
" Please upgrade your kernel. Aborting.\n", op);
exit(1);
}
fprintf(stderr, "Disk %s failed: result is %d, expected %d. Can't continue, sorry :-(\n", op, retval, expected);
fprintf(stderr, "Disk %s failed: %s (code %d). Can't continue, sorry :-(\n", op, strerror(-retval), retval);
exit(1);
}

View File

@@ -20,6 +20,7 @@
#include <unordered_map>
#include <unordered_set>
#include "disk_util.h"
#include "malloc_or_die.h"
class blockstore_impl_t;
@@ -132,21 +133,16 @@ public:
bool stop_sync_submitted = false;
inline struct io_uring_sqe* get_sqe()
{
return ringloop->get_sqe();
}
void open_data();
void open_meta();
void open_journal();
void disk_error_abort(const char *op, int retval, int expected);
void disk_error_abort(const char *op, int retval);
// Asynchronous init
int initialized;
int metadata_buf_size;
blockstore_init_meta* metadata_init_reader;
int initialized = 0;
uint64_t metadata_buf_size = 0;
blockstore_init_meta* metadata_init_reader = NULL;
void check_wait(blockstore_op_t *op);
void init_op(blockstore_op_t *op);
@@ -163,7 +159,7 @@ public:
void find_holes(std::vector<copy_buffer_t> & read_vec, uint32_t item_start, uint32_t item_end,
std::function<void(int&, uint32_t, uint32_t)> callback);
void free_read_buffers(std::vector<copy_buffer_t> & rv);
void handle_read_event(ring_data_t *data, blockstore_op_t *op);
void handle_read_event(int res, blockstore_op_t *op);
bool verify_read_checksums(blockstore_op_t *op);
// Write
@@ -173,11 +169,11 @@ public:
bool intent_write_allowed(blockstore_op_t *op, heap_entry_t *obj);
int dequeue_write(blockstore_op_t *op);
int continue_write(blockstore_op_t *op);
void handle_write_event(ring_data_t *data, blockstore_op_t *op);
void handle_write_event(int res, blockstore_op_t *op);
// Sync
int continue_sync(blockstore_op_t *op);
bool submit_fsyncs(int & wait_count);
int submit_fsyncs(int & wait_count);
int do_sync(blockstore_op_t *op, int base_state);
bool has_unsynced();

View File

@@ -11,12 +11,6 @@
#define INIT_META_READING 1
#define INIT_META_READ_DONE 2
#define GET_SQE() \
sqe = bs->get_sqe();\
if (!sqe)\
throw std::runtime_error("io_uring is full during initialization");\
data = ((ring_data_t*)sqe->user_data)
struct blockstore_recheck_io_t
{
bool is_data;
@@ -36,14 +30,13 @@ blockstore_init_meta::blockstore_init_meta(blockstore_impl_t *bs)
this->bs = bs;
}
void blockstore_init_meta::handle_event(ring_data_t *data, int buf_num, const char *op)
void blockstore_init_meta::handle_event(int res, int buf_num, const char *op)
{
if (data->res != data->iov.iov_len)
if (res < 0)
{
throw std::runtime_error(strprintf(
"%s failed at offset %ju: got %s (code %d), but expected %zu",
op, (buf_num >= 0 ? bufs[buf_num].offset : last_read_offset), strerror(-data->res),
data->res, data->iov.iov_len
"%s failed at offset %ju: %s (code %d)",
op, (buf_num >= 0 ? bufs[buf_num].offset : last_read_offset), strerror(-res), res
));
}
if (buf_num >= 0)
@@ -67,16 +60,16 @@ int blockstore_init_meta::loop()
else if (wait_state == 7) goto resume_7;
else if (wait_state == 8) goto resume_8;
else if (wait_state == 9) goto resume_9;
metadata_buffer = memalign(MEM_ALIGNMENT, 2*bs->metadata_buf_size);
if (bs->metadata_buf_size > bs->dsk.meta_disk->get_max_io_size())
bs->metadata_buf_size = bs->dsk.meta_disk->get_max_io_size();
metadata_buffer = bs->dsk.meta_disk->alloc_iomem(2*bs->metadata_buf_size);
if (!metadata_buffer)
throw std::runtime_error("Failed to allocate metadata read buffer");
// Read metadata superblock
GET_SQE();
last_read_offset = 0;
data->iov = { bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
data->callback = [this](ring_data_t *data) { handle_event(data, -1, "read metadata header"); };
io_uring_prep_readv(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
bs->ringloop->submit();
res = bs->dsk.meta_disk->read_async(bs->dsk.meta_offset, bs->dsk.meta_block_size, bs->meta_superblock,
[this](int res) { handle_event(res, -1, "read metadata header"); });
assert(res == 0);
submitted++;
resume_1:
if (submitted > 0)
@@ -165,16 +158,16 @@ resume_1:
md_offset = 0;
while (md_offset < bs->dsk.journal_len)
{
GET_SQE();
data->iov = (iovec){
submit_len = (bs->dsk.journal_len - md_offset < bs->metadata_buf_size ? bs->dsk.journal_len - md_offset : bs->metadata_buf_size);
last_read_offset = md_offset;
res = bs->dsk.journal_disk->read_async(
bs->dsk.journal_offset + md_offset, (size_t)submit_len,
bs->buffer_area + md_offset,
(size_t)(bs->dsk.journal_len - md_offset < bs->metadata_buf_size ? bs->dsk.journal_len - md_offset : bs->metadata_buf_size),
};
data->callback = [this](ring_data_t *data) { handle_event(data, -1, "read buffer area"); };
io_uring_prep_readv(sqe, bs->dsk.journal_fd, &data->iov, 1, bs->dsk.journal_offset + md_offset);
md_offset += data->iov.iov_len;
[this](int res) { handle_event(res, -1, "read buffer area"); }
);
assert(res == 0);
md_offset += submit_len;
submitted++;
bs->ringloop->submit();
resume_3:
if (submitted > 0)
{
@@ -201,26 +194,31 @@ resume_4:
bufs[i].size = bs->dsk.meta_area_size-next_offset > bs->metadata_buf_size
? bs->metadata_buf_size : bs->dsk.meta_area_size-next_offset;
bufs[i].state = INIT_META_READING;
submitted++;
next_offset += bufs[i].size;
GET_SQE();
assert(bufs[i].size <= 0x7fffffff);
data->iov = { bufs[i].buf, (size_t)bufs[i].size };
if (!zero_on_init)
{
data->callback = [this, i](ring_data_t *data) { handle_event(data, i, "read metadata"); };
io_uring_prep_readv(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + bufs[i].offset);
res = bs->dsk.meta_disk->read_async(
bs->dsk.meta_offset + bufs[i].offset,
(size_t)bufs[i].size, bufs[i].buf,
[this, i](int res) { handle_event(res, i, "read metadata"); }
);
assert(res == 0);
}
else
{
// Fill metadata with empty block pattern
data->callback = [this, i](ring_data_t *data) { handle_event(data, i, "clear metadata"); };
memset(bufs[i].buf, 0, bufs[i].size);
for (uint64_t o = 0; o < bufs[i].size; o += bs->dsk.meta_block_size)
bs->heap->fill_block_empty_space(bufs[i].buf + o, 0);
io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset + bufs[i].offset);
res = bs->dsk.meta_disk->write_async(
bs->dsk.meta_offset + bufs[i].offset,
(size_t)bufs[i].size, bufs[i].buf,
[this, i](int res) { handle_event(res, i, "clear metadata"); }
);
assert(res == 0);
}
bs->ringloop->submit();
submitted++;
next_offset += bufs[i].size;
break;
}
}
@@ -266,26 +264,27 @@ resume_6:
while (recheck_ios.size())
{
blockstore_recheck_io_t* rc = recheck_ios.back();
sqe = bs->get_sqe();
if (!sqe)
break;
disk_i *di = (rc->is_data ? bs->dsk.data_disk : bs->dsk.journal_disk);
if (!rc->buf)
rc->buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, rc->len);
data = ((ring_data_t*)sqe->user_data);
data->iov = (iovec){ rc->buf, rc->len };
data->callback = [this, rc](ring_data_t *data)
{
if (data->res < 0)
rc->buf = (uint8_t*)di->alloc_iomem(rc->len);
int res = di->read_async(
(rc->is_data ? bs->dsk.data_offset : bs->dsk.journal_offset) + rc->offset,
rc->len, rc->buf, [this, rc](int res)
{
fprintf(stderr, "Buffer area read failed at offset %ju: %d\n", rc->offset, data->res);
exit(1);
if (res < 0)
{
fprintf(stderr, "Buffer area read failed at offset %ju: %d\n", rc->offset, res);
exit(1);
}
rc->cb(rc->buf);
disk_i *di = (rc->is_data ? bs->dsk.data_disk : bs->dsk.journal_disk);
di->free_iomem(rc->buf);
delete rc;
}
rc->cb(rc->buf);
free(rc->buf);
delete rc;
};
io_uring_prep_readv(sqe, (rc->is_data ? bs->dsk.data_fd : bs->dsk.journal_fd), &data->iov, 1,
(rc->is_data ? bs->dsk.data_offset : bs->dsk.journal_offset) + rc->offset);
);
if (res == -EAGAIN)
break;
assert(res == 0);
recheck_ios.pop_back();
}
return 1;
@@ -307,30 +306,35 @@ resume_7:
for (i = 0; i < recheck_mod.size(); i++)
{
resume_8:
if (wait_count >= bs->meta_write_recheck_parallelism || !(sqe = bs->get_sqe()))
if (wait_count >= bs->meta_write_recheck_parallelism)
{
bs->ringloop->submit();
wait_state = 8;
return 1;
}
uint32_t block_num = recheck_mod[i];
uint64_t block_offset = bs->dsk.meta_offset + (uint64_t)(block_num+1) * bs->dsk.meta_block_size;
data = ((ring_data_t*)sqe->user_data);
uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, bs->dsk.meta_block_size);
uint8_t *buf = (uint8_t*)bs->dsk.meta_disk->alloc_iomem(bs->dsk.meta_block_size);
bs->heap->get_meta_block(block_num, buf);
data->iov = { buf, bs->dsk.meta_block_size };
data->callback = [this, buf, block_offset](ring_data_t *data)
{
wait_count--;
free(buf);
if (data->res != bs->dsk.meta_block_size)
res = bs->dsk.meta_disk->write_async(block_offset, bs->dsk.meta_block_size, buf,
[this, buf, block_offset](int res)
{
throw std::runtime_error(
"write metadata failed at offset " + std::to_string(block_offset) + ": " + strerror(-data->res)
);
wait_count--;
bs->dsk.meta_disk->free_iomem(buf);
if (res < 0)
{
throw std::runtime_error(
"write metadata failed at offset " + std::to_string(block_offset) + ": " + strerror(-res)
);
}
}
};
io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, block_offset);
);
if (res == -EAGAIN)
{
bs->dsk.meta_disk->free_iomem(buf);
wait_state = 8;
return 1;
}
assert(!res);
wait_count++;
}
resume_9:
@@ -340,18 +344,18 @@ resume_9:
wait_state = 9;
return 1;
}
free(metadata_buffer);
bs->dsk.meta_disk->free_iomem(metadata_buffer);
metadata_buffer = NULL;
do_fsync:
if (!bs->dsk.disable_meta_fsync && !bs->readonly)
{
GET_SQE();
io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
last_read_offset = 0;
data->iov = { 0 };
data->callback = [this](ring_data_t *data) { handle_event(data, -1, "fsync metadata"); };
submitted++;
bs->ringloop->submit();
res = bs->dsk.meta_disk->flush_async([this](int res) { handle_event(res, -1, "fsync metadata"); });
if (res != -EALREADY)
{
assert(res == 0);
submitted++;
}
resume_5:
if (submitted > 0)
{
@@ -361,13 +365,13 @@ do_fsync:
}
if (zero_on_init && !header_written && !bs->readonly)
{
GET_SQE();
header_written = true;
last_read_offset = 0;
data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
data->callback = [this](ring_data_t *data) { handle_event(data, -1, "write metadata header"); };
io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
bs->ringloop->submit();
res = bs->dsk.meta_disk->write_async(
bs->dsk.meta_offset, (size_t)bs->dsk.meta_block_size, bs->meta_superblock,
[this](int res) { handle_event(res, -1, "write metadata header"); }
);
assert(res == 0);
header_written = true;
submitted++;
resume_2:
if (submitted > 0)

View File

@@ -21,19 +21,19 @@ class blockstore_init_meta
bool zero_on_init = false;
bool header_written = false;
void *metadata_buffer = NULL;
int res = 0;
blockstore_init_meta_buf bufs[2] = {};
int submitted = 0;
struct io_uring_sqe *sqe;
struct ring_data_t *data;
uint64_t md_offset = 0;
uint64_t next_offset = 0;
uint64_t last_read_offset = 0;
uint64_t entries_loaded = 0;
uint64_t submit_len = 0;
std::vector<uint32_t> recheck_mod;
std::vector<blockstore_recheck_io_t*> recheck_ios;
int i = 0, j = 0;
bool handle_meta_block(uint8_t *buf, uint64_t count, uint64_t done_cnt);
void handle_event(ring_data_t *data, int buf_num, const char *op);
void handle_event(int res, int buf_num, const char *op);
public:
blockstore_init_meta(blockstore_impl_t *bs);
int loop();

View File

@@ -3,44 +3,40 @@
#pragma once
#define BS_SUBMIT_CHECK_SQES(n) \
if (ringloop->space_left() < (n))\
#define BS_SUBMIT_CHECK_JOURNAL_QUEUE(n) {\
if (dsk.journal_disk->get_queue_space() < (n))\
{\
/* Pause until there are more requests available */\
PRIV(op)->wait_detail = (n);\
PRIV(op)->wait_for = WAIT_SQE;\
PRIV(op)->wait_for = WAIT_JOURNAL_QUEUE;\
return 0;\
}
} }
#define BS_SUBMIT_GET_SQE(sqe, data) \
BS_SUBMIT_GET_ONLY_SQE(sqe); \
struct ring_data_t *data = ((ring_data_t*)sqe->user_data)
#define BS_SUBMIT_GET_ONLY_SQE(sqe) \
struct io_uring_sqe *sqe = get_sqe();\
if (!sqe)\
#define BS_SUBMIT_CHECK_DATA_QUEUE(n) {\
if (dsk.data_disk->get_queue_space() < (n))\
{\
/* Pause until there are more requests available */\
PRIV(op)->wait_detail = 1;\
PRIV(op)->wait_for = WAIT_SQE;\
PRIV(op)->wait_detail = (n);\
PRIV(op)->wait_for = WAIT_DATA_QUEUE;\
return 0;\
}
} }
#define BS_SUBMIT_GET_SQE_DECL(sqe) \
sqe = get_sqe();\
if (!sqe)\
#define BS_SUBMIT_CHECK_META_QUEUE(n) {\
if (dsk.meta_disk->get_queue_space() < (n))\
{\
/* Pause until there are more requests available */\
PRIV(op)->wait_detail = 1;\
PRIV(op)->wait_for = WAIT_SQE;\
PRIV(op)->wait_detail = (n);\
PRIV(op)->wait_for = WAIT_META_QUEUE;\
return 0;\
}
} }
#define PRIV(op) ((blockstore_op_private_t*)(op)->private_data)
#define FINISH_OP(op) PRIV(op)->~blockstore_op_private_t(); std::function<void (blockstore_op_t*)>(op->callback)(op)
// Suspend operation until there are more free SQEs
#define WAIT_SQE 1
#define WAIT_DATA_QUEUE 1
#define WAIT_META_QUEUE 2
#define WAIT_JOURNAL_QUEUE 3
// Suspend operation until there are <wait_detail> bytes of free space in the journal on disk
#define WAIT_COMPACTION 2
// Suspend operation until LSN <wait_detail> becomes durable
@@ -65,11 +61,3 @@
#define COPY_BUF_COALESCED 0x10
#define COPY_BUF_PADDED 0x20
#define COPY_BUF_SKIP_CSUM 0x40
#ifndef RWF_ATOMIC
#define RWF_ATOMIC 0x40
#endif
#ifndef RWF_DSYNC
#define RWF_DSYNC 0x02
#endif

View File

@@ -88,7 +88,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
if (need_wait)
{
undo_wait:
// Need to wait. undo added requests, unlock lsn
// Unlock LSN and wait
heap->unlock_entry(op->oid);
free_read_buffers(rv);
rv.clear();
@@ -106,12 +106,13 @@ undo_wait:
assert(fulfilled == op->len);
if (!fulfill_read(op))
{
// Not enough SQEs. Unlock lsn and wait
goto undo_wait;
}
op->version = result_version;
if (!PRIV(op)->pending_ops)
{
// everything is fulfilled from memory
// Everything is fulfilled from memory
heap->unlock_entry(op->oid);
op->retval = op->len;
free_read_buffers(rv);
@@ -125,6 +126,24 @@ undo_wait:
int blockstore_impl_t::fulfill_read(blockstore_op_t *op)
{
int ndata = 0, njournal = 0;
uint32_t nodsk = (COPY_BUF_COALESCED|COPY_BUF_CSUM_FILL|COPY_BUF_ZERO) |
(dsk.inmemory_journal ? COPY_BUF_JOURNAL : 0);
for (auto & vec: PRIV(op)->read_vec)
{
if (!(vec.copy_flags & nodsk))
{
auto & n = (vec.copy_flags & COPY_BUF_JOURNAL) ? njournal : ndata;
n++;
}
}
if (!dsk.data_off_ring && !dsk.journal_off_ring || dsk.data_disk == dsk.journal_disk)
{
ndata += njournal;
njournal = 0;
}
BS_SUBMIT_CHECK_DATA_QUEUE(ndata);
BS_SUBMIT_CHECK_JOURNAL_QUEUE(njournal);
for (auto & vec: PRIV(op)->read_vec)
{
if (vec.copy_flags & (COPY_BUF_COALESCED|COPY_BUF_CSUM_FILL))
@@ -142,16 +161,14 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op)
}
else
{
BS_SUBMIT_GET_SQE(sqe, data);
data->iov = (struct iovec){ vec.buf ? vec.buf : (op->buf + vec.offset - op->offset), (size_t)vec.disk_len };
PRIV(op)->pending_ops++;
io_uring_prep_readv(
sqe,
(vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_fd : dsk.data_fd,
&data->iov, 1,
((vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_offset : dsk.data_offset) + vec.disk_loc + vec.disk_offset
disk_i *di = ((vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_disk : dsk.data_disk);
int res = di->read_async(
((vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_offset : dsk.data_offset) + vec.disk_loc + vec.disk_offset,
vec.disk_len, vec.buf ? vec.buf : (op->buf + vec.offset - op->offset),
[this, op](int res) { handle_read_event(res, op); }
);
data->callback = [this, op](ring_data_t *data) { handle_read_event(data, op); };
assert(res == 0);
PRIV(op)->pending_ops++;
}
}
return 1;
@@ -332,7 +349,7 @@ void blockstore_impl_t::prepare_disk_read(std::vector<copy_buffer_t> & read_vec,
}
else
{
vec.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, vec.disk_len);
vec.buf = (uint8_t*)((vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_disk : dsk.data_disk)->alloc_iomem(vec.disk_len);
}
}
read_vec.insert(read_vec.begin() + pos, vec);
@@ -383,21 +400,22 @@ void blockstore_impl_t::free_read_buffers(std::vector<copy_buffer_t> & rv)
if (!(vec.copy_flags & COPY_BUF_COALESCED) && vec.buf &&
(!buffer_area || vec.buf < buffer_area || vec.buf >= (uint8_t*)buffer_area + dsk.journal_len))
{
free(vec.buf);
disk_i *di = (vec.copy_flags & COPY_BUF_JOURNAL) ? dsk.journal_disk : dsk.data_disk;
di->free_iomem(vec.buf);
vec.buf = NULL;
}
}
}
}
void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op)
void blockstore_impl_t::handle_read_event(int res, blockstore_op_t *op)
{
live = true;
PRIV(op)->pending_ops--;
if (data->res != data->iov.iov_len)
if (res < 0)
{
// read error
op->retval = data->res;
op->retval = res;
}
if (PRIV(op)->pending_ops == 0)
{

View File

@@ -86,7 +86,7 @@ resume_1:
if (priv->modified_block != UINT32_MAX && priv->modified_block2 != priv->modified_block)
{
priv->op_state = 1;
BS_SUBMIT_CHECK_SQES(1);
BS_SUBMIT_CHECK_META_QUEUE(1);
prepare_meta_block_write(priv->modified_block);
resume_2:
if (meta_block_is_pending(priv->modified_block))

View File

@@ -27,71 +27,72 @@ bool blockstore_impl_t::has_unsynced()
return data || buffer || meta;
}
bool blockstore_impl_t::submit_fsyncs(int & wait_count)
int blockstore_impl_t::submit_fsyncs(int & wait_count)
{
int n = (unsynced_meta_write_count > 0 && !dsk.disable_meta_fsync ? 1 : 0) +
(unsynced_buffer_write_count > 0 && !dsk.disable_journal_fsync &&
(!unsynced_meta_write_count || dsk.journal_fd != dsk.meta_fd) ? 1 : 0) +
(unsynced_data_write_count > 0 && !dsk.disable_data_fsync &&
(!unsynced_meta_write_count || dsk.data_fd != dsk.meta_fd) &&
(!unsynced_buffer_write_count || dsk.data_fd != dsk.journal_fd) ? 1 : 0);
if (ringloop->space_left() < n)
int nm = (unsynced_meta_write_count > 0 && !dsk.disable_meta_fsync ? 1 : 0);
int nj = (unsynced_buffer_write_count > 0 && !dsk.disable_journal_fsync &&
(!unsynced_meta_write_count || dsk.journal_disk != dsk.meta_disk) ? 1 : 0);
int nd = (unsynced_data_write_count > 0 && !dsk.disable_data_fsync &&
(!unsynced_meta_write_count || dsk.data_disk != dsk.meta_disk) &&
(!unsynced_buffer_write_count || dsk.data_disk != dsk.journal_disk) ? 1 : 0);
if (!nm && !nj && !nd)
return 0;
int rnm = nm, rnj = nj, rnd = nd;
if (!dsk.meta_off_ring && !dsk.journal_off_ring)
rnm += rnj, rnj = 0;
if (!dsk.meta_off_ring && !dsk.data_off_ring)
rnm += rnd, rnd = 0;
if (!dsk.journal_off_ring && !dsk.data_off_ring)
rnj += rnd, rnd = 0;
if (rnm > 0 && dsk.meta_disk->get_queue_space() < rnm)
return WAIT_META_QUEUE;
if (rnj > 0 && dsk.journal_disk->get_queue_space() < rnj)
return WAIT_JOURNAL_QUEUE;
if (rnd > 0 && dsk.data_disk->get_queue_space() < rnd)
return WAIT_DATA_QUEUE;
auto cb = [this, & wait_count](int res)
{
return false;
}
if (!n)
{
return true;
}
auto cb = [this, & wait_count](ring_data_t *data)
{
if (data->res != 0)
disk_error_abort("sync meta", data->res, 0);
if (res != 0)
disk_error_abort("sync meta", res);
wait_count--;
assert(wait_count >= 0);
if (!wait_count)
ringloop->wakeup();
};
if (unsynced_meta_write_count > 0 && !dsk.disable_meta_fsync)
if (nm)
{
// fsync meta
io_uring_sqe *sqe = get_sqe();
assert(sqe);
ring_data_t *data = ((ring_data_t*)sqe->user_data);
io_uring_prep_fsync(sqe, dsk.meta_fd, IORING_FSYNC_DATASYNC);
data->iov = { 0 };
data->callback = cb;
wait_count++;
int res = dsk.meta_disk->flush_async(cb);
if (res != -EALREADY)
{
assert(res == 0);
wait_count++;
}
}
if (unsynced_buffer_write_count > 0 && !dsk.disable_journal_fsync &&
(!unsynced_meta_write_count || dsk.journal_fd != dsk.meta_fd))
if (nj)
{
// fsync buffer
io_uring_sqe *sqe = get_sqe();
assert(sqe);
ring_data_t *data = ((ring_data_t*)sqe->user_data);
io_uring_prep_fsync(sqe, dsk.journal_fd, IORING_FSYNC_DATASYNC);
data->iov = { 0 };
data->callback = cb;
wait_count++;
int res = dsk.journal_disk->flush_async(cb);
if (res != -EALREADY)
{
assert(res == 0);
wait_count++;
}
}
if (unsynced_data_write_count > 0 && !dsk.disable_data_fsync &&
(!unsynced_meta_write_count || dsk.data_fd != dsk.meta_fd) &&
(!unsynced_buffer_write_count || dsk.data_fd != dsk.journal_fd))
if (nd)
{
// fsync data
io_uring_sqe *sqe = get_sqe();
assert(sqe);
ring_data_t *data = ((ring_data_t*)sqe->user_data);
io_uring_prep_fsync(sqe, dsk.data_fd, IORING_FSYNC_DATASYNC);
data->iov = { 0 };
data->callback = cb;
wait_count++;
int res = dsk.data_disk->flush_async(cb);
if (res != -EALREADY)
{
assert(res == 0);
wait_count++;
}
}
unsynced_data_write_count = 0;
unsynced_buffer_write_count = 0;
unsynced_meta_write_count = 0;
return true;
return 0;
}
int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state)
@@ -113,11 +114,10 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state)
}
assert(!PRIV(op)->lsn);
PRIV(op)->lsn = heap->get_completed_lsn();
if (!submit_fsyncs(PRIV(op)->pending_ops))
PRIV(op)->wait_for = submit_fsyncs(PRIV(op)->pending_ops);
if (PRIV(op)->wait_for)
{
PRIV(op)->lsn = 0;
PRIV(op)->wait_detail = 1;
PRIV(op)->wait_for = WAIT_SQE;
return 0;
}
resume_1:

View File

@@ -16,32 +16,31 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
{
if (modified_blocks.find(modified_block) != modified_blocks.end())
return;
io_uring_sqe *sqe = get_sqe();
assert(sqe != NULL);
ring_data_t *data = ((ring_data_t*)sqe->user_data);
uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
data->iov = (struct iovec){ buf, (size_t)dsk.meta_block_size };
data->callback = [this, modified_block](ring_data_t *data)
{
live = true;
if (data->res != data->iov.iov_len)
{
// FIXME: our state becomes corrupted after a write error. maybe do something better than just die
disk_error_abort("data write", data->res, data->iov.iov_len);
}
auto it = modified_blocks.find(modified_block);
assert(it != modified_blocks.end());
free(it->second.buf);
modified_blocks.erase(it);
unsynced_meta_write_count++;
heap->complete_block_write(modified_block);
ringloop->wakeup();
};
return;
}
uint8_t *buf = (uint8_t*)dsk.meta_disk->alloc_iomem(dsk.meta_block_size);
assert(((uint64_t)modified_block+2)*dsk.meta_block_size <= dsk.meta_area_size);
io_uring_prep_writev(
sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + ((uint64_t)modified_block+1)*dsk.meta_block_size
int res = dsk.meta_disk->write_async(
dsk.meta_offset + ((uint64_t)modified_block+1)*dsk.meta_block_size, dsk.meta_block_size, buf,
[this, modified_block](int res)
{
live = true;
if (res < 0)
{
// FIXME: our state becomes corrupted after a write error. maybe do something better than just die
disk_error_abort("data write", res);
}
auto it = modified_blocks.find(modified_block);
assert(it != modified_blocks.end());
dsk.meta_disk->free_iomem(it->second.buf);
modified_blocks.erase(it);
unsynced_meta_write_count++;
heap->complete_block_write(modified_block);
ringloop->wakeup();
}
);
assert(res == 0);
pending_modified_blocks.push_back(modified_block);
modified_blocks[modified_block] = { .sent = false, .buf = buf };
}
@@ -135,7 +134,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
return 2;
}
PRIV(op)->write_type = BS_HEAP_DELETE;
BS_SUBMIT_CHECK_SQES(1);
BS_SUBMIT_CHECK_META_QUEUE(1);
int res = heap->add_delete(obj, &PRIV(op)->modified_block);
assert(res != EAGAIN);
if (res == ENOSPC)
@@ -152,7 +151,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
{
// Big (redirect) write
PRIV(op)->write_type = dsk.disable_data_fsync || op->opcode != BS_OP_WRITE_STABLE ? BS_HEAP_BIG_WRITE : _REDIRECT_INTENT;
BS_SUBMIT_CHECK_SQES(1);
BS_SUBMIT_CHECK_DATA_QUEUE(1);
PRIV(op)->location = heap->find_free_data();
if (PRIV(op)->location == UINT64_MAX)
{
@@ -177,12 +176,12 @@ enospc:
);
#endif
heap->use_data(op->oid.inode, PRIV(op)->location);
io_uring_sqe *sqe = get_sqe();
ring_data_t *data = ((ring_data_t*)sqe->user_data);
data->iov = (struct iovec){ op->buf, op->len };
data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
assert(loc+op->offset+op->len <= dsk.block_count*dsk.data_block_size);
io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset);
int res = dsk.data_disk->write_async(
dsk.data_offset + loc + op->offset, op->len, op->buf,
[this, op](int res) { handle_write_event(res, op); }
);
assert(res == 0);
if (!dsk.disable_data_fsync)
{
// use PRIV->lsn for fsync_data_id
@@ -199,7 +198,7 @@ enospc:
else if (intent_write_allowed(op, obj))
{
// Direct intent-write
BS_SUBMIT_CHECK_SQES(1);
BS_SUBMIT_CHECK_META_QUEUE(1);
int res = 0;
if (dsk.csum_block_size <= dsk.bitmap_granularity &&
(obj->entry_type == (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE) ||
@@ -260,7 +259,15 @@ enospc:
return 0;
}
// There is sufficient space. Check SQE(s)
BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
if (!op->len || dsk.journal_disk == dsk.meta_disk || !dsk.journal_off_ring && !dsk.meta_off_ring)
{
BS_SUBMIT_CHECK_META_QUEUE(1 + (op->len > 0 ? 1 : 0));
}
else
{
BS_SUBMIT_CHECK_META_QUEUE(1);
BS_SUBMIT_CHECK_JOURNAL_QUEUE(1);
}
int res = heap->add_small_write(op->oid, &obj, (BS_HEAP_SMALL_WRITE | (op->opcode == BS_OP_WRITE_STABLE ? BS_HEAP_STABLE : 0)),
op->version, op->offset, op->len, loc, op->bitmap, (uint8_t*)op->buf, &PRIV(op)->modified_block);
BS_SUBMIT_CHECK_PLACEMENT(res, obj->lsn);
@@ -278,11 +285,12 @@ enospc:
{
memcpy((uint8_t*)buffer_area + loc, op->buf, op->len);
}
BS_SUBMIT_GET_SQE(sqe2, data2);
data2->iov = (struct iovec){ op->buf, op->len };
data2->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
assert(loc+op->len <= dsk.journal_len);
io_uring_prep_writev(sqe2, dsk.journal_fd, &data2->iov, 1, dsk.journal_offset + loc);
res = dsk.journal_disk->write_async(
dsk.journal_offset + loc, op->len, op->buf,
[this, op](int res) { handle_write_event(res, op); }
);
assert(res == 0);
PRIV(op)->pending_ops++;
}
else
@@ -298,6 +306,7 @@ enospc:
int blockstore_impl_t::continue_write(blockstore_op_t *op)
{
int res = 0;
int op_state = PRIV(op)->op_state;
again:
if (op_state == 2)
@@ -350,21 +359,29 @@ resume_11:
}
if (PRIV(op)->lsn > data_fsync_sent)
{
BS_SUBMIT_GET_SQE(sqe, data);
io_uring_prep_fsync(sqe, dsk.data_fd, IORING_FSYNC_DATASYNC);
data->iov = { 0 };
data->callback = [this, op, fs = data_fsync_cur](ring_data_t *data)
res = dsk.data_disk->flush_async([this, op, fs = data_fsync_cur](int res)
{
live = true;
if (data->res != 0)
disk_error_abort("sync data", data->res, 0);
if (res != 0)
disk_error_abort("sync data", res);
if (fs > data_fsync_done)
{
data_fsync_done = fs;
ringloop->wakeup();
}
};
data_fsync_sent = data_fsync_cur;
});
if (res == -EAGAIN)
{
return 1;
}
else
{
data_fsync_sent = data_fsync_cur;
if (res == -EALREADY)
data_fsync_done = data_fsync_cur;
else
assert(res == 0);
}
}
if (PRIV(op)->lsn > data_fsync_done)
{
@@ -374,9 +391,8 @@ resume_11:
}
resume_4:
{
BS_SUBMIT_CHECK_SQES(1);
BS_SUBMIT_CHECK_META_QUEUE(1);
auto obj = heap->read_entry(op->oid);
int res = 0;
if (PRIV(op)->write_type == _REDIRECT_INTENT)
{
res = heap->add_redirect_intent(op->oid, &obj, op->version, op->offset, op->len,
@@ -476,25 +492,28 @@ resume_8:
resume_10:
// Direct intent-write
// LSN is not marked as completed so big_write won't be freed
BS_SUBMIT_GET_SQE(sqe, data);
data->iov = (struct iovec){ op->buf, op->len };
data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
BS_SUBMIT_CHECK_DATA_QUEUE(1);
assert(PRIV(op)->location + op->offset <= dsk.block_count*dsk.data_block_size);
io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + PRIV(op)->location + op->offset);
if (dsk.use_atomic_flag)
sqe->rw_flags = RWF_ATOMIC;
// FIXME: We tried to use RWF_ATOMIC here, but it's useless because Linux
// incorrectly enforces 2^N atomic write alignment which is a blocker for us.
// So now it's removed.
res = dsk.data_disk->write_async(
dsk.data_offset + PRIV(op)->location + op->offset, op->len, op->buf,
[this, op](int res) { handle_write_event(res, op); }
);
assert(res == 0);
PRIV(op)->pending_ops++;
PRIV(op)->op_state = 7;
return 1;
}
void blockstore_impl_t::handle_write_event(ring_data_t *data, blockstore_op_t *op)
void blockstore_impl_t::handle_write_event(int res, blockstore_op_t *op)
{
live = true;
if (data->res != data->iov.iov_len)
if (res < 0)
{
// FIXME: our state becomes corrupted after a write error. maybe do something better than just die
disk_error_abort("data write", data->res, data->iov.iov_len);
disk_error_abort("data write", res);
}
PRIV(op)->pending_ops--;
assert(PRIV(op)->pending_ops >= 0);

View File

@@ -1,6 +1,7 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#include "disk_util.h"
#include "impl.h"
#include "internal.h"
@@ -22,6 +23,9 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
dsk.open_meta();
dsk.open_journal();
calc_lengths();
dsk.data_fd = dsk.data_disk->get_fd();
dsk.meta_fd = dsk.meta_disk->get_fd();
dsk.journal_fd = dsk.journal_disk->get_fd();
alloc_dyn_data = dsk.clean_dyn_size > sizeof(void*) || dsk.csum_block_size > 0;
zero_object = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.data_block_size);
memset(zero_object, 0, dsk.data_block_size);

View File

@@ -48,7 +48,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
msgr.exec_op = [this](osd_op_t *op)
{
// Garbage in
fprintf(stderr, "Can't handle incoming operation from client %lu\n", op->client_id);
fprintf(stderr, "Can't handle incoming operation from client %ju\n", op->client_id);
msgr.stop_client(op->client_id);
delete op;
};
@@ -591,7 +591,7 @@ void cluster_client_t::on_change_pool_config_hook()
{
if (log_level > 2 && pg_counts[pool_item.first])
{
fprintf(stderr, "Pool %u (%s) PG count changed from %lu to %lu\n", pool_item.first, pool_item.second.name.c_str(),
fprintf(stderr, "Pool %u (%s) PG count changed from %ju to %ju\n", pool_item.first, pool_item.second.name.c_str(),
pg_counts[pool_item.first], pool_item.second.real_pg_count);
}
// At this point, all pool operations should have been suspended

View File

@@ -700,15 +700,13 @@ bool osd_messenger_t::handle_hdr(osd_client_t *cl)
}
else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC)
{
if (cl->check_sequencing)
if (cl->check_sequencing && cl->read_op->req.hdr.id != cl->read_op_id)
{
if (cl->read_op->req.hdr.id != cl->read_op_id)
{
fprintf(stderr, "Warning: operation sequencing is broken on client %d: expected num %ju, got %ju, stopping client\n", cl->peer_fd, cl->read_op_id, cl->read_op->req.hdr.id);
return false;
}
cl->read_op_id++;
fprintf(stderr, "Warning: operation sequencing is broken on client %ju: expected num %ju, got %ju, stopping client\n", cl->client_id, cl->read_op_id, cl->read_op->req.hdr.id);
return false;
}
// Track expected op_ids before check_sequencing to not break if a PING sneaks before SHOW_CONFIG reply
cl->read_op_id = cl->read_op->req.hdr.id + 1;
if (!allocate_op_buffers(cl))
{
return false;
@@ -775,7 +773,7 @@ bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl)
{
if (log_level > 1)
{
fprintf(stderr, "Error: peer %ju stabilize request size too large (%lu > %u bytes), stopping\n", cl->client_id,
fprintf(stderr, "Error: peer %ju stabilize request size too large (%ju > %u bytes), stopping\n", cl->client_id,
cur_op->req.sec_stab.len, MAX_SIMPLE_PAYLOAD_SIZE);
}
return false;
@@ -792,7 +790,7 @@ bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl)
{
if (log_level > 1)
{
fprintf(stderr, "Error: peer %ju sec_read_bmp request size too large (%lu > %u bytes), stopping\n", cl->client_id,
fprintf(stderr, "Error: peer %ju sec_read_bmp request size too large (%ju > %u bytes), stopping\n", cl->client_id,
cur_op->req.sec_read_bmp.len, MAX_SIMPLE_PAYLOAD_SIZE);
}
return false;
@@ -826,7 +824,7 @@ bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl)
{
if (log_level > 1)
{
fprintf(stderr, "Error: peer %ju show_config request length too large (%lu > %u bytes), stopping\n", cl->client_id,
fprintf(stderr, "Error: peer %ju show_config request length too large (%ju > %u bytes), stopping\n", cl->client_id,
cur_op->req.show_conf.json_len, MAX_SIMPLE_PAYLOAD_SIZE);
}
return false;

View File

@@ -329,7 +329,9 @@ static void coroutine_fn vitastor_co_resize(void *opaque)
}
#else
BdrvChild *c;
#if QEMU_VERSION_MAJOR >= 8
WITH_GRAPH_RDLOCK_GUARD() {
#endif
#if QEMU_VERSION_MAJOR >= 9
assert_bdrv_graph_readable();
#endif
@@ -344,7 +346,9 @@ static void coroutine_fn vitastor_co_resize(void *opaque)
}
#endif
}
#if QEMU_VERSION_MAJOR >= 8
}
#endif
#endif
bdrv_dec_in_flight(bs);
}

View File

@@ -706,7 +706,7 @@ protected:
{
if (data->res != 0)
{
fprintf(stderr, "Failed to update size of /dev/ublkb%d to %lu bytes: %s (code %d)\n", ublk_dev.dev_id, size, strerror(-data->res), data->res);
fprintf(stderr, "Failed to update size of /dev/ublkb%d to %ju bytes: %s (code %d)\n", ublk_dev.dev_id, size, strerror(-data->res), data->res);
}
};
}

View File

@@ -583,7 +583,7 @@ struct snap_merger_t
if (use_cas && subop->retval == -EINTR)
{
// CAS failure - reread and repeat optimistically
assert(rwo->todo == 1); // initial refcount from read_and_write
assert(!rwo->todo);
rwo->error_code = -EINTR;
rwo->start = rwo->end = 0;
rwo->op.version = 0;

View File

@@ -10,6 +10,7 @@ add_executable(vitastor-disk
disk_tool.h
disk_tool_journal.cpp
disk_tool_meta.cpp
disk_tool_nvme.cpp
disk_tool_prepare.cpp
disk_tool_resize_auto.cpp
disk_tool_resize.cpp

View File

@@ -1,16 +1,11 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#include <fcntl.h>
#include <sys/ioctl.h>
#include <ctype.h>
#include <unistd.h>
#include <sys/stat.h>
#include "json11/json11.hpp"
#include "str_util.h"
#include "blockstore.h"
#include "blockstore_disk.h"
#include "disk_util.h"
#include "blockstore_heap.h"
#include "ondisk_formats.h"
@@ -62,39 +57,23 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output)
fprintf(stderr, "Device path is missing\n");
exit(1);
}
struct stat st;
if (stat(device.c_str(), &st) < 0)
disk_i *disk = disk_i::create_and_open(device, "cached,ro", true);
if (!disk)
{
fprintf(stderr, "Can't stat %s: %s\n", device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open %s\n", device.c_str());
exit(1);
}
if (S_ISBLK(st.st_mode))
device_size = disk->get_size();
uint64_t blksize = disk->get_block_size();
if (blksize < device_block_size)
{
int fd = open(device.c_str(), O_DIRECT|O_RDONLY);
if (fd < 0 || ioctl(fd, BLKGETSIZE64, &device_size) < 0)
{
fprintf(stderr, "Failed to get device size for %s: %s\n", device.c_str(), strerror(errno));
exit(1);
}
close(fd);
if (st.st_blksize < device_block_size)
{
fprintf(
stderr, "Warning: %s reports %ju byte blocks, but we use %ju."
" Set --device_block_size=%ju if you're sure it works well with %ju byte blocks.\n",
device.c_str(), (uint64_t)st.st_blksize, device_block_size, (uint64_t)st.st_blksize, (uint64_t)st.st_blksize
);
}
}
else if (S_ISREG(st.st_mode))
{
device_size = st.st_size;
}
else
{
fprintf(stderr, "%s is neither a block device nor a regular file\n", device.c_str());
exit(1);
fprintf(
stderr, "Warning: %s reports %ju byte blocks, but we use %ju."
" Set --device_block_size=%ju if you're sure it works well with %ju byte blocks.\n",
device.c_str(), blksize, device_block_size, blksize, blksize
);
}
delete disk;
}
if (!device_size)
{

View File

@@ -207,6 +207,9 @@ static const char *help_text =
" Write metadata from JSON taken from standard input in the same format as produced by `dump-meta`.\n"
" You can specify any OSD device (data, metadata or journal), or the layout manually.\n"
"\n"
"vitastor-disk find-pci <serial> [nsid]\n"
" Find PCI ID and attributes of an NVMe device with a specified serial number.\n"
"\n"
"vitastor-disk simple-offsets <device>\n"
" Calculate offsets for old simple&stupid (no superblock) OSD deployment. Options:\n"
" --object_size 128k Set blockstore block size\n"
@@ -473,6 +476,16 @@ int main(int argc, char *argv[])
{
return self.raw_resize();
}
else if (!strcmp(cmd[0], "find-pci"))
{
if (cmd.size() != 2 && cmd.size() != 3)
{
fprintf(stderr, "S/N and optionally NS number is required\n");
return 1;
}
self.find_pci_id(cmd[1], cmd.size() == 3 ? stoull_full(cmd[2]) : 0);
return 0;
}
else if (!strcmp(cmd[0], "simple-offsets"))
{
// Calculate offsets for simple & stupid OSD deployment without superblock

View File

@@ -17,6 +17,7 @@
#include "blockstore_disk.h"
#include "blockstore.h"
#include "blockstore_heap.h"
#include "disk_util.h"
#include "ondisk_formats.h"
#include "crc32c.h"
#include "allocator.h"
@@ -85,7 +86,7 @@ struct disk_tool_t
uint64_t total_blocks = 0, free_first = 0, free_last = 0;
uint64_t new_clean_entry_bitmap_size = 0, new_data_csum_size = 0, new_clean_entry_size = 0, new_entries_per_block = 0;
uint32_t new_meta_format = 0;
int new_journal_fd = -1, new_meta_fd = -1;
disk_i *new_journal_disk = NULL, *new_meta_disk = NULL;
resizer_data_moving_t *moving_blocks = NULL;
bool atomic_warned = false;
@@ -152,6 +153,8 @@ struct disk_tool_t
int clear_osd_superblock(const std::string & dev);
int trim_data(std::string device);
std::string find_pci_id(const std::string & nvme_serial, uint32_t nvme_nsid);
json11::Json read_osd_superblock(std::string device, bool expect_exist = true, bool ignore_nonref = false);
uint32_t write_osd_superblock(std::string device, json11::Json params);
@@ -177,7 +180,6 @@ uint64_t get_atomic_write_size(const std::string & dev);
uint64_t get_device_size(const std::string & dev, bool should_exist = false);
std::string get_parent_device(std::string dev);
int shell_exec(const std::vector<std::string> & cmd, const std::string & in, std::string *out, std::string *err);
int write_zero(int fd, uint64_t offset, uint64_t size);
json11::Json read_parttable(std::string dev);
uint64_t dev_size_from_parttable(json11::Json pt);
uint64_t free_from_parttable(json11::Json pt);

View File

@@ -22,18 +22,22 @@ int disk_tool_t::dump_journal()
printf("[\n");
if (all)
{
dsk.journal_fd = open(dsk.journal_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDONLY);
if (dsk.journal_fd < 0)
dsk.journal_disk = disk_i::create_and_open(dsk.journal_device, (options["io"] == "cached" ? "cached,ro" : "ro"), true);
if (!dsk.journal_disk)
{
fprintf(stderr, "Failed to open journal device %s: %s\n", dsk.journal_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open journal device %s\n", dsk.journal_device.c_str());
return 1;
}
void *journal_buf = memalign_or_die(MEM_ALIGNMENT, dsk.journal_block_size);
void *journal_buf = dsk.journal_disk->alloc_iomem(dsk.journal_block_size);
journal_pos = 0;
while (journal_pos < dsk.journal_len)
{
int r = pread(dsk.journal_fd, journal_buf, dsk.journal_block_size, dsk.journal_offset+journal_pos);
assert(r == dsk.journal_block_size);
int r = dsk.journal_disk->read_sync(dsk.journal_offset+journal_pos, dsk.journal_block_size, journal_buf);
if (r != 0)
{
fprintf(stderr, "Failed to read journal block at %ju: %s (code %u)\n", dsk.journal_offset+journal_pos, strerror(-r), -r);
break;
}
uint64_t s;
for (s = 0; s < dsk.journal_block_size; s += 8)
{
@@ -90,9 +94,9 @@ int disk_tool_t::dump_journal()
journal_pos += dsk.journal_block_size;
}
}
free(journal_buf);
close(dsk.journal_fd);
dsk.journal_fd = -1;
dsk.journal_disk->free_iomem(journal_buf);
delete dsk.journal_disk;
dsk.journal_disk = NULL;
}
else
{
@@ -127,23 +131,28 @@ int disk_tool_t::process_journal(std::function<int(void*)> block_fn, bool do_ope
{
if (do_open)
{
if (dsk.journal_fd >= 0)
if (dsk.journal_disk)
{
fprintf(stderr, "Bug: journal device is already opened\n");
return 1;
}
dsk.journal_fd = open(dsk.journal_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDONLY);
if (dsk.journal_fd < 0)
dsk.journal_disk = disk_i::create_and_open(dsk.journal_device, (options["io"] == "cached" ? "cached,ro" : "ro"), true);
if (!dsk.journal_disk)
{
fprintf(stderr, "Failed to open journal device %s: %s\n", dsk.journal_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open journal device %s\n", dsk.journal_device.c_str());
return 1;
}
}
void *data = memalign_or_die(MEM_ALIGNMENT, dsk.journal_block_size);
void *data = dsk.journal_disk->alloc_iomem(dsk.journal_block_size);
journal_pos = 0;
int r = pread(dsk.journal_fd, data, dsk.journal_block_size, dsk.journal_offset+journal_pos);
assert(r == dsk.journal_block_size);
journal_entry *je = (journal_entry*)(data);
journal_entry *je = NULL;
int r = dsk.journal_disk->read_sync(dsk.journal_offset+journal_pos, dsk.journal_block_size, data);
if (r != 0)
{
fprintf(stderr, "Failed to read journal header at %ju: %s (code %u)\n", dsk.journal_offset+journal_pos, strerror(-r), -r);
goto ret;
}
je = (journal_entry*)(data);
if (je->magic != JOURNAL_MAGIC || je->type != JE_START || je_crc32(je) != je->crc32)
{
fprintf(stderr, "offset %08jx: journal superblock is invalid\n", journal_pos);
@@ -175,19 +184,28 @@ int disk_tool_t::process_journal(std::function<int(void*)> block_fn, bool do_ope
{
if (journal_pos >= dsk.journal_len)
journal_pos = dsk.journal_block_size;
r = pread(dsk.journal_fd, data, dsk.journal_block_size, dsk.journal_offset+journal_pos);
assert(r == dsk.journal_block_size);
r = dsk.journal_disk->read_sync(dsk.journal_offset+journal_pos, dsk.journal_block_size, data);
if (r != 0)
{
fprintf(stderr, "Failed to read journal block at %ju: %s (code %u)\n", dsk.journal_offset+journal_pos, strerror(-r), -r);
break;
}
r = block_fn(data);
if (r <= 0)
break;
}
}
ret:
if (data)
{
dsk.journal_disk->free_iomem(data);
data = NULL;
}
if (do_open)
{
close(dsk.journal_fd);
dsk.journal_fd = -1;
delete dsk.journal_disk;
dsk.journal_disk = NULL;
}
free(data);
return r;
}
@@ -227,8 +245,14 @@ int disk_tool_t::process_journal_block(void *buf, std::function<void(int, journa
journal_pos = dsk.journal_block_size;
wrapped = true;
}
small_write_data = memalign_or_die(MEM_ALIGNMENT, je->small_write.len);
assert(pread(dsk.journal_fd, small_write_data, je->small_write.len, dsk.journal_offset+je->small_write.data_offset) == je->small_write.len);
small_write_data = dsk.journal_disk->alloc_iomem(je->small_write.len);
int r = dsk.journal_disk->read_sync(dsk.journal_offset+je->small_write.data_offset, je->small_write.len, small_write_data);
if (r != 0)
{
fprintf(stderr, "Failed to read %u journal bytes at %ju: %s (code %u)\n", je->small_write.len,
dsk.journal_offset+je->small_write.data_offset, strerror(-r), -r);
break;
}
data_crc32 = je_start.csum_block_size ? 0 : crc32c(0, small_write_data, je->small_write.len);
data_csum_valid = (data_crc32 == je->small_write.crc32_data);
if (je_start.csum_block_size && je->small_write.len > 0)
@@ -276,7 +300,7 @@ int disk_tool_t::process_journal_block(void *buf, std::function<void(int, journa
iter_fn(entry, je);
if (je->type == JE_SMALL_WRITE || je->type == JE_SMALL_WRITE_INSTANT)
{
free(small_write_data);
dsk.journal_disk->free_iomem(small_write_data);
small_write_data = NULL;
}
pos += je->size;
@@ -438,7 +462,7 @@ void disk_tool_t::dump_journal_entry(int num, journal_entry *je, bool json)
int disk_tool_t::write_json_journal(json11::Json entries)
{
new_journal_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_journal_len);
new_journal_buf = (uint8_t*)dsk.journal_disk->alloc_iomem(new_journal_len);
new_journal_ptr = new_journal_buf;
new_journal_data = new_journal_ptr + dsk.journal_block_size;
new_journal_in_pos = 0;
@@ -497,7 +521,7 @@ int disk_tool_t::write_json_journal(json11::Json entries)
if (new_journal_ptr-new_journal_buf >= new_journal_len)
{
fprintf(stderr, "Error: entries don't fit to the new journal\n");
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
return 1;
}
new_journal_data = new_journal_ptr+dsk.journal_block_size;
@@ -505,7 +529,7 @@ int disk_tool_t::write_json_journal(json11::Json entries)
if (dsk.journal_block_size < entry_size)
{
fprintf(stderr, "Error: journal entry too large (%u bytes)\n", entry_size);
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
return 1;
}
}
@@ -515,7 +539,7 @@ int disk_tool_t::write_json_journal(json11::Json entries)
if (new_journal_data - new_journal_buf + len > new_journal_len)
{
fprintf(stderr, "Error: entries don't fit to the new journal\n");
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
return 1;
}
*((journal_entry_small_write*)ne) = (journal_entry_small_write){
@@ -538,7 +562,7 @@ int disk_tool_t::write_json_journal(json11::Json entries)
if (ne->small_write.len > 0 && !rec["data"].is_string())
{
fprintf(stderr, "Error: entry data is missing, please generate the dump with --json --format data\n");
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
return 1;
}
if (ne->small_write.len > 0)
@@ -599,6 +623,6 @@ int disk_tool_t::write_json_journal(json11::Json entries)
new_journal_in_pos += ne->size;
}
int r = resize_write_new_journal();
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
return r;
}

View File

@@ -22,31 +22,36 @@ int disk_tool_t::process_meta(std::function<void(blockstore_meta_header_v3_t *)>
if (buf_size % dsk.meta_block_size)
buf_size = 8*dsk.meta_block_size;
uint8_t *data = NULL;
data = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, buf_size);
blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
blockstore_meta_header_v3_t *hdr = NULL;
if (do_open)
{
if (dsk.meta_fd >= 0)
if (dsk.meta_disk)
{
fprintf(stderr, "Bug: Metadata device is already opened\n");
close_error:
r = 1;
goto close_free;
}
dsk.meta_fd = open(dsk.meta_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDONLY);
if (dsk.meta_fd < 0)
dsk.meta_disk = disk_i::create_and_open(dsk.meta_device, (options["io"] == "cached" ? "cached,ro" : "ro"), true);
if (!dsk.meta_disk)
{
fprintf(stderr, "Failed to open metadata device %s: %s\n", dsk.meta_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open metadata device %s\n", dsk.meta_device.c_str());
goto close_error;
}
}
else if (dsk.meta_fd < 0)
else if (!dsk.meta_disk)
{
fprintf(stderr, "Bug: Metadata device is not opened\n");
goto close_error;
}
data = (uint8_t*)dsk.meta_disk->alloc_iomem(buf_size);
hdr = (blockstore_meta_header_v3_t *)dsk.meta_disk->alloc_iomem(dsk.meta_block_size);
// Check superblock
pread_blocking(dsk.meta_fd, hdr, dsk.meta_block_size, dsk.meta_offset);
if ((r = dsk.meta_disk->read_sync(dsk.meta_offset, dsk.meta_block_size, hdr)) != 0)
{
fprintf(stderr, "Failed to read metadata header: %s (code %d)\n", strerror(-r), -r);
goto close_error;
}
if (hdr->zero == 0 && hdr->magic == BLOCKSTORE_META_MAGIC_V1 && hdr->version == BLOCKSTORE_META_FORMAT_HEAP)
{
if (hdr->data_csum_type != 0 &&
@@ -62,35 +67,40 @@ close_error:
// Load buffer_area
if (with_data)
{
buffer_area = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.journal_len);
if (dsk.journal_device == dsk.meta_device || dsk.journal_device == "")
{
dsk.journal_fd = dsk.meta_fd;
dsk.journal_disk = dsk.meta_disk;
}
else if (do_open)
{
if (dsk.journal_fd >= 0)
if (dsk.journal_disk)
{
fprintf(stderr, "Bug: Metadata device is already opened\n");
goto close_error;
}
dsk.journal_fd = open(dsk.journal_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDONLY);
if (dsk.journal_fd < 0)
dsk.journal_disk = disk_i::create_and_open(dsk.journal_device, (options["io"] == "cached" ? "cached,ro" : "ro"), true);
if (!dsk.journal_disk)
{
fprintf(stderr, "Failed to open journal device %s: %s\n", dsk.journal_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open journal device %s\n", dsk.journal_device.c_str());
goto close_error;
}
}
else if (dsk.journal_fd < 0)
else if (!dsk.journal_disk)
{
fprintf(stderr, "Bug: journal device is not opened\n");
goto close_error;
}
buffer_area = (uint8_t*)dsk.journal_disk->alloc_iomem(dsk.journal_len);
uint64_t journal_pos = 0;
while (journal_pos < dsk.journal_len)
{
uint64_t read_len = buf_size < dsk.journal_len-journal_pos ? buf_size : dsk.journal_len-journal_pos;
pread_blocking(dsk.journal_fd, buffer_area+journal_pos, read_len, dsk.journal_offset+journal_pos);
if ((r = dsk.journal_disk->read_sync(dsk.journal_offset+journal_pos, read_len, buffer_area+journal_pos)) != 0)
{
fprintf(stderr, "Failed to read %ju bytes at %ju from journal device %s: %s (code %d)\n",
read_len, dsk.journal_offset+journal_pos, dsk.journal_device.c_str(), strerror(-r), -r);
goto close_error;
}
journal_pos += read_len;
}
}
@@ -108,7 +118,12 @@ close_error:
while (meta_pos < hdr->meta_area_size)
{
uint64_t read_len = buf_size < hdr->meta_area_size-meta_pos ? buf_size : hdr->meta_area_size-meta_pos;
pread_blocking(dsk.meta_fd, data, read_len, dsk.meta_offset+meta_pos);
if ((r = dsk.meta_disk->read_sync(dsk.meta_offset+meta_pos, read_len, data)) != 0)
{
fprintf(stderr, "Failed to read %ju bytes at %ju from metadata device %s: %s (code %d)\n",
read_len, dsk.meta_offset+meta_pos, dsk.meta_device.c_str(), strerror(-r), -r);
goto close_error;
}
r = heap->load_blocks(meta_pos-dsk.meta_block_size, read_len, data, true, entries_loaded);
meta_pos += read_len;
}
@@ -170,7 +185,12 @@ csum_unknown:
while (meta_pos < dsk.min_meta_len)
{
uint64_t read_len = buf_size < dsk.min_meta_len-meta_pos ? buf_size : dsk.min_meta_len-meta_pos;
pread_blocking(dsk.meta_fd, data, read_len, dsk.meta_offset+meta_pos);
if ((r = dsk.meta_disk->read_sync(dsk.meta_offset+meta_pos, read_len, data)) != 0)
{
fprintf(stderr, "Failed to read %ju bytes at %ju from metadata device %s: %s (code %d)\n",
read_len, dsk.meta_offset+meta_pos, dsk.meta_device.c_str(), strerror(-r), -r);
goto close_error;
}
meta_pos += read_len;
for (uint64_t blk = 0; blk < read_len; blk += dsk.meta_block_size)
{
@@ -184,7 +204,7 @@ csum_unknown:
uint32_t *entry_csum = (uint32_t*)((uint8_t*)entry + dsk.clean_entry_size - 4);
if (*entry_csum != crc32c(0, entry, dsk.clean_entry_size - 4))
{
fprintf(stderr, "Metadata entry %lu is corrupt (checksum mismatch), skipping\n", block_num);
fprintf(stderr, "Metadata entry %ju is corrupt (checksum mismatch), skipping\n", block_num);
continue;
}
}
@@ -205,7 +225,12 @@ csum_unknown:
while (meta_pos < dsk.min_meta_len)
{
uint64_t read_len = buf_size < dsk.min_meta_len-meta_pos ? buf_size : dsk.min_meta_len-meta_pos;
pread_blocking(dsk.meta_fd, data, read_len, dsk.meta_offset+meta_pos);
if ((r = dsk.meta_disk->read_sync(dsk.meta_offset+meta_pos, read_len, data)) != 0)
{
fprintf(stderr, "Failed to read %ju bytes at %ju from metadata device %s: %s (code %d)\n",
read_len, dsk.meta_offset+meta_pos, dsk.meta_device.c_str(), strerror(-r), -r);
goto close_error;
}
meta_pos += read_len;
for (uint64_t blk = 0; blk < read_len; blk += dsk.meta_block_size)
{
@@ -221,23 +246,29 @@ csum_unknown:
}
}
close_free:
free(data);
free(hdr);
if (data)
{
dsk.meta_disk->free_iomem(data);
data = NULL;
}
if (hdr)
{
dsk.meta_disk->free_iomem(hdr);
hdr = NULL;
}
if (buffer_area)
{
free(buffer_area);
dsk.journal_disk->free_iomem(buffer_area);
buffer_area = NULL;
}
if (do_open)
{
close(dsk.meta_fd);
dsk.meta_fd = -1;
if (dsk.journal_fd >= 0)
{
if (dsk.journal_fd != dsk.meta_fd)
close(dsk.journal_fd);
dsk.journal_fd = -1;
}
if (dsk.journal_disk && dsk.journal_disk != dsk.meta_disk)
delete dsk.journal_disk;
if (dsk.meta_disk)
delete dsk.meta_disk;
dsk.meta_disk = NULL;
dsk.journal_disk = NULL;
}
return r;
}
@@ -520,7 +551,7 @@ int disk_tool_t::write_json_meta(json11::Json meta)
{
return write_json_heap(meta, meta["journal"]);
}
new_meta_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_meta_len);
new_meta_buf = (uint8_t*)dsk.meta_disk->alloc_iomem(new_meta_len);
memset(new_meta_buf, 0, new_meta_len);
blockstore_meta_header_v2_t *new_hdr = (blockstore_meta_header_v2_t *)new_meta_buf;
new_hdr->zero = 0;
@@ -558,7 +589,7 @@ int disk_tool_t::write_json_meta(json11::Json meta)
uint64_t mb = 1 + data_block/new_entries_per_block;
if (mb >= new_meta_len/new_hdr->meta_block_size)
{
free(new_meta_buf);
dsk.meta_disk->free_iomem(new_meta_buf);
new_meta_buf = NULL;
fprintf(stderr, "Metadata (data block %ju) doesn't fit into the new area\n", data_block);
return 1;
@@ -592,7 +623,7 @@ int disk_tool_t::write_json_meta(json11::Json meta)
int disk_tool_t::write_json_heap(json11::Json meta, json11::Json journal)
{
assert(new_meta_len >= sizeof(blockstore_meta_header_v3_t));
new_meta_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_meta_len);
new_meta_buf = (uint8_t*)dsk.meta_disk->alloc_iomem(new_meta_len);
memset(new_meta_buf, 0, new_meta_len);
new_meta_hdr = (blockstore_meta_header_v3_t *)new_meta_buf;
new_meta_hdr->zero = 0;
@@ -627,7 +658,7 @@ int disk_tool_t::write_json_heap(json11::Json meta, json11::Json journal)
new_journal_buf = NULL;
if (new_journal_len)
{
new_journal_buf = (uint8_t*)memalign(MEM_ALIGNMENT, new_journal_len);
new_journal_buf = (uint8_t*)dsk.journal_disk->alloc_iomem(new_journal_len);
memset(new_journal_buf, 0, new_journal_len);
}
uint64_t total_used_space = 0;
@@ -684,8 +715,7 @@ int disk_tool_t::write_json_heap(json11::Json meta, json11::Json journal)
fprintf(stderr, "Write entry %s has invalid type: %s, aborting\n",
write_entry.dump().c_str(), write_entry["type"].dump().c_str());
close_err0:
free(new_meta_buf);
new_meta_buf = NULL;
free_new_meta();
return 1;
}
uint64_t wr_offset = write_entry["offset"].uint64_value();
@@ -772,8 +802,7 @@ close_err0:
{
fprintf(stderr, "Metadata should include journal in you want to convert it to the \"heap\" format\n");
close_err:
free(new_meta_buf);
new_meta_buf = NULL;
free_new_meta();
return 1;
}
std::map<object_id, std::vector<json11::Json::object>> journal_by_object;

View File

@@ -0,0 +1,158 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#ifdef WITH_VFN
#include "disk_tool.h"
#include "str_util.h"
#include "json_util.h"
#include "disk_util.h"
#include "malloc_or_die.h"
#define NVME_CLASS 0x010800
#define NVME_CLASS_MASK 0xffff00
const char *oncs_bit_names[] = {
"compare",
"write uncorrectable",
"dsm (deallocate)",
"write zeroes",
"save features",
"reservations",
"timestamp",
"verify",
"copy",
"copies are atomic",
"copies are fast",
"write zeroes with deallocate",
"namespaces zeroes"
};
bool bind_to_vfio(const std::string & dev)
{
int r;
if (access("/sys/bus/pci/drivers/vfio-pci", F_OK) != 0 && errno == ENOENT)
{
r = system("modprobe vfio-pci");
if (r != 0)
{
fprintf(stderr, "Failed to load vfio-pci kernel module: %s (code %u)\n", strerror(errno), errno);
return false;
}
}
r = write_file("/sys/bus/pci/drivers/vfio-pci/bind", dev);
if (r == -ENODEV)
{
auto vid = str_replace(trim(read_file("/sys/bus/pci/devices/"+dev+"/vendor")), "0x", "");
auto pid = str_replace(trim(read_file("/sys/bus/pci/devices/"+dev+"/device")), "0x", "");
r = write_file("/sys/bus/pci/drivers/vfio-pci/new_id", vid+" "+pid);
if (r != 0)
{
fprintf(stderr, "Failed to register VID:PID %s:%s in vfio-pci: %s (code %u)\n",
vid.c_str(), pid.c_str(), strerror(-r), -r);
return false;
}
r = write_file("/sys/bus/pci/drivers/vfio-pci/bind", dev);
}
if (r != 0)
{
fprintf(stderr, "Failed to bind %s to vfio-pci driver: %s (code %u)\n",
dev.c_str(), strerror(-r), -r);
return false;
}
return true;
}
std::string disk_tool_t::find_pci_id(const std::string & nvme_serial, uint32_t nvme_nsid)
{
auto dev = vfn_disk_i::find_pci_id(nvme_serial, nvme_nsid);
if (dev == "")
{
return dev;
}
std::unique_ptr<vfn_disk_i> dsk(vfn_disk_i::create());
if (!dsk->open_dev(dev))
{
continue;
}
const nvme_id_ctrl* id_ctrl = dsk->get_id_ctrl();
std::string serial = trim(std::string(id_ctrl->sn, 20));
fprintf(stderr, "NVMe device %s found - PCI ID %s\n", serial.c_str(), dev.c_str());
// Interesting fields: mdts, tnvmcap, nn, vwc, awupf, sgls, msdbd, cap.cmbs, hmpre, hmmin, oncs
fprintf(stderr, "Maximum transfer size: %ju bytes\n", dsk->get_mdts());
fprintf(stderr, "Total capacity: %ju bytes\n", *(uint64_t*)id_ctrl->tnvmcap);
fprintf(stderr, "Max number of namespaces: %u\n", id_ctrl->nn);
fprintf(stderr, "Volatile cache: %s\n", (id_ctrl->vwc & NVME_CTRL_VWC_PRESENT) ? "yes" : "no");
fprintf(stderr, "Atomic write unit power fail: %u blocks (probably 4k)\n", id_ctrl->awupf+1);
fprintf(stderr, "SGL support: %s",
(id_ctrl->sgls & NVME_CTRL_SGLS_SUPPORTED) == 1 ? "unaligned" :
((id_ctrl->sgls & NVME_CTRL_SGLS_SUPPORTED) == 2 ? "8 byte aligned" : "no support"));
if (id_ctrl->sgls & NVME_CTRL_SGLS_KEYED)
fprintf(stderr, ", keyed (RDMA)");
if (id_ctrl->sgls & NVME_CTRL_SGLS_BIT_BUCKET)
fprintf(stderr, ", bit bucket");
if (id_ctrl->sgls & NVME_CTRL_SGLS_MPTR_BYTE_ALIGNED)
fprintf(stderr, ", byte-aligned mptr");
if (id_ctrl->sgls & NVME_CTRL_SGLS_OVERSIZE)
fprintf(stderr, ", oversized");
if (id_ctrl->sgls & NVME_CTRL_SGLS_MPTR_SGL)
fprintf(stderr, ", SGL mptr");
if (id_ctrl->sgls & NVME_CTRL_SGLS_OFFSET)
fprintf(stderr, ", offset");
if (id_ctrl->sgls & NVME_CTRL_SGLS_TPORT)
fprintf(stderr, ", transport data block");
fprintf(stderr, "\nMax SGL data block descriptors: %u\n", id_ctrl->msdbd);
fprintf(stderr, "Optional command support:");
int oncs_n = 0;
for (int i = 0; i < sizeof(oncs_bit_names)/sizeof(oncs_bit_names[0]); i++)
{
if ((id_ctrl->oncs >> i) & 1)
fprintf(stderr, "%s%s", oncs_n++ > 0 ? ", " : " ", oncs_bit_names[i]);
}
fprintf(stderr, "%s\n", oncs_n > 0 ? "" : "none");
fprintf(stderr, "CMB: %s\n", NVME_CAP_CMBS(dsk->get_cap()) ? "supported" : "not supported");
fprintf(stderr, "HMB: min %u bytes, preferred %u bytes\n", id_ctrl->hmmin*4096, id_ctrl->hmpre*4096);
if (nvme_nsid > 0)
{
if (!dsk->open_ns(nvme_nsid))
{
return "";
}
// Interesting fields: nsze, nsfeat (bit 3 - nguid, bit 1 - nawupf), nlbaf,
// flbas, mc (bit 1 - OK, bit 0 - not OK), nawupf, nabspf, nabo, nguid, lbaf0-lbaf15
const nvme_id_ns *id_ns = dsk->get_id_ns();
uint8_t fmt = NVME_FLBAS_LOWER(id_ns->flbas);
uint64_t blksize = 1 << id_ns->lbaf[fmt].ds;
uint64_t ms = id_ns->lbaf[fmt].ms;
fprintf(stderr, "Namespace size: %ju bytes\n", (uint64_t)id_ns->nsze * blksize);
fprintf(stderr, "Format %u/%u: %ju byte sectors + %ju byte metadata\n", id_ns->flbas, id_ns->nlbaf+1, blksize, ms);
if (id_ns->nsfeat & NVME_NS_FEAT_ID_REUSE)
{
fprintf(stderr, "Namespace GUID: ");
for (int i = 0; i < sizeof(id_ns->nguid); i++)
fprintf(stderr, "%02x", id_ns->nguid[i]);
fprintf(stderr, "\n");
}
if (id_ns->nsfeat & NVME_NS_FEAT_NATOMIC)
{
fprintf(stderr, "Namespace atomic write unit power fail: %u blocks\n", id_ns->nawupf+1);
fprintf(stderr, "Namespace atomic boundary power fail: %u blocks\n", id_ns->nabspf);
fprintf(stderr, "Namespace atomic boundary offset: %u blocks\n", id_ns->nabo);
}
}
return dev;
}
#else
#include "disk_tool.h"
#include <string>
std::string disk_tool_t::find_pci_id(const std::string & nvme_serial, uint32_t nvme_nsid)
{
fprintf(stderr, "Vitastor is built without userspace NVMe (libvfn) support\n");
exit(1);
return "";
}
#endif

View File

@@ -4,6 +4,7 @@
#include "disk_tool.h"
#include "str_util.h"
#include "json_util.h"
#include "malloc_or_die.h"
void disk_tool_t::parse_meta_reserve()
{
@@ -272,10 +273,24 @@ int disk_tool_t::prepare_one(std::map<std::string, std::string> options, int is_
}
}
// Zero out the first block of metadata and journal - OSD will zero the rest on the first run
if (write_zero(dsk.meta_fd, sb["meta_offset"].uint64_value(), 4096) != 0 ||
write_zero(dsk.journal_fd, sb["journal_offset"].uint64_value(), 4096) != 0)
int rz = 0;
if (dsk.meta_disk)
{
fprintf(stderr, "Failed to zero out metadata or journal: %s\n", strerror(errno));
void *zero_buf = dsk.meta_disk->alloc_iomem(4096);
memset(zero_buf, 0, 4096);
rz = dsk.meta_disk->write_sync(sb["meta_offset"].uint64_value(), 4096, zero_buf);
dsk.meta_disk->free_iomem(zero_buf);
}
if (!rz && dsk.journal_disk)
{
void *zero_buf = dsk.journal_disk->alloc_iomem(4096);
memset(zero_buf, 0, 4096);
rz = dsk.journal_disk->write_sync(sb["journal_offset"].uint64_value(), 4096, zero_buf);
dsk.journal_disk->free_iomem(zero_buf);
}
if (rz != 0)
{
fprintf(stderr, "Failed to zero out metadata or journal header: %s (code %d)\n", strerror(-rz), -rz);
dsk.close_all();
return 1;
}

View File

@@ -303,16 +303,17 @@ int disk_tool_t::resize_copy_data()
iodepth = 32;
}
ringloop = new ring_loop_t(iodepth < RINGLOOP_DEFAULT_SIZE ? RINGLOOP_DEFAULT_SIZE : iodepth);
dsk.data_fd = open(dsk.data_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDWR);
if (dsk.data_fd < 0)
dsk.data_disk = disk_i::create_and_open(dsk.data_device, (options["io"] == "cached" ? "cached" : ""), true);
if (!dsk.data_disk)
{
fprintf(stderr, "Failed to open data device %s: %s\n", dsk.data_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open data device %s\n", dsk.data_device.c_str());
delete ringloop;
ringloop = NULL;
return 1;
}
dsk.data_disk->set_ringloop(ringloop);
moving_blocks = new resizer_data_moving_t[iodepth];
moving_blocks[0].buf = memalign_or_die(MEM_ALIGNMENT, iodepth*dsk.data_block_size);
moving_blocks[0].buf = dsk.data_disk->alloc_iomem(iodepth*dsk.data_block_size);
for (int i = 1; i < iodepth; i++)
{
moving_blocks[i].buf = (uint8_t*)moving_blocks[0].buf + i*dsk.data_block_size;
@@ -334,52 +335,76 @@ int disk_tool_t::resize_copy_data()
}
if (moving_blocks[i].state == DM_ST_TO_READ)
{
struct io_uring_sqe *sqe = ringloop->get_sqe();
if (sqe)
{
moving_blocks[i].state = DM_ST_READING;
struct ring_data_t *data = ((ring_data_t*)sqe->user_data);
data->iov = (struct iovec){ moving_blocks[i].buf, dsk.data_block_size };
io_uring_prep_readv(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + moving_blocks[i].old_loc*dsk.data_block_size);
data->callback = [this, i](ring_data_t *data)
int res = dsk.data_disk->read_async(
dsk.data_offset + moving_blocks[i].old_loc*dsk.data_block_size,
dsk.data_block_size, moving_blocks[i].buf, [this, i](int res)
{
if (data->res != dsk.data_block_size)
if (res != 0)
{
fprintf(
stderr, "Failed to read %ju bytes at %ju from %s: %s\n", dsk.data_block_size,
dsk.data_offset + moving_blocks[i].old_loc*dsk.data_block_size, dsk.data_device.c_str(),
data->res < 0 ? strerror(-data->res) : "short read"
strerror(-res)
);
exit(1);
}
moving_blocks[i].state = DM_ST_TO_WRITE;
ringloop->wakeup();
};
}
}
);
if (res == 0)
{
moving_blocks[i].state = DM_ST_READING;
}
else if (res == -EAGAIN)
{
break;
}
else
{
fprintf(
stderr, "Failed to submit read of %ju bytes at %ju from %s: %s\n", dsk.data_block_size,
dsk.data_offset + moving_blocks[i].old_loc*dsk.data_block_size, dsk.data_device.c_str(),
strerror(-res)
);
exit(1);
}
}
if (moving_blocks[i].state == DM_ST_TO_WRITE)
{
struct io_uring_sqe *sqe = ringloop->get_sqe();
if (sqe)
{
moving_blocks[i].state = DM_ST_WRITING;
struct ring_data_t *data = ((ring_data_t*)sqe->user_data);
data->iov = (struct iovec){ moving_blocks[i].buf, dsk.data_block_size };
io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + moving_blocks[i].new_loc*dsk.data_block_size);
data->callback = [this, i](ring_data_t *data)
int res = dsk.data_disk->write_async(
dsk.data_offset + moving_blocks[i].new_loc*dsk.data_block_size,
dsk.data_block_size, moving_blocks[i].buf, [this, i](int res)
{
if (data->res != dsk.data_block_size)
if (res != 0)
{
fprintf(
stderr, "Failed to write %ju bytes at %ju to %s: %s\n", dsk.data_block_size,
dsk.data_offset + moving_blocks[i].new_loc*dsk.data_block_size, dsk.data_device.c_str(),
data->res < 0 ? strerror(-data->res) : "short write"
strerror(-res)
);
exit(1);
}
moving_blocks[i].state = DM_ST_EMPTY;
ringloop->wakeup();
};
}
);
if (res == 0)
{
moving_blocks[i].state = DM_ST_WRITING;
}
else if (res == -EAGAIN)
{
break;
}
else
{
fprintf(
stderr, "Failed to submit write of %ju bytes at %ju to %s: %s\n", dsk.data_block_size,
dsk.data_offset + moving_blocks[i].new_loc*dsk.data_block_size, dsk.data_device.c_str(),
strerror(-res)
);
exit(1);
}
}
remap_active += moving_blocks[i].state != DM_ST_EMPTY ? 1 : 0;
@@ -395,11 +420,11 @@ int disk_tool_t::resize_copy_data()
ringloop->wait();
}
ringloop->unregister_consumer(&ring_consumer);
free(moving_blocks[0].buf);
dsk.data_disk->free_iomem(moving_blocks[0].buf);
delete[] moving_blocks;
moving_blocks = NULL;
close(dsk.data_fd);
dsk.data_fd = -1;
delete dsk.data_disk;
dsk.data_disk = NULL;
delete ringloop;
ringloop = NULL;
return 0;
@@ -407,7 +432,7 @@ int disk_tool_t::resize_copy_data()
void disk_tool_t::resize_alloc_journal()
{
new_journal_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_journal_len);
new_journal_buf = (uint8_t*)dsk.journal_disk->alloc_iomem(new_journal_len);
memset(new_journal_buf, 0, new_journal_len);
new_journal_ptr = new_journal_buf;
new_journal_data = new_journal_ptr + dsk.journal_block_size;
@@ -512,17 +537,24 @@ int disk_tool_t::resize_rebuild_journal()
int disk_tool_t::resize_write_new_journal()
{
new_journal_fd = open(new_journal_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDWR);
if (new_journal_fd < 0)
new_journal_disk = disk_i::create_and_open(new_journal_device, (options["io"] == "cached" ? "cached" : ""), true);
if (!new_journal_disk)
{
fprintf(stderr, "Failed to open new journal device %s: %s\n", new_journal_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open new journal device %s\n", new_journal_device.c_str());
return 1;
}
pwrite_blocking(new_journal_fd, new_journal_buf, new_journal_len, new_journal_offset);
fsync(new_journal_fd);
close(new_journal_fd);
new_journal_fd = -1;
return 0;
int r = new_journal_disk->write_sync(new_journal_offset, new_journal_len, new_journal_buf);
if (r == 0)
{
r = new_journal_disk->flush_sync();
}
if (r != 0)
{
fprintf(stderr, "Failed to write new journal to %s: %s (code %d)\n", new_journal_device.c_str(), strerror(-r), -r);
}
delete new_journal_disk;
new_journal_disk = NULL;
return r;
}
void disk_tool_t::remap_big_write(blockstore_heap_t *heap, heap_entry_t *wr)
@@ -559,7 +591,7 @@ void disk_tool_t::remap_small_write(blockstore_heap_t *heap, heap_entry_t *wr)
int disk_tool_t::resize_rebuild_meta()
{
new_meta_buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, new_meta_len);
new_meta_buf = (uint8_t*)dsk.meta_disk->alloc_iomem(new_meta_len);
memset(new_meta_buf, 0, new_meta_len);
new_meta_hdr = (blockstore_meta_header_v3_t *)new_meta_buf;
uint64_t new_meta_pos = dsk.meta_block_size;
@@ -691,34 +723,41 @@ int disk_tool_t::resize_rebuild_meta()
int disk_tool_t::resize_write_new_meta()
{
new_meta_fd = open(new_meta_device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDWR);
if (new_meta_fd < 0)
new_meta_disk = disk_i::create_and_open(new_meta_device, (options["io"] == "cached" ? "cached" : ""), true);
if (!new_meta_disk)
{
fprintf(stderr, "Failed to open new metadata device %s: %s\n", new_meta_device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open new metadata device %s\n", new_meta_device.c_str());
return 1;
}
pwrite_blocking(new_meta_fd, new_meta_buf, new_meta_len, new_meta_offset);
fsync(new_meta_fd);
close(new_meta_fd);
new_meta_fd = -1;
return 0;
int r = new_meta_disk->write_sync(new_meta_offset, new_meta_len, new_meta_buf);
if (r == 0)
{
r = new_meta_disk->flush_sync();
}
if (r != 0)
{
fprintf(stderr, "Failed to write new metadata to %s: %s (code %d)\n", new_meta_device.c_str(), strerror(-r), -r);
}
delete new_meta_disk;
new_meta_disk = NULL;
return r;
}
void disk_tool_t::free_new_meta()
{
if ((uint8_t*)new_meta_hdr != new_meta_buf)
{
free(new_meta_hdr);
dsk.meta_disk->free_iomem(new_meta_hdr);
new_meta_hdr = NULL;
}
if (new_meta_buf)
{
free(new_meta_buf);
dsk.meta_disk->free_iomem(new_meta_buf);
new_meta_buf = NULL;
}
if (new_journal_buf)
{
free(new_journal_buf);
dsk.journal_disk->free_iomem(new_journal_buf);
new_journal_buf = NULL;
}
}

View File

@@ -37,8 +37,6 @@ int disk_tool_t::resize_data(std::string device)
fprintf(stderr, "%s\n", e.what());
return 1;
}
// Save FD numbers because calc_lengths() relies on them
int old_journal_fd = dsk.journal_fd, old_meta_fd = dsk.meta_fd, old_data_fd = dsk.data_fd;
dsk.close_all();
bool dry_run = options.find("dry_run") != options.end();
auto old_journal_device = dsk.journal_device;
@@ -58,13 +56,7 @@ int disk_tool_t::resize_data(std::string device)
? dsk.data_device_size : new_data_dev_size;
dsk.data_device_size = new_data_dev_size;
dsk.cfg_data_size = 0;
dsk.journal_fd = old_journal_fd;
dsk.meta_fd = old_meta_fd;
dsk.data_fd = old_data_fd;
dsk.calc_lengths(true);
dsk.journal_fd = -1;
dsk.meta_fd = -1;
dsk.data_fd = -1;
}
std::map<std::string, std::string> move_options;
if (options.find("move_journal") != options.end())

View File

@@ -155,15 +155,16 @@ json11::Json disk_tool_t::read_osd_superblock(std::string device, bool expect_ex
json11::Json osd_params;
std::string json_err;
std::string real_device, device_type, real_data, real_meta, real_journal;
int r, fd = open(device.c_str(), (options["io"] == "cached" ? 0 : O_DIRECT) | O_RDWR);
if (fd < 0)
int r = 0;
auto di = disk_i::create_and_open(device, (options["io"] == "cached" ? "cached,ro" : "ro"), true);
if (!di)
{
fprintf(stderr, "Failed to open device %s: %s\n", device.c_str(), strerror(errno));
fprintf(stderr, "Failed to open device %s\n", device.c_str());
return osd_params;
}
buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
r = pread_blocking(fd, buf, 4096, 0);
if (r != 4096)
buf = (uint8_t*)di->alloc_iomem(4096);
r = di->read_sync(0, 4096, buf);
if (r != 0)
{
fprintf(stderr, "Failed to read OSD superblock from %s: %s\n", device.c_str(), strerror(errno));
goto ex;
@@ -186,10 +187,10 @@ json11::Json disk_tool_t::read_osd_superblock(std::string device, bool expect_ex
if (sb->size > 4096)
{
uint64_t sb_size = ((sb->size+4095)/4096)*4096;
free(buf);
buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, sb_size);
r = pread_blocking(fd, buf, sb_size, 0);
if (r != sb_size)
di->free_iomem(buf);
buf = (uint8_t*)di->alloc_iomem(sb_size);
r = di->read_sync(0, sb_size, buf);
if (r != 0)
{
fprintf(stderr, "Failed to read OSD superblock from %s: %s\n", device.c_str(), strerror(errno));
goto ex;
@@ -224,12 +225,12 @@ json11::Json disk_tool_t::read_osd_superblock(std::string device, bool expect_ex
osd_params = json11::Json();
goto ex;
}
real_device = realpath_str(device);
real_data = realpath_str(osd_params["data_device"].string_value());
real_device = disk_i::realpath(device);
real_data = disk_i::realpath(osd_params["data_device"].string_value());
real_meta = osd_params["meta_device"].string_value() != "" && osd_params["meta_device"] != osd_params["data_device"]
? realpath_str(osd_params["meta_device"].string_value()) : "";
? disk_i::realpath(osd_params["meta_device"].string_value()) : "";
real_journal = osd_params["journal_device"].string_value() != "" && osd_params["journal_device"] != osd_params["meta_device"]
? realpath_str(osd_params["journal_device"].string_value()) : "";
? disk_i::realpath(osd_params["journal_device"].string_value()) : "";
if (real_journal == real_meta)
{
real_journal = "";
@@ -265,8 +266,12 @@ json11::Json disk_tool_t::read_osd_superblock(std::string device, bool expect_ex
{ "real_journal_device", real_journal },
};
ex:
free(buf);
close(fd);
if (buf)
{
di->free_iomem(buf);
buf = NULL;
}
delete di;
return osd_params;
}

View File

@@ -3,7 +3,6 @@
#include <sys/wait.h>
#include <sys/ioctl.h>
#include <dirent.h>
#include "disk_tool.h"
#include "rw_blocking.h"
@@ -234,30 +233,6 @@ err_pipe1:
return 255;
}
int write_zero(int fd, uint64_t offset, uint64_t size)
{
uint64_t buf_len = 1024*1024;
void *zero_buf = memalign_or_die(MEM_ALIGNMENT, buf_len);
memset(zero_buf, 0, buf_len);
ssize_t r;
while (size > 0)
{
r = pwrite(fd, zero_buf, size > buf_len ? buf_len : size, offset);
if (r > 0)
{
size -= r;
offset += r;
}
else if (errno != EAGAIN && errno != EINTR)
{
free(zero_buf);
return -1;
}
}
free(zero_buf);
return 0;
}
// Returns false in case of an error
// Returns null if there is no partition table
json11::Json read_parttable(std::string dev)

View File

@@ -485,24 +485,26 @@ pg_osd_set_state_t *osd_t::mark_partial_write(pg_t & pg, osd_op_t *cur_op)
int changes = 0;
for (auto & chunk: new_set)
{
if (chunk.loc_bad != LOC_OUTDATED)
bool success = false;
for (int i = 0; i < op_data->n_subops; i++)
{
bool success = false;
for (int i = 0; i < op_data->n_subops; i++)
if (op_data->subops[i].osd_num == chunk.osd_num &&
op_data->subops[i].reply.hdr.retval == op_data->subops[i].req.sec_rw.len)
{
if (op_data->subops[i].osd_num == chunk.osd_num &&
op_data->subops[i].reply.hdr.retval == op_data->subops[i].req.sec_rw.len)
{
success = true;
break;
}
}
if (!success)
{
changes++;
chunk.loc_bad = LOC_OUTDATED;
success = true;
break;
}
}
if (!success && !(chunk.loc_bad & LOC_OUTDATED))
{
changes++;
chunk.loc_bad = LOC_OUTDATED;
}
else if (success && (chunk.loc_bad & LOC_OUTDATED))
{
changes++;
chunk.loc_bad &= ~LOC_OUTDATED;
}
}
return changes;
});

View File

@@ -370,7 +370,6 @@ void osd_t::exec_show_config(osd_op_t *cur_op)
if (req_json["features"]["check_sequencing"].bool_value())
{
cl->check_sequencing = true;
cl->read_op_id = cur_op->req.hdr.id + 1;
}
auto features = json11::Json::object{ { "pg_locks", true } };
if (msgr.use_proto_checksums)

View File

@@ -144,6 +144,20 @@ void test_replicated_write()
printf("test_replicated_write passed\n");
}
// Find the object's current pg_osd_set_state_t in whatever "not clean" index
// the PG keeps it in. NULL means the object is considered clean.
static pg_osd_set_state_t *find_object_state(pg_t & pg, object_id oid)
{
for (auto *idx: { &pg.inconsistent_objects, &pg.incomplete_objects,
&pg.degraded_objects, &pg.misplaced_objects })
{
auto it = idx->find(oid);
if (it != idx->end())
return it->second;
}
return NULL;
}
// Regression test for a bug where a second scrub on a still-failing replica
// (already marked LOC_CORRUPTED) would clear the corruption flag.
void test_scrub_corruption_persists()
@@ -2035,6 +2049,113 @@ void test_ec42_inconsistent_read()
printf("test_ec42_inconsistent_read passed\n");
}
// Regression test for a bug where a repeated partially failed write to the same
// object didn't clear LOC_OUTDATED from the replicas that succeeded this time —
// so a copy stayed marked outdated even though it had just received the newest
// version, and every failing write only ever added outdated copies.
//
// Layout: pool 1, replicated x3, primary = OSD 1 (us), secondaries = OSD 2 and 3.
//
// Write 1 fails with ENOSPC on OSD 3 -> OSD 3 is marked outdated.
// Write 2 fails with ENOSPC on OSD 1 (us) but succeeds on OSD 2 and OSD 3
// -> OSD 1 must become outdated AND OSD 3 must stop being outdated.
void test_repeated_partial_write_unmarks_outdated()
{
printf("test_repeated_partial_write_unmarks_outdated\n");
osd_test_fixture_t f;
f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 3, /*pg_minsize*/ 1, /*pg_count*/ 1,
{ { 1, 2, 3 } });
f.start(json11::Json::object {
{ "osd_num", 1 },
{ "etcd_address", "127.0.0.1:2379" },
{ "immediate_commit", "all" },
{ "block_size", 131072 },
{ "bitmap_granularity", 4096 },
});
f.connect_peer(2);
f.connect_peer(3);
f.complete_peering_empty();
assert(f.pg(1, 1).state & PG_ACTIVE);
inode_t inode = INODE_WITH_POOL(1, 1);
object_id oid = { .inode = inode, .stripe = 0 };
// Is the copy on <osd_num> marked outdated? A copy that vanished from the
// set entirely counts as outdated, too.
auto outdated_on = [&](osd_num_t osd_num)
{
auto *st = find_object_state(f.pg(1, 1), oid);
assert(st);
for (auto & chunk: st->osd_set)
if (chunk.osd_num == osd_num)
return (chunk.loc_bad & LOC_OUTDATED) != 0;
return true;
};
// ============ WRITE 1 — ENOSPC on OSD 3 ============
// Full-block writes on purpose: a sub-block write to a degraded object
// would turn into a read-modify-write and only obscure what's tested here.
auto *write1 = make_write_op(inode, 0, 131072, 0xab);
int retval1 = -1;
write1->callback = [&retval1](osd_op_t *op) { retval1 = op->reply.hdr.retval; };
f.exec(write1);
// Zero-length read resolving the (not yet existing) object version
f.bs_zero_read_ok(0);
// One local write + one write per peer
assert(f.bs->queued.size() == 1);
f.bs_write_ok(BS_OP_WRITE_STABLE, 1);
f.peer_write_ok(2, OSD_OP_SEC_WRITE_STABLE, 1);
auto *peer3_w1 = f.peer_take(3, OSD_OP_SEC_WRITE_STABLE);
assert(retval1 == -1); // still waiting for OSD 3
f.peer_complete(peer3_w1, -ENOSPC);
assert(retval1 == -ENOSPC);
assert(f.pg(1, 1).state & PG_HAS_DEGRADED);
assert(!outdated_on(1));
assert(!outdated_on(2));
assert(outdated_on(3));
printf("test_repeated_partial_write_unmarks_outdated: write 1 -> OSD 3 outdated\n");
delete write1;
// ============ WRITE 2 — ENOSPC locally, OK on both peers ============
auto *write2 = make_write_op(inode, 0, 131072, 0xcd);
int retval2 = -1;
write2->callback = [&retval2](osd_op_t *op) { retval2 = op->reply.hdr.retval; };
f.exec(write2);
// The object exists at version 1 now
f.bs_zero_read_ok(1);
assert(f.bs->queued.size() == 1);
auto *local_w2 = f.bs->take(BS_OP_WRITE_STABLE);
assert(local_w2->version == 2);
local_w2->retval = -ENOSPC;
local_w2->callback(local_w2);
f.peer_write_ok(2, OSD_OP_SEC_WRITE_STABLE, 2);
f.peer_write_ok(3, OSD_OP_SEC_WRITE_STABLE, 2);
assert(retval2 == -ENOSPC);
// THE KEY ASSERTION — OSD 3 took the new version, it isn't outdated anymore
if (outdated_on(3))
printf("BUG: OSD 3 is still marked outdated after successfully taking the new version\n");
assert(!outdated_on(3));
// ... and the OSD that missed this write is
assert(outdated_on(1));
assert(!outdated_on(2));
assert(f.pg(1, 1).inflight == 0);
assert(f.pg(1, 1).write_queue.empty());
delete write2;
printf("test_repeated_partial_write_unmarks_outdated passed\n");
}
int main(int narg, char *args[])
{
test_load_global_config();
@@ -2058,5 +2179,6 @@ int main(int narg, char *args[])
test_no_new_write_on_non_active_pg();
test_pg_epoch_bump_blocks_write_until_reported();
test_ec42_inconsistent_read();
test_repeated_partial_write_unmarks_outdated();
return 0;
}

View File

@@ -1010,6 +1010,211 @@ void test_iterate_compaction()
assert(small_writes == 1);
}
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) BIG_STABLE(v3 l3) un-fsynced
// -> skip compaction of l2 into l1 EVEN WHEN under pressure,
// because add_compact rejects stable overwrites with EBUSY
_test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
heap.use_data(INODE_WITH_POOL(1, 1), 128*1024);
uint32_t mblock = 999999;
res = _test_do_big_write(heap, dsk, 1, 0, 3, 128*1024, true, 0, 0, buffer_area.data(), &mblock);
assert(res == 0);
heap.start_block_write(mblock);
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
auto obj = heap.read_entry(oid);
assert(obj->lsn == 3);
assert(obj->is_overwrite());
assert(heap.get_fsynced_lsn() == 2);
// When not under pressure: postpone compaction
int small_writes = 0;
auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// When under pressure: MUST ALSO postpone compaction because obj is STABLE!
// Before the fix, this would incorrectly compact l2 (compact_lsn == 2)
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// Verify that add_compact() rejects stable overwrites with EBUSY
uint32_t compact_mblock = 999999;
uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
assert(res == EBUSY);
heap.complete_block_write(mblock);
assert(heap.get_fsynced_lsn() == 3);
// Once fsynced, l3 is durable and becomes the clean_wr base
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.clean_wr && compact_info.clean_wr->lsn == 3);
assert(small_writes == 0);
}
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) BIG_INTENT(v3 l3) un-fsynced
// -> skip compaction of l2 into l1 both when not under pressure and under pressure
// Before the fix, BIG_INTENT was missing from the check and was never skipped.
_test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
auto obj = heap.read_entry(oid);
uint8_t ext_bitmap[dsk.clean_entry_bitmap_size];
memset(ext_bitmap, 0x8e, dsk.clean_entry_bitmap_size);
heap.use_data(INODE_WITH_POOL(1, 1), 128*1024);
uint32_t mblock = 999999;
res = heap.add_redirect_intent(oid, &obj, 3, 0, dsk.data_block_size, 128*1024, ext_bitmap, buffer_area.data(), &mblock);
assert(res == 0);
heap.start_block_write(mblock);
obj = heap.read_entry(oid);
assert(obj->lsn == 3);
assert(obj->type() == BS_HEAP_BIG_INTENT);
assert(obj->is_overwrite());
assert(heap.get_fsynced_lsn() == 2);
// When not under pressure: MUST postpone compaction!
// Before the fix, BIG_INTENT was not checked, so it incorrectly compacted l2 (compact_lsn == 2)
int small_writes = 0;
auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// When under pressure: MUST ALSO postpone compaction!
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// Verify add_compact() rejects stable overwrite with EBUSY
uint32_t compact_mblock = 999999;
uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
assert(res == EBUSY);
// Complete block write
heap.complete_block_write(mblock);
assert(heap.get_fsynced_lsn() == 3);
// Once fsynced, BIG_INTENT is clean_wr base
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.clean_wr && compact_info.clean_wr->lsn == 3);
assert(small_writes == 0);
}
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) DELETE_STABLE(l3) un-fsynced
// -> skip compaction of l2 into l1 EVEN WHEN under pressure,
// because add_compact rejects stable overwrites with EBUSY
_test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
auto obj = heap.read_entry(oid);
uint32_t mblock = 999999;
res = heap.add_delete(obj, &mblock);
assert(res == 0);
heap.start_block_write(mblock);
obj = heap.read_entry(oid);
assert(obj->lsn == 3);
assert(obj->type() == BS_HEAP_DELETE);
assert(obj->is_overwrite());
assert(heap.get_fsynced_lsn() == 2);
// When not under pressure: postpone compaction
int small_writes = 0;
auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// When under pressure: MUST ALSO postpone compaction because DELETE is STABLE!
// Before the fix, this would incorrectly compact l2 (compact_lsn == 2)
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.compact_lsn == 0);
assert(compact_info.compact_version == 0);
assert(!compact_info.clean_wr);
assert(!compact_info.do_delete);
assert(small_writes == 0);
// Verify add_compact() rejects stable overwrite with EBUSY
uint32_t compact_mblock = 999999;
uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
assert(res == EBUSY);
// Complete block write
heap.complete_block_write(mblock);
assert(heap.get_fsynced_lsn() == 3);
// Once fsynced, DELETE is processed
compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(compact_info.do_delete);
assert(!compact_info.clean_wr);
assert(small_writes == 0);
}
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();

View File

@@ -19,20 +19,27 @@ add_library(util STATIC
../../json11/json11.cpp
xxh_x86dispatch.c
)
if (RDMACM_LIBRARIES)
if (RDMACM_LIBRARIES OR (VFN_LIBRARIES AND NVME_LIBRARIES))
target_sources(util PRIVATE
dma_alloc_impl.cpp
)
endif (RDMACM_LIBRARIES)
endif (RDMACM_LIBRARIES OR (VFN_LIBRARIES AND NVME_LIBRARIES))
if (RDMACM_LIBRARIES)
target_sources(util PRIVATE
rdma_alloc.cpp
)
endif (RDMACM_LIBRARIES)
if (VFN_LIBRARIES AND NVME_LIBRARIES)
target_sources(util PRIVATE
vfn_alloc.cpp
)
endif (VFN_LIBRARIES AND NVME_LIBRARIES)
target_compile_options(util PUBLIC -fPIC)
target_link_libraries(util
${LIBURING_LIBRARIES}
${OPENSSL_LIBRARIES}
${VFN_LIBRARIES}
${NVME_LIBRARIES}
)
# libcrc32c.a to isolate the isa-l dependency

File diff suppressed because it is too large Load Diff

View File

@@ -4,6 +4,76 @@
#pragma once
#include <string>
#include <map>
#include <functional>
#include <stdint.h>
#include <time.h>
#ifdef WITH_VFN
#include <nvme/types.h>
#endif
class ring_loop_i;
class disk_i
{
public:
static disk_i *create_and_open(const std::string & path, const std::string & mode, bool disable_flock = false);
static disk_i *create_mock(int mock_fd, uint64_t mock_size, uint64_t mock_sectsize);
static std::string realpath(const std::string & path);
virtual ~disk_i() = default;
virtual void set_ringloop(ring_loop_i *ringloop) = 0;
virtual bool is_off_ring() const = 0;
virtual int get_fd() const = 0;
virtual void close() = 0;
virtual uint64_t get_size() const = 0;
virtual uint64_t get_block_size() const = 0;
virtual uint64_t get_max_io_size() const = 0;
virtual void *alloc_iomem(size_t size) = 0;
virtual void free_iomem(void *buf) = 0;
virtual uint32_t get_queue_space() const = 0;
// all errors are passed as negative errno's
virtual int rw_sync(bool is_wr, uint64_t offset, size_t len, void *buf) = 0;
virtual int flush_sync() = 0;
virtual int discard_sync(uint64_t offset, uint64_t len) = 0;
virtual int rw_async(bool is_wr, uint64_t offset, size_t len, void *buf, bool fua, std::function<void(int res)> cb) = 0;
virtual int flush_async(std::function<void(int res)> cb) = 0;
virtual int discard_async(uint64_t offset, uint64_t len, std::function<void(int res)> cb) = 0;
inline int read_sync(uint64_t offset, size_t len, void *buf)
{
return rw_sync(false, offset, len, buf);
}
inline int write_sync(uint64_t offset, size_t len, void *buf)
{
return rw_sync(true, offset, len, buf);
}
inline int read_async(uint64_t offset, size_t len, void *buf, std::function<void(int res)> cb)
{
return rw_async(false, offset, len, buf, false, cb);
}
inline int write_async(uint64_t offset, size_t len, void *buf, std::function<void(int res)> cb, bool fua = false)
{
return rw_async(true, offset, len, buf, fua, cb);
}
};
void check_size(int fd, uint64_t *size, uint64_t *sectsize, const std::string & name);
#ifdef WITH_VFN
class vfn_disk_i: public disk_i
{
public:
static vfn_disk_i *create();
virtual ~vfn_disk_i() = default;
virtual bool open_dev(const std::string & pci_id) = 0;
virtual bool open_ns(uint32_t ns_num) = 0;
virtual const nvme_id_ctrl* get_id_ctrl() const = 0;
virtual const nvme_id_ns* get_id_ns() const = 0;
virtual uint64_t get_mdts() const = 0;
virtual uint64_t get_cap() const = 0;
static bool is_vfn_path(const std::string & path);
static bool bind_to_vfio(const std::string & dev);
static std::string find_pci_id(const std::string & nvme_serial, uint32_t nvme_nsid);
};
#endif

View File

@@ -52,6 +52,8 @@ void ring_loop_t::register_consumer(ring_consumer_t *consumer)
{
unregister_consumer(consumer);
consumers.push_back(consumer);
if (consumer->is_idle)
idle_consumers++;
}
void ring_loop_t::wakeup()
@@ -65,6 +67,8 @@ void ring_loop_t::unregister_consumer(ring_consumer_t *consumer)
{
if (consumers[i] == consumer)
{
if (consumer->is_idle)
idle_consumers--;
consumers.erase(consumers.begin()+i, consumers.begin()+i+1);
break;
}
@@ -154,10 +158,30 @@ void ring_loop_t::loop()
}
do
{
loop_again = false;
for (int i = 0; i < consumers.size(); i++)
do
{
consumers[i]->loop();
loop_again = false;
for (int i = 0; i < consumers.size(); i++)
{
if (!consumers[i]->is_idle)
consumers[i]->loop();
}
if (immediate_queue.size())
{
immediate_queue2.swap(immediate_queue);
for (auto & cb: immediate_queue2)
cb();
immediate_queue2.clear();
}
} while (loop_again);
if (idle_consumers > 0)
{
// idle consumers are those who poll devices when everything is idle
for (int i = 0; i < consumers.size(); i++)
{
if (consumers[i]->is_idle)
consumers[i]->loop();
}
if (immediate_queue.size())
{
immediate_queue2.swap(immediate_queue);

View File

@@ -32,6 +32,7 @@ struct ring_data_t
struct ring_consumer_t
{
std::function<void(void)> loop;
bool is_idle = false;
};
class __attribute__((visibility("default"))) ring_loop_i
@@ -60,6 +61,7 @@ class __attribute__((visibility("default"))) ring_loop_t: public ring_loop_i
std::vector<ring_consumer_t*> consumers;
std::vector<ring_data_t> ring_datas;
std::vector<int> free_ring_data;
int idle_consumers = 0;
std::mutex mu;
bool mt;
bool in_loop;

56
src/util/vfn_alloc.cpp Normal file
View File

@@ -0,0 +1,56 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// libvfn (IOMMU) variant of our simple allocator
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <vfn/iommu.h>
#include "dma_alloc_impl.h"
static_assert(sizeof(iova_t) == sizeof(void*));
class vfn_allocator_t: public dma_allocator_t
{
protected:
iommu_ctx *ctx = NULL;
void* reg_mr(void *buf, size_t len) override
{
iova_t iova = 0;
if (iommu_map_vaddr(ctx, buf, len, &iova, 0))
{
fprintf(stderr, "iommu_map_vaddr %zu bytes failed: %s\n", len, strerror(errno));
exit(1);
}
return (void*)iova;
}
void dereg_mr(void *buf, size_t len, void *handle) override
{
if (iommu_unmap_vaddr(ctx, buf, &len))
{
fprintf(stderr, "iommu_unmap_vaddr failed: %s\n", strerror(errno));
exit(1);
}
}
public:
vfn_allocator_t(iommu_ctx *ctx, size_t region_size, size_t max_unused)
: dma_allocator_t(region_size, max_unused, 1), ctx(ctx)
{
alignment = page_size;
}
~vfn_allocator_t()
{
free_unused_buffers(0, true);
}
};
dma_allocator_i *vfn_malloc_create(iommu_ctx *ctx, size_t region_size, size_t max_unused)
{
return new vfn_allocator_t(ctx, region_size, max_unused);
}

11
src/util/vfn_alloc.h Normal file
View File

@@ -0,0 +1,11 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// libvfn (IOMMU) variant of our simple allocator
#pragma once
#include "rdma_alloc.h"
struct iommu_ctx;
dma_allocator_i *vfn_malloc_create(iommu_ctx *ctx, size_t rdma_alloc_size, size_t rdma_max_unused);

View File

@@ -72,4 +72,19 @@ qemu-img convert -S 4096 -f qcow2 ./testdata/layer1.qcow2 -O raw ./testdata/bin/
cmp ./testdata/bin/layer1.bin ./testdata/bin/rebased.bin
# Test CAS merge
$VITASTOR_CLI create -s 1M microimg
$VITASTOR_FIO -bs=1M -direct=1 -iodepth=1 -rw=write -image=microimg
$VITASTOR_CLI snap-create microimg@0
$VITASTOR_FIO -bs=16k -direct=1 -iodepth=1 -rw=randwrite -image=microimg -size=16k -loops=1000000 -runtime=10 &
FIO_PID=$!
sleep 5
$VITASTOR_CLI rm microimg@0
wait $FIO_PID
format_green OK