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