1 /* GIO - GLib Input, Output and Streaming Library
2 *
3 * Copyright (C) 2006-2007 Red Hat, Inc.
4 *
5 * This library is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Lesser General Public
7 * License as published by the Free Software Foundation; either
8 * version 2 of the License, or (at your option) any later version.
9 *
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Lesser General Public License for more details.
14 *
15 * You should have received a copy of the GNU Lesser General
16 * Public License along with this library; if not, write to the
17 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
18 * Boston, MA 02111-1307, USA.
19 *
20 * Author: Alexander Larsson <alexl@redhat.com>
21 */
22
23 #include "config.h"
24
25 #include "gioscheduler.h"
26 #include "gcancellable.h"
27
28 #include "gioalias.h"
29
30 /**
31 * SECTION:gioscheduler
32 * @short_description: I/O Scheduler
33 * @include: gio/gio.h
34 *
35 * Schedules asynchronous I/O operations. #GIOScheduler integrates
36 * into the main event loop (#GMainLoop) and may use threads if they
37 * are available.
38 *
39 * <para id="io-priority"><indexterm><primary>I/O priority</primary></indexterm>
40 * Each I/O operation has a priority, and the scheduler uses the priorities
41 * to determine the order in which operations are executed. They are
42 * <emphasis>not</emphasis> used to determine system-wide I/O scheduling.
43 * Priorities are integers, with lower numbers indicating higher priority.
44 * It is recommended to choose priorities between %G_PRIORITY_LOW and
45 * %G_PRIORITY_HIGH, with %G_PRIORITY_DEFAULT as a default.
46 * </para>
47 **/
48
49 struct _GIOSchedulerJob {
50 GSList *active_link;
51 GIOSchedulerJobFunc job_func;
52 GSourceFunc cancel_func; /* Runs under job map lock */
53 gpointer data;
54 GDestroyNotify destroy_notify;
55
56 gint io_priority;
57 GCancellable *cancellable;
58
59 guint idle_tag;
60 };
61
62 G_LOCK_DEFINE_STATIC(active_jobs);
63 static GSList *active_jobs = NULL;
64
65 static GThreadPool *job_thread_pool = NULL;
66
67 static void io_job_thread (gpointer data,
68 gpointer user_data);
69
70 static void
g_io_job_free(GIOSchedulerJob * job)71 g_io_job_free (GIOSchedulerJob *job)
72 {
73 if (job->cancellable)
74 g_object_unref (job->cancellable);
75 g_free (job);
76 }
77
78 static gint
g_io_job_compare(gconstpointer a,gconstpointer b,gpointer user_data)79 g_io_job_compare (gconstpointer a,
80 gconstpointer b,
81 gpointer user_data)
82 {
83 const GIOSchedulerJob *aa = a;
84 const GIOSchedulerJob *bb = b;
85
86 /* Cancelled jobs are set prio == -1, so that
87 they are executed as quickly as possible */
88
89 /* Lower value => higher priority */
90 if (aa->io_priority < bb->io_priority)
91 return -1;
92 if (aa->io_priority == bb->io_priority)
93 return 0;
94 return 1;
95 }
96
97 static gpointer
init_scheduler(gpointer arg)98 init_scheduler (gpointer arg)
99 {
100 if (job_thread_pool == NULL)
101 {
102 /* TODO: thread_pool_new can fail */
103 job_thread_pool = g_thread_pool_new (io_job_thread,
104 NULL,
105 10,
106 FALSE,
107 NULL);
108 if (job_thread_pool != NULL)
109 {
110 g_thread_pool_set_sort_function (job_thread_pool,
111 g_io_job_compare,
112 NULL);
113 /* It's kinda weird that this is a global setting
114 * instead of per threadpool. However, we really
115 * want to cache some threads, but not keep around
116 * those threads forever. */
117 g_thread_pool_set_max_idle_time (15 * 1000);
118 g_thread_pool_set_max_unused_threads (2);
119 }
120 }
121 return NULL;
122 }
123
124 static void
remove_active_job(GIOSchedulerJob * job)125 remove_active_job (GIOSchedulerJob *job)
126 {
127 GIOSchedulerJob *other_job;
128 GSList *l;
129 gboolean resort_jobs;
130
131 G_LOCK (active_jobs);
132 active_jobs = g_slist_delete_link (active_jobs, job->active_link);
133
134 resort_jobs = FALSE;
135 for (l = active_jobs; l != NULL; l = l->next)
136 {
137 other_job = l->data;
138 if (other_job->io_priority >= 0 &&
139 g_cancellable_is_cancelled (other_job->cancellable))
140 {
141 other_job->io_priority = -1;
142 resort_jobs = TRUE;
143 }
144 }
145 G_UNLOCK (active_jobs);
146
147 if (resort_jobs &&
148 job_thread_pool != NULL)
149 g_thread_pool_set_sort_function (job_thread_pool,
150 g_io_job_compare,
151 NULL);
152
153 }
154
155 static void
job_destroy(gpointer data)156 job_destroy (gpointer data)
157 {
158 GIOSchedulerJob *job = data;
159
160 if (job->destroy_notify)
161 job->destroy_notify (job->data);
162
163 remove_active_job (job);
164 g_io_job_free (job);
165 }
166
167 static void
io_job_thread(gpointer data,gpointer user_data)168 io_job_thread (gpointer data,
169 gpointer user_data)
170 {
171 GIOSchedulerJob *job = data;
172 gboolean result;
173
174 if (job->cancellable)
175 g_cancellable_push_current (job->cancellable);
176
177 do
178 {
179 result = job->job_func (job, job->cancellable, job->data);
180 }
181 while (result);
182
183 if (job->cancellable)
184 g_cancellable_pop_current (job->cancellable);
185
186 job_destroy (job);
187 }
188
189 static gboolean
run_job_at_idle(gpointer data)190 run_job_at_idle (gpointer data)
191 {
192 GIOSchedulerJob *job = data;
193 gboolean result;
194
195 if (job->cancellable)
196 g_cancellable_push_current (job->cancellable);
197
198 result = job->job_func (job, job->cancellable, job->data);
199
200 if (job->cancellable)
201 g_cancellable_pop_current (job->cancellable);
202
203 return result;
204 }
205
206 /**
207 * g_io_scheduler_push_job:
208 * @job_func: a #GIOSchedulerJobFunc.
209 * @user_data: data to pass to @job_func
210 * @notify: a #GDestroyNotify for @user_data, or %NULL
211 * @io_priority: the <link linkend="gioscheduler">I/O priority</link>
212 * of the request.
213 * @cancellable: optional #GCancellable object, %NULL to ignore.
214 *
215 * Schedules the I/O job to run.
216 *
217 * @notify will be called on @user_data after @job_func has returned,
218 * regardless whether the job was cancelled or has run to completion.
219 *
220 * If @cancellable is not %NULL, it can be used to cancel the I/O job
221 * by calling g_cancellable_cancel() or by calling
222 * g_io_scheduler_cancel_all_jobs().
223 **/
224 void
g_io_scheduler_push_job(GIOSchedulerJobFunc job_func,gpointer user_data,GDestroyNotify notify,gint io_priority,GCancellable * cancellable)225 g_io_scheduler_push_job (GIOSchedulerJobFunc job_func,
226 gpointer user_data,
227 GDestroyNotify notify,
228 gint io_priority,
229 GCancellable *cancellable)
230 {
231 static GOnce once_init = G_ONCE_INIT;
232 GIOSchedulerJob *job;
233
234 g_return_if_fail (job_func != NULL);
235
236 job = g_new0 (GIOSchedulerJob, 1);
237 job->job_func = job_func;
238 job->data = user_data;
239 job->destroy_notify = notify;
240 job->io_priority = io_priority;
241
242 if (cancellable)
243 job->cancellable = g_object_ref (cancellable);
244
245 G_LOCK (active_jobs);
246 active_jobs = g_slist_prepend (active_jobs, job);
247 job->active_link = active_jobs;
248 G_UNLOCK (active_jobs);
249
250 if (g_thread_supported())
251 {
252 g_once (&once_init, init_scheduler, NULL);
253 g_thread_pool_push (job_thread_pool, job, NULL);
254 }
255 else
256 {
257 /* Threads not available, instead do the i/o sync inside a
258 * low prio idle handler
259 */
260 job->idle_tag = g_idle_add_full (G_PRIORITY_DEFAULT_IDLE + 1 + io_priority / 10,
261 run_job_at_idle,
262 job, job_destroy);
263 }
264 }
265
266 /**
267 * g_io_scheduler_cancel_all_jobs:
268 *
269 * Cancels all cancellable I/O jobs.
270 *
271 * A job is cancellable if a #GCancellable was passed into
272 * g_io_scheduler_push_job().
273 **/
274 void
g_io_scheduler_cancel_all_jobs(void)275 g_io_scheduler_cancel_all_jobs (void)
276 {
277 GSList *cancellable_list, *l;
278
279 G_LOCK (active_jobs);
280 cancellable_list = NULL;
281 for (l = active_jobs; l != NULL; l = l->next)
282 {
283 GIOSchedulerJob *job = l->data;
284 if (job->cancellable)
285 cancellable_list = g_slist_prepend (cancellable_list,
286 g_object_ref (job->cancellable));
287 }
288 G_UNLOCK (active_jobs);
289
290 for (l = cancellable_list; l != NULL; l = l->next)
291 {
292 GCancellable *c = l->data;
293 g_cancellable_cancel (c);
294 g_object_unref (c);
295 }
296 g_slist_free (cancellable_list);
297 }
298
299 typedef struct {
300 GSourceFunc func;
301 gboolean ret_val;
302 gpointer data;
303 GDestroyNotify notify;
304
305 GMutex *ack_lock;
306 GCond *ack_condition;
307 } MainLoopProxy;
308
309 static gboolean
mainloop_proxy_func(gpointer data)310 mainloop_proxy_func (gpointer data)
311 {
312 MainLoopProxy *proxy = data;
313
314 proxy->ret_val = proxy->func (proxy->data);
315
316 if (proxy->notify)
317 proxy->notify (proxy->data);
318
319 if (proxy->ack_lock)
320 {
321 g_mutex_lock (proxy->ack_lock);
322 g_cond_signal (proxy->ack_condition);
323 g_mutex_unlock (proxy->ack_lock);
324 }
325
326 return FALSE;
327 }
328
329 static void
mainloop_proxy_free(MainLoopProxy * proxy)330 mainloop_proxy_free (MainLoopProxy *proxy)
331 {
332 if (proxy->ack_lock)
333 {
334 g_mutex_free (proxy->ack_lock);
335 g_cond_free (proxy->ack_condition);
336 }
337
338 g_free (proxy);
339 }
340
341 /**
342 * g_io_scheduler_job_send_to_mainloop:
343 * @job: a #GIOSchedulerJob
344 * @func: a #GSourceFunc callback that will be called in the main thread
345 * @user_data: data to pass to @func
346 * @notify: a #GDestroyNotify for @user_data, or %NULL
347 *
348 * Used from an I/O job to send a callback to be run in the
349 * main loop (main thread), waiting for the result (and thus
350 * blocking the I/O job).
351 *
352 * Returns: The return value of @func
353 **/
354 gboolean
g_io_scheduler_job_send_to_mainloop(GIOSchedulerJob * job,GSourceFunc func,gpointer user_data,GDestroyNotify notify)355 g_io_scheduler_job_send_to_mainloop (GIOSchedulerJob *job,
356 GSourceFunc func,
357 gpointer user_data,
358 GDestroyNotify notify)
359 {
360 GSource *source;
361 MainLoopProxy *proxy;
362 guint id;
363 gboolean ret_val;
364
365 g_return_val_if_fail (job != NULL, FALSE);
366 g_return_val_if_fail (func != NULL, FALSE);
367
368 if (job->idle_tag)
369 {
370 /* We just immediately re-enter in the case of idles (non-threads)
371 * Anything else would just deadlock. If you can't handle this, enable threads.
372 */
373 ret_val = func (user_data);
374 if (notify)
375 notify (user_data);
376 return ret_val;
377 }
378
379 proxy = g_new0 (MainLoopProxy, 1);
380 proxy->func = func;
381 proxy->data = user_data;
382 proxy->notify = notify;
383 proxy->ack_lock = g_mutex_new ();
384 proxy->ack_condition = g_cond_new ();
385 g_mutex_lock (proxy->ack_lock);
386
387 source = g_idle_source_new ();
388 g_source_set_priority (source, G_PRIORITY_DEFAULT);
389 g_source_set_callback (source, mainloop_proxy_func, proxy,
390 NULL);
391
392 id = g_source_attach (source, NULL);
393 g_source_unref (source);
394
395 g_cond_wait (proxy->ack_condition, proxy->ack_lock);
396 g_mutex_unlock (proxy->ack_lock);
397
398 ret_val = proxy->ret_val;
399 mainloop_proxy_free (proxy);
400
401 return ret_val;
402 }
403
404 /**
405 * g_io_scheduler_job_send_to_mainloop_async:
406 * @job: a #GIOSchedulerJob
407 * @func: a #GSourceFunc callback that will be called in the main thread
408 * @user_data: data to pass to @func
409 * @notify: a #GDestroyNotify for @user_data, or %NULL
410 *
411 * Used from an I/O job to send a callback to be run asynchronously
412 * in the main loop (main thread). The callback will be run when the
413 * main loop is available, but at that time the I/O job might have
414 * finished. The return value from the callback is ignored.
415 *
416 * Note that if you are passing the @user_data from g_io_scheduler_push_job()
417 * on to this function you have to ensure that it is not freed before
418 * @func is called, either by passing %NULL as @notify to
419 * g_io_scheduler_push_job() or by using refcounting for @user_data.
420 **/
421 void
g_io_scheduler_job_send_to_mainloop_async(GIOSchedulerJob * job,GSourceFunc func,gpointer user_data,GDestroyNotify notify)422 g_io_scheduler_job_send_to_mainloop_async (GIOSchedulerJob *job,
423 GSourceFunc func,
424 gpointer user_data,
425 GDestroyNotify notify)
426 {
427 GSource *source;
428 MainLoopProxy *proxy;
429 guint id;
430
431 g_return_if_fail (job != NULL);
432 g_return_if_fail (func != NULL);
433
434 if (job->idle_tag)
435 {
436 /* We just immediately re-enter in the case of idles (non-threads)
437 * Anything else would just deadlock. If you can't handle this, enable threads.
438 */
439 func (user_data);
440 if (notify)
441 notify (user_data);
442 return;
443 }
444
445 proxy = g_new0 (MainLoopProxy, 1);
446 proxy->func = func;
447 proxy->data = user_data;
448 proxy->notify = notify;
449
450 source = g_idle_source_new ();
451 g_source_set_priority (source, G_PRIORITY_DEFAULT);
452 g_source_set_callback (source, mainloop_proxy_func, proxy,
453 (GDestroyNotify)mainloop_proxy_free);
454
455 id = g_source_attach (source, NULL);
456 g_source_unref (source);
457 }
458
459
460 #define __G_IO_SCHEDULER_C__
461 #include "gioaliasdef.c"
462