Code Commits61
t
Download ZIP
139 lines2.9 KBC
1/*
2jobs.c: worker threads.
3*/
4#include <stdio.h>
5#include <stdlib.h>
6#include <string.h>
7#include "plat.h"
8#include "jobs.h"
9
10static const int lane_threads[NUM_LANES] = {1, 2, 1};
11
12static struct {
13 plat_mutex_t *lock;
14 job_t *head, *tail;
15 volatile int count;
16 volatile int busy; // jobs running now
17} lanes[NUM_LANES];
18
19static void Job_Unref(job_t *j)
20{
21 if (__sync_sub_and_fetch(&j->refs, 1) == 0) {
22 if (j->cleanup)
23 j->cleanup(j);
24 free(j);
25 }
26}
27
28static job_t *Take(int lane)
29{
30 job_t *j;
31 plat_mutex_lock(lanes[lane].lock);
32 j = lanes[lane].head;
33 if (j) {
34 lanes[lane].head = j->next;
35 if (!lanes[lane].head)
36 lanes[lane].tail = NULL;
37 lanes[lane].count--;
38 }
39 plat_mutex_unlock(lanes[lane].lock);
40 return j;
41}
42
43static void Worker(void *arg)
44{
45 int lane = (int)(intptr_t)arg;
46 for (;;) {
47 __sync_fetch_and_add(&lanes[lane].busy, 1);
48 job_t *j = Take(lane);
49 if (!j) {
50 __sync_fetch_and_sub(&lanes[lane].busy, 1);
51 plat_sleep_us(5000);
52 continue;
53 }
54 if (!j->cancel)
55 j->run(j);
56 __sync_fetch_and_sub(&lanes[lane].busy, 1);
57 __atomic_store_n(&j->done, 1, __ATOMIC_RELEASE);
58 Job_Unref(j);
59 }
60}
61
62void jobs_init(void)
63{
64 static const char *names[NUM_LANES] = {"jobs_api", "jobs_art", "jobs_dl"};
65 for (int l = 0; l < NUM_LANES; l++) {
66 lanes[l].lock = plat_mutex_new();
67 for (int t = 0; t < lane_threads[l]; t++)
68 plat_thread_start(Worker, (void *)(intptr_t)l, 1700, 256, names[l]); // art decoding etc: after the frames
69 }
70}
71
72job_t *job_new(void (*run)(job_t *), void (*cleanup)(job_t *))
73{
74 job_t *j = calloc(1, sizeof(*j));
75 if (!j) return NULL;
76 j->run = run;
77 j->cleanup = cleanup;
78 j->refs = 2; // the poster and the worker
79 return j;
80}
81
82void job_post(job_t *j, int lane)
83{
84 plat_mutex_lock(lanes[lane].lock);
85 if (lane == LANE_ART) {
86 j->next = lanes[lane].head;
87 lanes[lane].head = j;
88 if (!lanes[lane].tail)
89 lanes[lane].tail = j;
90 } else {
91 j->next = NULL;
92 if (lanes[lane].tail)
93 lanes[lane].tail->next = j;
94 else
95 lanes[lane].head = j;
96 lanes[lane].tail = j;
97 }
98 lanes[lane].count++;
99 plat_mutex_unlock(lanes[lane].lock);
100}
101
102void job_release(job_t *j)
103{
104 if (!j) return;
105 j->cancel = 1;
106 Job_Unref(j);
107}
108
109// fire and forget: the poster lets go but the job still runs (job_release would cancel it)
110void job_detach(job_t *j)
111{
112 if (j) Job_Unref(j);
113}
114
115void jobs_cancel_lane(int lane)
116{
117 job_t *list;
118 plat_mutex_lock(lanes[lane].lock);
119 list = lanes[lane].head;
120 lanes[lane].head = lanes[lane].tail = NULL;
121 lanes[lane].count = 0;
122 plat_mutex_unlock(lanes[lane].lock);
123 // never ran: the worker's reference goes now
124 while (list) {
125 job_t *next = list->next;
126 list->cancel = 1;
127 __atomic_store_n(&list->done, 1, __ATOMIC_RELEASE);
128 Job_Unref(list);
129 list = next;
130 }
131}
132
133// at exit: give a lane's last jobs a moment to finish
134void jobs_wait_lane(int lane, unsigned max_us)
135{
136 uint64_t until = plat_time_us() + max_us;
137 while ((lanes[lane].count > 0 || lanes[lane].busy > 0) && plat_time_us() < until)
138 plat_sleep_us(20000);
139}