rt/snapraid
1
0
Fork 0
mirror of https://github.com/amadvance/snapraid.git synced 2026-10-26 19:01:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

IO functions renaming to prepare for parity write support

Browse source 
No functional change.
This commit is contained in:
Andrea Mazzoleni 2016-01-04 11:24:04 +01:00
parent d9f7657107
commit 62b7587c3a
4 changed files with 116 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

143
cmdline/io.c
View file

@ -28,8 +28,8 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
unsigned i;
pthread_mutex_init(&io->mutex, 0);
pthread_cond_init(&io->not_empty, 0);
pthread_cond_init(&io->not_full, 0);
pthread_cond_init(&io->read_done, 0);
pthread_cond_init(&io->read_sched, 0);
io->state = state;
@ -40,21 +40,22 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
mtest_vector(io->buffer_max, state->block_size, io->buffer_map[i]);
}
io->worker_max = handle_max + parity_handle_max;
io->worker_map = malloc_nofail(sizeof(struct snapraid_worker) * io->worker_max);
io->worker_list = malloc_nofail(io->worker_max + 1);
io->reader_max = handle_max + parity_handle_max;
io->reader_map = malloc_nofail(sizeof(struct snapraid_worker) * io->reader_max);
io->reader_list = malloc_nofail(io->reader_max + 1);
io->data_base = 0;
io->data_count = handle_max;
io->parity_base = handle_max;
io->parity_count = parity_handle_max;
for (i = 0; i < io->worker_max; ++i) {
struct snapraid_worker* worker = &io->worker_map[i];
for (i = 0; i < io->reader_max; ++i) {
struct snapraid_worker* worker = &io->reader_map[i];
worker->io = io;
if (i < handle_max) {
/* it's a data read */
worker->handle = &handle_map[i];
worker->parity_handle = 0;
worker->func = data_reader;
@ -62,6 +63,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
/* data read is put in lower buffer index */
worker->buffer_skew = 0;
} else {
/* it's a parity read */
worker->handle = 0;
worker->parity_handle = &parity_handle_map[i - handle_max];
worker->func = parity_reader;
@ -81,12 +83,12 @@ void io_done(struct snapraid_io* io)
free(io->buffer_alloc_map[i]);
}
free(io->worker_map);
free(io->worker_list);
free(io->reader_map);
free(io->reader_list);
pthread_mutex_destroy(&io->mutex);
pthread_cond_destroy(&io->not_empty);
pthread_cond_destroy(&io->not_full);
pthread_cond_destroy(&io->read_done);
pthread_cond_destroy(&io->read_sched);
}
/**
@ -109,14 +111,14 @@ static block_off_t io_position_next(struct snapraid_io* io)
}
/**
* Setup the next pending task for all workers.
* Setup the next pending task for all readers.
*/
static void io_pending_next(struct snapraid_io* io, int index, block_off_t blockcur)
static void io_reader_sched(struct snapraid_io* io, int index, block_off_t blockcur)
{
unsigned i;
for (i = 0; i < io->worker_max; ++i) {
struct snapraid_worker* worker = &io->worker_map[i];
for (i = 0; i < io->reader_max; ++i) {
struct snapraid_worker* worker = &io->reader_map[i];
struct snapraid_task* task = &worker->task_map[index];
/* setup the new pending task */
@ -141,11 +143,11 @@ static void io_pending_next(struct snapraid_io* io, int index, block_off_t block
}
/**
* Get the next task to work on for a worker.
* Get the next task to work on for a reader.
*
* This is the synchronization point for workers with the io.
*/
static struct snapraid_task* io_worker_next(struct snapraid_worker* worker)
static struct snapraid_task* io_reader_step(struct snapraid_worker* worker)
{
struct snapraid_io* io = worker->io;
struct snapraid_task* task = 0;
@ -157,6 +159,7 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker)
unsigned index;
/* check if the worker has to exit */
/* even if there is work to do */
if (io->done)
break;
@ -164,20 +167,20 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker)
index = (worker->index + 1) % IO_MAX;
/* if the queue of pending tasks is not empty */
if (index != io->index) {
if (index != io->reader_index) {
/* get the new working task */
worker->index = index;
task = &worker->task_map[worker->index];
/* notify the io that we started working */
pthread_cond_signal(&io->not_empty);
/* notify the io that a new read is complete */
pthread_cond_signal(&io->read_done);
/* return the new task */
break;
}
/* otherwise wait for a not_full event */
pthread_cond_wait(&io->not_full, &io->mutex);
/* otherwise wait for a read_sched event */
pthread_cond_wait(&io->read_sched, &io->mutex);
}
pthread_mutex_unlock(&io->mutex);
@ -190,75 +193,85 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker)
*
* This is the synchronization point for workers with the io.
*/
block_off_t io_next(struct snapraid_io* io, void*** buffer)
block_off_t io_read_next(struct snapraid_io* io, void*** buffer)
{
block_off_t blockcur_schedule;
block_off_t blockcur_caller;
unsigned i;
/* get the next parity position to process */
blockcur_schedule = io_position_next(io);
/* ensure that all data/parity was read */
assert(io->reader_list[0] == io->reader_max);
/* setup the list of workers to process */
for (i = 0; i <= io->worker_max; ++i)
io->worker_list[i] = i;
for (i = 0; i <= io->reader_max; ++i)
io->reader_list[i] = i;
/* the synchronization is protected by the io mutex */
pthread_mutex_lock(&io->mutex);
/* reset the task just used to be the next pending task */
io_pending_next(io, io->index, blockcur_schedule);
/* schedule the next read */
io_reader_sched(io, io->reader_index, blockcur_schedule);
/* get the next task to return to the caller */
io->index = (io->index + 1) % IO_MAX;
io->reader_index = (io->reader_index + 1) % IO_MAX;
/* get the position to operate at high level from one task */
blockcur_caller = io->worker_map[0].task_map[io->index].position;
blockcur_caller = io->reader_map[0].task_map[io->reader_index].position;
/* set the buffer to use */
*buffer = io->buffer_map[io->index];
*buffer = io->buffer_map[io->reader_index];
/* signal all the workers that there is a new pending task */
pthread_cond_broadcast(&io->not_full);
pthread_cond_broadcast(&io->read_sched);
pthread_mutex_unlock(&io->mutex);
return blockcur_caller;
}
static struct snapraid_task* io_task_next(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos)
static struct snapraid_task* io_task_read(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos)
{
/* the synchronization is protected by the io mutex */
pthread_mutex_lock(&io->mutex);
while (1) {
unsigned char* let;
unsigned index;
/* get the index the caller is using */
/* we must ensure that this index has not a read in progress */
/* to avoid a concurrent access */
index = io->reader_index;
/* search for a worker that has already finished */
let = &io->worker_list[0];
let = &io->reader_list[0];
while (1) {
unsigned i = *let;
/* if we are at the end */
if (i == io->worker_max)
if (i == io->reader_max)
break;
/* if it's in range */
if (base <= i && i < base + count) {
struct snapraid_worker* worker;
worker = &io->worker_map[i];
worker = &io->reader_map[i];
/* if the worker has finished this index */
if (io->index != worker->index) {
if (index != worker->index) {
struct snapraid_task* task;
task = &worker->task_map[io->index];
task = &worker->task_map[io->reader_index];
pthread_mutex_unlock(&io->mutex);
/* mark the worker as processed */
/* setting the previous one to point at the next one */
*let = io->worker_list[i + 1];
*let = io->reader_list[i + 1];
/* return the position */
*pos = i - base;
@ -268,46 +281,47 @@ static struct snapraid_task* io_task_next(struct snapraid_io* io, unsigned base,
}
/* next position to check */
let = &io->worker_list[i + 1];
let = &io->reader_list[i + 1];
}
/* if not worker is ready, wait for an event */
pthread_cond_wait(&io->not_empty, &io->mutex);
pthread_cond_wait(&io->read_done, &io->mutex);
}
}
struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos)
struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* pos)
{
return io_task_next(io, io->data_base, io->data_count, pos);
return io_task_read(io, io->data_base, io->data_count, pos);
}
struct snapraid_task* io_parity_next(struct snapraid_io* io, unsigned* pos)
struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* pos)
{
return io_task_next(io, io->parity_base, io->parity_count, pos);
return io_task_read(io, io->parity_base, io->parity_count, pos);
}
static void io_worker(struct snapraid_worker* worker, struct snapraid_task* task)
static void io_reader_worker(struct snapraid_worker* worker, struct snapraid_task* task)
{
/* if we reached the end */
if (task->position >= worker->io->block_max) {
/* complete a dummy task */
task->state = TASK_STATE_EMPTY;
} else {
worker->func(worker, task);
}
}
static void* io_thread(void* arg)
static void* io_reader_thread(void* arg)
{
struct snapraid_worker* worker = arg;
/* force completion of the first task */
io_worker(worker, &worker->task_map[0]);
io_reader_worker(worker, &worker->task_map[0]);
while (1) {
struct snapraid_task* task;
/* get the new task */
task = io_worker_next(worker);
task = io_reader_step(worker);
/* if no task, it means to exit */
if (!task)
@ -320,7 +334,7 @@ static void* io_thread(void* arg)
assert(task->state == TASK_STATE_READY);
/* work on the assigned task */
io_worker(worker, task);
io_reader_worker(worker, task);
}
return 0;
@ -339,25 +353,28 @@ void io_start(struct snapraid_io* io,
io->block_next = blockstart;
io->done = 0;
io->index = IO_MAX - 1;
io->reader_index = IO_MAX - 1;
/* setup the initial pending tasks, except the latest one, */
/* the latest will be initialized at the fist io_next() call */
/* setup the initial read pending tasks, except the latest one, */
/* the latest will be initialized at the fist io_read_next() call */
for (i = 0; i < IO_MAX - 1; ++i) {
block_off_t blockcur = io_position_next(io);
io_pending_next(io, i, blockcur);
io_reader_sched(io, i, blockcur);
}
/* start the worker threads */
for (i = 0; i < io->worker_max; ++i) {
struct snapraid_worker* worker = &io->worker_map[i];
/* setup the lists of workers to process */
io->reader_list[0] = io->reader_max;
/* start the reader threads */
for (i = 0; i < io->reader_max; ++i) {
struct snapraid_worker* worker = &io->reader_map[i];
worker->index = 0;
if (pthread_create(&worker->thread, 0, io_thread, worker) != 0) {
if (pthread_create(&worker->thread, 0, io_reader_thread, worker) != 0) {
/* LCOV_EXCL_START */
log_fatal("Failed to create thread.\n");
log_fatal("Failed to create reader thread.\n");
exit(EXIT_FAILURE);
/* LCOV_EXCL_STOP */
}
@ -374,19 +391,19 @@ void io_stop(struct snapraid_io* io)
io->done = 1;
/* signal all the threads to recognize the new state */
pthread_cond_broadcast(&io->not_full);
pthread_cond_broadcast(&io->read_sched);
pthread_mutex_unlock(&io->mutex);
/* wait for all threads to terminate */
for (i = 0; i < io->worker_max; ++i) {
struct snapraid_worker* worker = &io->worker_map[i];
/* wait for all readers to terminate */
for (i = 0; i < io->reader_max; ++i) {
struct snapraid_worker* worker = &io->reader_map[i];
void* retval;
/* wait for thread termination */
if (pthread_join(worker->thread, &retval) != 0) {
/* LCOV_EXCL_START */
log_fatal("Failed to join thread.\n");
log_fatal("Failed to join reader thread.\n");
exit(EXIT_FAILURE);
/* LCOV_EXCL_STOP */
}

54
cmdline/io.h
View file

@ -134,20 +134,22 @@ struct snapraid_io {
pthread_mutex_t mutex;
/**
* Condition for (not) empty queue of tasks.
* Condition for a new read is completed.
*
* The io waits on this condition when the queue is empty,
* and it's signaled by workers.
* The workers signal this condition when a new read is completed.
* The IO waits on this condition when it's waiting for
* a new read to be completed.
*/
pthread_cond_t not_empty;
pthread_cond_t read_done;
/**
* Condition for (not) full queue of tasks.
* Condition for a new read scheduled.
*
* The workers wait on this condition when the queue is full,
* and it's signaled by the io.
* The workers wait on this condition when they are waiting for a new
* read to process.
* The IO signals this condition when new reads are scheduled.
*/
pthread_cond_t not_full;
pthread_cond_t read_sched;
struct snapraid_state* state;
@ -191,22 +193,22 @@ struct snapraid_io {
/**
* Workers.
*
* A vector of workers, each one representing a different thread.
* A vector of readers, each one representing a different thread.
*/
unsigned worker_max; /**< Number of workers. */
struct snapraid_worker* worker_map; /**< Vector of workers. */
unsigned reader_max; /**< Number of workers. */
struct snapraid_worker* reader_map; /**< Vector of workers. */
/**
* List of not yet processed workers.
*
* The list has ::worker_max + 1 elements. Each element contains
* the number of the worker to process.
* The list has ::reader_max + 1 elements. Each element contains
* the number of the reader to process.
*
* At initialization the list is filled with [0..worker_max].
* At initialization the list is filled with [0..reader_max].
* To get the next element to process we use i = list[i + 1].
* The end is when i == worker_max.
* The end is when i == reader_max.
*/
unsigned char* worker_list;
unsigned char* reader_list;
/**
* Exit condition for all threads.
@ -218,8 +220,11 @@ struct snapraid_io {
*
* It's a rolling counter, when reaching IO_MAX
* it goes again to 0.
*
* When the caller finish with the current index,
* it's incremented, and a read_sched() signal is sent.
*/
unsigned index;
unsigned reader_index;
};
/**
@ -252,16 +257,19 @@ void io_start(struct snapraid_io* io,
void io_stop(struct snapraid_io* io);
/**
* Get the next parity position to process.
* Next read position.
*
* This call starts the reading process.
* It must be called before io_data_read() and io_parity_read().
*
* \param io InputOutput context.
* \param buffer The data buffers to use for this position.
* \return The parity position.
*/
block_off_t io_next(struct snapraid_io* io, void*** buffer);
block_off_t io_read_next(struct snapraid_io* io, void*** buffer);
/**
* Get the next data block to process.
* Read a data block.
*
* It must be called exactly ::handle_max times.
*
@ -269,10 +277,10 @@ block_off_t io_next(struct snapraid_io* io, void*** buffer);
* \param pos The position of the data block in the ::handle_map vector.
* \return The completed task.
*/
struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos);
struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* diskcur);
/**
* Get the next parity block to process.
* Read a parity block.
*
* It must be called exactly ::parity_handle_max times.
*
@ -280,6 +288,6 @@ struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos);
* \param pos The position of the parity block in the ::parity_handle_map vector.
* \return The completed task.
*/
struct snapraid_task* io_parity_next(struct snapraid_io* io, unsigned* pos);
struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* levcur);
#endif

6
cmdline/scrub.c
View file

@ -319,7 +319,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par
void** buffer;
/* go to the next block */
blockcur = io_next(&io, &buffer);
blockcur = io_read_next(&io, &buffer);
if (blockcur >= blockmax)
break;
@ -362,7 +362,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par
state_usage_cpu(state);
/* get the next task */
task = io_data_next(&io, &diskcur);
task = io_data_read(&io, &diskcur);
/* get the task results */
disk = task->disk;
@ -485,7 +485,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par
struct snapraid_task* task;
unsigned levcur;
task = io_parity_next(&io, &levcur);
task = io_parity_read(&io, &levcur);
/* until now is parity */
state_usage_parity(state, levcur);

4
cmdline/sync.c
View file

@ -715,7 +715,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
void** buffer;
/* go to the next block */
blockcur = io_next(&io, &buffer);
blockcur = io_read_next(&io, &buffer);
if (blockcur >= blockmax)
break;
@ -766,7 +766,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
/* until now is CPU */
state_usage_cpu(state);
task = io_data_next(&io, &diskcur);
task = io_data_read(&io, &diskcur);
/* get the results */
disk = task->disk;