9:42 a.m.
Make this a (fairly) abstract structure. At least hide the subtype
fields from the main program. This change is pure refactoring and
doesn’t change the semantics.
---
copy/file-ops.c | 164 +++++++++++++++++--
copy/main.c | 315 ++++++++----------------------------
copy/multi-thread-copying.c | 104 ++++++------
copy/nbd-ops.c | 248 +++++++++++++++++++++++++---
copy/nbdcopy.h | 53 +++---
copy/null-ops.c | 50 +++++-
copy/pipe-ops.c | 64 +++++++-
copy/synch-copying.c | 30 ++--
8 files changed, 649 insertions(+), 379 deletions(-)
diff --git a/copy/file-ops.c b/copy/file-ops.c
index 2a239d0..81b08ce 100644
--- a/copy/file-ops.c
+++ b/copy/file-ops.c
@@ -36,35 +36,159 @@
#include "isaligned.h"
#include "nbdcopy.h"
+static struct rw_ops file_ops;
+
+struct rw_file {
+ struct rw rw;
+ int fd;
+ struct stat stat;
+ bool seek_hole_supported;
+ int sector_size;
+};
+
+static bool
+seek_hole_supported (int fd)
+{
+#ifndef SEEK_HOLE
+ return false;
+#else
+ off_t r = lseek (fd, 0, SEEK_HOLE);
+ return r >= 0;
+#endif
+}
+
+struct rw *
+file_create (const char *name, const struct stat *stat, int fd)
+{
+ struct rw_file *rwf = calloc (1, sizeof *rwf);
+ if (rwf == NULL) { perror ("calloc"); exit (EXIT_FAILURE); }
+
+ rwf->rw.ops = &file_ops;
+ rwf->rw.name = name;
+ rwf->stat = *stat;
+ rwf->fd = fd;
+
+ if (S_ISBLK (stat->st_mode)) {
+ /* Block device. */
+ rwf->rw.size = lseek (fd, 0, SEEK_END);
+ if (rwf->rw.size == -1) {
+ perror ("lseek");
+ exit (EXIT_FAILURE);
+ }
+ if (lseek (fd, 0, SEEK_SET) == -1) {
+ perror ("lseek");
+ exit (EXIT_FAILURE);
+ }
+ rwf->seek_hole_supported = seek_hole_supported (fd);
+ rwf->sector_size = 4096;
+#ifdef BLKSSZGET
+ if (ioctl (fd, BLKSSZGET, &rwf->sector_size))
+ fprintf (stderr, "warning: cannot get sector size: %s: %m", name);
+#endif
+ }
+ else if (S_ISREG (stat->st_mode)) {
+ /* Regular file. */
+ rwf->rw.size = stat->st_size;
+ rwf->seek_hole_supported = seek_hole_supported (fd);
+ }
+ else
+ abort ();
+
+ return (struct rw *) rwf;
+}
+
static void
file_close (struct rw *rw)
{
- if (close (rw->u.local.fd) == -1) {
+ struct rw_file *rwf = (struct rw_file *)rw;
+
+ if (close (rwf->fd) == -1) {
fprintf (stderr, "%s: close: %m\n", rw->name);
exit (EXIT_FAILURE);
}
+ free (rw);
+}
+
+static void
+file_truncate (struct rw *rw, int64_t size)
+{
+ struct rw_file *rwf = (struct rw_file *) rw;
+
+ /* If the destination is an ordinary file then the original file
+ * size doesn't matter. Truncate it to the source size. But
+ * truncate it to zero first so the file is completely empty and
+ * sparse.
+ */
+ if (! S_ISREG (rwf->stat.st_mode))
+ return;
+
+ if (ftruncate (rwf->fd, 0) == -1 ||
+ ftruncate (rwf->fd, size) == -1) {
+ perror ("truncate");
+ exit (EXIT_FAILURE);
+ }
+ rwf->rw.size = size;
+
+ /* We can assume the destination is zero. */
+ destination_is_zero = true;
}
static void
file_flush (struct rw *rw)
{
- if ((S_ISREG (rw->u.local.stat.st_mode) ||
- S_ISBLK (rw->u.local.stat.st_mode)) &&
- fsync (rw->u.local.fd) == -1) {
+ struct rw_file *rwf = (struct rw_file *)rw;
+
+ if ((S_ISREG (rwf->stat.st_mode) ||
+ S_ISBLK (rwf->stat.st_mode)) &&
+ fsync (rwf->fd) == -1) {
perror (rw->name);
exit (EXIT_FAILURE);
}
}
+static bool
+file_is_read_only (struct rw *rw)
+{
+ /* Permissions are hard, and this is only used as an early check
+ * before the copy. Proceed with the copy and fail if it fails.
+ */
+ return false;
+}
+
+static bool
+file_can_extents (struct rw *rw)
+{
+#ifdef SEEK_HOLE
+ return true;
+#else
+ return false;
+#endif
+}
+
+static bool
+file_can_multi_conn (struct rw *rw)
+{
+ return true;
+}
+
+static void
+file_start_multi_conn (struct rw *rw)
+{
+ /* Don't need to do anything for files since we can read/write on a
+ * single file descriptor.
+ */
+}
+
static size_t
file_synch_read (struct rw *rw,
void *data, size_t len, uint64_t offset)
{
+ struct rw_file *rwf = (struct rw_file *)rw;
size_t n = 0;
ssize_t r;
while (len > 0) {
- r = pread (rw->u.local.fd, data, len, offset);
+ r = pread (rwf->fd, data, len, offset);
if (r == -1) {
perror (rw->name);
exit (EXIT_FAILURE);
@@ -85,10 +209,11 @@ static void
file_synch_write (struct rw *rw,
const void *data, size_t len, uint64_t offset)
{
+ struct rw_file *rwf = (struct rw_file *)rw;
ssize_t r;
while (len > 0) {
- r = pwrite (rw->u.local.fd, data, len, offset);
+ r = pwrite (rwf->fd, data, len, offset);
if (r == -1) {
perror (rw->name);
exit (EXIT_FAILURE);
@@ -103,7 +228,8 @@ static bool
file_synch_trim (struct rw *rw, uint64_t offset, uint64_t count)
{
#ifdef FALLOC_FL_PUNCH_HOLE
- int fd = rw->u.local.fd;
+ struct rw_file *rwf = (struct rw_file *)rw;
+ int fd = rwf->fd;
int r;
r = fallocate (fd, FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE,
@@ -121,9 +247,11 @@ file_synch_trim (struct rw *rw, uint64_t offset, uint64_t count)
static bool
file_synch_zero (struct rw *rw, uint64_t offset, uint64_t count)
{
- if (S_ISREG (rw->u.local.stat.st_mode)) {
+ struct rw_file *rwf = (struct rw_file *)rw;
+
+ if (S_ISREG (rwf->stat.st_mode)) {
#ifdef FALLOC_FL_ZERO_RANGE
- int fd = rw->u.local.fd;
+ int fd = rwf->fd;
int r;
r = fallocate (fd, FALLOC_FL_ZERO_RANGE, offset, count);
@@ -134,10 +262,10 @@ file_synch_zero (struct rw *rw, uint64_t offset, uint64_t count)
return true;
#endif
}
- else if (S_ISBLK (rw->u.local.stat.st_mode) &&
- IS_ALIGNED (offset | count, rw->u.local.sector_size)) {
+ else if (S_ISBLK (rwf->stat.st_mode) &&
+ IS_ALIGNED (offset | count, rwf->sector_size)) {
#ifdef BLKZEROOUT
- int fd = rw->u.local.fd;
+ int fd = rwf->fd;
int r;
uint64_t range[2] = {offset, count};
@@ -223,11 +351,12 @@ file_get_extents (struct rw *rw, uintptr_t index,
ret->size = 0;
#ifdef SEEK_HOLE
+ struct rw_file *rwf = (struct rw_file *)rw;
static pthread_mutex_t lseek_lock = PTHREAD_MUTEX_INITIALIZER;
- if (rw->u.local.seek_hole_supported) {
+ if (rwf->seek_hole_supported) {
uint64_t end = offset + count;
- int fd = rw->u.local.fd;
+ int fd = rwf->fd;
off_t pos;
struct extent e;
size_t last;
@@ -302,9 +431,14 @@ file_get_extents (struct rw *rw, uintptr_t index,
default_get_extents (rw, index, offset, count, ret);
}
-struct rw_ops file_ops = {
+static struct rw_ops file_ops = {
.ops_name = "file_ops",
.close = file_close,
+ .is_read_only = file_is_read_only,
+ .can_extents = file_can_extents,
+ .can_multi_conn = file_can_multi_conn,
+ .start_multi_conn = file_start_multi_conn,
+ .truncate = file_truncate,
.flush = file_flush,
.synch_read = file_synch_read,
.synch_write = file_synch_write,
diff --git a/copy/main.c b/copy/main.c
index 68a6030..6fdc6fd 100644
--- a/copy/main.c
+++ b/copy/main.c
@@ -53,19 +53,11 @@ int progress_fd = -1; /* --progress=FD */
unsigned sparse_size = 4096; /* --sparse */
bool synchronous; /* --synchronous flag */
unsigned threads; /* --threads */
-struct rw src, dst; /* The source and destination. */
+struct rw *src, *dst; /* The source and destination. */
bool verbose; /* --verbose flag */
static bool is_nbd_uri (const char *s);
-static bool seek_hole_supported (int fd);
-static void open_null (struct rw *rw);
-static int open_local (const char *prog,
- const char *filename, bool writing, struct rw *rw);
-static void open_nbd_uri (const char *prog,
- const char *uri, bool writing, struct rw *rw);
-static void open_nbd_subprocess (const char *prog,
- const char **argv, size_t argc,
- bool writing, struct rw *rw);
+static struct rw *open_local (const char *filename, bool writing);
static void print_rw (struct rw *rw, const char *prefix, FILE *fp);
static void __attribute__((noreturn))
@@ -242,18 +234,18 @@ main (int argc, char *argv[])
found1:
connections = 1; /* multi-conn not supported */
- src.name = argv[optind+1];
- open_nbd_subprocess (argv[0],
- (const char **) &argv[optind+1], i-optind-1,
- false, &src);
+ src =
+ nbd_rw_create_subprocess ((const char **) &argv[optind+1], i-optind-1,
+ false);
optind = i+1;
}
else { /* Source is not [...]. */
- src.name = argv[optind++];
- if (! is_nbd_uri (src.name))
- src.u.local.fd = open_local (argv[0], src.name, false, &src);
+ const char *src_name = argv[optind++];
+
+ if (! is_nbd_uri (src_name))
+ src = open_local (src_name, false);
else
- open_nbd_uri (argv[0], src.name, false, &src);
+ src = nbd_rw_create_uri (src_name, src_name, false);
}
if (optind >= argc)
@@ -267,48 +259,46 @@ main (int argc, char *argv[])
found2:
connections = 1; /* multi-conn not supported */
- dst.name = argv[optind+1];
- open_nbd_subprocess (argv[0],
- (const char **) &argv[optind+1], i-optind-1,
- true, &dst);
+ dst =
+ nbd_rw_create_subprocess ((const char **) &argv[optind+1], i-optind-1,
+ true);
optind = i+1;
}
else { /* Destination is not [...] */
- dst.name = argv[optind++];
- if (strcmp (dst.name, "null:") == 0)
- open_null (&dst);
- else if (! is_nbd_uri (dst.name))
- dst.u.local.fd = open_local (argv[0], dst.name, true /* writing */, &dst);
- else {
- open_nbd_uri (argv[0], dst.name, true, &dst);
+ const char *dst_name = argv[optind++];
- /* Obviously this is not going to work if the server is
- * advertising read-only, so fail early with a nice error message.
- */
- if (nbd_is_read_only (dst.u.nbd.handles.ptr[0])) {
- fprintf (stderr, "%s: %s: "
- "this NBD server is read-only, cannot write to it\n",
- argv[0], dst.name);
- exit (EXIT_FAILURE);
- }
- }
+ if (strcmp (dst_name, "null:") == 0)
+ dst = null_create (dst_name);
+ else if (! is_nbd_uri (dst_name))
+ dst = open_local (dst_name, true /* writing */);
+ else
+ dst = nbd_rw_create_uri (dst_name, dst_name, true);
}
/* There must be no extra parameters. */
if (optind != argc)
usage (stderr, EXIT_FAILURE);
- /* Check we've set the fields of src and dst. */
- assert (src.ops);
- assert (src.name);
- assert (dst.ops);
- assert (dst.name);
+ /* Check we've created src and dst and set the expected fields. */
+ assert (src != NULL);
+ assert (dst != NULL);
+ assert (src->ops != NULL);
+ assert (src->name != NULL);
+ assert (dst->ops != NULL);
+ assert (dst->name != NULL);
+
+ /* Obviously this is not going to work if the destination is
+ * read-only, so fail early with a nice error message.
+ */
+ if (dst->ops->is_read_only (dst)) {
+ fprintf (stderr, "%s: %s: "
+ "the destination is read-only, cannot write to it\n",
+ argv[0], dst->name);
+ exit (EXIT_FAILURE);
+ }
/* If multi-conn is not supported, force connections to 1. */
- if ((src.ops == &nbd_ops &&
- ! nbd_can_multi_conn (src.u.nbd.handles.ptr[0])) ||
- (dst.ops == &nbd_ops &&
- ! nbd_can_multi_conn (dst.u.nbd.handles.ptr[0])))
+ if (! src->ops->can_multi_conn (src) || ! dst->ops->can_multi_conn (dst))
connections = 1;
/* Calculate the number of threads from the number of connections. */
@@ -335,44 +325,17 @@ main (int argc, char *argv[])
if (threads < connections)
connections = threads;
- /* Calculate the source and destination sizes. We set these to -1
- * if the size is not known (because it's a stream). Note that for
- * local types, open_local set something in *.size already.
+ /* Truncate the destination to the same size as the source. Only
+ * has an effect on regular files.
*/
- if (src.ops == &nbd_ops) {
- src.size = nbd_get_size (src.u.nbd.handles.ptr[0]);
- if (src.size == -1) {
- fprintf (stderr, "%s: %s: %s\n", argv[0], src.name, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
- }
- if (dst.ops != &nbd_ops && S_ISREG (dst.u.local.stat.st_mode)) {
- /* If the destination is an ordinary file then the original file
- * size doesn't matter. Truncate it to the source size. But
- * truncate it to zero first so the file is completely empty and
- * sparse.
- */
- dst.size = src.size;
- if (ftruncate (dst.u.local.fd, 0) == -1 ||
- ftruncate (dst.u.local.fd, dst.size) == -1) {
- perror ("truncate");
- exit (EXIT_FAILURE);
- }
- destination_is_zero = true;
- }
- else if (dst.ops == &nbd_ops) {
- dst.size = nbd_get_size (dst.u.nbd.handles.ptr[0]);
- if (dst.size == -1) {
- fprintf (stderr, "%s: %s: %s\n", argv[0], dst.name, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
- }
+ if (dst->ops->truncate)
+ dst->ops->truncate (dst, src->size);
/* Check if the source is bigger than the destination, since that
* would truncate (ie. lose) data. Copying from smaller to larger
* is OK.
*/
- if (src.size >= 0 && dst.size >= 0 && src.size > dst.size) {
+ if (src->size >= 0 && dst->size >= 0 && src->size >
dst->size) {
fprintf (stderr,
"nbdcopy: error: destination size is smaller than source
size\n");
exit (EXIT_FAILURE);
@@ -383,37 +346,29 @@ main (int argc, char *argv[])
* settings.
*/
if (verbose) {
- print_rw (&src, "nbdcopy: src", stderr);
- print_rw (&dst, "nbdcopy: dst", stderr);
+ print_rw (src, "nbdcopy: src", stderr);
+ print_rw (dst, "nbdcopy: dst", stderr);
fprintf (stderr, "nbdcopy: connections=%u requests=%u threads=%u "
"synchronous=%s\n",
connections, max_requests, threads,
synchronous ? "true" : "false");
}
- /* If #connections > 1 then multi-conn is enabled at both ends and
- * we need to open further connections.
+ /* If multi-conn is enabled on either side, then at this point we
+ * need to ask the backend to open the extra connections.
*/
if (connections > 1) {
assert (threads == connections);
-
- if (src.ops == &nbd_ops) {
- for (i = 1; i < connections; ++i)
- open_nbd_uri (argv[0], src.name, false, &src);
- assert (src.u.nbd.handles.size == connections);
- }
- if (dst.ops == &nbd_ops) {
- for (i = 1; i < connections; ++i)
- open_nbd_uri (argv[0], dst.name, true, &dst);
- assert (dst.u.nbd.handles.size == connections);
- }
+ if (src->ops->can_multi_conn (src))
+ src->ops->start_multi_conn (src);
+ if (dst->ops->can_multi_conn (dst))
+ dst->ops->start_multi_conn (dst);
}
/* If the source is NBD and we couldn't negotiate meta
* base:allocation then turn off extents.
*/
- if (src.ops == &nbd_ops &&
- !nbd_can_meta_context (src.u.nbd.handles.ptr[0], "base:allocation"))
+ if (! src->ops->can_extents (src))
extents = false;
/* Always set the progress bar to 0% at the start of the copy. */
@@ -429,12 +384,12 @@ main (int argc, char *argv[])
progress_bar (1, 1);
/* Shut down the source side. */
- src.ops->close (&src);
+ src->ops->close (src);
/* Shut down the destination side. */
if (flush)
- dst.ops->flush (&dst);
- dst.ops->close (&dst);
+ dst->ops->flush (dst);
+ dst->ops->close (dst);
exit (EXIT_SUCCESS);
}
@@ -452,33 +407,25 @@ is_nbd_uri (const char *s)
strncmp (s, "nbds+vsock:", 11) == 0;
}
-/* Open null: (destination only). */
-static void
-open_null (struct rw *rw)
-{
- rw->ops = &null_ops;
- rw->size = INT64_MAX;
-}
-
/* Open a local (non-NBD) file, ie. a file, device, or "-" for stdio.
- * Returns the open file descriptor which the caller must close.
+ * Returns the struct rw * which the caller must close.
*
* “writing” is true if this is the destination parameter.
* “rw->u.local.stat” and “rw->size” return the file stat and size,
* but size can be returned as -1 if we don't know the size (if it's a
* pipe or stdio).
*/
-static int
-open_local (const char *prog,
- const char *filename, bool writing, struct rw *rw)
+static struct rw *
+open_local (const char *filename, bool writing)
{
int flags, fd;
+ struct stat stat;
if (strcmp (filename, "-") == 0) {
synchronous = true;
fd = writing ? STDOUT_FILENO : STDIN_FILENO;
if (writing && isatty (fd)) {
- fprintf (stderr, "%s: refusing to write to tty\n", prog);
+ fprintf (stderr, "%s: refusing to write to tty\n", "nbdcopy");
exit (EXIT_FAILURE);
}
}
@@ -502,146 +449,17 @@ open_local (const char *prog,
}
}
- if (fstat (fd, &rw->u.local.stat) == -1) {
+ if (fstat (fd, &stat) == -1) {
perror (filename);
exit (EXIT_FAILURE);
}
- if (S_ISBLK (rw->u.local.stat.st_mode)) {
- /* Block device. */
- rw->ops = &file_ops;
- rw->size = lseek (fd, 0, SEEK_END);
- if (rw->size == -1) {
- perror ("lseek");
- exit (EXIT_FAILURE);
- }
- if (lseek (fd, 0, SEEK_SET) == -1) {
- perror ("lseek");
- exit (EXIT_FAILURE);
- }
- rw->u.local.seek_hole_supported = seek_hole_supported (fd);
- rw->u.local.sector_size = 4096;
-#ifdef BLKSSZGET
- if (ioctl (fd, BLKSSZGET, &rw->u.local.sector_size))
- fprintf (stderr, "warning: cannot get sector size: %s: %m",
rw->name);
-#endif
- }
- else if (S_ISREG (rw->u.local.stat.st_mode)) {
- /* Regular file. */
- rw->ops = &file_ops;
- rw->size = rw->u.local.stat.st_size;
- rw->u.local.seek_hole_supported = seek_hole_supported (fd);
- }
+ if (S_ISBLK (stat.st_mode) || S_ISREG (stat.st_mode))
+ return file_create (filename, &stat, fd);
else {
- /* Probably stdin/stdout, a pipe or a socket. Set size == -1
- * which means don't know, and force synchronous mode.
- */
- synchronous = true;
- rw->ops = &pipe_ops;
- rw->size = -1;
- rw->u.local.seek_hole_supported = false;
+ /* Probably stdin/stdout, a pipe or a socket. */
+ synchronous = true; /* Force synchronous mode for pipes. */
+ return pipe_create (filename, fd);
}
-
- return fd;
-}
-
-static bool
-seek_hole_supported (int fd)
-{
-#ifndef SEEK_HOLE
- return false;
-#else
- off_t r = lseek (fd, 0, SEEK_HOLE);
- return r >= 0;
-#endif
-}
-
-static void
-open_nbd_uri (const char *prog,
- const char *uri, bool writing, struct rw *rw)
-{
- struct nbd_handle *nbd;
-
- rw->ops = &nbd_ops;
- nbd = nbd_create ();
- if (nbd == NULL) {
- fprintf (stderr, "%s: %s\n", prog, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
- nbd_set_debug (nbd, verbose);
- nbd_set_uri_allow_local_file (nbd, true); /* Allow ?tls-psk-file. */
- if (extents && !writing &&
- nbd_add_meta_context (nbd, "base:allocation") == -1) {
- fprintf (stderr, "%s: %s\n", prog, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
-
- if (handles_append (&rw->u.nbd.handles, nbd) == -1) {
- perror ("realloc");
- exit (EXIT_FAILURE);
- }
-
- if (nbd_connect_uri (nbd, uri) == -1) {
- fprintf (stderr, "%s: %s: %s\n", prog, uri, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
-
- /* Cache these. We assume with multi-conn that each handle will act
- * the same way.
- */
- rw->u.nbd.can_trim = nbd_can_trim (nbd) > 0;
- rw->u.nbd.can_zero = nbd_can_zero (nbd) > 0;
-}
-
-DEFINE_VECTOR_TYPE (const_string_vector, const char *);
-
-static void
-open_nbd_subprocess (const char *prog,
- const char **argv, size_t argc,
- bool writing, struct rw *rw)
-{
- struct nbd_handle *nbd;
- const_string_vector copy = empty_vector;
- size_t i;
-
- rw->ops = &nbd_ops;
- nbd = nbd_create ();
- if (nbd == NULL) {
- fprintf (stderr, "%s: %s\n", prog, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
- nbd_set_debug (nbd, verbose);
- if (extents && !writing &&
- nbd_add_meta_context (nbd, "base:allocation") == -1) {
- fprintf (stderr, "%s: %s\n", prog, nbd_get_error ());
- exit (EXIT_FAILURE);
- }
-
- if (handles_append (&rw->u.nbd.handles, nbd) == -1) {
- memory_error:
- perror ("realloc");
- exit (EXIT_FAILURE);
- }
-
- /* We have to copy the args so we can null-terminate them. */
- for (i = 0; i < argc; ++i) {
- if (const_string_vector_append (©, argv[i]) == -1)
- goto memory_error;
- }
- if (const_string_vector_append (©, NULL) == -1)
- goto memory_error;
-
- if (nbd_connect_systemd_socket_activation (nbd, (char **) copy.ptr) == -1) {
- fprintf (stderr, "%s: %s: %s\n", prog, argv[0], nbd_get_error ());
- exit (EXIT_FAILURE);
- }
-
- /* Cache these. We assume with multi-conn that each handle will act
- * the same way.
- */
- rw->u.nbd.can_trim = nbd_can_trim (nbd) > 0;
- rw->u.nbd.can_zero = nbd_can_zero (nbd) > 0;
-
- free (copy.ptr);
}
/* Print an rw struct, used in --verbose mode. */
@@ -650,7 +468,6 @@ print_rw (struct rw *rw, const char *prefix, FILE *fp)
{
fprintf (fp, "%s: %s \"%s\"\n", prefix, rw->ops->ops_name,
rw->name);
fprintf (fp, "%s: size=%" PRIi64 "\n", prefix, rw->size);
- /* Could print other stuff here, but that's enough for debugging. */
}
/* Default implementation of rw->ops->get_extents for backends which
diff --git a/copy/multi-thread-copying.c b/copy/multi-thread-copying.c
index 98b4056..4f57054 100644
--- a/copy/multi-thread-copying.c
+++ b/copy/multi-thread-copying.c
@@ -50,13 +50,13 @@ get_next_offset (uint64_t *offset, uint64_t *count)
bool r = false; /* returning false means no more work */
pthread_mutex_lock (&lock);
- if (next_offset < src.size) {
+ if (next_offset < src->size) {
*offset = next_offset;
/* Work out how large this range is. The last range may be
* smaller than THREAD_WORK_SIZE.
*/
- *count = src.size - *offset;
+ *count = src->size - *offset;
if (*count > THREAD_WORK_SIZE)
*count = THREAD_WORK_SIZE;
@@ -69,7 +69,7 @@ get_next_offset (uint64_t *offset, uint64_t *count)
* are called from threads and not necessarily in monotonic order
* so the progress bar would move erratically.
*/
- progress_bar (*offset, dst.size);
+ progress_bar (*offset, dst->size);
}
pthread_mutex_unlock (&lock);
return r;
@@ -89,11 +89,13 @@ multi_thread_copying (void)
*/
assert (threads > 0);
assert (threads == connections);
+/*
if (src.ops == &nbd_ops)
assert (src.u.nbd.handles.size == connections);
if (dst.ops == &nbd_ops)
assert (dst.u.nbd.handles.size == connections);
- assert (src.size != -1);
+*/
+ assert (src->size != -1);
workers = malloc (sizeof (pthread_t) * threads);
if (workers == NULL) {
@@ -147,9 +149,9 @@ worker_thread (void *indexp)
assert (0 < count && count <= THREAD_WORK_SIZE);
if (extents)
- src.ops->get_extents (&src, index, offset, count, &exts);
+ src->ops->get_extents (src, index, offset, count, &exts);
else
- default_get_extents (&src, index, offset, count, &exts);
+ default_get_extents (src, index, offset, count, &exts);
for (i = 0; i < exts.size; ++i) {
struct command *command;
@@ -208,11 +210,11 @@ worker_thread (void *indexp)
wait_for_request_slots (index);
/* Begin the asynch read operation. */
- src.ops->asynch_read (&src, command,
- (nbd_completion_callback) {
- .callback = finished_read,
- .user_data = command,
- });
+ src->ops->asynch_read (src, command,
+ (nbd_completion_callback) {
+ .callback = finished_read,
+ .user_data = command,
+ });
exts.ptr[i].offset += len;
exts.ptr[i].length -= len;
@@ -254,7 +256,7 @@ wait_for_request_slots (uintptr_t index)
static unsigned
in_flight (uintptr_t index)
{
- return src.ops->in_flight (&src, index) + dst.ops->in_flight (&dst,
index);
+ return src->ops->in_flight (src, index) + dst->ops->in_flight (dst,
index);
}
/* Poll (optional) NBD src and NBD dst, moving the state machine(s)
@@ -271,7 +273,7 @@ poll_both_ends (uintptr_t index)
/* Note: if polling is not supported, this function will
* set fd == -1 which poll ignores.
*/
- src.ops->get_polling_fd (&src, index, &fds[0].fd, &direction);
+ src->ops->get_polling_fd (src, index, &fds[0].fd, &direction);
if (fds[0].fd >= 0) {
switch (direction) {
case LIBNBD_AIO_DIRECTION_READ:
@@ -286,7 +288,7 @@ poll_both_ends (uintptr_t index)
}
}
- dst.ops->get_polling_fd (&dst, index, &fds[1].fd, &direction);
+ dst->ops->get_polling_fd (dst, index, &fds[1].fd, &direction);
if (fds[1].fd >= 0) {
switch (direction) {
case LIBNBD_AIO_DIRECTION_READ:
@@ -311,24 +313,24 @@ poll_both_ends (uintptr_t index)
if (fds[0].fd >= 0) {
if ((fds[0].revents & (POLLIN | POLLHUP)) != 0)
- src.ops->asynch_notify_read (&src, index);
+ src->ops->asynch_notify_read (src, index);
else if ((fds[0].revents & POLLOUT) != 0)
- src.ops->asynch_notify_write (&src, index);
+ src->ops->asynch_notify_write (src, index);
else if ((fds[0].revents & (POLLERR | POLLNVAL)) != 0) {
errno = ENOTCONN;
- perror (src.name);
+ perror (src->name);
exit (EXIT_FAILURE);
}
}
if (fds[1].fd >= 0) {
if ((fds[1].revents & (POLLIN | POLLHUP)) != 0)
- dst.ops->asynch_notify_read (&dst, index);
+ dst->ops->asynch_notify_read (dst, index);
else if ((fds[1].revents & POLLOUT) != 0)
- dst.ops->asynch_notify_write (&dst, index);
+ dst->ops->asynch_notify_write (dst, index);
else if ((fds[1].revents & (POLLERR | POLLNVAL)) != 0) {
errno = ENOTCONN;
- perror (dst.name);
+ perror (dst->name);
exit (EXIT_FAILURE);
}
}
@@ -377,11 +379,11 @@ finished_read (void *vp, int *error)
/* If sparseness detection (see below) is turned off then we write
* the whole command.
*/
- dst.ops->asynch_write (&dst, command,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = command,
- });
+ dst->ops->asynch_write (dst, command,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = command,
+ });
}
else { /* Sparseness detection. */
const uint64_t start = command->offset;
@@ -408,11 +410,11 @@ finished_read (void *vp, int *error)
newcommand = copy_subcommand (command,
last_offset, i - last_offset,
false);
- dst.ops->asynch_write (&dst, newcommand,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = newcommand,
- });
+ dst->ops->asynch_write (dst, newcommand,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = newcommand,
+ });
}
/* Start the new hole. */
last_offset = i;
@@ -445,11 +447,11 @@ finished_read (void *vp, int *error)
newcommand = copy_subcommand (command,
last_offset, i - last_offset,
false);
- dst.ops->asynch_write (&dst, newcommand,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = newcommand,
- });
+ dst->ops->asynch_write (dst, newcommand,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = newcommand,
+ });
}
else {
newcommand = copy_subcommand (command,
@@ -462,11 +464,11 @@ finished_read (void *vp, int *error)
/* There may be an unaligned tail, so write that. */
if (end - i > 0) {
newcommand = copy_subcommand (command, i, end - i, false);
- dst.ops->asynch_write (&dst, newcommand,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = newcommand,
- });
+ dst->ops->asynch_write (dst, newcommand,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = newcommand,
+ });
}
/* Free the original command since it has been split into
@@ -503,20 +505,20 @@ fill_dst_range_with_zeroes (struct command *command)
if (!allocated) {
/* Try trimming. */
- if (dst.ops->asynch_trim (&dst, command,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = command,
- }))
+ if (dst->ops->asynch_trim (dst, command,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = command,
+ }))
return;
}
/* Try efficient zeroing. */
- if (dst.ops->asynch_zero (&dst, command,
- (nbd_completion_callback) {
- .callback = free_command,
- .user_data = command,
- }))
+ if (dst->ops->asynch_zero (dst, command,
+ (nbd_completion_callback) {
+ .callback = free_command,
+ .user_data = command,
+ }))
return;
/* Fall back to loop writing zeroes. This is going to be slow
@@ -533,7 +535,7 @@ fill_dst_range_with_zeroes (struct command *command)
if (len > MAX_REQUEST_SIZE)
len = MAX_REQUEST_SIZE;
- dst.ops->synch_write (&dst, data, len, command->offset);
+ dst->ops->synch_write (dst, data, len, command->offset);
command->slice.len -= len;
command->offset += len;
}
diff --git a/copy/nbd-ops.c b/copy/nbd-ops.c
index 7b48cbc..24970c2 100644
--- a/copy/nbd-ops.c
+++ b/copy/nbd-ops.c
@@ -27,45 +27,221 @@
#include "nbdcopy.h"
+static struct rw_ops nbd_ops;
+
+DEFINE_VECTOR_TYPE (handles, struct nbd_handle *)
+DEFINE_VECTOR_TYPE (const_string_vector, const char *);
+
+struct rw_nbd {
+ struct rw rw;
+
+ /* Because of multi-conn we have to remember enough state in this
+ * handle in order to be able to open another connection with the
+ * same parameters after nbd_rw_create* has been called once.
+ */
+ enum { CREATE_URI, CREATE_SUBPROCESS } create_t;
+ const char *uri; /* For CREATE_URI */
+ const_string_vector argv; /* For CREATE_SUBPROCESS */
+ bool writing;
+
+ handles handles; /* One handle per connection. */
+ bool can_trim, can_zero; /* Cached nbd_can_trim, nbd_can_zero. */
+};
+
+static void
+open_one_nbd_handle (struct rw_nbd *rwn)
+{
+ struct nbd_handle *nbd;
+
+ nbd = nbd_create ();
+ if (nbd == NULL) {
+ fprintf (stderr, "%s: %s\n", "nbdcopy", nbd_get_error ());
+ exit (EXIT_FAILURE);
+ }
+
+ nbd_set_debug (nbd, verbose);
+
+ if (extents && !rwn->writing &&
+ nbd_add_meta_context (nbd, "base:allocation") == -1) {
+ fprintf (stderr, "%s: %s\n", "nbdcopy", nbd_get_error ());
+ exit (EXIT_FAILURE);
+ }
+
+ switch (rwn->create_t) {
+ case CREATE_URI:
+ nbd_set_uri_allow_local_file (nbd, true); /* Allow ?tls-psk-file. */
+
+ if (nbd_connect_uri (nbd, rwn->uri) == -1) {
+ fprintf (stderr, "%s: %s: %s\n", "nbdcopy", rwn->uri,
nbd_get_error ());
+ exit (EXIT_FAILURE);
+ }
+ break;
+
+ case CREATE_SUBPROCESS:
+ if (nbd_connect_systemd_socket_activation (nbd,
+ (char **) rwn->argv.ptr)
+ == -1) {
+ fprintf (stderr, "%s: %s: %s\n", "nbdcopy",
rwn->argv.ptr[0],
+ nbd_get_error ());
+ exit (EXIT_FAILURE);
+ }
+ }
+
+ /* Cache these. We assume with multi-conn that each handle will act
+ * the same way.
+ */
+ if (rwn->handles.size == 0) {
+ rwn->can_trim = nbd_can_trim (nbd) > 0;
+ rwn->can_zero = nbd_can_zero (nbd) > 0;
+ rwn->rw.size = nbd_get_size (nbd);
+ if (rwn->rw.size == -1) {
+ fprintf (stderr, "%s: %s: %s\n", "nbdcopy", rwn->rw.name,
+ nbd_get_error ());
+ exit (EXIT_FAILURE);
+ }
+ }
+
+ if (handles_append (&rwn->handles, nbd) == -1) {
+ perror ("realloc");
+ exit (EXIT_FAILURE);
+ }
+}
+
+struct rw *
+nbd_rw_create_uri (const char *name, const char *uri, bool writing)
+{
+ struct rw_nbd *rwn = calloc (1, sizeof *rwn);
+ if (rwn == NULL) { perror ("calloc"); exit (EXIT_FAILURE); }
+
+ rwn->rw.ops = &nbd_ops;
+ rwn->rw.name = name;
+ rwn->create_t = CREATE_URI;
+ rwn->uri = uri;
+ rwn->writing = writing;
+
+ open_one_nbd_handle (rwn);
+
+ return (struct rw *) rwn;
+}
+
+struct rw *
+nbd_rw_create_subprocess (const char **argv, size_t argc, bool writing)
+{
+ size_t i;
+ struct rw_nbd *rwn = calloc (1, sizeof *rwn);
+ if (rwn == NULL) { perror ("calloc"); exit (EXIT_FAILURE); }
+
+ rwn->rw.ops = &nbd_ops;
+ rwn->rw.name = argv[0];
+ rwn->create_t = CREATE_SUBPROCESS;
+ rwn->writing = writing;
+
+ /* We have to copy the args so we can null-terminate them. */
+ for (i = 0; i < argc; ++i) {
+ if (const_string_vector_append (&rwn->argv, argv[i]) == -1) {
+ memory_error:
+ perror ("realloc");
+ exit (EXIT_FAILURE);
+ }
+ }
+ if (const_string_vector_append (&rwn->argv, NULL) == -1)
+ goto memory_error;
+
+ open_one_nbd_handle (rwn);
+
+ return (struct rw *) rwn;
+}
+
static void
nbd_ops_close (struct rw *rw)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
size_t i;
- for (i = 0; i < rw->u.nbd.handles.size; ++i) {
- if (nbd_shutdown (rw->u.nbd.handles.ptr[i], 0) == -1) {
+ for (i = 0; i < rwn->handles.size; ++i) {
+ if (nbd_shutdown (rwn->handles.ptr[i], 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
- nbd_close (rw->u.nbd.handles.ptr[i]);
+ nbd_close (rwn->handles.ptr[i]);
}
- handles_reset (&rw->u.nbd.handles);
+ handles_reset (&rwn->handles);
+ const_string_vector_reset (&rwn->argv);
+ free (rw);
}
static void
nbd_ops_flush (struct rw *rw)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
size_t i;
- for (i = 0; i < rw->u.nbd.handles.size; ++i) {
- if (nbd_flush (rw->u.nbd.handles.ptr[i], 0) == -1) {
+ for (i = 0; i < rwn->handles.size; ++i) {
+ if (nbd_flush (rwn->handles.ptr[i], 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
}
}
+static bool
+nbd_ops_is_read_only (struct rw *rw)
+{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (rwn->handles.size > 0)
+ return nbd_is_read_only (rwn->handles.ptr[0]);
+ else
+ return false;
+}
+
+static bool
+nbd_ops_can_extents (struct rw *rw)
+{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (rwn->handles.size > 0)
+ return nbd_can_meta_context (rwn->handles.ptr[0], "base:allocation");
+ else
+ return false;
+}
+
+static bool
+nbd_ops_can_multi_conn (struct rw *rw)
+{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (rwn->handles.size > 0)
+ return nbd_can_multi_conn (rwn->handles.ptr[0]);
+ else
+ return false;
+}
+
+static void
+nbd_ops_start_multi_conn (struct rw *rw)
+{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+ size_t i;
+
+ for (i = 1; i < connections; ++i)
+ open_one_nbd_handle (rwn);
+
+ assert (rwn->handles.size == connections);
+}
+
static size_t
nbd_ops_synch_read (struct rw *rw,
void *data, size_t len, uint64_t offset)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
if (len > rw->size - offset)
len = rw->size - offset;
if (len == 0)
return 0;
- if (nbd_pread (rw->u.nbd.handles.ptr[0], data, len, offset, 0) == -1) {
+ if (nbd_pread (rwn->handles.ptr[0], data, len, offset, 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
@@ -77,7 +253,9 @@ static void
nbd_ops_synch_write (struct rw *rw,
const void *data, size_t len, uint64_t offset)
{
- if (nbd_pwrite (rw->u.nbd.handles.ptr[0], data, len, offset, 0) == -1) {
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (nbd_pwrite (rwn->handles.ptr[0], data, len, offset, 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
@@ -86,10 +264,12 @@ nbd_ops_synch_write (struct rw *rw,
static bool
nbd_ops_synch_trim (struct rw *rw, uint64_t offset, uint64_t count)
{
- if (!rw->u.nbd.can_trim)
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (!rwn->can_trim)
return false;
- if (nbd_trim (rw->u.nbd.handles.ptr[0], count, offset, 0) == -1) {
+ if (nbd_trim (rwn->handles.ptr[0], count, offset, 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
@@ -99,10 +279,12 @@ nbd_ops_synch_trim (struct rw *rw, uint64_t offset, uint64_t count)
static bool
nbd_ops_synch_zero (struct rw *rw, uint64_t offset, uint64_t count)
{
- if (!rw->u.nbd.can_zero)
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (!rwn->can_zero)
return false;
- if (nbd_zero (rw->u.nbd.handles.ptr[0],
+ if (nbd_zero (rwn->handles.ptr[0],
count, offset, LIBNBD_CMD_FLAG_NO_HOLE) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
@@ -115,7 +297,9 @@ nbd_ops_asynch_read (struct rw *rw,
struct command *command,
nbd_completion_callback cb)
{
- if (nbd_aio_pread (rw->u.nbd.handles.ptr[command->index],
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (nbd_aio_pread (rwn->handles.ptr[command->index],
slice_ptr (command->slice),
command->slice.len, command->offset,
cb, 0) == -1) {
@@ -129,7 +313,9 @@ nbd_ops_asynch_write (struct rw *rw,
struct command *command,
nbd_completion_callback cb)
{
- if (nbd_aio_pwrite (rw->u.nbd.handles.ptr[command->index],
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (nbd_aio_pwrite (rwn->handles.ptr[command->index],
slice_ptr (command->slice),
command->slice.len, command->offset,
cb, 0) == -1) {
@@ -142,12 +328,14 @@ static bool
nbd_ops_asynch_trim (struct rw *rw, struct command *command,
nbd_completion_callback cb)
{
- if (!rw->u.nbd.can_trim)
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (!rwn->can_trim)
return false;
assert (command->slice.len <= UINT32_MAX);
- if (nbd_aio_trim (rw->u.nbd.handles.ptr[command->index],
+ if (nbd_aio_trim (rwn->handles.ptr[command->index],
command->slice.len, command->offset,
cb, 0) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
@@ -160,12 +348,14 @@ static bool
nbd_ops_asynch_zero (struct rw *rw, struct command *command,
nbd_completion_callback cb)
{
- if (!rw->u.nbd.can_zero)
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
+ if (!rwn->can_zero)
return false;
assert (command->slice.len <= UINT32_MAX);
- if (nbd_aio_zero (rw->u.nbd.handles.ptr[command->index],
+ if (nbd_aio_zero (rwn->handles.ptr[command->index],
command->slice.len, command->offset,
cb, LIBNBD_CMD_FLAG_NO_HOLE) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
@@ -212,19 +402,22 @@ add_extent (void *vp, const char *metacontext,
static unsigned
nbd_ops_in_flight (struct rw *rw, uintptr_t index)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+
/* Since the commands are auto-retired in the callbacks we don't
* need to count "done" commands.
*/
- return nbd_aio_in_flight (rw->u.nbd.handles.ptr[index]);
+ return nbd_aio_in_flight (rwn->handles.ptr[index]);
}
static void
nbd_ops_get_polling_fd (struct rw *rw, uintptr_t index,
int *fd, int *direction)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
struct nbd_handle *nbd;
- nbd = rw->u.nbd.handles.ptr[index];
+ nbd = rwn->handles.ptr[index];
*fd = nbd_aio_get_fd (nbd);
if (*fd == -1) {
@@ -240,7 +433,8 @@ nbd_ops_get_polling_fd (struct rw *rw, uintptr_t index,
static void
nbd_ops_asynch_notify_read (struct rw *rw, uintptr_t index)
{
- if (nbd_aio_notify_read (rw->u.nbd.handles.ptr[index]) == -1) {
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+ if (nbd_aio_notify_read (rwn->handles.ptr[index]) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
@@ -249,7 +443,8 @@ nbd_ops_asynch_notify_read (struct rw *rw, uintptr_t index)
static void
nbd_ops_asynch_notify_write (struct rw *rw, uintptr_t index)
{
- if (nbd_aio_notify_write (rw->u.nbd.handles.ptr[index]) == -1) {
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
+ if (nbd_aio_notify_write (rwn->handles.ptr[index]) == -1) {
fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ());
exit (EXIT_FAILURE);
}
@@ -266,10 +461,11 @@ nbd_ops_get_extents (struct rw *rw, uintptr_t index,
uint64_t offset, uint64_t count,
extent_list *ret)
{
+ struct rw_nbd *rwn = (struct rw_nbd *) rw;
extent_list exts = empty_vector;
struct nbd_handle *nbd;
- nbd = rw->u.nbd.handles.ptr[index];
+ nbd = rwn->handles.ptr[index];
ret->size = 0;
@@ -331,9 +527,13 @@ nbd_ops_get_extents (struct rw *rw, uintptr_t index,
free (exts.ptr);
}
-struct rw_ops nbd_ops = {
+static struct rw_ops nbd_ops = {
.ops_name = "nbd_ops",
.close = nbd_ops_close,
+ .is_read_only = nbd_ops_is_read_only,
+ .can_extents = nbd_ops_can_extents,
+ .can_multi_conn = nbd_ops_can_multi_conn,
+ .start_multi_conn = nbd_ops_start_multi_conn,
.flush = nbd_ops_flush,
.synch_read = nbd_ops_synch_read,
.synch_write = nbd_ops_synch_write,
diff --git a/copy/nbdcopy.h b/copy/nbdcopy.h
index 69fac2a..94fbdeb 100644
--- a/copy/nbdcopy.h
+++ b/copy/nbdcopy.h
@@ -36,8 +36,6 @@
*/
#define THREAD_WORK_SIZE (128 * 1024 * 1024)
-DEFINE_VECTOR_TYPE (handles, struct nbd_handle *)
-
/* Abstracts the input (src) and output (dst) parameters on the
* command line.
*/
@@ -45,21 +43,20 @@ struct rw {
struct rw_ops *ops; /* Operations. */
const char *name; /* Printable name, for error messages etc. */
int64_t size; /* May be -1 for streams. */
- union {
- struct { /* For files and pipes. */
- int fd;
- struct stat stat;
- bool seek_hole_supported;
- int sector_size;
- } local;
- struct {
- handles handles; /* For NBD, one handle per connection. */
- bool can_trim, can_zero; /* Cached nbd_can_trim, nbd_can_zero. */
- } nbd;
- } u;
+ /* Followed by private data for the particular subtype. */
};
-extern struct rw src, dst;
+extern struct rw *src, *dst;
+
+/* Create subtypes. */
+extern struct rw *file_create (const char *name,
+ const struct stat *stat, int fd);
+extern struct rw *nbd_rw_create_uri (const char *name,
+ const char *uri, bool writing);
+extern struct rw *nbd_rw_create_subprocess (const char **argv, size_t argc,
+ bool writing);
+extern struct rw *null_create (const char *name);
+extern struct rw *pipe_create (const char *name, int fd);
/* Underlying data buffers. */
struct buffer {
@@ -117,6 +114,28 @@ struct rw_ops {
/* Close the connection and free up associated resources. */
void (*close) (struct rw *rw);
+ /* Return true if this is a read-only connection. */
+ bool (*is_read_only) (struct rw *rw);
+
+ /* For source only, does it support reading extents? */
+ bool (*can_extents) (struct rw *rw);
+
+ /* Return true if the connection can do multi-conn. This is true
+ * for files, false for streams, and passed through for NBD.
+ */
+ bool (*can_multi_conn) (struct rw *rw);
+
+ /* For multi-conn capable backends, before copying we must call this
+ * to begin multi-conn. For NBD this means opening the additional
+ * connections.
+ */
+ void (*start_multi_conn) (struct rw *rw);
+
+ /* Truncate, only called on output files. This callback can be NULL
+ * for types that don't support this.
+ */
+ void (*truncate) (struct rw *rw, int64_t size);
+
/* Flush pending writes to permanent storage. */
void (*flush) (struct rw *rw);
@@ -188,10 +207,6 @@ struct rw_ops {
uint64_t offset, uint64_t count,
extent_list *ret);
};
-extern struct rw_ops file_ops;
-extern struct rw_ops nbd_ops;
-extern struct rw_ops pipe_ops;
-extern struct rw_ops null_ops;
extern void default_get_extents (struct rw *rw, uintptr_t index,
uint64_t offset, uint64_t count,
diff --git a/copy/null-ops.c b/copy/null-ops.c
index b2ca66f..3262fb5 100644
--- a/copy/null-ops.c
+++ b/copy/null-ops.c
@@ -30,10 +30,28 @@
* and fast zeroing.
*/
+static struct rw_ops null_ops;
+
+struct rw_null {
+ struct rw rw;
+};
+
+struct rw *
+null_create (const char *name)
+{
+ struct rw_null *rw = calloc (1, sizeof *rw);
+ if (rw == NULL) { perror ("calloc"); exit (EXIT_FAILURE); }
+
+ rw->rw.ops = &null_ops;
+ rw->rw.name = name;
+ rw->rw.size = INT64_MAX;
+ return (struct rw *) rw;
+}
+
static void
null_close (struct rw *rw)
{
- /* nothing */
+ free (rw);
}
static void
@@ -42,6 +60,30 @@ null_flush (struct rw *rw)
/* nothing */
}
+static bool
+null_is_read_only (struct rw *rw)
+{
+ return false;
+}
+
+static bool
+null_can_extents (struct rw *rw)
+{
+ return false;
+}
+
+static bool
+null_can_multi_conn (struct rw *rw)
+{
+ return true;
+}
+
+static void
+null_start_multi_conn (struct rw *rw)
+{
+ /* nothing */
+}
+
static size_t
null_synch_read (struct rw *rw,
void *data, size_t len, uint64_t offset)
@@ -126,9 +168,13 @@ null_get_extents (struct rw *rw, uintptr_t index,
abort ();
}
-struct rw_ops null_ops = {
+static struct rw_ops null_ops = {
.ops_name = "null_ops",
.close = null_close,
+ .is_read_only = null_is_read_only,
+ .can_extents = null_can_extents,
+ .can_multi_conn = null_can_multi_conn,
+ .start_multi_conn = null_start_multi_conn,
.flush = null_flush,
.synch_read = null_synch_read,
.synch_write = null_synch_write,
diff --git a/copy/pipe-ops.c b/copy/pipe-ops.c
index d127dad..286e6c0 100644
--- a/copy/pipe-ops.c
+++ b/copy/pipe-ops.c
@@ -26,10 +26,33 @@
#include "nbdcopy.h"
+static struct rw_ops pipe_ops;
+
+struct rw_pipe {
+ struct rw rw;
+ int fd;
+};
+
+struct rw *
+pipe_create (const char *name, int fd)
+{
+ struct rw_pipe *rwp = calloc (1, sizeof *rwp);
+ if (rwp == NULL) { perror ("calloc"); exit (EXIT_FAILURE); }
+
+ /* Set size == -1 which means don't know. */
+ rwp->rw.ops = &pipe_ops;
+ rwp->rw.name = name;
+ rwp->rw.size = -1;
+ rwp->fd = fd;
+ return (struct rw *) rwp;
+}
+
static void
pipe_close (struct rw *rw)
{
- if (close (rw->u.local.fd) == -1) {
+ struct rw_pipe *rwp = (struct rw_pipe *) rw;
+
+ if (close (rwp->fd) == -1) {
fprintf (stderr, "%s: close: %m\n", rw->name);
exit (EXIT_FAILURE);
}
@@ -43,13 +66,39 @@ pipe_flush (struct rw *rw)
*/
}
+static bool
+pipe_is_read_only (struct rw *rw)
+{
+ return false;
+}
+
+static bool
+pipe_can_extents (struct rw *rw)
+{
+ return false;
+}
+
+static bool
+pipe_can_multi_conn (struct rw *rw)
+{
+ return false;
+}
+
+static void
+pipe_start_multi_conn (struct rw *rw)
+{
+ /* Should never be called. */
+ abort ();
+}
+
static size_t
pipe_synch_read (struct rw *rw,
void *data, size_t len, uint64_t offset)
{
+ struct rw_pipe *rwp = (struct rw_pipe *) rw;
ssize_t r;
- r = read (rw->u.local.fd, data, len);
+ r = read (rwp->fd, data, len);
if (r == -1) {
perror (rw->name);
exit (EXIT_FAILURE);
@@ -61,10 +110,11 @@ static void
pipe_synch_write (struct rw *rw,
const void *data, size_t len, uint64_t offset)
{
+ struct rw_pipe *rwp = (struct rw_pipe *) rw;
ssize_t r;
while (len > 0) {
- r = write (rw->u.local.fd, data, len);
+ r = write (rwp->fd, data, len);
if (r == -1) {
perror (rw->name);
exit (EXIT_FAILURE);
@@ -109,10 +159,16 @@ pipe_in_flight (struct rw *rw, uintptr_t index)
return 0;
}
-struct rw_ops pipe_ops = {
+static struct rw_ops pipe_ops = {
.ops_name = "pipe_ops",
.close = pipe_close,
+
+ .is_read_only = pipe_is_read_only,
+ .can_extents = pipe_can_extents,
+ .can_multi_conn = pipe_can_multi_conn,
+ .start_multi_conn = pipe_start_multi_conn,
+
.flush = pipe_flush,
.synch_read = pipe_synch_read,
diff --git a/copy/synch-copying.c b/copy/synch-copying.c
index 2712c10..985f005 100644
--- a/copy/synch-copying.c
+++ b/copy/synch-copying.c
@@ -38,13 +38,13 @@ synch_copying (void)
/* If the source size is unknown then we copy data and cannot use
* extent information.
*/
- if (src.size == -1) {
+ if (src->size == -1) {
size_t r;
- while ((r = src.ops->synch_read (&src, buf, sizeof buf, offset)) > 0) {
- dst.ops->synch_write (&dst, buf, r, offset);
+ while ((r = src->ops->synch_read (src, buf, sizeof buf, offset)) > 0) {
+ dst->ops->synch_write (dst, buf, r, offset);
offset += r;
- progress_bar (offset, src.size);
+ progress_bar (offset, src->size);
}
}
@@ -52,47 +52,47 @@ synch_copying (void)
* blocks and use extent information to optimize the case.
*/
else {
- while (offset < src.size) {
+ while (offset < src->size) {
extent_list exts = empty_vector;
- uint64_t count = src.size - offset;
+ uint64_t count = src->size - offset;
size_t i, r;
if (count > sizeof buf)
count = sizeof buf;
if (extents)
- src.ops->get_extents (&src, 0, offset, count, &exts);
+ src->ops->get_extents (src, 0, offset, count, &exts);
else
- default_get_extents (&src, 0, offset, count, &exts);
+ default_get_extents (src, 0, offset, count, &exts);
for (i = 0; i < exts.size; ++i) {
assert (exts.ptr[i].length <= count);
if (exts.ptr[i].zero) {
- if (!dst.ops->synch_trim (&dst, offset, exts.ptr[i].length) &&
- !dst.ops->synch_zero (&dst, offset, exts.ptr[i].length)) {
+ if (!dst->ops->synch_trim (dst, offset, exts.ptr[i].length) &&
+ !dst->ops->synch_zero (dst, offset, exts.ptr[i].length)) {
/* If neither trimming nor efficient zeroing are possible,
* write zeroes the hard way.
*/
memset (buf, 0, exts.ptr[i].length);
- dst.ops->synch_write (&dst, buf, exts.ptr[i].length, offset);
+ dst->ops->synch_write (dst, buf, exts.ptr[i].length, offset);
}
offset += exts.ptr[i].length;
}
else /* data */ {
- r = src.ops->synch_read (&src, buf, exts.ptr[i].length, offset);
+ r = src->ops->synch_read (src, buf, exts.ptr[i].length, offset);
/* These cases should never happen unless the file is
* truncated underneath us.
*/
if (r == 0 || r < exts.ptr[i].length) {
- fprintf (stderr, "%s: unexpected end of file\n", src.name);
+ fprintf (stderr, "%s: unexpected end of file\n", src->name);
exit (EXIT_FAILURE);
}
- dst.ops->synch_write (&dst, buf, r, offset);
+ dst->ops->synch_write (dst, buf, r, offset);
offset += r;
- progress_bar (offset, src.size);
+ progress_bar (offset, src->size);
}
}
--
2.29.0.rc2