1 | #undef G_DISABLE_ASSERT |
2 | #undef G_LOG_DOMAIN |
3 | |
4 | #include <time.h> |
5 | #include <stdlib.h> |
6 | |
7 | #include <glib.h> |
8 | |
9 | #define DEBUG_MSG(args) |
10 | /* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n"); */ |
11 | #define PRINT_MSG(args) |
12 | /* #define PRINT_MSG(args) g_printerr args ; g_printerr ("\n"); */ |
13 | |
14 | #define MAX_THREADS 50 |
15 | #define MAX_SORTS 5 /* only applies if |
16 | ASYC_QUEUE_DO_SORT is set to 1 */ |
17 | #define MAX_TIME 20 /* seconds */ |
18 | #define MIN_TIME 5 /* seconds */ |
19 | |
20 | #define SORT_QUEUE_AFTER 1 |
21 | #define SORT_QUEUE_ON_PUSH 1 /* if this is done, the |
22 | SORT_QUEUE_AFTER is ignored */ |
23 | #define QUIT_WHEN_DONE 1 |
24 | |
25 | |
26 | #if SORT_QUEUE_ON_PUSH == 1 |
27 | # undef SORT_QUEUE_AFTER |
28 | # define SORT_QUEUE_AFTER 0 |
29 | #endif |
30 | |
31 | |
32 | static GMainLoop *main_loop = NULL; |
33 | static GThreadPool *thread_pool = NULL; |
34 | static GAsyncQueue *async_queue = NULL; |
35 | |
36 | |
37 | static gint |
38 | sort_compare (gconstpointer p1, gconstpointer p2, gpointer user_data) |
39 | { |
40 | gint32 id1; |
41 | gint32 id2; |
42 | |
43 | id1 = GPOINTER_TO_INT (p1); |
44 | id2 = GPOINTER_TO_INT (p2); |
45 | |
46 | DEBUG_MSG (("comparing #1:%d and #2:%d, returning %d" , |
47 | id1, id2, (id1 > id2 ? +1 : id1 == id2 ? 0 : -1))); |
48 | |
49 | return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1); |
50 | } |
51 | |
52 | static gboolean |
53 | sort_queue (gpointer user_data) |
54 | { |
55 | static gint sorts = 0; |
56 | static gpointer last_p = NULL; |
57 | gpointer p; |
58 | gboolean can_quit = FALSE; |
59 | gint sort_multiplier; |
60 | gint len; |
61 | gint i; |
62 | |
63 | sort_multiplier = GPOINTER_TO_INT (user_data); |
64 | |
65 | if (SORT_QUEUE_AFTER) { |
66 | PRINT_MSG (("sorting async queue..." )); |
67 | g_async_queue_sort (queue: async_queue, func: sort_compare, NULL); |
68 | |
69 | sorts++; |
70 | |
71 | if (sorts >= sort_multiplier) { |
72 | can_quit = TRUE; |
73 | } |
74 | |
75 | g_async_queue_sort (queue: async_queue, func: sort_compare, NULL); |
76 | len = g_async_queue_length (queue: async_queue); |
77 | |
78 | PRINT_MSG (("sorted queue (for %d/%d times, size:%d)..." , sorts, MAX_SORTS, len)); |
79 | } else { |
80 | can_quit = TRUE; |
81 | len = g_async_queue_length (queue: async_queue); |
82 | DEBUG_MSG (("printing queue (size:%d)..." , len)); |
83 | } |
84 | |
85 | for (i = 0, last_p = NULL; i < len; i++) { |
86 | p = g_async_queue_pop (queue: async_queue); |
87 | DEBUG_MSG (("item %d ---> %d" , i, GPOINTER_TO_INT (p))); |
88 | |
89 | if (last_p) { |
90 | g_assert (GPOINTER_TO_INT (last_p) <= GPOINTER_TO_INT (p)); |
91 | } |
92 | |
93 | last_p = p; |
94 | } |
95 | |
96 | if (can_quit && QUIT_WHEN_DONE) { |
97 | g_main_loop_quit (loop: main_loop); |
98 | } |
99 | |
100 | return !can_quit; |
101 | } |
102 | |
103 | static void |
104 | enter_thread (gpointer data, gpointer user_data) |
105 | { |
106 | gint len G_GNUC_UNUSED; |
107 | gint id; |
108 | gulong ms; |
109 | |
110 | id = GPOINTER_TO_INT (data); |
111 | |
112 | ms = g_random_int_range (MIN_TIME * 1000, MAX_TIME * 1000); |
113 | DEBUG_MSG (("entered thread with id:%d, adding to queue in:%ld ms" , id, ms)); |
114 | |
115 | g_usleep (microseconds: ms * 1000); |
116 | |
117 | if (SORT_QUEUE_ON_PUSH) { |
118 | g_async_queue_push_sorted (queue: async_queue, GINT_TO_POINTER (id), func: sort_compare, NULL); |
119 | } else { |
120 | g_async_queue_push (queue: async_queue, GINT_TO_POINTER (id)); |
121 | } |
122 | |
123 | len = g_async_queue_length (queue: async_queue); |
124 | |
125 | DEBUG_MSG (("thread id:%d added to async queue (size:%d)" , |
126 | id, len)); |
127 | } |
128 | |
129 | static gint destroy_count = 0; |
130 | |
131 | static void |
132 | counting_destroy (gpointer item) |
133 | { |
134 | destroy_count++; |
135 | } |
136 | |
137 | static void |
138 | basic_tests (void) |
139 | { |
140 | GAsyncQueue *q; |
141 | gpointer item; |
142 | |
143 | destroy_count = 0; |
144 | |
145 | q = g_async_queue_new_full (item_free_func: counting_destroy); |
146 | g_async_queue_lock (queue: q); |
147 | g_async_queue_ref (queue: q); |
148 | g_async_queue_unlock (queue: q); |
149 | g_async_queue_lock (queue: q); |
150 | g_async_queue_ref_unlocked (queue: q); |
151 | g_async_queue_unref_and_unlock (queue: q); |
152 | |
153 | item = g_async_queue_try_pop (queue: q); |
154 | g_assert (item == NULL); |
155 | |
156 | g_async_queue_lock (queue: q); |
157 | item = g_async_queue_try_pop_unlocked (queue: q); |
158 | g_async_queue_unlock (queue: q); |
159 | g_assert (item == NULL); |
160 | |
161 | g_async_queue_push (queue: q, GINT_TO_POINTER (1)); |
162 | g_async_queue_push (queue: q, GINT_TO_POINTER (2)); |
163 | g_async_queue_push (queue: q, GINT_TO_POINTER (3)); |
164 | g_assert_cmpint (destroy_count, ==, 0); |
165 | |
166 | g_async_queue_unref (queue: q); |
167 | g_assert_cmpint (destroy_count, ==, 0); |
168 | |
169 | item = g_async_queue_pop (queue: q); |
170 | g_assert_cmpint (GPOINTER_TO_INT (item), ==, 1); |
171 | g_assert_cmpint (destroy_count, ==, 0); |
172 | |
173 | g_async_queue_unref (queue: q); |
174 | g_assert_cmpint (destroy_count, ==, 2); |
175 | } |
176 | |
177 | int |
178 | main (int argc, char *argv[]) |
179 | { |
180 | gint i; |
181 | gint max_threads = MAX_THREADS; |
182 | gint max_unused_threads = MAX_THREADS; |
183 | gint sort_multiplier = MAX_SORTS; |
184 | gint sort_interval; |
185 | gchar *msg G_GNUC_UNUSED; |
186 | |
187 | basic_tests (); |
188 | |
189 | PRINT_MSG (("creating async queue..." )); |
190 | async_queue = g_async_queue_new (); |
191 | |
192 | g_return_val_if_fail (async_queue != NULL, EXIT_FAILURE); |
193 | |
194 | PRINT_MSG (("creating thread pool with max threads:%d, max unused threads:%d..." , |
195 | max_threads, max_unused_threads)); |
196 | thread_pool = g_thread_pool_new (func: enter_thread, |
197 | user_data: async_queue, |
198 | max_threads, |
199 | FALSE, |
200 | NULL); |
201 | |
202 | g_return_val_if_fail (thread_pool != NULL, EXIT_FAILURE); |
203 | |
204 | g_thread_pool_set_max_unused_threads (max_threads: max_unused_threads); |
205 | |
206 | PRINT_MSG (("creating threads..." )); |
207 | for (i = 1; i <= max_threads; i++) { |
208 | GError *error = NULL; |
209 | |
210 | g_thread_pool_push (pool: thread_pool, GINT_TO_POINTER (i), error: &error); |
211 | |
212 | g_assert_no_error (error); |
213 | } |
214 | |
215 | if (!SORT_QUEUE_AFTER) { |
216 | sort_multiplier = 1; |
217 | } |
218 | |
219 | sort_interval = ((MAX_TIME / sort_multiplier) + 2) * 1000; |
220 | g_timeout_add (interval: sort_interval, function: sort_queue, GINT_TO_POINTER (sort_multiplier)); |
221 | |
222 | if (SORT_QUEUE_ON_PUSH) { |
223 | msg = "sorting when pushing into the queue, checking queue is sorted" ; |
224 | } else { |
225 | msg = "sorting" ; |
226 | } |
227 | |
228 | PRINT_MSG (("%s %d %s %d ms" , |
229 | msg, |
230 | sort_multiplier, |
231 | sort_multiplier == 1 ? "time in" : "times, once every" , |
232 | sort_interval)); |
233 | |
234 | DEBUG_MSG (("entering main event loop" )); |
235 | |
236 | main_loop = g_main_loop_new (NULL, FALSE); |
237 | g_main_loop_run (loop: main_loop); |
238 | |
239 | g_main_loop_unref (loop: main_loop); |
240 | g_thread_pool_free (pool: thread_pool, TRUE, TRUE); |
241 | g_async_queue_unref (queue: async_queue); |
242 | |
243 | return EXIT_SUCCESS; |
244 | } |
245 | |