quorum: Avoid bdrv_aio_writev() for rewrites
[qemu.git] / io / task.c
1 /*
2 * QEMU I/O task
3 *
4 * Copyright (c) 2015 Red Hat, Inc.
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, see <http://www.gnu.org/licenses/>.
18 *
19 */
20
21 #include "qemu/osdep.h"
22 #include "io/task.h"
23 #include "qapi/error.h"
24 #include "qemu/thread.h"
25 #include "trace.h"
26
27 struct QIOTask {
28 Object *source;
29 QIOTaskFunc func;
30 gpointer opaque;
31 GDestroyNotify destroy;
32 };
33
34
35 QIOTask *qio_task_new(Object *source,
36 QIOTaskFunc func,
37 gpointer opaque,
38 GDestroyNotify destroy)
39 {
40 QIOTask *task;
41
42 task = g_new0(QIOTask, 1);
43
44 task->source = source;
45 object_ref(source);
46 task->func = func;
47 task->opaque = opaque;
48 task->destroy = destroy;
49
50 trace_qio_task_new(task, source, func, opaque);
51
52 return task;
53 }
54
55 static void qio_task_free(QIOTask *task)
56 {
57 if (task->destroy) {
58 task->destroy(task->opaque);
59 }
60 object_unref(task->source);
61
62 g_free(task);
63 }
64
65
66 struct QIOTaskThreadData {
67 QIOTask *task;
68 QIOTaskWorker worker;
69 gpointer opaque;
70 GDestroyNotify destroy;
71 Error *err;
72 int ret;
73 };
74
75
76 static gboolean gio_task_thread_result(gpointer opaque)
77 {
78 struct QIOTaskThreadData *data = opaque;
79
80 trace_qio_task_thread_result(data->task);
81 if (data->ret == 0) {
82 qio_task_complete(data->task);
83 } else {
84 qio_task_abort(data->task, data->err);
85 }
86
87 error_free(data->err);
88 if (data->destroy) {
89 data->destroy(data->opaque);
90 }
91
92 g_free(data);
93
94 return FALSE;
95 }
96
97
98 static gpointer qio_task_thread_worker(gpointer opaque)
99 {
100 struct QIOTaskThreadData *data = opaque;
101
102 trace_qio_task_thread_run(data->task);
103 data->ret = data->worker(data->task, &data->err, data->opaque);
104 if (data->ret < 0 && data->err == NULL) {
105 error_setg(&data->err, "Task worker failed but did not set an error");
106 }
107
108 /* We're running in the background thread, and must only
109 * ever report the task results in the main event loop
110 * thread. So we schedule an idle callback to report
111 * the worker results
112 */
113 trace_qio_task_thread_exit(data->task);
114 g_idle_add(gio_task_thread_result, data);
115 return NULL;
116 }
117
118
119 void qio_task_run_in_thread(QIOTask *task,
120 QIOTaskWorker worker,
121 gpointer opaque,
122 GDestroyNotify destroy)
123 {
124 struct QIOTaskThreadData *data = g_new0(struct QIOTaskThreadData, 1);
125 QemuThread thread;
126
127 data->task = task;
128 data->worker = worker;
129 data->opaque = opaque;
130 data->destroy = destroy;
131
132 trace_qio_task_thread_start(task, worker, opaque);
133 qemu_thread_create(&thread,
134 "io-task-worker",
135 qio_task_thread_worker,
136 data,
137 QEMU_THREAD_DETACHED);
138 }
139
140
141 void qio_task_complete(QIOTask *task)
142 {
143 task->func(task->source, NULL, task->opaque);
144 trace_qio_task_complete(task);
145 qio_task_free(task);
146 }
147
148 void qio_task_abort(QIOTask *task,
149 Error *err)
150 {
151 task->func(task->source, err, task->opaque);
152 trace_qio_task_abort(task);
153 qio_task_free(task);
154 }
155
156
157 Object *qio_task_get_source(QIOTask *task)
158 {
159 object_ref(task->source);
160 return task->source;
161 }