qemu

task.c (5799B)

---

 /*
  * QEMU I/O task
  *
  * Copyright (c) 2015 Red Hat, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
  * License as published by the Free Software Foundation; either
  * version 2.1 of the License, or (at your option) any later version.
  *
  * This library is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  * Lesser General Public License for more details.
  *
  * You should have received a copy of the GNU Lesser General Public
  * License along with this library; if not, see <http://www.gnu.org/licenses/>.
  *
  */
 
 #include "qemu/osdep.h"
 #include "io/task.h"
 #include "qapi/error.h"
 #include "qemu/thread.h"
 #include "qom/object.h"
 #include "trace.h"
 
 struct QIOTaskThreadData {
     QIOTaskWorker worker;
     gpointer opaque;
     GDestroyNotify destroy;
     GMainContext *context;
     GSource *completion;
 };
 
 
 struct QIOTask {
     Object *source;
     QIOTaskFunc func;
     gpointer opaque;
     GDestroyNotify destroy;
     Error *err;
     gpointer result;
     GDestroyNotify destroyResult;
     QemuMutex thread_lock;
     QemuCond thread_cond;
     struct QIOTaskThreadData *thread;
 };
 
 
 QIOTask *qio_task_new(Object *source,
                       QIOTaskFunc func,
                       gpointer opaque,
                       GDestroyNotify destroy)
 {
     QIOTask *task;
 
     task = g_new0(QIOTask, 1);
 
     task->source = source;
     object_ref(source);
     task->func = func;
     task->opaque = opaque;
     task->destroy = destroy;
     qemu_mutex_init(&task->thread_lock);
     qemu_cond_init(&task->thread_cond);
 
     trace_qio_task_new(task, source, func, opaque);
 
     return task;
 }
 
 static void qio_task_free(QIOTask *task)
 {
     qemu_mutex_lock(&task->thread_lock);
     if (task->thread) {
         if (task->thread->destroy) {
             task->thread->destroy(task->thread->opaque);
         }
 
         if (task->thread->context) {
             g_main_context_unref(task->thread->context);
         }
 
         g_free(task->thread);
     }
 
     if (task->destroy) {
         task->destroy(task->opaque);
     }
     if (task->destroyResult) {
         task->destroyResult(task->result);
     }
     if (task->err) {
         error_free(task->err);
     }
     object_unref(task->source);
 
     qemu_mutex_unlock(&task->thread_lock);
     qemu_mutex_destroy(&task->thread_lock);
     qemu_cond_destroy(&task->thread_cond);
 
     g_free(task);
 }
 
 
 static gboolean qio_task_thread_result(gpointer opaque)
 {
     QIOTask *task = opaque;
 
     trace_qio_task_thread_result(task);
     qio_task_complete(task);
 
     return FALSE;
 }
 
 
 static gpointer qio_task_thread_worker(gpointer opaque)
 {
     QIOTask *task = opaque;
 
     trace_qio_task_thread_run(task);
 
     task->thread->worker(task, task->thread->opaque);
 
     /* We're running in the background thread, and must only
      * ever report the task results in the main event loop
      * thread. So we schedule an idle callback to report
      * the worker results
      */
     trace_qio_task_thread_exit(task);
 
     qemu_mutex_lock(&task->thread_lock);
 
     task->thread->completion = g_idle_source_new();
     g_source_set_callback(task->thread->completion,
                           qio_task_thread_result, task, NULL);
     g_source_attach(task->thread->completion,
                     task->thread->context);
     g_source_unref(task->thread->completion);
     trace_qio_task_thread_source_attach(task, task->thread->completion);
 
     qemu_cond_signal(&task->thread_cond);
     qemu_mutex_unlock(&task->thread_lock);
 
     return NULL;
 }
 
 
 void qio_task_run_in_thread(QIOTask *task,
                             QIOTaskWorker worker,
                             gpointer opaque,
                             GDestroyNotify destroy,
                             GMainContext *context)
 {
     struct QIOTaskThreadData *data = g_new0(struct QIOTaskThreadData, 1);
     QemuThread thread;
 
     if (context) {
         g_main_context_ref(context);
     }
 
     data->worker = worker;
     data->opaque = opaque;
     data->destroy = destroy;
     data->context = context;
 
     task->thread = data;
 
     trace_qio_task_thread_start(task, worker, opaque);
     qemu_thread_create(&thread,
                        "io-task-worker",
                        qio_task_thread_worker,
                        task,
                        QEMU_THREAD_DETACHED);
 }
 
 
 void qio_task_wait_thread(QIOTask *task)
 {
     qemu_mutex_lock(&task->thread_lock);
     g_assert(task->thread != NULL);
     while (task->thread->completion == NULL) {
         qemu_cond_wait(&task->thread_cond, &task->thread_lock);
     }
 
     trace_qio_task_thread_source_cancel(task, task->thread->completion);
     g_source_destroy(task->thread->completion);
     qemu_mutex_unlock(&task->thread_lock);
 
     qio_task_thread_result(task);
 }
 
 
 void qio_task_complete(QIOTask *task)
 {
     task->func(task, task->opaque);
     trace_qio_task_complete(task);
     qio_task_free(task);
 }
 
 
 void qio_task_set_error(QIOTask *task,
                         Error *err)
 {
     error_propagate(&task->err, err);
 }
 
 
 bool qio_task_propagate_error(QIOTask *task,
                               Error **errp)
 {
     if (task->err) {
         error_propagate(errp, task->err);
         task->err = NULL;
         return true;
     }
 
     return false;
 }
 
 
 void qio_task_set_result_pointer(QIOTask *task,
                                  gpointer result,
                                  GDestroyNotify destroy)
 {
     task->result = result;
     task->destroyResult = destroy;
 }
 
 
 gpointer qio_task_get_result_pointer(QIOTask *task)
 {
     return task->result;
 }
 
 
 Object *qio_task_get_source(QIOTask *task)
 {
     return task->source;
 }