xref: /qemu/iothread.c (revision 603476c2)
1 /*
2  * Event loop thread
3  *
4  * Copyright Red Hat Inc., 2013
5  *
6  * Authors:
7  *  Stefan Hajnoczi   <stefanha@redhat.com>
8  *
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.
11  *
12  */
13 
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 "sysemu/iothread.h"
20 #include "qmp-commands.h"
21 #include "qemu/error-report.h"
22 #include "qemu/rcu.h"
23 
24 typedef ObjectClass IOThreadClass;
25 
26 #define IOTHREAD_GET_CLASS(obj) \
27    OBJECT_GET_CLASS(IOThreadClass, obj, TYPE_IOTHREAD)
28 #define IOTHREAD_CLASS(klass) \
29    OBJECT_CLASS_CHECK(IOThreadClass, klass, TYPE_IOTHREAD)
30 
31 static void *iothread_run(void *opaque)
32 {
33     IOThread *iothread = opaque;
34     bool blocking;
35 
36     rcu_register_thread();
37 
38     qemu_mutex_lock(&iothread->init_done_lock);
39     iothread->thread_id = qemu_get_thread_id();
40     qemu_cond_signal(&iothread->init_done_cond);
41     qemu_mutex_unlock(&iothread->init_done_lock);
42 
43     while (!iothread->stopping) {
44         aio_context_acquire(iothread->ctx);
45         blocking = true;
46         while (!iothread->stopping && aio_poll(iothread->ctx, blocking)) {
47             /* Progress was made, keep going */
48             blocking = false;
49         }
50         aio_context_release(iothread->ctx);
51     }
52 
53     rcu_unregister_thread();
54     return NULL;
55 }
56 
57 static int iothread_stop(Object *object, void *opaque)
58 {
59     IOThread *iothread;
60 
61     iothread = (IOThread *)object_dynamic_cast(object, TYPE_IOTHREAD);
62     if (!iothread || !iothread->ctx) {
63         return 0;
64     }
65     iothread->stopping = true;
66     aio_notify(iothread->ctx);
67     qemu_thread_join(&iothread->thread);
68     return 0;
69 }
70 
71 static void iothread_instance_finalize(Object *obj)
72 {
73     IOThread *iothread = IOTHREAD(obj);
74 
75     iothread_stop(obj, NULL);
76     qemu_cond_destroy(&iothread->init_done_cond);
77     qemu_mutex_destroy(&iothread->init_done_lock);
78     if (!iothread->ctx) {
79         return;
80     }
81     aio_context_unref(iothread->ctx);
82 }
83 
84 static void iothread_complete(UserCreatable *obj, Error **errp)
85 {
86     Error *local_error = NULL;
87     IOThread *iothread = IOTHREAD(obj);
88     char *name, *thread_name;
89 
90     iothread->stopping = false;
91     iothread->thread_id = -1;
92     iothread->ctx = aio_context_new(&local_error);
93     if (!iothread->ctx) {
94         error_propagate(errp, local_error);
95         return;
96     }
97 
98     qemu_mutex_init(&iothread->init_done_lock);
99     qemu_cond_init(&iothread->init_done_cond);
100 
101     /* This assumes we are called from a thread with useful CPU affinity for us
102      * to inherit.
103      */
104     name = object_get_canonical_path_component(OBJECT(obj));
105     thread_name = g_strdup_printf("IO %s", name);
106     qemu_thread_create(&iothread->thread, thread_name, iothread_run,
107                        iothread, QEMU_THREAD_JOINABLE);
108     g_free(thread_name);
109     g_free(name);
110 
111     /* Wait for initialization to complete */
112     qemu_mutex_lock(&iothread->init_done_lock);
113     while (iothread->thread_id == -1) {
114         qemu_cond_wait(&iothread->init_done_cond,
115                        &iothread->init_done_lock);
116     }
117     qemu_mutex_unlock(&iothread->init_done_lock);
118 }
119 
120 static void iothread_class_init(ObjectClass *klass, void *class_data)
121 {
122     UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
123     ucc->complete = iothread_complete;
124 }
125 
126 static const TypeInfo iothread_info = {
127     .name = TYPE_IOTHREAD,
128     .parent = TYPE_OBJECT,
129     .class_init = iothread_class_init,
130     .instance_size = sizeof(IOThread),
131     .instance_finalize = iothread_instance_finalize,
132     .interfaces = (InterfaceInfo[]) {
133         {TYPE_USER_CREATABLE},
134         {}
135     },
136 };
137 
138 static void iothread_register_types(void)
139 {
140     type_register_static(&iothread_info);
141 }
142 
143 type_init(iothread_register_types)
144 
145 char *iothread_get_id(IOThread *iothread)
146 {
147     return object_get_canonical_path_component(OBJECT(iothread));
148 }
149 
150 AioContext *iothread_get_aio_context(IOThread *iothread)
151 {
152     return iothread->ctx;
153 }
154 
155 static int query_one_iothread(Object *object, void *opaque)
156 {
157     IOThreadInfoList ***prev = opaque;
158     IOThreadInfoList *elem;
159     IOThreadInfo *info;
160     IOThread *iothread;
161 
162     iothread = (IOThread *)object_dynamic_cast(object, TYPE_IOTHREAD);
163     if (!iothread) {
164         return 0;
165     }
166 
167     info = g_new0(IOThreadInfo, 1);
168     info->id = iothread_get_id(iothread);
169     info->thread_id = iothread->thread_id;
170 
171     elem = g_new0(IOThreadInfoList, 1);
172     elem->value = info;
173     elem->next = NULL;
174 
175     **prev = elem;
176     *prev = &elem->next;
177     return 0;
178 }
179 
180 IOThreadInfoList *qmp_query_iothreads(Error **errp)
181 {
182     IOThreadInfoList *head = NULL;
183     IOThreadInfoList **prev = &head;
184     Object *container = object_get_objects_root();
185 
186     object_child_foreach(container, query_one_iothread, &prev);
187     return head;
188 }
189 
190 void iothread_stop_all(void)
191 {
192     Object *container = object_get_objects_root();
193 
194     object_child_foreach(container, iothread_stop, NULL);
195 }
196