@@ -39,6 +39,22 @@ typedef struct task_queue {
3939 int tail ;
4040} task_queue ;
4141
42+ static int task_queue_init (task_queue * tasks , pthread_t thread ) {
43+ task * task_buffer = malloc (sizeof (task ) * TASK_QUEUE_INITIAL_CAPACITY );
44+ if (task_buffer == NULL ) {
45+ return 0 ;
46+ }
47+ * tasks = (task_queue ){.thread = thread ,
48+ .processing = 0 ,
49+ .tasks = task_buffer ,
50+ .capacity = TASK_QUEUE_INITIAL_CAPACITY ,
51+ .head = 0 ,
52+ .tail = 0 };
53+ return 1 ;
54+ }
55+
56+ static void task_queue_deinit (task_queue * tasks ) { free (tasks -> tasks ); }
57+
4258// Not thread safe.
4359static int task_queue_empty (task_queue * tasks ) {
4460 return tasks -> head == tasks -> tail ;
@@ -137,7 +153,7 @@ void em_proxying_queue_destroy(em_proxying_queue* q) {
137153 // of the queue.
138154 pthread_mutex_destroy (& q -> mutex );
139155 for (int i = 0 ; i < q -> size ; i ++ ) {
140- free ( q -> task_queues [i ]. tasks );
156+ task_queue_deinit ( & q -> task_queues [i ]);
141157 }
142158 free (q -> task_queues );
143159 free (q );
@@ -178,14 +194,9 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
178194 }
179195 // Initialize the next available task queue.
180196 tasks = & q -> task_queues [q -> size ];
181- tasks -> thread = thread ;
182- tasks -> processing = 0 ;
183- tasks -> tasks = malloc (sizeof (task ) * TASK_QUEUE_INITIAL_CAPACITY );
184- if (tasks -> tasks == NULL ) {
197+ if (!task_queue_init (tasks , thread )) {
185198 return NULL ;
186199 }
187- tasks -> head = 0 ;
188- tasks -> tail = 0 ;
189200 q -> size ++ ;
190201 return tasks ;
191202}
0 commit comments