1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225
|
/**
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
* SPDX-License-Identifier: Apache-2.0.
*/
#include <aws/common/clock.h>
#include <aws/common/condition_variable.h>
#include <aws/common/mutex.h>
#include <aws/common/ref_count.h>
#include <aws/common/task_scheduler.h>
#include <aws/common/thread.h>
#include <aws/common/thread_scheduler.h>
struct aws_thread_scheduler {
struct aws_allocator *allocator;
struct aws_ref_count ref_count;
struct aws_thread thread;
struct aws_task_scheduler scheduler;
struct aws_atomic_var should_exit;
struct {
struct aws_linked_list scheduling_queue;
struct aws_linked_list cancel_queue;
struct aws_mutex mutex;
struct aws_condition_variable c_var;
} thread_data;
};
struct cancellation_node {
struct aws_task *task_to_cancel;
struct aws_linked_list_node node;
};
static void s_destroy_callback(void *arg) {
struct aws_thread_scheduler *scheduler = arg;
aws_atomic_store_int(&scheduler->should_exit, 1U);
aws_condition_variable_notify_all(&scheduler->thread_data.c_var);
aws_thread_join(&scheduler->thread);
aws_task_scheduler_clean_up(&scheduler->scheduler);
aws_condition_variable_clean_up(&scheduler->thread_data.c_var);
aws_mutex_clean_up(&scheduler->thread_data.mutex);
aws_thread_clean_up(&scheduler->thread);
aws_mem_release(scheduler->allocator, scheduler);
}
static bool s_thread_should_wake(void *arg) {
struct aws_thread_scheduler *scheduler = arg;
uint64_t current_time = 0;
aws_high_res_clock_get_ticks(¤t_time);
uint64_t next_scheduled_task = 0;
aws_task_scheduler_has_tasks(&scheduler->scheduler, &next_scheduled_task);
return aws_atomic_load_int(&scheduler->should_exit) ||
!aws_linked_list_empty(&scheduler->thread_data.scheduling_queue) ||
!aws_linked_list_empty(&scheduler->thread_data.cancel_queue) || (next_scheduled_task <= current_time);
}
static void s_thread_fn(void *arg) {
struct aws_thread_scheduler *scheduler = arg;
while (!aws_atomic_load_int(&scheduler->should_exit)) {
/* move tasks from the mutex protected list to the scheduler. This is because we don't want to hold the lock
* for the scheduler during run_all and then try and acquire the lock from another thread to schedule something
* because that potentially would block the calling thread. */
struct aws_linked_list list_cpy;
aws_linked_list_init(&list_cpy);
struct aws_linked_list cancel_list_cpy;
aws_linked_list_init(&cancel_list_cpy);
AWS_FATAL_ASSERT(!aws_mutex_lock(&scheduler->thread_data.mutex) && "mutex lock failed!");
aws_linked_list_swap_contents(&scheduler->thread_data.scheduling_queue, &list_cpy);
aws_linked_list_swap_contents(&scheduler->thread_data.cancel_queue, &cancel_list_cpy);
AWS_FATAL_ASSERT(!aws_mutex_unlock(&scheduler->thread_data.mutex) && "mutex unlock failed!");
while (!aws_linked_list_empty(&list_cpy)) {
struct aws_linked_list_node *node = aws_linked_list_pop_front(&list_cpy);
struct aws_task *task = AWS_CONTAINER_OF(node, struct aws_task, node);
if (task->timestamp) {
aws_task_scheduler_schedule_future(&scheduler->scheduler, task, task->timestamp);
} else {
aws_task_scheduler_schedule_now(&scheduler->scheduler, task);
}
}
/* now cancel the tasks. */
while (!aws_linked_list_empty(&cancel_list_cpy)) {
struct aws_linked_list_node *node = aws_linked_list_pop_front(&cancel_list_cpy);
struct cancellation_node *cancellation_node = AWS_CONTAINER_OF(node, struct cancellation_node, node);
aws_task_scheduler_cancel_task(&scheduler->scheduler, cancellation_node->task_to_cancel);
aws_mem_release(scheduler->allocator, cancellation_node);
}
/* now run everything */
uint64_t current_time = 0;
aws_high_res_clock_get_ticks(¤t_time);
aws_task_scheduler_run_all(&scheduler->scheduler, current_time);
uint64_t next_scheduled_task = 0;
aws_task_scheduler_has_tasks(&scheduler->scheduler, &next_scheduled_task);
int64_t timeout = 0;
if (next_scheduled_task == UINT64_MAX) {
/* at least wake up once per 30 seconds. */
timeout = (int64_t)30 * (int64_t)AWS_TIMESTAMP_NANOS;
} else {
timeout = (int64_t)(next_scheduled_task - current_time);
}
if (timeout > 0) {
AWS_FATAL_ASSERT(!aws_mutex_lock(&scheduler->thread_data.mutex) && "mutex lock failed!");
aws_condition_variable_wait_for_pred(
&scheduler->thread_data.c_var, &scheduler->thread_data.mutex, timeout, s_thread_should_wake, scheduler);
AWS_FATAL_ASSERT(!aws_mutex_unlock(&scheduler->thread_data.mutex) && "mutex unlock failed!");
}
}
}
struct aws_thread_scheduler *aws_thread_scheduler_new(
struct aws_allocator *allocator,
const struct aws_thread_options *thread_options) {
struct aws_thread_scheduler *scheduler = aws_mem_calloc(allocator, 1, sizeof(struct aws_thread_scheduler));
if (!scheduler) {
return NULL;
}
if (aws_thread_init(&scheduler->thread, allocator)) {
goto clean_up;
}
AWS_FATAL_ASSERT(!aws_mutex_init(&scheduler->thread_data.mutex) && "mutex init failed!");
AWS_FATAL_ASSERT(!aws_condition_variable_init(&scheduler->thread_data.c_var) && "condition variable init failed!");
if (aws_task_scheduler_init(&scheduler->scheduler, allocator)) {
goto thread_init;
}
scheduler->allocator = allocator;
aws_atomic_init_int(&scheduler->should_exit, 0U);
aws_ref_count_init(&scheduler->ref_count, scheduler, s_destroy_callback);
aws_linked_list_init(&scheduler->thread_data.scheduling_queue);
aws_linked_list_init(&scheduler->thread_data.cancel_queue);
if (aws_thread_launch(&scheduler->thread, s_thread_fn, scheduler, thread_options)) {
goto scheduler_init;
}
return scheduler;
scheduler_init:
aws_task_scheduler_clean_up(&scheduler->scheduler);
thread_init:
aws_condition_variable_clean_up(&scheduler->thread_data.c_var);
aws_mutex_clean_up(&scheduler->thread_data.mutex);
aws_thread_clean_up(&scheduler->thread);
clean_up:
aws_mem_release(allocator, scheduler);
return NULL;
}
void aws_thread_scheduler_acquire(struct aws_thread_scheduler *scheduler) {
aws_ref_count_acquire(&scheduler->ref_count);
}
void aws_thread_scheduler_release(const struct aws_thread_scheduler *scheduler) {
aws_ref_count_release((struct aws_ref_count *)&scheduler->ref_count);
}
void aws_thread_scheduler_schedule_future(
struct aws_thread_scheduler *scheduler,
struct aws_task *task,
uint64_t time_to_run) {
task->timestamp = time_to_run;
AWS_FATAL_ASSERT(!aws_mutex_lock(&scheduler->thread_data.mutex) && "mutex lock failed!");
aws_linked_list_push_back(&scheduler->thread_data.scheduling_queue, &task->node);
AWS_FATAL_ASSERT(!aws_mutex_unlock(&scheduler->thread_data.mutex) && "mutex unlock failed!");
aws_condition_variable_notify_one(&scheduler->thread_data.c_var);
}
void aws_thread_scheduler_schedule_now(struct aws_thread_scheduler *scheduler, struct aws_task *task) {
aws_thread_scheduler_schedule_future(scheduler, task, 0U);
}
void aws_thread_scheduler_cancel_task(struct aws_thread_scheduler *scheduler, struct aws_task *task) {
struct cancellation_node *cancellation_node =
aws_mem_calloc(scheduler->allocator, 1, sizeof(struct cancellation_node));
AWS_FATAL_ASSERT(cancellation_node && "allocation failed for cancellation node!");
AWS_FATAL_ASSERT(!aws_mutex_lock(&scheduler->thread_data.mutex) && "mutex lock failed!");
struct aws_task *found_task = NULL;
/* remove tasks that are still in the scheduling queue, but haven't made it to the scheduler yet. */
struct aws_linked_list_node *node = aws_linked_list_empty(&scheduler->thread_data.scheduling_queue)
? NULL
: aws_linked_list_front(&scheduler->thread_data.scheduling_queue);
while (node != NULL) {
struct aws_task *potential_task = AWS_CONTAINER_OF(node, struct aws_task, node);
if (potential_task == task) {
found_task = potential_task;
break;
}
if (aws_linked_list_node_next_is_valid(node)) {
node = aws_linked_list_next(node);
} else {
node = NULL;
}
}
if (found_task) {
aws_linked_list_remove(&found_task->node);
}
cancellation_node->task_to_cancel = task;
/* regardless put it in the cancel queue so the thread can call the task with canceled status. */
aws_linked_list_push_back(&scheduler->thread_data.cancel_queue, &cancellation_node->node);
AWS_FATAL_ASSERT(!aws_mutex_unlock(&scheduler->thread_data.mutex) && "mutex unlock failed!");
/* notify so the loop knows to wakeup and process the cancellations. */
aws_condition_variable_notify_one(&scheduler->thread_data.c_var);
}
|