io/task.c

/*
 * 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;
}