task.pmc

/*
Copyright (C) 2001-2014, Parrot Foundation.

=head1 NAME

src/pmc/task.pmc - Task PMC

=head1 DESCRIPTION

A Task PMC represents a concurrent native thread,
initialized at startup in a threadpool.

=head2 Functions

=over 4

=cut

*/

#include "parrot/scheduler_private.h"
#include "pmc/pmc_scheduler.h"
#include "pmc/pmc_proxy.h"

/* HEADERIZER HFILE: none */
/* HEADERIZER BEGIN: static */
/* HEADERIZER END: static */

pmclass Task provides invokable auto_attrs {
    ATTR UINTVAL       id;        /* Unique identifier for this task */
    ATTR FLOATVAL      birthtime; /* The creation time stamp of the task */
    ATTR Parrot_Interp interp;    /* The interpreter that created the task */
    ATTR PMC          *code;      /* An (optional) code for the task */
    ATTR PMC          *data;      /* Additional data for the task */
    ATTR INTVAL        killed;    /* Dead tasks don't get run */
    ATTR PMC          *mailbox;   /* List of incoming messages */
    ATTR Parrot_mutex mailbox_lock;
    ATTR PMC          *waiters;   /* Tasks waiting on this one */
    ATTR Parrot_mutex waiters_lock;
    ATTR PMC          *shared;    /* List of variables shared with this task */
    ATTR PMC          *partner;   /* Copy of this task on the other side of a GC barrier,
                                     meaning in another thread */

/*

=item C<void init()>

Initialize a concurrency task object.

=cut

*/

    VTABLE void init() {
        Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);
        Parrot_Scheduler_attributes * const sched_data =
                PARROT_SCHEDULER(interp->scheduler);

        /* Set flags for custom GC mark. */
        PObj_custom_mark_SET(SELF);

        /* Set up the core struct. */
        core_struct->birthtime = Parrot_floatval_time();
        core_struct->code      = PMCNULL;
        core_struct->data      = PMCNULL;
        core_struct->interp    = INTERP;
        core_struct->killed    = 0;
        core_struct->mailbox   = PMCNULL; /* Created lazily on demand */
        core_struct->waiters   = PMCNULL; /* Created lazily on demand */
        core_struct->shared    = Parrot_pmc_new(INTERP, enum_class_ResizablePMCArray);
        core_struct->partner   = NULL; /* Set by Parrot_thread_create_local_task */

        MUTEX_INIT(core_struct->mailbox_lock);
        MUTEX_INIT(core_struct->waiters_lock);

        /* Assign a unique ID */
        /* TODO: Fix collisions. */
        core_struct->id = sched_data->next_task_id;
        sched_data->next_task_id += 1;

        /* By default, all flags are clear. */
        TASK_active_CLEAR(SELF);
        TASK_in_preempt_CLEAR(SELF);
        TASK_recv_block_CLEAR(SELF);
    }

/*

=item C<void init_pmc(PMC *data)>

Initializes a new Task with a C<Hash> PMC with any or all of the keys:

=over 4

=item C<birthtime>

The time at which this Task was created.

=item C<code>

An C<invokable> PMC related to this task.

=item C<data>

Some data that will be passed to C<code> when invoked.

=back

=cut

*/

    VTABLE void init_pmc(PMC *data) {
        Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);

        SELF.init();

        if (VTABLE_isa(INTERP, data, CONST_STRING(INTERP, "Sub"))) {
            core_struct->code = data;
        }
        else if (VTABLE_isa(INTERP, data, CONST_STRING(INTERP, "Hash"))) {
            PMC * elem = VTABLE_get_pmc_keyed_str(INTERP, data, CONST_STRING(INTERP, "birthtime"));
            if (! PMC_IS_NULL(elem))
                core_struct->birthtime = VTABLE_get_number(INTERP, elem);

            elem = VTABLE_get_pmc_keyed_str(INTERP, data, CONST_STRING(INTERP, "code"));
            if (! PMC_IS_NULL(elem))
                core_struct->code = elem;

            elem = VTABLE_get_pmc_keyed_str(INTERP, data, CONST_STRING(INTERP, "data"));
            if (! PMC_IS_NULL(elem))
                core_struct->data = elem;
        }
        else {
            Parrot_ex_throw_from_c_noargs(INTERP, EXCEPTION_INVALID_OPERATION,
                "Task initializer must be a Hash or Sub");
        }
    }

/*

=item C<opcode_t *invoke(void *next)>

Invokes whatever is in the Task's associated code.

If the Task's data attribute is not null, pass it to the
code as the first argument.

=cut

*/

    VTABLE opcode_t *invoke(void *next) :no_wb {
        Parrot_Task_attributes * const task = PARROT_TASK(SELF);
        Parrot_Scheduler_attributes * const sdata =
                PARROT_SCHEDULER(interp->scheduler);

        PMC * const active_tasks = sdata->all_tasks;

        PARROT_ASSERT(! PMC_IS_NULL(task->code));

        /* If a task is pre-empted, this will be set again. */
        TASK_in_preempt_CLEAR(SELF);

        if (! task->killed) {
            const INTVAL current_depth =
                Parrot_pcc_get_recursion_depth(interp, CURRENT_CONTEXT(interp));
            /* Add the task to the set of active Tasks */
            PMC * const task_id = Parrot_pmc_new(interp, enum_class_Integer);
            VTABLE_set_integer_native(interp, task_id, task->id);
            VTABLE_set_pmc_keyed(interp, active_tasks, task_id, SELF);
            TASK_active_SET(SELF);

            /* Actually run the task */
            Parrot_ext_call(interp, task->code, "P->", task->data);
            /* Restore recursion_depth since Parrot_Sub_invoke increments recursion_depth
               which would not be decremented anymore if the sub is preempted */
            Parrot_pcc_set_recursion_depth(interp, CURRENT_CONTEXT(interp), current_depth);
        }

        /* Fixed possibly racy read with write in send TASK_recv_block_CLEAR(partner) */
        if (task->killed || !TASK_in_preempt_TEST(SELF)) {
            /* The task is done. */
            /* Remove it from the set of active Tasks */

            INTVAL i, n = 0;
            PMC * const task_id = Parrot_pmc_new(interp, enum_class_Integer);
            VTABLE_set_integer_native(interp, task_id, task->id);
            TASK_active_CLEAR(SELF);
            VTABLE_delete_keyed(interp, active_tasks, task_id);
            task->killed = 1;

            /* schedule any waiters. */
            if (!PMC_IS_NULL(task->waiters))
                n = VTABLE_get_integer(interp, task->waiters);

            for (i = 0; i < n; ++i) {
                PMC * const wtask = VTABLE_get_pmc_keyed_int(interp, task->waiters, i);
                Parrot_cx_schedule_task(interp, wtask);
            }

            if (task->partner) { /* TODO how can we know if the partner's still alive? */
                Parrot_Task_attributes * const partner_task = PARROT_TASK(task->partner);
                LOCK(partner_task->waiters_lock);

                if (!PMC_IS_NULL(partner_task->waiters)) {
                    Parrot_block_GC_mark_locked(partner_task->interp);
                    partner_task->killed = 1;

                    n = VTABLE_get_integer(interp, partner_task->waiters);

                    for (i = 0; i < n; ++i) {
                        PMC * const wtask =
                            VTABLE_get_pmc_keyed_int(interp, partner_task->waiters, i);
                        Parrot_cx_schedule_immediate(partner_task->interp, wtask);
                    }
                    Parrot_unblock_GC_mark_locked(partner_task->interp);
                }
                else
                    partner_task->killed = 1;

                UNLOCK(partner_task->waiters_lock);
            }
        }

        return (opcode_t*) next;
    }

/*

=item C<PMC *clone()>

Create a copy of the task, resetting status, ID, and birthtime.

=cut

*/

    VTABLE PMC *clone() :no_wb {
        /* Create the new task PMC, of the same type of this one (we may
         * have been subclassed). */
        PMC * const copy  = Parrot_pmc_new(INTERP, SELF->vtable->base_type);
        Parrot_Task_attributes * const new_struct = PARROT_TASK(copy);
        const Parrot_Task_attributes * const old_struct = PARROT_TASK(SELF);

        new_struct->code   = VTABLE_clone(INTERP, old_struct->code);
        new_struct->data   = VTABLE_clone(INTERP, old_struct->data);
        new_struct->shared = VTABLE_clone(INTERP, old_struct->shared);

        return copy;
    }

/*

=item C<PMC *get_attr_str(STRING *name)>

Gets the value of an attribute for this task.

=cut

*/

    VTABLE PMC *get_attr_str(STRING *name) :no_wb {
        Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);
        PMC * value = PMCNULL;

        if (Parrot_str_equal(INTERP, name, CONST_STRING(INTERP, "birthtime"))) {
            value = Parrot_pmc_new(INTERP, enum_class_Float);
            VTABLE_set_number_native(INTERP, value, core_struct->birthtime);
        }
        else if (Parrot_str_equal(INTERP, name, CONST_STRING(INTERP, "code"))) {
            value = core_struct->code;
        }
        else if (Parrot_str_equal(INTERP, name, CONST_STRING(INTERP, "data"))) {
            value = core_struct->data;
        }

        return value;
    }

/*

=item C<void set_attr_str(STRING *name, PMC *value)>

Sets the value of an attribute for this task.

=cut

*/

    VTABLE void set_attr_str(STRING *name, PMC *value) {
        Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);

        if (Parrot_str_equal(INTERP, name, CONST_STRING(INTERP, "birthtime"))) {
            core_struct->birthtime = VTABLE_get_number(INTERP, value);
        }
        else if (Parrot_str_equal(INTERP, name, CONST_STRING(INTERP, "code"))) {
            core_struct->code = value;
        }
        else if (STRING_equal(INTERP, name, CONST_STRING(INTERP, "data"))) {
            core_struct->data = value;
        }
    }

/*

=item C<void push_pmc(PMC *value)>

Add value to the list of PMCs shared with this task.

=cut

*/

    VTABLE void push_pmc(PMC *value) :manual_wb {
        VTABLE_push_pmc(interp, PARROT_TASK(SELF)->shared, value);
    }

/*

=item C<PMC *pop_pmc()>

Return a proxy for the last shared PMC.

=cut

*/

    VTABLE PMC *pop_pmc() :manual_wb {
        PMC *ret = VTABLE_pop_pmc(interp, PARROT_TASK(SELF)->shared);
        RETURN(PMC *ret);
    }

/*

=item C<void mark()>

Mark any referenced strings and PMCs.

=cut

*/
    VTABLE void mark() :no_wb {
        Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);
        if (core_struct) {
            Parrot_gc_mark_PMC_alive(INTERP, core_struct->code);
            Parrot_gc_mark_PMC_alive(INTERP, core_struct->data);
            Parrot_gc_mark_PMC_alive(INTERP, core_struct->mailbox);
            Parrot_gc_mark_PMC_alive(INTERP, core_struct->waiters);
            Parrot_gc_mark_PMC_alive(INTERP, core_struct->shared);
            /* don't mark our partner, since it belongs to another GC */
        }
    }

/*

=item C<void visit(PMC *info)>

This is used by freeze/thaw to visit the contents of the task.

C<*info> is the visit info, (see F<include/parrot/pmc_freeze.h>).

=cut

*/

    VTABLE void visit(PMC *info) :no_wb {
        /* 1) visit code block */
        VISIT_PMC_ATTR(INTERP, info, SELF, Task, code);
        VISIT_PMC_ATTR(INTERP, info, SELF, Task, data);
        VISIT_PMC_ATTR(INTERP, info, SELF, Task, mailbox);
        VISIT_PMC_ATTR(INTERP, info, SELF, Task, waiters);
    }

/*

=item C<void freeze(PMC *info)>

Used to archive the task.

=cut

*/

    VTABLE void freeze(PMC *info) :no_wb {
        const Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF);

        VTABLE_push_float(INTERP, info, core_struct->birthtime);
    }

/*

=item C<void thaw(PMC *info)>

Used to unarchive the task.

=cut

*/

    VTABLE void thaw(PMC *info) {
        const FLOATVAL birthtime = VTABLE_shift_float(INTERP, info);

        /* Allocate the task's core data struct and set custom flags. */
        SELF.init();

        /* Set the task's birthtime to the frozen birthtime */
        PARROT_TASK(SELF)->birthtime = birthtime;
    }

/*

=item C<void thawfinish(PMC *info)>

Called after the task has been thawed.

=cut

*/

    VTABLE void thawfinish(PMC *info) :no_wb {
        UNUSED(INTERP)
        UNUSED(SELF)
        UNUSED(info)
        /* Parrot_Task_attributes * const core_struct = PARROT_TASK(SELF); */
    }

/*

=back

=head2 Methods

=over 4

=item METHOD send(PMC *message)

Send a message to this task.

=cut

*/

    METHOD send(PMC *message) {
        Parrot_Task_attributes * const tdata = PARROT_TASK(SELF);
        LOCK(tdata->mailbox_lock);
        if (PMC_IS_NULL(tdata->mailbox)) {
            tdata->mailbox = Parrot_pmc_new(interp, enum_class_PMCList);
            PARROT_GC_WRITE_BARRIER(interp, SELF);
        }
        VTABLE_push_pmc(interp, tdata->mailbox, message);
        UNLOCK(tdata->mailbox_lock);

        if (tdata->partner) {
            PMC *                    const partner = tdata->partner;
            Parrot_Task_attributes * const pdata = PARROT_TASK(partner);
            LOCK(tdata->mailbox_lock);
            Parrot_block_GC_mark_locked(pdata->interp);
            if (TASK_recv_block_TEST(partner)) {
                /* Was: racy write with read in invoke task->killed || in_preempt */
                /* TASK_recv_block_CLEAR(partner); */
                Parrot_cx_schedule_immediate(pdata->interp, partner);
                TASK_recv_block_CLEAR(partner);
            }
            Parrot_unblock_GC_mark_locked(pdata->interp);
            UNLOCK(tdata->mailbox_lock);
        }
        else {
            if (TASK_recv_block_TEST(SELF)) {
                TASK_recv_block_CLEAR(SELF);
                Parrot_cx_schedule_task(interp, SELF);
            }
        }
    }

/*
    METHOD receive() {
    }
*/

/*

=item METHOD code(PMC * code :optional)

Read or write optional code, an C<invokable> PMC related to this task.

=item METHOD data(PMC * data :optional)

Reads or writes optional task-specific data, that will be passed to C<code> when invoked.

=cut

*/

    METHOD code(PMC * code :optional, INTVAL has_code :opt_flag) :manual_wb {
        Parrot_Task_attributes * const tdata = PARROT_TASK(SELF);
        if (has_code) {
            tdata->code = code;
            PARROT_GC_WRITE_BARRIER(INTERP, SELF);
        }
        code = tdata->code;
        RETURN(PMC *code);
    }

    METHOD data(PMC * data :optional, INTVAL has_data :opt_flag) :manual_wb {
        Parrot_Task_attributes * const tdata = PARROT_TASK(SELF);
        if (has_data) {
            tdata->data = data;
            PARROT_GC_WRITE_BARRIER(INTERP, SELF);
        }
        data = tdata->data;
        RETURN(PMC *data);
    }

/*

=item METHOD kill()

Kill this task.

=cut

*/

    METHOD kill() :no_wb {
        Parrot_Task_attributes * const tdata = PARROT_TASK(SELF);
        tdata->killed = 1;
    }
}

/*

=back

=head1 SEE ALSO

F<docs/pdds/pdd25_concurrency.pod>.

=cut

*/

/*
 * Local variables:
 *   c-file-style: "parrot"
 * End:
 * vim: expandtab shiftwidth=4 cinoptions='\:2=2' :
 */