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 }