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     Error *err;
  33     gpointer result;
  34     GDestroyNotify destroyResult;
  35 };
  36
  37
  38 QIOTask *qio_task_new(Object *source,
  39                       QIOTaskFunc func,
  40                       gpointer opaque,
  41                       GDestroyNotify destroy)
  42 {
  43     QIOTask *task;
  44
  45     task = g_new0(QIOTask, 1);
  46
  47     task->source = source;
  48     object_ref(source);
  49     task->func = func;
  50     task->opaque = opaque;
  51     task->destroy = destroy;
  52
  53     trace_qio_task_new(task, source, func, opaque);
  54
  55     return task;
  56 }
  57
  58 static void qio_task_free(QIOTask *task)
  59 {
  60     if (task->destroy) {
  61         task->destroy(task->opaque);
  62     }
  63     if (task->destroyResult) {
  64         task->destroyResult(task->result);
  65     }
  66     if (task->err) {
  67         error_free(task->err);
  68     }
  69     object_unref(task->source);
  70
  71     g_free(task);
  72 }
  73
  74
  75 struct QIOTaskThreadData {
  76     QIOTask *task;
  77     QIOTaskWorker worker;
  78     gpointer opaque;
  79     GDestroyNotify destroy;
  80     Error *err;
  81     int ret;
  82 };
  83
  84
  85 static gboolean gio_task_thread_result(gpointer opaque)
  86 {
  87     struct QIOTaskThreadData *data = opaque;
  88
  89     trace_qio_task_thread_result(data->task);
  90     if (data->ret == 0) {
  91         qio_task_complete(data->task);
  92     } else {
  93         qio_task_abort(data->task, data->err);
  94     }
  95
  96     error_free(data->err);
  97     if (data->destroy) {
  98         data->destroy(data->opaque);
  99     }
 100
 101     g_free(data);
 102
 103     return FALSE;
 104 }
 105
 106
 107 static gpointer qio_task_thread_worker(gpointer opaque)
 108 {
 109     struct QIOTaskThreadData *data = opaque;
 110
 111     trace_qio_task_thread_run(data->task);
 112     data->ret = data->worker(data->task, &data->err, data->opaque);
 113     if (data->ret < 0 && data->err == NULL) {
 114         error_setg(&data->err, "Task worker failed but did not set an error");
 115     }
 116
 117     /* We're running in the background thread, and must only
 118      * ever report the task results in the main event loop
 119      * thread. So we schedule an idle callback to report
 120      * the worker results
 121      */
 122     trace_qio_task_thread_exit(data->task);
 123     g_idle_add(gio_task_thread_result, data);
 124     return NULL;
 125 }
 126
 127
 128 void qio_task_run_in_thread(QIOTask *task,
 129                             QIOTaskWorker worker,
 130                             gpointer opaque,
 131                             GDestroyNotify destroy)
 132 {
 133     struct QIOTaskThreadData *data = g_new0(struct QIOTaskThreadData, 1);
 134     QemuThread thread;
 135
 136     data->task = task;
 137     data->worker = worker;
 138     data->opaque = opaque;
 139     data->destroy = destroy;
 140
 141     trace_qio_task_thread_start(task, worker, opaque);
 142     qemu_thread_create(&thread,
 143                        "io-task-worker",
 144                        qio_task_thread_worker,
 145                        data,
 146                        QEMU_THREAD_DETACHED);
 147 }
 148
 149
 150 void qio_task_complete(QIOTask *task)
 151 {
 152     task->func(task->source, NULL, task->opaque);
 153     trace_qio_task_complete(task);
 154     qio_task_free(task);
 155 }
 156
 157 void qio_task_abort(QIOTask *task,
 158                     Error *err)
 159 {
 160     task->func(task->source, err, task->opaque);
 161     trace_qio_task_abort(task);
 162     qio_task_free(task);
 163 }
 164
 165
 166 void qio_task_set_error(QIOTask *task,
 167                         Error *err)
 168 {
 169     error_propagate(&task->err, err);
 170 }
 171
 172
 173 bool qio_task_propagate_error(QIOTask *task,
 174                               Error **errp)
 175 {
 176     if (task->err) {
 177         error_propagate(errp, task->err);
 178         return true;
 179     }
 180
 181     return false;
 182 }
 183
 184
 185 void qio_task_set_result_pointer(QIOTask *task,
 186                                  gpointer result,
 187                                  GDestroyNotify destroy)
 188 {
 189     task->result = result;
 190     task->destroyResult = destroy;
 191 }
 192
 193
 194 gpointer qio_task_get_result_pointer(QIOTask *task)
 195 {
 196     return task->result;
 197 }
 198
 199
 200 Object *qio_task_get_source(QIOTask *task)
 201 {
 202     return task->source;
 203 }