diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index e8738a50..c839c1f3 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -558,6 +558,24 @@ jobs: echo "" done + test_dd: + runs-on: ubuntu-latest + needs: build + container: ${{env.TEST_IMAGE}}:${{github.sha}} + steps: + - name: Run test + id: test + timeout-minutes: 3 + run: /root/vitastor/tests/test_dd.sh + - name: Print logs + if: always() && steps.test.outcome == 'failure' + run: | + for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do + echo "-------- $i --------" + cat $i + echo "" + done + test_root_node: runs-on: ubuntu-latest needs: build diff --git a/src/cmd/CMakeLists.txt b/src/cmd/CMakeLists.txt index ba4f970a..100b7c27 100644 --- a/src/cmd/CMakeLists.txt +++ b/src/cmd/CMakeLists.txt @@ -11,6 +11,7 @@ add_library(vitastor_cli STATIC cli_fix.cpp cli_ls.cpp cli_create.cpp + cli_dd.cpp cli_modify.cpp cli_modify_osd.cpp cli_osd_tree.cpp diff --git a/src/cmd/cli.cpp b/src/cmd/cli.cpp index 47ff1c12..63cc5113 100644 --- a/src/cmd/cli.cpp +++ b/src/cmd/cli.cpp @@ -62,6 +62,31 @@ static const char* help_text = " Other options:\n" " --down-ok Continue deletion/merging even if some data will be left on unavailable OSDs.\n" "\n" + "vitastor-cli dd [iimg= | if=] [oimg= | of=] [bs=1M]\n" + " [count=N] [seek/oseek=N] [skip/iseek=M] [iodepth=N] [status=progress]\n" + " [conv=nocreat,noerror,nofsync,trunc,nosparse] [iflag=direct] [oflag=direct,append]\n" + " Copy data between Vitastor images, files and pipes.\n" + " Options can be specified in classic dd style (key=value) or like usual (--key value).\n" + " iimg= Copy from Vitastor image \n" + " if= Copy from file \n" + " oimg= Copy to Vitastor image \n" + " of= Copy to file \n" + " bs=1M Set copy block size\n" + " count=N Copy only N input blocks. If N ends in B it counts bytes, not blocks\n" + " seek/oseek=N Skip N output blocks. If N ends in B it counts bytes, not blocks\n" + " skip/iseek=N Skip N input blocks. If N ends in B it counts bytes, not blocks\n" + " iodepth=N Send N reads or writes in parallel (default 4)\n" + " status=LEVEL The LEVEL of information to print to stderr: none/noxfer/progress\n" + " size=N Specify size for the created output file/image (defaults to input size)\n" + " iflag=direct For files only: use direct I/O\n" + " oflag=direct For files only: use direct I/O\n" + " oflag=append For files only: append to output file\n" + " conv=nocreat Do not create output file/image\n" + " conv=trunc For files only: truncate output file\n" + " conv=noerror Continue read after errors\n" + " conv=nofsync Do not call fsync before finishing (default behaviour is fsync)\n" + " conv=nosparse Write all output blocks including all-zero blocks\n" + "\n" "vitastor-cli flatten \n" " Flatten a layer, i.e. merge data and detach it from parents.\n" "\n" @@ -382,6 +407,20 @@ static int run(cli_tool_t *p, json11::Json::object cfg) } action_cb = p->start_flatten(cfg); } + else if (cmd[0] == "dd") + { + // Read or write to/from cluster + for (int i = 0; i < cmd.size(); i++) + { + auto arg = cmd[i].string_value(); + ssize_t p = arg.find("="); + if (p != std::string::npos) + { + cfg[arg.substr(0, p)] = arg.substr(p+1); + } + } + action_cb = p->start_dd(cfg); + } else if (cmd[0] == "rm") { // Remove multiple snapshots and rebase their children diff --git a/src/cmd/cli.h b/src/cmd/cli.h index 7f3be58c..2c02f2bc 100644 --- a/src/cmd/cli.h +++ b/src/cmd/cli.h @@ -76,6 +76,7 @@ public: std::function start_rm_data(json11::Json); std::function start_rm_osd(json11::Json); std::function start_status(json11::Json); + std::function start_dd(json11::Json); // Should be called like loop_and_wait(start_status(), ) void loop_and_wait(std::function loop_cb, std::function complete_cb); diff --git a/src/cmd/cli_create.cpp b/src/cmd/cli_create.cpp index e1b1a5e0..9d211587 100644 --- a/src/cmd/cli_create.cpp +++ b/src/cmd/cli_create.cpp @@ -26,7 +26,7 @@ struct image_creator_t std::string new_pool_name; std::string image_name, new_snap, new_parent; json11::Json new_meta; - uint64_t size; + uint64_t size = 0; bool force = false; bool force_size = false; @@ -554,10 +554,10 @@ std::function cli_tool_t::start_create(json11::Json cfg) image_creator->new_snap = cfg["snapshot"].string_value(); } image_creator->new_parent = cfg["parent"].string_value(); - if (cfg["size"].string_value() != "") + if (!cfg["size"].is_null()) { bool ok; - image_creator->size = parse_size(cfg["size"].string_value(), &ok); + image_creator->size = parse_size(cfg["size"].as_string(), &ok); if (!ok) { return [size = cfg["size"].string_value()](cli_result_t & result) diff --git a/src/cmd/cli_dd.cpp b/src/cmd/cli_dd.cpp new file mode 100644 index 00000000..53a7a67e --- /dev/null +++ b/src/cmd/cli_dd.cpp @@ -0,0 +1,950 @@ +// Copyright (c) Vitaliy Filippov, 2019+ +// License: VNPL-1.1 (see README.md for details) + +#include "cli.h" +#include "cluster_client.h" +#include "str_util.h" +#include +#include +#include +#include +#include + +// Copy data between Vitastor images, files and pipes +// A showpiece implementation of dd :-) with iodepth, asynchrony, pipe support and so on + +struct dd_buf_t +{ + void *buf = NULL; + uint64_t offset = 0, len = 0, max = 0; + + dd_buf_t(uint64_t offset, uint64_t max) + { + this->offset = offset; + this->max = max; + this->buf = malloc_or_die(max); + } + + ~dd_buf_t() + { + free(this->buf); + this->buf = NULL; + } +}; + +struct dd_in_info_t +{ + // in + std::string iimg, ifile; + bool in_direct = false; + bool detect_size = true; + + // out + cli_result_t result; + inode_watch_t *iwatch = NULL; + int ifd = -1; + uint64_t in_size = 0; + uint32_t in_granularity = 0; + bool in_seekable = false; + + void open_input(cli_tool_t *parent) + { + in_seekable = true; + if (iimg != "") + { + iwatch = parent->cli->st_cli.watch_inode(iimg); + if (!iwatch->cfg.num) + { + result = (cli_result_t){ .err = ENOENT, .text = "Image "+iimg+" does not exist" }; + parent->cli->st_cli.close_watch(iwatch); + iwatch = NULL; + return; + } + auto pool_it = parent->cli->st_cli.pool_config.find(INODE_POOL(iwatch->cfg.num)); + if (pool_it == parent->cli->st_cli.pool_config.end()) + { + result = (cli_result_t){ .err = ENOENT, .text = "Pool of image "+iimg+" does not exist" }; + parent->cli->st_cli.close_watch(iwatch); + iwatch = NULL; + return; + } + in_granularity = pool_it->second.bitmap_granularity; + if (detect_size) + { + in_size = iwatch->cfg.size; + } + } + else if (ifile != "") + { + ifd = open(ifile.c_str(), (in_direct ? O_DIRECT : 0) | O_RDONLY); + if (ifd < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to open "+ifile+": "+std::string(strerror(errno)) }; + return; + } + if (detect_size) + { + struct stat st; + if (fstat(ifd, &st) < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to stat "+ifile+": "+std::string(strerror(errno)) }; + close(ifd); + ifd = -1; + return; + } + if (S_ISREG(st.st_mode)) + { + in_size = st.st_size; + } + else if (S_ISBLK(st.st_mode)) + { + if (ioctl(ifd, BLKGETSIZE64, &in_size) < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to get "+ifile+" size: "+std::string(strerror(errno)) }; + close(ifd); + ifd = -1; + return; + } + } + } + if (in_direct) + { + in_granularity = 512; + } + if (lseek(ifd, 1, SEEK_SET) == (off_t)-1) + { + in_seekable = false; + } + else + { + lseek(ifd, 0, SEEK_SET); + } + } + else + { + ifd = 0; + in_seekable = false; + } + } + + void close_input(cli_tool_t *parent) + { + if (iimg != "") + { + parent->cli->st_cli.close_watch(iwatch); + iwatch = NULL; + } + else if (ifile != "") + { + close(ifd); + ifd = -1; + } + } +}; + +struct dd_out_info_t +{ + std::string oimg, ofile; + std::string out_pool; + bool out_direct = false; + bool out_create = true; + bool out_trunc = false; + bool out_append = false; + bool end_fsync = true; + uint64_t out_size = 0; + + cli_result_t result; + inode_watch_t *owatch = NULL; + int ofd = -1; + uint32_t out_granularity = 0; + bool out_seekable = false; + std::function sub_cb; + + pool_config_t *find_pool(cli_tool_t *parent, const std::string & name) + { + if (name == "" && parent->cli->st_cli.pool_config.size() == 1) + { + return &parent->cli->st_cli.pool_config.begin()->second; + } + for (auto & pp: parent->cli->st_cli.pool_config) + { + if (pp.second.name == name) + { + return &pp.second; + } + } + return NULL; + } + + bool open_output(cli_tool_t *parent, int & state, int base_state) + { + if (state == base_state) + goto resume_1; + else if (state == base_state+1) + goto resume_2; + if (oimg != "") + { + out_seekable = true; + owatch = parent->cli->st_cli.watch_inode(oimg); + if (owatch->cfg.num) + { + auto pool_it = parent->cli->st_cli.pool_config.find(INODE_POOL(owatch->cfg.num)); + if (pool_it == parent->cli->st_cli.pool_config.end()) + { + result = (cli_result_t){ .err = ENOENT, .text = "Pool of image "+oimg+" does not exist" }; + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + out_granularity = pool_it->second.bitmap_granularity; + } + else + { + auto pool_cfg = find_pool(parent, out_pool); + if (pool_cfg) + { + out_granularity = pool_cfg->bitmap_granularity; + } + else + { + result = (cli_result_t){ .err = ENOENT, .text = "Pool to create output image "+oimg+" is not specified" }; + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + } + if (out_size % 4096) + { + out_size += (4096 - (out_size % 4096)); + } + if (!owatch->cfg.num) + { + if (!out_create) + { + result = (cli_result_t){ .err = ENOENT, .text = "Image "+oimg+" does not exist" }; + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + if (!out_size) + { + result = (cli_result_t){ .err = ENOENT, .text = "Input size is unknown, specify size to create output image "+oimg }; + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + // Create output image + sub_cb = parent->start_create(json11::Json::object { + { "image", oimg }, + { "pool", out_pool }, + { "size", out_size }, + }); + } + else if (owatch->cfg.size < out_size || out_trunc) + { + if (!out_size) + { + result = (cli_result_t){ .err = ENOENT, .text = "Input size is unknown, specify size to truncate output image" }; + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + // Resize output image + sub_cb = parent->start_create(json11::Json::object { + { "image", oimg }, + { "resize", out_size }, + }); + } + else + { + // ok + return true; + } + // Wait for sub-command +resume_1: + while (!sub_cb(result)) + { + state = base_state; + return false; + } + sub_cb = NULL; + if (result.err) + { + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + return true; + } + // Wait until output image actually appears +resume_2: + while (!owatch->cfg.num) + { + state = base_state+1; + return false; + } + } + else if (ofile != "") + { + ofd = open(ofile.c_str(), (out_direct ? O_DIRECT : 0) | (out_append ? O_APPEND : O_RDWR) | (out_create ? O_CREAT : 0), 0666); + if (ofd < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to open "+ofile+": "+std::string(strerror(errno)) }; + return true; + } + if (out_trunc && ftruncate(ofd, out_size) < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to truncate "+ofile+": "+std::string(strerror(errno)) }; + return true; + } + if (out_direct) + { + out_granularity = 512; + } + out_seekable = !out_append; + } + else + { + ofd = 1; + out_seekable = false; + } + return true; + } + + bool fsync_output(cli_tool_t *parent, int & state, int base_state) + { + if (state == base_state) + goto resume_1; + if (oimg != "") + { + { + cluster_op_t *sync_op = new cluster_op_t; + sync_op->opcode = OSD_OP_SYNC; + parent->waiting++; + sync_op->callback = [this, parent](cluster_op_t *sync_op) + { + parent->waiting--; + delete sync_op; + parent->ringloop->wakeup(); + }; + parent->cli->execute(sync_op); + } + resume_1: + if (parent->waiting > 0) + { + state = base_state; + return false; + } + } + else + { + int res = fsync(ofd); + if (res < 0) + { + result = (cli_result_t){ .err = errno, .text = "Failed to fsync "+ofile+": "+std::string(strerror(errno)) }; + } + } + return true; + } + + void close_output(cli_tool_t *parent) + { + if (oimg != "") + { + parent->cli->st_cli.close_watch(owatch); + owatch = NULL; + } + else + { + if (ofile != "") + close(ofd); + ofd = -1; + } + } +}; + +struct cli_dd_t +{ + cli_tool_t *parent; + + dd_in_info_t iinfo; + dd_out_info_t oinfo; + + uint64_t blocksize = 0, bytelimit = 0, iseek = 0, oseek = 0, iodepth = 0; + bool end_status = true, ignore_errors = false; + bool write_zero = false; + + uint64_t in_iodepth = 0, out_iodepth = 0; + uint64_t read_offset = 0, read_end = 0; + std::vector read_buffers, short_reads, short_writes; + bool in_eof = false; + uint64_t written_size = 0; + uint64_t written_progress = 0; + timespec tv_begin = {}, tv_progress = {}; + int state = 0; + int copy_error = 0; + int in_waiting = 0, out_waiting = 0; + cli_result_t result; + + bool is_done() + { + return state == 100; + } + + int skip_read(int fd, uint64_t to_skip) + { + void *buf = malloc_or_die(blocksize); + while (to_skip > 0) + { + auto res = read(fd, buf, blocksize < to_skip ? blocksize : to_skip); + if (res <= 0) + { + return res == 0 ? -EPIPE : -errno; + } + to_skip -= res; + } + free(buf); + return 0; + } + + void vitastor_read_bitmap(dd_buf_t *cur_read) + { + cluster_op_t *read_op = new cluster_op_t; + read_op->opcode = OSD_OP_READ_CHAIN_BITMAP; + read_op->inode = iinfo.iwatch->cfg.num; + // FIXME: Support unaligned read? + read_op->offset = cur_read->offset + iseek; + read_op->len = cur_read->max; + in_waiting++; + read_op->callback = [this, cur_read](cluster_op_t *read_op) + { + in_waiting--; + if (read_op->retval < 0) + { + if (ignore_errors) + { + fprintf( + stderr, "Failed to read bitmap for %lu bytes from image %s at offset %lu: %s (code %d)\n", + read_op->len, iinfo.iimg.c_str(), read_op->offset, + strerror(read_op->retval < 0 ? -read_op->retval : EIO), read_op->retval + ); + } + else + { + copy_error = read_op->retval < 0 ? -read_op->retval : EIO; + } + delete cur_read; + } + else if (!is_zero(read_op->bitmap_buf, read_op->len/iinfo.in_granularity/8)) + { + vitastor_read(cur_read); + } + else + { + delete cur_read; + } + delete read_op; + parent->ringloop->wakeup(); + }; + parent->cli->execute(read_op); + } + + void vitastor_read(dd_buf_t *cur_read) + { + cluster_op_t *read_op = new cluster_op_t; + read_op->opcode = OSD_OP_READ; + read_op->inode = iinfo.iwatch->cfg.num; + // FIXME: Support unaligned read? + read_op->offset = cur_read->offset + iseek; + read_op->len = cur_read->max; + read_op->iov.push_back(cur_read->buf, cur_read->max); + in_waiting++; + read_op->callback = [this, cur_read](cluster_op_t *read_op) + { + in_waiting--; + if (read_op->retval != read_op->len) + { + if (ignore_errors) + { + fprintf( + stderr, "Failed to read %lu bytes from image %s at offset %lu: %s (code %d)\n", + read_op->len, iinfo.iimg.c_str(), read_op->offset, + strerror(read_op->retval < 0 ? -read_op->retval : EIO), read_op->retval + ); + } + else + { + copy_error = read_op->retval < 0 ? -read_op->retval : EIO; + } + delete cur_read; + } + else + { + cur_read->len = cur_read->max; + add_finished_read(cur_read); + } + delete read_op; + parent->ringloop->wakeup(); + }; + parent->cli->execute(read_op); + } + + bool add_read_op() + { + if (iinfo.iwatch) + { + dd_buf_t *cur_read = new dd_buf_t(read_offset, read_offset + blocksize > read_end ? read_end - read_offset : blocksize); + read_offset += cur_read->max; + in_eof = read_offset >= read_end; + cur_read->len = cur_read->max; + if (!write_zero) + { + vitastor_read_bitmap(cur_read); + } + else + { + vitastor_read(cur_read); + } + } + else + { + io_uring_sqe *sqe = parent->ringloop->get_sqe(); + if (!sqe) + { + return false; + } + dd_buf_t *cur_read; + if (short_reads.size()) + { + cur_read = short_reads[0]; + short_reads.erase(short_reads.begin(), short_reads.begin()+1); + // reset eof flag + if (!short_reads.size() && read_offset >= read_end) + in_eof = true; + } + else + { + cur_read = new dd_buf_t(read_offset, iinfo.in_seekable && read_offset + blocksize > read_end ? read_end-read_offset : blocksize); + read_offset += cur_read->max; + if (read_offset >= read_end) + in_eof = true; + } + ring_data_t *data = ((ring_data_t*)sqe->user_data); + data->iov = (iovec){ cur_read->buf + cur_read->len, cur_read->max - cur_read->len }; + my_uring_prep_readv(sqe, iinfo.ifd, &data->iov, 1, iinfo.in_seekable ? iseek + cur_read->offset + cur_read->len : -1); + in_waiting++; + data->callback = [this, cur_read](ring_data_t *data) + { + in_waiting--; + if (data->res < 0) + { + if (ignore_errors) + { + fprintf( + stderr, "Failed to read %lu bytes from %s at offset %lu: %s (code %d)\n", + data->iov.iov_len, iinfo.ifile == "" ? "stdin" : iinfo.ifile.c_str(), cur_read->offset, + strerror(-data->res), data->res + ); + } + else + { + copy_error = -data->res; + } + } + else if (data->res == 0) + { + in_eof = true; + } + if (data->res <= 0) + { + if (cur_read->len > 0) + add_finished_read(cur_read); + else + delete cur_read; + } + else + { + cur_read->len += data->res; + if (cur_read->len < cur_read->max) + { + // short read, retry + short_reads.push_back(cur_read); + // reset eof flag to signal that there's still something to read + in_eof = false; + } + else + { + add_finished_read(cur_read); + } + } + parent->ringloop->wakeup(); + }; + } + return true; + } + + void add_finished_read(dd_buf_t *cur_read) + { + if (!write_zero && is_zero(cur_read->buf, cur_read->max)) + { + // do not write all-zero buffer + delete cur_read; + return; + } + auto it = std::lower_bound(read_buffers.begin(), read_buffers.end(), cur_read, [](dd_buf_t *item, dd_buf_t *ref) + { + return item->offset < ref->offset; + }); + read_buffers.insert(it, cur_read); + } + + bool add_write_op() + { + dd_buf_t *cur_read; + if (short_writes.size()) + { + cur_read = short_writes[0]; + short_writes.erase(short_writes.begin(), short_writes.begin()+1); + } + else + { + cur_read = read_buffers[0]; + if (!oinfo.out_seekable && cur_read->offset > written_size) + { + // can't write - input buffers are out of order + return false; + } + cur_read->max = cur_read->len; + cur_read->len = 0; + read_buffers.erase(read_buffers.begin(), read_buffers.begin()+1); + } + if (oinfo.owatch) + { + cluster_op_t *write_op = new cluster_op_t; + write_op->opcode = OSD_OP_WRITE; + write_op->inode = oinfo.owatch->cfg.num; + // FIXME: Support unaligned write? + write_op->offset = cur_read->offset + oseek; + write_op->len = cur_read->max; + write_op->iov.push_back(cur_read->buf, cur_read->max); + out_waiting++; + write_op->callback = [this, cur_read](cluster_op_t *write_op) + { + out_waiting--; + if (write_op->retval != write_op->len) + { + if (ignore_errors) + { + fprintf( + stderr, "Failed to write %lu bytes to image %s at offset %lu: %s (code %d)\n", + write_op->len, oinfo.oimg.c_str(), write_op->offset, + strerror(write_op->retval < 0 ? -write_op->retval : EIO), write_op->retval + ); + } + else + { + copy_error = write_op->retval < 0 ? -write_op->retval : EIO; + } + } + else + { + written_size += write_op->len; + } + delete cur_read; + delete write_op; + parent->ringloop->wakeup(); + }; + parent->cli->execute(write_op); + } + else + { + io_uring_sqe *sqe = parent->ringloop->get_sqe(); + if (!sqe) + { + return false; + } + ring_data_t *data = ((ring_data_t*)sqe->user_data); + data->iov = (iovec){ .iov_base = cur_read->buf+cur_read->len, .iov_len = cur_read->max-cur_read->len }; + my_uring_prep_writev(sqe, oinfo.ofd, &data->iov, 1, oinfo.out_seekable ? cur_read->offset+cur_read->len+oseek : -1); + out_waiting++; + data->callback = [this, cur_read](ring_data_t *data) + { + out_waiting--; + if (data->res < 0) + { + if (ignore_errors) + { + fprintf( + stderr, "Failed to write %lu bytes to %s at offset %lu: %s (code %d)\n", + data->iov.iov_len, oinfo.ofile == "" ? "stdout" : oinfo.ofile.c_str(), + oinfo.out_seekable ? cur_read->offset+cur_read->len+oseek : 0, + strerror(-data->res), data->res + ); + } + else + { + copy_error = -data->res; + } + delete cur_read; + } + else + { + written_size += data->res; + cur_read->len += data->res; + if (cur_read->len < cur_read->max) + short_writes.push_back(cur_read); + else + delete cur_read; + } + parent->ringloop->wakeup(); + }; + } + return true; + } + + void print_progress(bool end) + { + if (!parent->progress && (!end || !end_status && !parent->json_output)) + { + return; + } + timespec tv_now; + clock_gettime(CLOCK_REALTIME, &tv_now); + double sec_delta = ((tv_now.tv_sec - tv_progress.tv_sec) + (double)(tv_now.tv_nsec - tv_progress.tv_nsec)/1000000000.0); + if (sec_delta < 1 && !end) + { + return; + } + double sec_total = ((tv_now.tv_sec - tv_begin.tv_sec) + (double)(tv_now.tv_nsec - tv_begin.tv_nsec)/1000000000.0); + uint64_t delta = written_size-written_progress; + tv_progress = tv_now; + written_progress = written_size; + if (end) + { + char buf[256]; + snprintf( + buf, sizeof(buf), "%lu bytes (%s) copied, %.1f s, %sB/s", + written_size, format_size(written_size).c_str(), sec_total, + format_size((uint64_t)(written_size/sec_total), true).c_str() + ); + if (parent->json_output) + { + if (parent->progress) + fprintf(stderr, "\n"); + result.text = buf; + result.data = json11::Json::object { + { "copied", written_size }, + { "seconds", sec_total }, + }; + } + else + { + fprintf(stderr, (parent->progress ? ("\r%s\033[K\n") : ("%s\n")), buf); + } + } + else + { + fprintf( + stderr, "\r%lu bytes (%s) copied, %.1f s, %sB/s, avg %sB/s\033[K", + written_size, format_size(written_size).c_str(), sec_total, + format_size((uint64_t)(delta/sec_delta), true).c_str(), + format_size((uint64_t)(written_size/sec_total), true).c_str() + ); + } + } + + void loop() + { + if (state == 1) + goto resume_1; + else if (state == 2) + goto resume_2; + else if (state == 3) + goto resume_3; + else if (state == 4) + goto resume_4; + if ((oinfo.oimg != "" && oinfo.ofile != "") || (iinfo.iimg != "" && iinfo.ifile != "")) + { + result = (cli_result_t){ .err = EINVAL, .text = "Image and file can't be specified at the same time" }; + state = 100; + return; + } + if ((iinfo.iimg != "" ? "i"+iinfo.iimg : "f"+iinfo.ifile) == (oinfo.oimg != "" ? "i"+oinfo.oimg : "f"+oinfo.ofile)) + { + result = (cli_result_t){ .err = EINVAL, .text = "Input and output image/file can't be equal" }; + state = 100; + return; + } + // Open input and output + iinfo.open_input(parent); + if (iinfo.result.err) + { + result = iinfo.result; + state = 100; + return; + } + if (!oinfo.out_size) + { + oinfo.out_size = oseek + (iinfo.in_seekable && (!bytelimit || iinfo.in_size-iseek < bytelimit) ? iinfo.in_size-iseek : bytelimit); + } +resume_1: +resume_2: + if (!oinfo.open_output(parent, state, 1)) + { + return; + } + if (oinfo.result.err) + { + iinfo.close_input(parent); + result = oinfo.result; + state = 100; + return; + } + // Copy data + if (iinfo.in_seekable && iseek >= iinfo.in_size) + { + result = (cli_result_t){ .err = -EINVAL, .text = "Input seek position is beyond end of input" }; + goto close_end; + } + if (!iinfo.iwatch && !iinfo.in_seekable && iseek) + { + // Read and ignore some data from input + int res = skip_read(iinfo.ifd, iseek); + if (res < 0) + { + result = (cli_result_t){ .err = -res, .text = "Failed to skip "+std::to_string(iseek)+" input bytes: "+std::string(strerror(-res)) }; + goto close_end; + } + } + in_iodepth = iinfo.in_seekable ? iodepth : 1; + out_iodepth = oinfo.out_seekable ? iodepth : 1; + write_zero = write_zero || !oinfo.out_seekable; + oinfo.end_fsync = oinfo.end_fsync && oinfo.out_seekable; + read_offset = 0; + read_end = iinfo.in_seekable ? iinfo.in_size-iseek : 0; + if (bytelimit && (!read_end || read_end > bytelimit)) + read_end = bytelimit; + clock_gettime(CLOCK_REALTIME, &tv_begin); + tv_progress = tv_begin; +resume_3: + while ((ignore_errors || !copy_error) && (!in_eof || read_buffers.size() || in_waiting > 0 || out_waiting > 0)) + { + print_progress(false); + while ((ignore_errors || !copy_error) && + (!in_eof && in_waiting < in_iodepth && read_buffers.size() < out_iodepth || + read_buffers.size() && out_waiting < out_iodepth)) + { + if (!in_eof && in_waiting < in_iodepth && read_buffers.size() < out_iodepth) + { + if (!add_read_op()) + { + break; + } + } + if (read_buffers.size() && out_waiting < out_iodepth) + { + if (!add_write_op()) + { + break; + } + } + } + if (in_waiting > 0 || out_waiting > 0) + { + state = 3; + return; + } + } + if (oinfo.end_fsync) + { +resume_4: + if (!oinfo.fsync_output(parent, state, 4)) + { + return; + } + } + print_progress(true); +close_end: + oinfo.close_output(parent); + iinfo.close_input(parent); + // Done + result.err = copy_error; + state = 100; + } +}; + +// parse B or blocks of size `bs` +static uint64_t parse_blocks(json11::Json v, uint64_t bs, uint64_t def) +{ + uint64_t res; + if (!v.is_string() && !v.is_number() || + v.is_string() && v.string_value() == "" || + v.is_number() && !v.uint64_value()) + return def; + auto num = v.uint64_value(); + if (num) + return num * bs; + auto s = v.string_value(); + if (s != "" && (s[s.size()-1] == 'b' || s[s.size()-1] == 'B')) + res = stoull_full(s.substr(0, s.size()-1)); + else + res = parse_size(s); + return res; +} + +std::function cli_tool_t::start_dd(json11::Json cfg) +{ + auto dd = new cli_dd_t(); + dd->parent = this; + dd->iinfo.iimg = cfg["iimg"].string_value(); + dd->oinfo.oimg = cfg["oimg"].string_value(); + dd->iinfo.ifile = cfg["if"].string_value(); + dd->oinfo.ofile = cfg["of"].string_value(); + dd->blocksize = parse_size(cfg["bs"].string_value()); + if (!dd->blocksize) + dd->blocksize = 1048576; + dd->bytelimit = parse_blocks(cfg["count"], dd->blocksize, 0); + dd->oseek = parse_blocks(cfg["oseek"], dd->blocksize, 0); + if (!dd->oseek) + dd->oseek = parse_blocks(cfg["seek"], dd->blocksize, 0); + dd->iseek = parse_blocks(cfg["oseek"], dd->blocksize, 0); + if (!dd->iseek) + dd->iseek = parse_blocks(cfg["skip"], dd->blocksize, 0); + dd->iodepth = cfg["iodepth"].uint64_value(); + if (!dd->iodepth) + dd->iodepth = 4; + if (cfg["status"] == "none") + dd->end_status = false; + else if (cfg["status"] == "progress") + progress = true; + dd->iinfo.detect_size = cfg["size"].is_null(); + dd->oinfo.out_size = parse_size(cfg["size"].as_string()); + std::vector conv = explode(",", cfg["conv"].string_value(), true); + if (std::find(conv.begin(), conv.end(), "nofsync") != conv.end()) + dd->oinfo.end_fsync = false; + if (std::find(conv.begin(), conv.end(), "trunc") != conv.end()) + dd->oinfo.out_trunc = true; + if (std::find(conv.begin(), conv.end(), "nocreat") != conv.end()) + dd->oinfo.out_create = false; + if (std::find(conv.begin(), conv.end(), "noerror") != conv.end()) + dd->ignore_errors = true; + if (std::find(conv.begin(), conv.end(), "nosparse") != conv.end()) + dd->write_zero = true; + conv = explode(",", cfg["iflag"].string_value(), true); + if (std::find(conv.begin(), conv.end(), "direct") != conv.end()) + dd->iinfo.in_direct = true; + conv = explode(",", cfg["oflag"].string_value(), true); + if (std::find(conv.begin(), conv.end(), "direct") != conv.end()) + dd->oinfo.out_direct = true; + if (std::find(conv.begin(), conv.end(), "append") != conv.end()) + dd->oinfo.out_append = true; + return [dd](cli_result_t & result) + { + dd->loop(); + if (dd->is_done()) + { + result = dd->result; + delete dd; + return true; + } + return false; + }; +} diff --git a/src/kv/kv_db.cpp b/src/kv/kv_db.cpp index e2f588c9..57af47ad 100644 --- a/src/kv/kv_db.cpp +++ b/src/kv/kv_db.cpp @@ -671,17 +671,6 @@ void kv_db_t::stop_writing_new(uint64_t offset) } } -static bool is_zero(void *buf, int size) -{ - assert(!(size % 8)); - size /= 8; - uint64_t *ptr = (uint64_t*)buf; - for (int i = 0; i < size/8; i++) - if (ptr[i]) - return false; - return true; -} - // Find approximate index size // Phase 1: try 2^i-1 for i=0,1,2,... * ino_block_size // Phase 2: binary search between 2^(N-1)-1 and 2^N-1 * ino_block_size diff --git a/src/util/str_util.cpp b/src/util/str_util.cpp index 505f73ff..3f823f6c 100644 --- a/src/util/str_util.cpp +++ b/src/util/str_util.cpp @@ -485,3 +485,21 @@ std::string format_datetime(uint64_t unixtime) int len = strftime(buf, 128, "%Y-%m-%d %H:%M:%S", <); return std::string(buf, len); } + +bool is_zero(void *buf, size_t size) +{ + size_t i = 0; + while (i <= size-8) + { + if (*(uint64_t*)((uint8_t*)buf + i)) + return false; + i += 8; + } + while (i < size) + { + if (*((uint8_t*)buf + i)) + return false; + i++; + } + return true; +} diff --git a/src/util/str_util.h b/src/util/str_util.h index c789627b..5e8b5cb3 100644 --- a/src/util/str_util.h +++ b/src/util/str_util.h @@ -31,3 +31,4 @@ std::string auto_addslashes(const std::string & str, const char *toescape = "\\\ std::string addslashes(const std::string & str, const char *toescape = "\\\""); std::string realpath_str(std::string path, bool nofail = true); std::string format_datetime(uint64_t unixtime); +bool is_zero(void *buf, size_t size); diff --git a/tests/run_tests.sh b/tests/run_tests.sh index 97d730e5..b193cb9e 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -46,6 +46,8 @@ IMMEDIATE_COMMIT=1 ./test_rebalance_verify.sh SCHEME=ec ./test_rebalance_verify.sh SCHEME=ec IMMEDIATE_COMMIT=1 ./test_rebalance_verify.sh +./test_dd.sh + ./test_root_node.sh ./test_switch_primary.sh diff --git a/tests/test_dd.sh b/tests/test_dd.sh new file mode 100755 index 00000000..87346b98 --- /dev/null +++ b/tests/test_dd.sh @@ -0,0 +1,20 @@ +#!/bin/bash -ex + +. `dirname $0`/run_3osds.sh + +# pipe in - pipe out +dd if=/dev/urandom of=./testdata/testfile bs=1M count=128 +build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd oimg=testimg iodepth=4 bs=1M count=128 < ./testdata/testfile +build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd iimg=testimg iodepth=4 bs=1M count=128 > ./testdata/testfile1 +diff ./testdata/testfile ./testdata/testfile1 +rm ./testdata/testfile1 + +# snapshot +dd if=/dev/urandom of=./testdata/over bs=1M count=4 +dd if=./testdata/over of=./testdata/testfile bs=1M seek=17 conv=notrunc +build/src/cmd/vitastor-cli --etcd_address $ETCD_URL snap-create testimg@snap1 +build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd iodepth=4 if=./testdata/over oimg=testimg bs=1M seek=17 +build/src/cmd/vitastor-cli --etcd_address $ETCD_URL dd iodepth=4 iimg=testimg of=./testdata/testfile1 +diff ./testdata/testfile ./testdata/testfile1 + +format_green OK