Replace the command index with a worker pointer. The nbd-ops access the index via the worker pointer. This allows commands to modify worker state during processing. Signed-off-by: Nir Soffer <nsoffer@redhat.com> --- copy/multi-thread-copying.c | 12 ++++++------ copy/nbd-ops.c | 6 +++--- copy/nbdcopy.h | 2 +- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/copy/multi-thread-copying.c b/copy/multi-thread-copying.c index a1a8d09c..8ba721fe 100644 --- a/copy/multi-thread-copying.c +++ b/copy/multi-thread-copying.c @@ -131,21 +131,21 @@ multi_thread_copying (void) } static void wait_for_request_slots (size_t index); static unsigned in_flight (size_t index); static void poll_both_ends (size_t index); static int finished_read (void *vp, int *error); static int finished_command (void *vp, int *error); static void free_command (struct command *command); static void fill_dst_range_with_zeroes (struct command *command); static struct command *create_command (uint64_t offset, size_t len, bool zero, - size_t index); + struct worker *worker); /* There are 'threads' worker threads, each copying work ranges from * src to dst until there are no more work ranges. */ static void * worker_thread (void *wp) { struct worker *w = wp; uint64_t offset, count; extent_list exts = empty_vector; @@ -161,36 +161,36 @@ worker_thread (void *wp) for (i = 0; i < exts.len; ++i) { struct command *command; size_t len; if (exts.ptr[i].zero) { /* The source is zero so we can proceed directly to skipping, * fast zeroing, or writing zeroes at the destination. */ command = create_command (exts.ptr[i].offset, exts.ptr[i].length, - true, w->index); + true, w); fill_dst_range_with_zeroes (command); } else /* data */ { /* As the extent might be larger than permitted for a single * command, we may have to split this into multiple read * requests. */ while (exts.ptr[i].length > 0) { len = exts.ptr[i].length; if (len > request_size) len = request_size; command = create_command (exts.ptr[i].offset, len, - false, w->index); + false, w); wait_for_request_slots (w->index); /* Begin the asynch read operation. */ src->ops->asynch_read (src, command, (nbd_completion_callback) { .callback = finished_read, .user_data = command, }); @@ -332,38 +332,38 @@ create_buffer (size_t len) exit (EXIT_FAILURE); } buffer->refs = 1; return buffer; } /* Create a new command for read or zero. */ static struct command * -create_command (uint64_t offset, size_t len, bool zero, size_t index) +create_command (uint64_t offset, size_t len, bool zero, struct worker *worker) { struct command *command; command = calloc (1, sizeof *command); if (command == NULL) { perror ("calloc"); exit (EXIT_FAILURE); } command->offset = offset; command->slice.len = len; command->slice.base = 0; if (!zero) command->slice.buffer = create_buffer (len); - command->index = index; + command->worker = worker; return command; } /* Create a sub-command of an existing command. This creates a slice * referencing the buffer of the existing command without copying. */ static struct command * create_subcommand (struct command *command, uint64_t offset, size_t len, bool zero) @@ -379,21 +379,21 @@ create_subcommand (struct command *command, uint64_t offset, size_t len, perror ("calloc"); exit (EXIT_FAILURE); } newcommand->offset = offset; newcommand->slice.len = len; if (!zero) { newcommand->slice.buffer = command->slice.buffer; newcommand->slice.buffer->refs++; newcommand->slice.base = offset - command->offset; } - newcommand->index = command->index; + newcommand->worker = command->worker; return newcommand; } /* Callback called when src has finished one read command. This * initiates a write. */ static int finished_read (void *vp, int *error) { diff --git a/copy/nbd-ops.c b/copy/nbd-ops.c index dca86e88..adfe4de5 100644 --- a/copy/nbd-ops.c +++ b/copy/nbd-ops.c @@ -296,57 +296,57 @@ nbd_ops_synch_zero (struct rw *rw, uint64_t offset, uint64_t count, return true; } static void nbd_ops_asynch_read (struct rw *rw, struct command *command, nbd_completion_callback cb) { struct rw_nbd *rwn = (struct rw_nbd *) rw; - if (nbd_aio_pread (rwn->handles.ptr[command->index], + if (nbd_aio_pread (rwn->handles.ptr[command->worker->index], slice_ptr (command->slice), command->slice.len, command->offset, cb, 0) == -1) { fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ()); exit (EXIT_FAILURE); } } static void nbd_ops_asynch_write (struct rw *rw, struct command *command, nbd_completion_callback cb) { struct rw_nbd *rwn = (struct rw_nbd *) rw; - if (nbd_aio_pwrite (rwn->handles.ptr[command->index], + if (nbd_aio_pwrite (rwn->handles.ptr[command->worker->index], slice_ptr (command->slice), command->slice.len, command->offset, cb, 0) == -1) { fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ()); exit (EXIT_FAILURE); } } static bool nbd_ops_asynch_zero (struct rw *rw, struct command *command, nbd_completion_callback cb, bool allocate) { struct rw_nbd *rwn = (struct rw_nbd *) rw; if (!rwn->can_zero) return false; assert (command->slice.len <= UINT32_MAX); - if (nbd_aio_zero (rwn->handles.ptr[command->index], + if (nbd_aio_zero (rwn->handles.ptr[command->worker->index], command->slice.len, command->offset, cb, allocate ? LIBNBD_CMD_FLAG_NO_HOLE : 0) == -1) { fprintf (stderr, "%s: %s\n", rw->name, nbd_get_error ()); exit (EXIT_FAILURE); } return true; } static int add_extent (void *vp, const char *metacontext, diff --git a/copy/nbdcopy.h b/copy/nbdcopy.h index 4fe8bee6..8027836b 100644 --- a/copy/nbdcopy.h +++ b/copy/nbdcopy.h @@ -92,21 +92,21 @@ struct worker { * * slice.buffer may be NULL for commands (like zero) that have no * associated data. * * A separate set of commands, slices and buffers is maintained per * thread so no locking is necessary. */ struct command { uint64_t offset; /* Offset relative to start of disk. */ struct slice slice; /* Data slice. */ - size_t index; /* Thread number. */ + struct worker *worker; /* The worker owning this command. */ }; /* List of extents for rw->ops->get_extents. */ struct extent { uint64_t offset; uint64_t length; bool zero; }; DEFINE_VECTOR_TYPE(extent_list, struct extent); -- 2.35.1