4 * Copyright Red Hat Inc., 2013
7 * Stefan Hajnoczi <stefanha@redhat.com>
9 * This work is licensed under the terms of the GNU GPL, version 2 or later.
10 * See the COPYING file in the top-level directory.
14 #include "qemu/osdep.h"
15 #include "qom/object.h"
16 #include "qom/object_interfaces.h"
17 #include "qemu/module.h"
18 #include "block/aio.h"
19 #include "block/block.h"
20 #include "sysemu/iothread.h"
21 #include "qapi/error.h"
22 #include "qapi/qapi-commands-misc.h"
23 #include "qemu/error-report.h"
25 #include "qemu/main-loop.h"
27 typedef ObjectClass IOThreadClass
;
29 #define IOTHREAD_GET_CLASS(obj) \
30 OBJECT_GET_CLASS(IOThreadClass, obj, TYPE_IOTHREAD)
31 #define IOTHREAD_CLASS(klass) \
32 OBJECT_CLASS_CHECK(IOThreadClass, klass, TYPE_IOTHREAD)
35 /* Benchmark results from 2016 on NVMe SSD drives show max polling times around
36 * 16-32 microseconds yield IOPS improvements for both iodepth=1 and iodepth=32
39 #define IOTHREAD_POLL_MAX_NS_DEFAULT 32768ULL
41 #define IOTHREAD_POLL_MAX_NS_DEFAULT 0ULL
44 static __thread IOThread
*my_iothread
;
46 AioContext
*qemu_get_current_aio_context(void)
48 return my_iothread
? my_iothread
->ctx
: qemu_get_aio_context();
51 static void *iothread_run(void *opaque
)
53 IOThread
*iothread
= opaque
;
55 rcu_register_thread();
57 my_iothread
= iothread
;
58 iothread
->thread_id
= qemu_get_thread_id();
59 qemu_sem_post(&iothread
->init_done_sem
);
61 while (iothread
->running
) {
62 aio_poll(iothread
->ctx
, true);
65 * We must check the running state again in case it was
66 * changed in previous aio_poll()
68 if (iothread
->running
&& atomic_read(&iothread
->worker_context
)) {
71 g_main_context_push_thread_default(iothread
->worker_context
);
73 g_main_loop_new(iothread
->worker_context
, TRUE
);
74 loop
= iothread
->main_loop
;
76 g_main_loop_run(iothread
->main_loop
);
77 iothread
->main_loop
= NULL
;
78 g_main_loop_unref(loop
);
80 g_main_context_pop_thread_default(iothread
->worker_context
);
84 rcu_unregister_thread();
88 /* Runs in iothread_run() thread */
89 static void iothread_stop_bh(void *opaque
)
91 IOThread
*iothread
= opaque
;
93 iothread
->running
= false; /* stop iothread_run() */
95 if (iothread
->main_loop
) {
96 g_main_loop_quit(iothread
->main_loop
);
100 void iothread_stop(IOThread
*iothread
)
102 if (!iothread
->ctx
|| iothread
->stopping
) {
105 iothread
->stopping
= true;
106 aio_bh_schedule_oneshot(iothread
->ctx
, iothread_stop_bh
, iothread
);
107 qemu_thread_join(&iothread
->thread
);
110 static void iothread_instance_init(Object
*obj
)
112 IOThread
*iothread
= IOTHREAD(obj
);
114 iothread
->poll_max_ns
= IOTHREAD_POLL_MAX_NS_DEFAULT
;
115 iothread
->thread_id
= -1;
116 qemu_sem_init(&iothread
->init_done_sem
, 0);
119 static void iothread_instance_finalize(Object
*obj
)
121 IOThread
*iothread
= IOTHREAD(obj
);
123 iothread_stop(iothread
);
126 * Before glib2 2.33.10, there is a glib2 bug that GSource context
127 * pointer may not be cleared even if the context has already been
128 * destroyed (while it should). Here let's free the AIO context
129 * earlier to bypass that glib bug.
131 * We can remove this comment after the minimum supported glib2
132 * version boosts to 2.33.10. Before that, let's free the
133 * GSources first before destroying any GMainContext.
136 aio_context_unref(iothread
->ctx
);
137 iothread
->ctx
= NULL
;
139 if (iothread
->worker_context
) {
140 g_main_context_unref(iothread
->worker_context
);
141 iothread
->worker_context
= NULL
;
143 qemu_sem_destroy(&iothread
->init_done_sem
);
146 static void iothread_complete(UserCreatable
*obj
, Error
**errp
)
148 Error
*local_error
= NULL
;
149 IOThread
*iothread
= IOTHREAD(obj
);
150 char *name
, *thread_name
;
152 iothread
->stopping
= false;
153 iothread
->running
= true;
154 iothread
->ctx
= aio_context_new(&local_error
);
155 if (!iothread
->ctx
) {
156 error_propagate(errp
, local_error
);
160 aio_context_set_poll_params(iothread
->ctx
,
161 iothread
->poll_max_ns
,
163 iothread
->poll_shrink
,
166 error_propagate(errp
, local_error
);
167 aio_context_unref(iothread
->ctx
);
168 iothread
->ctx
= NULL
;
172 iothread
->once
= (GOnce
) G_ONCE_INIT
;
174 /* This assumes we are called from a thread with useful CPU affinity for us
177 name
= object_get_canonical_path_component(OBJECT(obj
));
178 thread_name
= g_strdup_printf("IO %s", name
);
179 qemu_thread_create(&iothread
->thread
, thread_name
, iothread_run
,
180 iothread
, QEMU_THREAD_JOINABLE
);
184 /* Wait for initialization to complete */
185 while (iothread
->thread_id
== -1) {
186 qemu_sem_wait(&iothread
->init_done_sem
);
192 ptrdiff_t offset
; /* field's byte offset in IOThread struct */
195 static PollParamInfo poll_max_ns_info
= {
196 "poll-max-ns", offsetof(IOThread
, poll_max_ns
),
198 static PollParamInfo poll_grow_info
= {
199 "poll-grow", offsetof(IOThread
, poll_grow
),
201 static PollParamInfo poll_shrink_info
= {
202 "poll-shrink", offsetof(IOThread
, poll_shrink
),
205 static void iothread_get_poll_param(Object
*obj
, Visitor
*v
,
206 const char *name
, void *opaque
, Error
**errp
)
208 IOThread
*iothread
= IOTHREAD(obj
);
209 PollParamInfo
*info
= opaque
;
210 int64_t *field
= (void *)iothread
+ info
->offset
;
212 visit_type_int64(v
, name
, field
, errp
);
215 static void iothread_set_poll_param(Object
*obj
, Visitor
*v
,
216 const char *name
, void *opaque
, Error
**errp
)
218 IOThread
*iothread
= IOTHREAD(obj
);
219 PollParamInfo
*info
= opaque
;
220 int64_t *field
= (void *)iothread
+ info
->offset
;
221 Error
*local_err
= NULL
;
224 visit_type_int64(v
, name
, &value
, &local_err
);
230 error_setg(&local_err
, "%s value must be in range [0, %"PRId64
"]",
231 info
->name
, INT64_MAX
);
238 aio_context_set_poll_params(iothread
->ctx
,
239 iothread
->poll_max_ns
,
241 iothread
->poll_shrink
,
246 error_propagate(errp
, local_err
);
249 static void iothread_class_init(ObjectClass
*klass
, void *class_data
)
251 UserCreatableClass
*ucc
= USER_CREATABLE_CLASS(klass
);
252 ucc
->complete
= iothread_complete
;
254 object_class_property_add(klass
, "poll-max-ns", "int",
255 iothread_get_poll_param
,
256 iothread_set_poll_param
,
257 NULL
, &poll_max_ns_info
, &error_abort
);
258 object_class_property_add(klass
, "poll-grow", "int",
259 iothread_get_poll_param
,
260 iothread_set_poll_param
,
261 NULL
, &poll_grow_info
, &error_abort
);
262 object_class_property_add(klass
, "poll-shrink", "int",
263 iothread_get_poll_param
,
264 iothread_set_poll_param
,
265 NULL
, &poll_shrink_info
, &error_abort
);
268 static const TypeInfo iothread_info
= {
269 .name
= TYPE_IOTHREAD
,
270 .parent
= TYPE_OBJECT
,
271 .class_init
= iothread_class_init
,
272 .instance_size
= sizeof(IOThread
),
273 .instance_init
= iothread_instance_init
,
274 .instance_finalize
= iothread_instance_finalize
,
275 .interfaces
= (InterfaceInfo
[]) {
276 {TYPE_USER_CREATABLE
},
281 static void iothread_register_types(void)
283 type_register_static(&iothread_info
);
286 type_init(iothread_register_types
)
288 char *iothread_get_id(IOThread
*iothread
)
290 return object_get_canonical_path_component(OBJECT(iothread
));
293 AioContext
*iothread_get_aio_context(IOThread
*iothread
)
295 return iothread
->ctx
;
298 static int query_one_iothread(Object
*object
, void *opaque
)
300 IOThreadInfoList
***prev
= opaque
;
301 IOThreadInfoList
*elem
;
305 iothread
= (IOThread
*)object_dynamic_cast(object
, TYPE_IOTHREAD
);
310 info
= g_new0(IOThreadInfo
, 1);
311 info
->id
= iothread_get_id(iothread
);
312 info
->thread_id
= iothread
->thread_id
;
313 info
->poll_max_ns
= iothread
->poll_max_ns
;
314 info
->poll_grow
= iothread
->poll_grow
;
315 info
->poll_shrink
= iothread
->poll_shrink
;
317 elem
= g_new0(IOThreadInfoList
, 1);
326 IOThreadInfoList
*qmp_query_iothreads(Error
**errp
)
328 IOThreadInfoList
*head
= NULL
;
329 IOThreadInfoList
**prev
= &head
;
330 Object
*container
= object_get_objects_root();
332 object_child_foreach(container
, query_one_iothread
, &prev
);
336 static gpointer
iothread_g_main_context_init(gpointer opaque
)
339 IOThread
*iothread
= opaque
;
342 iothread
->worker_context
= g_main_context_new();
344 ctx
= iothread_get_aio_context(iothread
);
345 source
= aio_get_g_source(ctx
);
346 g_source_attach(source
, iothread
->worker_context
);
347 g_source_unref(source
);
349 aio_notify(iothread
->ctx
);
353 GMainContext
*iothread_get_g_main_context(IOThread
*iothread
)
355 g_once(&iothread
->once
, iothread_g_main_context_init
, iothread
);
357 return iothread
->worker_context
;
360 IOThread
*iothread_create(const char *id
, Error
**errp
)
364 obj
= object_new_with_props(TYPE_IOTHREAD
,
365 object_get_internal_root(),
368 return IOTHREAD(obj
);
371 void iothread_destroy(IOThread
*iothread
)
373 object_unparent(OBJECT(iothread
));
376 /* Lookup IOThread by its id. Only finds user-created objects, not internal
377 * iothread_create() objects. */
378 IOThread
*iothread_by_id(const char *id
)
380 return IOTHREAD(object_resolve_path_type(id
, TYPE_IOTHREAD
, NULL
));