scheduler.pmc

/*
Copyright (C) 2001-2008, The Perl Foundation.
$Id$

=head1 NAME

src/pmc/scheduler.pmc - The concurrency scheduler

=head1 DESCRIPTION

Implements the core concurrency scheduler.

=head2 Vtable Functions

=over 4

=cut

*/

#include "parrot/parrot.h"
#include "parrot/scheduler_private.h"

pmclass Scheduler need_ext {
    ATTR INTVAL        id;         /* The scheduler's ID. */
    ATTR INTVAL        max_tid;    /* The highest assigned task ID. */
    ATTR INTVAL        pending;    /* A count of pending tasks (cached for fast
                                     lookup). */
    ATTR PMC          *task_list;  /* The current list of tasks. */
    ATTR PMC          *task_index; /* An index into the current list of tasks,
                                     ordered by priority. */
    ATTR PMC          *wait_index; /* An unordered index of inactive tasks. */
    ATTR PMC          *handlers;   /* The list of currently active handlers. */
    ATTR PMC          *messages;   /* A message queue used for communication
                                     between schedulers. */
    ATTR Parrot_mutex  msg_lock;   /* Lock to synchronize the message queue. */
    ATTR Parrot_Interp interp;     /* A link to the scheduler's interpreter. */

/*

=item C<void init()>

Initialize a concurrency scheduler object.

=cut

*/

    VTABLE void init() {
        Parrot_Scheduler * const core_struct =
            mem_allocate_zeroed_typed(Parrot_Scheduler);

        /* Set flags for custom DOD mark and destroy. */
        PObj_custom_mark_SET(SELF);
        PObj_active_destroy_SET(SELF);

        /* Set up the core struct. */
        PMC_data(SELF)           = core_struct;
        core_struct->id          = 0;
        core_struct->max_tid     = 0;
        core_struct->task_list   = pmc_new(interp, enum_class_Hash);
        core_struct->task_index  = pmc_new(interp, enum_class_ResizableIntegerArray);
        core_struct->wait_index  = pmc_new(interp, enum_class_ResizablePMCArray);
        core_struct->handlers    = pmc_new(interp, enum_class_ResizablePMCArray);
        core_struct->messages    = pmc_new(interp, enum_class_ResizablePMCArray);
        core_struct->interp      = INTERP;
        MUTEX_INIT(core_struct->msg_lock);
    }

/*

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

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

=over 4

=item C<id>

An C<Integer> representing the unique identifier for this scheduler.

=back

*/

    VTABLE void init_pmc(PMC *data) {
        PMC              *elem;
        Parrot_Scheduler *core_struct;

        if (! VTABLE_isa(INTERP, data, CONST_STRING(INTERP, "Hash")))
            real_exception(INTERP, NULL, INVALID_OPERATION,
                "Scheduler initializer must be a Hash");

        SELF.init();
        core_struct = PARROT_SCHEDULER(SELF);

        elem = VTABLE_get_pmc_keyed_str(INTERP, data, CONST_STRING(INTERP, "id"));

        if (! PMC_IS_NULL(elem))
            core_struct->id      = VTABLE_get_integer(INTERP, elem);
    }

/*

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

Insert a task into the task list, giving it a task ID one higher than the
current maximum, and a birthtime of the current time.

=cut

*/

    void push_pmc(PMC *task) {
        Parrot_Scheduler * const core_struct = PARROT_SCHEDULER(SELF);
        STRING *task_id_str;
        INTVAL new_tid;

        task = VTABLE_share_ro(INTERP, task);
        VTABLE_set_number_native(INTERP, task, Parrot_floatval_time());

        new_tid = ++(core_struct->max_tid);
        VTABLE_set_integer_native(INTERP, task, new_tid);
        task_id_str = string_from_int(INTERP, new_tid);

        VTABLE_set_pmc_keyed_str(INTERP, core_struct->task_list, task_id_str, task);


        if (task->vtable->base_type == enum_class_Timer) {
            VTABLE_push_integer(INTERP, core_struct->wait_index, new_tid);
        }
        else {
            VTABLE_push_integer(INTERP, core_struct->task_index, new_tid);
        }


        SCHEDULER_cache_valid_CLEAR(SELF);
        Parrot_cx_runloop_wake(core_struct->interp, SELF);
    }

/*

=item C<PMC *pop_pmc()>

Retrieve the next task from the task list. If the task index is invalid,
recalculate it before retrieving the next task.

=cut

*/

    VTABLE PMC *pop_pmc() {
        Parrot_Scheduler * core_struct = PARROT_SCHEDULER(SELF);
        PMC *task = PMCNULL;

        /* Pull the next valid task off the task list, skipping expired and
         * deleted tasks. */
        while (PMC_IS_NULL(task) && VTABLE_elements(INTERP, core_struct->task_index) > 0) {
            INTVAL tid = VTABLE_shift_integer(INTERP, core_struct->task_index);
            if (tid > 0)
                task = VTABLE_get_pmc_keyed_int(INTERP, core_struct->task_list, tid);
        }

        return task;
    }

/*

=item C<INTVAL get_integer()>

Retrieve the number of pending tasks in the scheduler's task list.

=cut

*/

    VTABLE INTVAL get_integer() {
        Parrot_Scheduler * core_struct = PARROT_SCHEDULER(SELF);
        return VTABLE_elements(INTERP, core_struct->task_index);
    }


/*

=item C<void delete_keyed_int(INTVAL key)>

Removes the task with the given task ID from the task list.

=cut

*/

    VTABLE void delete_keyed_int(INTVAL key) {
        Parrot_Scheduler * core_struct = PARROT_SCHEDULER(SELF);
        STRING *task_id_str = string_from_int(INTERP, key);
        VTABLE_delete_keyed_str(INTERP, core_struct->task_list, task_id_str);
        SCHEDULER_cache_valid_CLEAR(SELF);
    }

/*

=item C<PMC *share_ro()>

Set this PMC as shared.

=cut

*/

    VTABLE PMC *share_ro() {
        PMC *shared_self;
        Parrot_Scheduler *shared_struct;

        if (PObj_is_PMC_shared_TEST(SELF))
            return SELF;

        shared_self = pt_shared_fixup(INTERP, SELF);
        shared_struct = PARROT_SCHEDULER(shared_self);

        shared_struct->task_list  = pt_shared_fixup(INTERP, shared_struct->task_list);
        shared_struct->task_index = pt_shared_fixup(INTERP, shared_struct->task_index);
        shared_struct->wait_index = pt_shared_fixup(INTERP, shared_struct->wait_index);
        shared_struct->handlers   = pt_shared_fixup(INTERP, shared_struct->handlers);
        shared_struct->messages   = pt_shared_fixup(INTERP, shared_struct->messages);

        return shared_self;
    }

/*

=item C<void destroy()>

Free the scheduler's underlying struct.

=cut

*/
    VTABLE void destroy() {
        Parrot_Scheduler * const core_struct = PARROT_SCHEDULER(SELF);
        MUTEX_DESTROY(core_struct->msg_lock);
        mem_sys_free(core_struct);
    }

/*

=item C<void mark()>

Mark any referenced strings and PMCs.

=cut

*/
    VTABLE void mark() {
        if (PARROT_SCHEDULER(SELF)) {
            Parrot_Scheduler * const core_struct = PARROT_SCHEDULER(SELF);

            if (core_struct->task_list)
                pobject_lives(interp, (PObj*)core_struct->task_list);
            if (core_struct->task_index)
                pobject_lives(interp, (PObj*)core_struct->task_index);
            if (core_struct->wait_index)
                pobject_lives(interp, (PObj*)core_struct->wait_index);
            if (core_struct->handlers)
                pobject_lives(interp, (PObj*)core_struct->handlers);
            if (core_struct->messages)
                pobject_lives(interp, (PObj*)core_struct->messages);
        }
    }

/*

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

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

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

=cut

*/

    VTABLE void visit(visit_info *info) {
        Parrot_Scheduler * const core_struct = PARROT_SCHEDULER(SELF);
        PMC **pos;

        /* 1) visit task list */
        pos            = &core_struct->task_list;
        info->thaw_ptr = pos;
        (info->visit_pmc_now)(INTERP, *pos, info);

        /* 2) visit the handlers */
        pos            = &core_struct->handlers;
        info->thaw_ptr = pos;
        (info->visit_pmc_now)(INTERP, *pos, info);
    }

/*

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

Used to archive the scheduler.

=cut

*/

    VTABLE void freeze(visit_info *info) {
        IMAGE_IO *io = info->image_io;
        Parrot_Scheduler * const core_struct = PARROT_SCHEDULER(SELF);

        /* 1) freeze scheduler id */
        VTABLE_push_integer(INTERP, io, core_struct->id);

        /* 2) freeze maximum task id */
        VTABLE_push_integer(INTERP, io, core_struct->max_tid);
    }

/*

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

Used to unarchive the scheduler.

=cut

*/

    VTABLE void thaw(visit_info *info) {
        IMAGE_IO * const io = info->image_io;

        /* 1. thaw scheduler id */
        const INTVAL id = VTABLE_shift_integer(INTERP, io);

        /* 2. thaw maximum task id */
        const INTVAL max_tid = VTABLE_shift_integer(INTERP, io);

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

        /* Set the scheduler's id to the frozen id */
        PARROT_SCHEDULER(SELF)->id = id;

        /* Set the scheduler's maximum task id to the frozen tid */
        PARROT_SCHEDULER(SELF)->max_tid = max_tid;
    }

/*

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

Called after the scheduler has been thawed.

=cut

*/

    void thawfinish(visit_info *info) {
        Parrot_cx_refresh_task_list(INTERP, SELF);
    }

/*

=back

=head2 Methods

=over 4

=cut

*/

/*

=item C<METHOD add_handler(PMC *handler)>

Add a handler to the scheduler.

=cut

*/

    METHOD add_handler(PMC *handler) {
        Parrot_Scheduler * core_struct = PARROT_SCHEDULER(SELF);
        VTABLE_push_pmc(INTERP, core_struct->handlers, handler);
    }

/*

=item C<METHOD find_handler(PMC *task)>

Search for a handler for the given task. If no handler is found, return
PMCNULL.

=cut

*/

    METHOD find_handler(PMC *task) {
        Parrot_Scheduler * core_struct = PARROT_SCHEDULER(SELF);
        INTVAL elements = VTABLE_elements(INTERP, core_struct->handlers);
        INTVAL counter;

        for (counter = 0; counter < elements; ++counter) {
            PMC *handler = VTABLE_get_pmc_keyed_int(interp,
                    core_struct->handlers, counter);
            INTVAL valid_handler = 0;
            if (!PMC_IS_NULL(handler)) {
                (INTVAL valid_handler) = PCCINVOKE(interp, handler, "can_handle", PMC *task);
                if (valid_handler) {
                    RETURN(PMC *handler);
                }
            }
        }

        RETURN(PMC *PMCNULL);
    }


}

/*

=back

=head1 SEE ALSO

F<docs/pdds/pdd25_concurrency.pod>.

=cut

*/

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