diff options
| author | Paul Buetow <pbuetow@mimecast.com> | 2018-03-06 17:38:59 +0000 |
|---|---|---|
| committer | Paul Buetow <pbuetow@mimecast.com> | 2018-03-06 17:38:59 +0000 |
| commit | 26b3b3e368a79ce29df732ea04e72a4c002ae2ce (patch) | |
| tree | e3fc8d7461ab371279f7bf9c692096cd39cc92f6 /ioreplay/src/replay/rthread.c | |
| parent | ae2221660f9b411fa78cdf8034f0803e9a870cde (diff) | |
rename into ioriot
Diffstat (limited to 'ioreplay/src/replay/rthread.c')
| -rw-r--r-- | ioreplay/src/replay/rthread.c | 216 |
1 files changed, 0 insertions, 216 deletions
diff --git a/ioreplay/src/replay/rthread.c b/ioreplay/src/replay/rthread.c deleted file mode 100644 index 55364ec..0000000 --- a/ioreplay/src/replay/rthread.c +++ /dev/null @@ -1,216 +0,0 @@ -// Copyright 2018 Mimecast Ltd. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "rthread.h" - -#include <sys/types.h> - -#include "rworker.h" -#include "rprocess.h" - -#include "rioop.h" - -#ifdef THREAD_DEBUG -/** - * @brief For debugging purposes only - * - * @param t The responsible thread object - */ -static void _rthread_init_log(rthread_s *t) -{ - rworker_s *w = t->worker; - char *rthread_log = Calloc(1024, char); - snprintf(rthread_log, 1023, "/tmp/ioreplay/worker%d.thread%ld.debuglog", - w->rworker_num, (long)pthread_self()); - - ensure_dir_exists("/tmp/ioreplay"); - t->rthread_fd = Fopen(rthread_log, "a"); - - free(rthread_log); - fprintf(t->rthread_fd, "%ld: DEBUG: Created thread log\n", t->tid); -} -#endif - -void rthread_process_task(rthread_s* t, rtask_s *task, - pid_t pthread_id) -{ - char *next = task->line; - rworker_s *w = (rworker_s*) task->worker; - - // Tokenize the remaining elements of the line. - int ntoks = 0; - char *saveptr; - char *tok = strtok_r(next, "|", &saveptr); - - while (tok) { - if (ntoks > MAX_TOKENS) { - Error("worker(%d) pthread(%d): lineno:%lu, missing newline?", - w->rworker_num, pthread_id, task->lineno); - } - task->toks[ntoks++] = tok; - tok = strtok_r(NULL, "|", &saveptr); - } - // NULL marker (no more token from here) - task->toks[ntoks] = NULL; - -#ifdef THREAD_DEBUG - fprintf(t->rthread_fd, "%ld(%ld): %s", - t->tid, (long)pthread_self(), task->clone); - fflush(t->rthread_fd); - free(task->clone); - task->clone = NULL; -#endif -#ifndef NO_RIOOP - // Perform the corresponding I/O operation! - rioop_run(task->process, t, task); -#endif - - // Make the task object recyclable/reusable - pthread_mutex_lock(&w->task_buffer_mutex); - if (!rbuffer_insert(w->task_buffer, task)) - // We can't recycle the task object if the buffer is full! - rtask_destroy(task); - pthread_mutex_unlock(&w->task_buffer_mutex); -} - -void *rthread_pthread_start(void *data) -{ - rthread_s* t = (rthread_s*) data; - rworker_s *w = t->worker; - rtask_s *task = NULL; - pid_t pthread_id = pthread_self(); - -#ifdef THREAD_DEBUG - _rthread_init_log(t); -#endif - - do { - while (!rbuffer_has_next(t->tasks) && !t->terminate) - usleep(100); - - while ((task = rbuffer_get_next(t->tasks)) != NULL) - rthread_process_task(t, task, pthread_id); - -#ifdef THREAD_DEBUG - fprintf(t->rthread_fd, "%ld: DEBUG: Idling\n", t->tid); - fflush(t->rthread_fd); -#endif - - // Tell rworker_s that thread is not doing any work! - int inserted = false; - while (!inserted && !t->terminate) { - if (rbuffer_has_next(t->tasks)) - break; - - usleep(1000); - - if (rbuffer_has_next(t->tasks)) - break; - - // Make the rthread reusable, he is without any tasks - // for some time. - pthread_mutex_lock(&w->rthread_buffer_mutex); - inserted = rbuffer_insert(w->rthread_buffer, t); - pthread_mutex_unlock(&w->rthread_buffer_mutex); - } - -#ifdef THREAD_DEBUG - if (inserted) { - fprintf(t->rthread_fd, "%ld: DEBUG: Added to thread buffer\n", - t->tid); - } else { - fprintf(t->rthread_fd, "%ld: DEBUG: Idling thread recovered\n", - t->tid); - } - fflush(t->rthread_fd); -#endif - - } while (!t->terminate); - -#ifdef THREAD_DEBUG - fprintf(t->rthread_fd, "%ld: DEBUG: Terminating\n", t->tid); - fflush(t->rthread_fd); -#endif - - // Process the very last tasks - while (NULL != (task = rbuffer_get_next(t->tasks))) - rthread_process_task(t, task, pthread_id); - -#ifdef THREAD_DEBUG - fprintf(t->rthread_fd, "%ld: DEBUG: Done terminating\n", t->tid); - fflush(t->rthread_fd); -#endif - - return NULL; -} - -rthread_s* rthread_new(const long tid, void *worker) -{ - rthread_s *t = Malloc(rthread_s); - rworker_s *w = worker; - - t->single_threaded = w->opts->num_threads_per_worker == 1; - t->tasks = rbuffer_new(TASK_BUFFER_PER_THREAD); - t->terminate = false; - t->worker = worker; - rthread_update(t, tid); - - if (t->single_threaded) { -#ifdef THREAD_DEBUG - _rthread_init_log(t); -#endif - return t; - } - - start_pthread(&t->pthread, rthread_pthread_start, (void*)t); - return t; -} - -long rthread_update(rthread_s *t, const long tid) -{ - long prev_tid = t->tid; - t->tid = tid; - - return prev_tid; -} - -void rthread_destroy(rthread_s *t) -{ - if (rbuffer_has_next(t->tasks)) { - Error("Didn't expect to have any tasks left!"); - } - rbuffer_destroy(t->tasks); - -#ifdef THREAD_DEBUG - if (t->rthread_fd) - fclose(t->rthread_fd); -#endif - - free(t); -} - -bool rthread_insert_task(rthread_s* t, rtask_s* task) -{ - if (t->single_threaded) { - rthread_process_task(t, task, pthread_self()); - return true; - } - return rbuffer_insert(t->tasks, task); -} - -void rthread_terminate(rthread_s* t) -{ - t->terminate = true; - pthread_join(t->pthread, NULL); -} |
