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
->run_gcontext
)) {
69 g_main_context_push_thread_default(iothread
->worker_context
);
70 g_main_loop_run(iothread
->main_loop
);
71 g_main_context_pop_thread_default(iothread
->worker_context
);
75 rcu_unregister_thread();
79 /* Runs in iothread_run() thread */
80 static void iothread_stop_bh(void *opaque
)
82 IOThread
*iothread
= opaque
;
84 iothread
->running
= false; /* stop iothread_run() */
86 if (iothread
->main_loop
) {
87 g_main_loop_quit(iothread
->main_loop
);
91 void iothread_stop(IOThread
*iothread
)
93 if (!iothread
->ctx
|| iothread
->stopping
) {
96 iothread
->stopping
= true;
97 aio_bh_schedule_oneshot(iothread
->ctx
, iothread_stop_bh
, iothread
);
98 qemu_thread_join(&iothread
->thread
);
101 static void iothread_instance_init(Object
*obj
)
103 IOThread
*iothread
= IOTHREAD(obj
);
105 iothread
->poll_max_ns
= IOTHREAD_POLL_MAX_NS_DEFAULT
;
106 iothread
->thread_id
= -1;
107 qemu_sem_init(&iothread
->init_done_sem
, 0);
108 /* By default, we don't run gcontext */
109 atomic_set(&iothread
->run_gcontext
, 0);
112 static void iothread_instance_finalize(Object
*obj
)
114 IOThread
*iothread
= IOTHREAD(obj
);
116 iothread_stop(iothread
);
119 * Before glib2 2.33.10, there is a glib2 bug that GSource context
120 * pointer may not be cleared even if the context has already been
121 * destroyed (while it should). Here let's free the AIO context
122 * earlier to bypass that glib bug.
124 * We can remove this comment after the minimum supported glib2
125 * version boosts to 2.33.10. Before that, let's free the
126 * GSources first before destroying any GMainContext.
129 aio_context_unref(iothread
->ctx
);
130 iothread
->ctx
= NULL
;
132 if (iothread
->worker_context
) {
133 g_main_context_unref(iothread
->worker_context
);
134 iothread
->worker_context
= NULL
;
135 g_main_loop_unref(iothread
->main_loop
);
136 iothread
->main_loop
= NULL
;
138 qemu_sem_destroy(&iothread
->init_done_sem
);
141 static void iothread_init_gcontext(IOThread
*iothread
)
145 iothread
->worker_context
= g_main_context_new();
146 source
= aio_get_g_source(iothread_get_aio_context(iothread
));
147 g_source_attach(source
, iothread
->worker_context
);
148 g_source_unref(source
);
149 iothread
->main_loop
= g_main_loop_new(iothread
->worker_context
, TRUE
);
152 static void iothread_complete(UserCreatable
*obj
, Error
**errp
)
154 Error
*local_error
= NULL
;
155 IOThread
*iothread
= IOTHREAD(obj
);
156 char *name
, *thread_name
;
158 iothread
->stopping
= false;
159 iothread
->running
= true;
160 iothread
->ctx
= aio_context_new(&local_error
);
161 if (!iothread
->ctx
) {
162 error_propagate(errp
, local_error
);
167 * Init one GMainContext for the iothread unconditionally, even if
170 iothread_init_gcontext(iothread
);
172 aio_context_set_poll_params(iothread
->ctx
,
173 iothread
->poll_max_ns
,
175 iothread
->poll_shrink
,
178 error_propagate(errp
, local_error
);
179 aio_context_unref(iothread
->ctx
);
180 iothread
->ctx
= NULL
;
184 /* This assumes we are called from a thread with useful CPU affinity for us
187 name
= object_get_canonical_path_component(OBJECT(obj
));
188 thread_name
= g_strdup_printf("IO %s", name
);
189 qemu_thread_create(&iothread
->thread
, thread_name
, iothread_run
,
190 iothread
, QEMU_THREAD_JOINABLE
);
194 /* Wait for initialization to complete */
195 while (iothread
->thread_id
== -1) {
196 qemu_sem_wait(&iothread
->init_done_sem
);
202 ptrdiff_t offset
; /* field's byte offset in IOThread struct */
205 static PollParamInfo poll_max_ns_info
= {
206 "poll-max-ns", offsetof(IOThread
, poll_max_ns
),
208 static PollParamInfo poll_grow_info
= {
209 "poll-grow", offsetof(IOThread
, poll_grow
),
211 static PollParamInfo poll_shrink_info
= {
212 "poll-shrink", offsetof(IOThread
, poll_shrink
),
215 static void iothread_get_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
;
222 visit_type_int64(v
, name
, field
, errp
);
225 static void iothread_set_poll_param(Object
*obj
, Visitor
*v
,
226 const char *name
, void *opaque
, Error
**errp
)
228 IOThread
*iothread
= IOTHREAD(obj
);
229 PollParamInfo
*info
= opaque
;
230 int64_t *field
= (void *)iothread
+ info
->offset
;
231 Error
*local_err
= NULL
;
234 visit_type_int64(v
, name
, &value
, &local_err
);
240 error_setg(&local_err
, "%s value must be in range [0, %"PRId64
"]",
241 info
->name
, INT64_MAX
);
248 aio_context_set_poll_params(iothread
->ctx
,
249 iothread
->poll_max_ns
,
251 iothread
->poll_shrink
,
256 error_propagate(errp
, local_err
);
259 static void iothread_class_init(ObjectClass
*klass
, void *class_data
)
261 UserCreatableClass
*ucc
= USER_CREATABLE_CLASS(klass
);
262 ucc
->complete
= iothread_complete
;
264 object_class_property_add(klass
, "poll-max-ns", "int",
265 iothread_get_poll_param
,
266 iothread_set_poll_param
,
267 NULL
, &poll_max_ns_info
, &error_abort
);
268 object_class_property_add(klass
, "poll-grow", "int",
269 iothread_get_poll_param
,
270 iothread_set_poll_param
,
271 NULL
, &poll_grow_info
, &error_abort
);
272 object_class_property_add(klass
, "poll-shrink", "int",
273 iothread_get_poll_param
,
274 iothread_set_poll_param
,
275 NULL
, &poll_shrink_info
, &error_abort
);
278 static const TypeInfo iothread_info
= {
279 .name
= TYPE_IOTHREAD
,
280 .parent
= TYPE_OBJECT
,
281 .class_init
= iothread_class_init
,
282 .instance_size
= sizeof(IOThread
),
283 .instance_init
= iothread_instance_init
,
284 .instance_finalize
= iothread_instance_finalize
,
285 .interfaces
= (InterfaceInfo
[]) {
286 {TYPE_USER_CREATABLE
},
291 static void iothread_register_types(void)
293 type_register_static(&iothread_info
);
296 type_init(iothread_register_types
)
298 char *iothread_get_id(IOThread
*iothread
)
300 return object_get_canonical_path_component(OBJECT(iothread
));
303 AioContext
*iothread_get_aio_context(IOThread
*iothread
)
305 return iothread
->ctx
;
308 static int query_one_iothread(Object
*object
, void *opaque
)
310 IOThreadInfoList
***prev
= opaque
;
311 IOThreadInfoList
*elem
;
315 iothread
= (IOThread
*)object_dynamic_cast(object
, TYPE_IOTHREAD
);
320 info
= g_new0(IOThreadInfo
, 1);
321 info
->id
= iothread_get_id(iothread
);
322 info
->thread_id
= iothread
->thread_id
;
323 info
->poll_max_ns
= iothread
->poll_max_ns
;
324 info
->poll_grow
= iothread
->poll_grow
;
325 info
->poll_shrink
= iothread
->poll_shrink
;
327 elem
= g_new0(IOThreadInfoList
, 1);
336 IOThreadInfoList
*qmp_query_iothreads(Error
**errp
)
338 IOThreadInfoList
*head
= NULL
;
339 IOThreadInfoList
**prev
= &head
;
340 Object
*container
= object_get_objects_root();
342 object_child_foreach(container
, query_one_iothread
, &prev
);
346 GMainContext
*iothread_get_g_main_context(IOThread
*iothread
)
348 atomic_set(&iothread
->run_gcontext
, 1);
349 aio_notify(iothread
->ctx
);
350 return iothread
->worker_context
;
353 IOThread
*iothread_create(const char *id
, Error
**errp
)
357 obj
= object_new_with_props(TYPE_IOTHREAD
,
358 object_get_internal_root(),
361 return IOTHREAD(obj
);
364 void iothread_destroy(IOThread
*iothread
)
366 object_unparent(OBJECT(iothread
));
369 /* Lookup IOThread by its id. Only finds user-created objects, not internal
370 * iothread_create() objects. */
371 IOThread
*iothread_by_id(const char *id
)
373 return IOTHREAD(object_resolve_path_type(id
, TYPE_IOTHREAD
, NULL
));