bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fileiotest: limit concurrency

Browse Source 
Otherwise, all of the writes are submitted at once, consuming tons of memory,
and preventing reads from happening in parallel to writes.

Add a semaphore to limit the amount of parallel I/O.
This commit is contained in:
Avi Kivity
2014-08-24 10:30:54 +03:00
parent 91f10b8a9c
commit 7f609a1959
1 changed files with 18 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
fileiotest.cc
View File

@@ -9,26 +9,30 @@ struct file_test {
file_test(file&& f) : f(std::move(f)) {}
file f;
semaphore sem = { 0 };
semaphore par = { 1000 };
};
int main(int ac, char** av) {
static constexpr auto max = 10000;
the_reactor.open_file_dma("testfile.tmp").then([] (file f) {
auto ft = new file_test{std::move(f)};
for (size_t i = 0; i < 10000; ++i) {
auto wbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
std::fill(wbuf.get(), wbuf.get() + 4096, i);
auto wb = wbuf.get();
ft->f.dma_write(i * 4096, wb, 4096).then(
[ft, i, wbuf = std::move(wbuf)] (size_t ret) mutable {
assert(ret == 4096);
auto rbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
auto rb = rbuf.get();
ft->f.dma_read(i * 4096, rb, 4096).then(
[ft, i, rbuf = std::move(rbuf), wbuf = std::move(wbuf)] (size_t) mutable {
bool eq = std::equal(rbuf.get(), rbuf.get() + 4096, wbuf.get());
assert(eq);
ft->sem.signal(1);
for (size_t i = 0; i < max; ++i) {
ft->par.wait().then([ft, i] {
auto wbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
std::fill(wbuf.get(), wbuf.get() + 4096, i);
auto wb = wbuf.get();
ft->f.dma_write(i * 4096, wb, 4096).then(
[ft, i, wbuf = std::move(wbuf)] (size_t ret) mutable {
assert(ret == 4096);
auto rbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
auto rb = rbuf.get();
ft->f.dma_read(i * 4096, rb, 4096).then(
[ft, i, rbuf = std::move(rbuf), wbuf = std::move(wbuf)] (size_t) mutable {
bool eq = std::equal(rbuf.get(), rbuf.get() + 4096, wbuf.get());
assert(eq);
ft->sem.signal(1);
ft->par.signal();
});
});
});
}