monolithium/kernel/src/pipe.c

/*
 * pipe.c
 *
 * Copyright (C) 2015 Aleksandar Andrejevic <theflash@sdf.lonestar.org>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program 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 Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <pipe.h>
#include <exception.h>
#include <timer.h>
#include <process.h>
#include <thread.h>
#include <syscalls.h>
#include <heap.h>

static dword_t create_pipe_internal(pipe_t **_pipe, dword_t flags)
{
    pipe_t *pipe = (pipe_t*)malloc(sizeof(pipe_t));
    if (pipe == NULL) return ERR_NOMEMORY;

    init_object(&pipe->header, NULL, OBJECT_PIPE);
    init_pipe(pipe, flags);

    dword_t ret = create_object(&pipe->header);
    if (ret != ERR_SUCCESS)
    {
        free(pipe);
        return ret;
    }

    *_pipe = pipe;
    return ERR_SUCCESS;
}

dword_t read_pipe(pipe_t *pipe, void *buffer, size_t *size, dword_t timeout)
{
    dword_t ret = ERR_SUCCESS;
    size_t requested_size = *size;
    lock_acquire(&pipe->lock);

    if (pipe->flags & PIPE_MESSAGE)
    {
        if (pipe->fifo.next == &pipe->fifo)
        {
            lock_release(&pipe->lock);
            wait_condition_t condition = { .type = WAIT_UNTIL_NOT_EQUAL, .pointer = (dword_t*)&pipe->fifo.next, .value = 0 };
            wait_result_t status = scheduler_wait(&condition, timeout);
            if (status != ERR_SUCCESS) return (status == WAIT_TIMED_OUT) ? ERR_TIMEOUT : ERR_CANCELED;
            lock_acquire(&pipe->lock);
        }

        pipe_message_entry_t *entry = CONTAINER_OF(pipe->fifo.next, pipe_message_entry_t, link);
        *size = entry->size;

        if (requested_size < entry->size)
        {
            ret = ERR_SMALLBUF;
            goto cleanup;
        }

        memcpy(buffer, entry->data, entry->size);

        list_remove(&entry->link);
        heap_free(&evictable_heap, entry);
    }
    else
    {
        byte_t *data = (byte_t*)buffer;
        *size = 0;

        while (*size < requested_size)
        {
            if (pipe->fifo.next == &pipe->fifo)
            {
                lock_release(&pipe->lock);
                wait_condition_t condition = { .type = WAIT_UNTIL_NOT_EQUAL, .pointer = (dword_t*)&pipe->fifo.next, .value = 0 };
                wait_result_t status = scheduler_wait(&condition, timeout);
                if (status != ERR_SUCCESS) return (status == WAIT_TIMED_OUT) ? ERR_TIMEOUT : ERR_CANCELED;
                lock_acquire(&pipe->lock);
            }

            pipe_fifo_entry_t *entry = CONTAINER_OF(pipe->fifo.next, pipe_fifo_entry_t, link);

            while ((*size < requested_size) && ((entry->start != entry->end) || entry->full))
            {
                data[(*size)++] = entry->data[entry->start++];
                entry->start %= PIPE_BLOCK_SIZE;
                entry->full = FALSE;
            }

            if ((entry->start == entry->end) && !entry->full)
            {
                list_remove(&entry->link);
                heap_free(&evictable_heap, entry);
            }
        }
    }

    pipe->bytes_ready -= *size;

cleanup:
    lock_release(&pipe->lock);
    return ret;
}

dword_t write_pipe(pipe_t *pipe, const void *buffer, size_t size)
{
    dword_t ret = ERR_SUCCESS;
    dword_t count = 0;

    lock_acquire(&pipe->lock);

    if (pipe->flags & PIPE_MESSAGE)
    {
        pipe_message_entry_t *entry = heap_alloc(&evictable_heap, sizeof(pipe_message_entry_t) + size);
        if (!entry)
        {
            ret = ERR_NOMEMORY;
            goto cleanup;
        }

        entry->size = size;
        memcpy(entry->data, buffer, size);
        list_append(&pipe->fifo, &entry->link);
        count = size;
    }
    else
    {
        const byte_t *data = (const byte_t*)buffer;
        pipe_fifo_entry_t *entry = (pipe->fifo.prev != &pipe->fifo) ? (pipe_fifo_entry_t*)pipe->fifo.prev : NULL;

        while (count < size)
        {
            if (!entry || entry->full)
            {
                entry = (pipe_fifo_entry_t*)heap_alloc(&evictable_heap, sizeof(pipe_fifo_entry_t));
                if (!entry)
                {
                    ret = ERR_NOMEMORY;
                    break;
                }

                entry->start = entry->end = 0;
                entry->full = FALSE;
                list_append(&pipe->fifo, &entry->link);
            }

            while ((count < size) && (entry->start != entry->end))
            {
                entry->data[entry->end++] = data[count++];
                entry->end %= PIPE_BLOCK_SIZE;
            }

            if (entry->start == entry->end) entry->full = TRUE;
        }
    }


    pipe->bytes_ready += count;

cleanup:
    lock_release(&pipe->lock);
    return ret;
}

void pipe_cleanup(object_t *obj)
{
    pipe_t *pipe = (pipe_t*)obj;
    while (pipe->fifo.next != &pipe->fifo)
    {
        pipe_fifo_entry_t *entry = CONTAINER_OF(pipe->fifo.next, pipe_fifo_entry_t, link);
        list_remove(&entry->link);
        free(entry);
    }
}

dword_t pipe_pre_wait(object_t *obj, void *parameter, wait_condition_t *condition)
{
    pipe_t *pipe = (pipe_t*)obj;
    dword_t min_bytes = parameter ? (dword_t)parameter : 1;
    condition->type = WAIT_UNTIL_NOT_LESS;
    condition->pointer = &pipe->bytes_ready;
    condition->value = min_bytes;
    return ERR_SUCCESS;
}

sysret_t syscall_create_pipeline(const char *name, dword_t flags, access_flags_t access, handle_t *handle)
{
    dword_t ret;
    handle_t safe_handle;
    const char *safe_name = NULL;

    if (get_previous_mode() == USER_MODE)
    {
        dword_t name_length = 0;

        EH_TRY name_length = strlen(name);
        EH_CATCH EH_ESCAPE(return ERR_BADPTR);
        EH_DONE;

        if (!check_usermode(name, name_length + 1)) return ERR_BADPTR;
        if (!check_usermode(handle, sizeof(handle_t))) return ERR_BADPTR;

        safe_name = copy_user_string(name);
        if (safe_name == NULL) return ERR_BADPTR;
    }
    else
    {
        safe_name = (const char*)name;
    }

    pipeline_t *pipeline = (pipeline_t*)malloc(sizeof(pipeline_t));
    if (pipeline == NULL)
    {
        ret = ERR_NOMEMORY;
        goto cleanup;
    }

    init_object(&pipeline->header, safe_name, OBJECT_PIPELINE);
    pipeline->flags = flags;
    lock_init(&pipeline->lock);
    pipeline->status = PIPELINE_IDLE;

    ret = create_object(&pipeline->header);
    if (ret != ERR_SUCCESS)
    {
        free(pipeline->header.name);
        free(pipeline);
        goto cleanup;
    }

    ret = open_object(&pipeline->header, access, &safe_handle);
    if (ret != ERR_SUCCESS) goto cleanup;

    EH_TRY
    {
        *handle = safe_handle;
    }
    EH_CATCH
    {
        syscall_close_object(safe_handle);
        ret = ERR_BADPTR;
    }
    EH_DONE;

cleanup:
    dereference(&pipeline->header);
    if (get_previous_mode() == USER_MODE) free((void*)safe_name);
    return ret;
}

sysret_t syscall_open_pipeline(const char *name, access_flags_t access, handle_t *handle)
{
    dword_t ret;
    handle_t safe_handle;
    const char *safe_name = NULL;

    if (get_previous_mode() == USER_MODE)
    {
        dword_t name_length = 0;

        EH_TRY name_length = strlen(name);
        EH_CATCH EH_ESCAPE(return ERR_BADPTR);
        EH_DONE;

        if (!check_usermode(name, name_length + 1)) return ERR_BADPTR;
        if (!check_usermode(handle, sizeof(handle_t))) return ERR_BADPTR;

        safe_name = copy_user_string(name);
        if (safe_name == NULL) return ERR_BADPTR;
    }
    else
    {
        safe_name = (const char*)name;
    }

    ret = open_object_by_name(safe_name, OBJECT_PIPELINE, access, &safe_handle);
    if (ret != ERR_SUCCESS) goto cleanup;

    EH_TRY
    {
        *handle = safe_handle;
    }
    EH_CATCH
    {
        ret = ERR_BADPTR;
    }
    EH_DONE;

cleanup:
    if (get_previous_mode() == USER_MODE) free((void*)safe_name);
    return ret;
}

static dword_t pipeline_create_pipes(pipeline_t *pipeline)
{
    dword_t request_flags = (pipeline->flags & PIPELINE_REQUEST_MESSAGE) ? PIPE_MESSAGE : 0;
    dword_t response_flags = (pipeline->flags & PIPELINE_RESPONSE_MESSAGE) ? PIPE_MESSAGE : 0;
    dword_t ret = create_pipe_internal(&pipeline->request_pipe, request_flags);
    if (ret != ERR_SUCCESS) return ret;

    if (!(pipeline->flags & PIPELINE_REQUEST_ONLY))
    {
        ret = create_pipe_internal(&pipeline->response_pipe, response_flags);
        if (ret != ERR_SUCCESS)
        {
            dereference(&pipeline->request_pipe->header);
            pipeline->request_pipe = NULL;
        }
    }

    return ret;
}

static dword_t connect_to_pipeline(pipeline_t *pipeline, pipe_connection_t *conn)
{
    lock_acquire_smart(&pipeline->lock);

    dword_t ret = open_object(&pipeline->request_pipe->header, 0, &conn->request_pipe);
    if (ret != ERR_SUCCESS) goto cleanup;

    if (pipeline->response_pipe)
    {
        ret = open_object(&pipeline->response_pipe->header, 0, &conn->response_pipe);
        if (ret != ERR_SUCCESS) syscall_close_object(conn->response_pipe);
    }

cleanup:
    lock_release(&pipeline->lock);
    return ret;
}

sysret_t syscall_listen_pipeline(handle_t handle, dword_t timeout, pipe_connection_t *connection)
{
    dword_t ret;
    process_t *proc;

    if (get_previous_mode() == USER_MODE)
    {
        proc = get_current_process();
    }
    else
    {
        proc = kernel_process;
    }

    pipeline_t *pipeline;
    if (!reference_by_handle(handle, OBJECT_PIPELINE, (object_t**)&pipeline)) return ERR_INVALID;

    lock_acquire_smart(&pipeline->lock);

    if (pipeline->status != PIPELINE_IDLE)
    {
        lock_release(&pipeline->lock);
        dereference(&pipeline->header);
        return ERR_BUSY;
    }

    pipeline->master_pid = proc->pid;
    pipeline->slave_pid  = 0;
    pipeline->request_pipe = pipeline->response_pipe = NULL;

    pipeline->status = PIPELINE_ACCEPTING;
    lock_release(&pipeline->lock);

    wait_condition_t condition = { .type = WAIT_UNTIL_NOT_EQUAL, .pointer = &pipeline->status, .value = PIPELINE_ACCEPTING };
    wait_result_t result = scheduler_wait(&condition, timeout);
    lock_acquire_smart(&pipeline->lock);

    if (pipeline->status != PIPELINE_CONNECTING)
    {
        switch (result)
        {
        case WAIT_CONDITION_HIT:
            ret = pipeline->last_error;
            break;

        case WAIT_TIMED_OUT:
            ret = ERR_TIMEOUT;
            break;

        case WAIT_CANCELED:
            ret = ERR_CANCELED;
        }

        goto cleanup;
    }

    pipe_connection_t conn;
    conn.master_pid = pipeline->master_pid;
    conn.slave_pid  = pipeline->slave_pid;
    conn.request_pipe = conn.response_pipe = INVALID_HANDLE;

    ret = connect_to_pipeline(pipeline, &conn);
    if (ret == ERR_SUCCESS)
    {
        EH_TRY
        {
            *connection = conn;
        }
        EH_CATCH
        {
            ret = ERR_BADPTR;
        }
        EH_DONE;
    }

cleanup:
    pipeline->status = PIPELINE_IDLE;

    if (pipeline->request_pipe) dereference(&pipeline->request_pipe->header);
    if (pipeline->response_pipe) dereference(&pipeline->response_pipe->header);
    pipeline->request_pipe = pipeline->response_pipe = NULL;

    lock_release(&pipeline->lock);
    dereference(&pipeline->header);
    return ret;
}

sysret_t syscall_connect_pipeline(handle_t handle, access_flags_t access, pipe_connection_t *connection)
{
    dword_t ret;
    process_t *proc;

    if (get_previous_mode() == USER_MODE)
    {
        proc = get_current_process();
        if (!check_usermode(connection, sizeof(pipe_connection_t))) return ERR_BADPTR;
    }
    else
    {
        proc = kernel_process;
    }

    pipeline_t *pipeline = NULL;
    if (!reference_by_handle(handle, OBJECT_PIPELINE, (object_t**)&pipeline))
    {
        ret = ERR_NOTFOUND;
        goto cleanup;
    }

    lock_acquire_smart(&pipeline->lock);

    if (pipeline->status != PIPELINE_ACCEPTING)
    {
        ret = ERR_BUSY;
        goto cleanup;
    }

    pipeline->slave_pid = proc->pid;

    ret = pipeline_create_pipes(pipeline);
    if (ret != ERR_SUCCESS)
    {
        pipeline->last_error = ret;
        goto cleanup;
    }

    pipe_connection_t conn;
    conn.master_pid = pipeline->master_pid;
    conn.slave_pid  = pipeline->slave_pid;
    conn.request_pipe  = INVALID_HANDLE;
    conn.response_pipe = INVALID_HANDLE;

    ret = connect_to_pipeline(pipeline, &conn);
    if (ret == ERR_SUCCESS)
    {
        pipeline->status = PIPELINE_CONNECTING;

        EH_TRY
        {
            *connection = conn;
        }
        EH_CATCH
        {
            ret = ERR_BADPTR;
        }
        EH_DONE;
    }
    else
    {
        pipeline->last_error = ret;
        pipeline->status = PIPELINE_FAILED;
    }

cleanup:
    lock_release(&pipeline->lock);
    if (pipeline) dereference(&pipeline->header);
    return ret;
}

sysret_t syscall_read_pipe(handle_t handle, void *buffer, size_t *size, dword_t timeout)
{
    pipe_t *pipe;
    void *safe_buffer = NULL;
    size_t safe_size = 0;

    if (get_previous_mode() == USER_MODE)
    {
        if (!check_usermode(size, sizeof(size_t))) return ERR_BADPTR;

        EH_TRY safe_size = *size;
        EH_CATCH EH_ESCAPE(return ERR_BADPTR);
        EH_DONE;

        if (!check_usermode(buffer, safe_size)) return ERR_BADPTR;
        safe_buffer = malloc(safe_size);
        if (safe_buffer == NULL) return ERR_NOMEMORY;
    }
    else
    {
        safe_buffer = buffer;
        safe_size = *size;
    }

    if (!reference_by_handle(handle, OBJECT_PIPE, (object_t**)&pipe)) return ERR_INVALID;
    dword_t ret = read_pipe(pipe, (byte_t*)safe_buffer, &safe_size, timeout);
    dereference(&pipe->header);

    if (get_previous_mode() == USER_MODE)
    {
        EH_TRY
        {
            memcpy(buffer, safe_buffer, safe_size);
            *size = safe_size;
        }
        EH_CATCH
        {
            ret = ERR_BADPTR;
        }
        EH_DONE;

        free(safe_buffer);
    }
    else
    {
        *size = safe_size;
    }

    return ret;
}

sysret_t syscall_write_pipe(handle_t handle, void *buffer, size_t size)
{
    pipe_t *pipe;
    byte_t *safe_buffer = NULL;

    if (get_previous_mode() == USER_MODE)
    {
        if (!check_usermode(buffer, size)) return ERR_BADPTR;

        safe_buffer = (byte_t*)malloc(size);
        if (safe_buffer == NULL) return ERR_NOMEMORY;

        EH_TRY
        {
            memcpy(safe_buffer, buffer, size);
        }
        EH_CATCH
        {
            free(safe_buffer);
            EH_ESCAPE(return ERR_BADPTR);
        }
        EH_DONE;
    }
    else safe_buffer = (byte_t*)buffer;

    if (!reference_by_handle(handle, OBJECT_PIPE, (object_t**)&pipe)) return ERR_INVALID;
    dword_t ret = write_pipe(pipe, safe_buffer, size);
    dereference(&pipe->header);

    if (get_previous_mode() == USER_MODE)
    {
        EH_TRY memcpy(buffer, safe_buffer, size);
        EH_CATCH ret = ERR_BADPTR;
        EH_DONE;

        free(safe_buffer);
    }

    return ret;
}