tagged release 0.6.4

[parrot.git] / src / thread.c

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

=head1 NAME

src/thread.c - Thread handling stuff

=head1 DESCRIPTION

Threads are created by creating new C<ParrotInterpreter> objects.

=head2 Functions

=over 4

=cut

*/

#include "parrot/parrot.h"
#include "parrot/atomic.h"

/* HEADERIZER HFILE: include/parrot/thread.h */

/* HEADERIZER BEGIN: static */
/* Don't modify between HEADERIZER BEGIN / HEADERIZER END.  Your changes will be lost. */

static Parrot_Interp detach(UINTVAL tid);
PARROT_CAN_RETURN_NULL
static Shared_gc_info * get_pool(PARROT_INTERP)
        __attribute__nonnull__(1);

PARROT_WARN_UNUSED_RESULT
static int is_suspended_for_gc(PARROT_INTERP)
        __attribute__nonnull__(1);

PARROT_CAN_RETURN_NULL
static PMC * make_local_args_copy(PARROT_INTERP,
    ARGIN(Parrot_Interp old_interp),
    ARGIN_NULLOK(PMC *args))
        __attribute__nonnull__(1)
        __attribute__nonnull__(2);

PARROT_CAN_RETURN_NULL
static PMC * make_local_copy(PARROT_INTERP,
    ARGIN(Parrot_Interp from),
    ARGIN(PMC *arg))
        __attribute__nonnull__(1)
        __attribute__nonnull__(2)
        __attribute__nonnull__(3);

static void mutex_unlock(ARGMOD(void *arg))
        __attribute__nonnull__(1)
        FUNC_MODIFIES(*arg);

static Parrot_Interp pt_check_tid(UINTVAL tid, ARGIN(const char *from))
        __attribute__nonnull__(2);

static int pt_gc_count_threads(PARROT_INTERP)
        __attribute__nonnull__(1);

static void pt_gc_wait_for_stage(PARROT_INTERP,
    thread_gc_stage_enum from_stage,
    thread_gc_stage_enum to_stage)
        __attribute__nonnull__(1);

static void pt_gc_wakeup_check(PARROT_INTERP)
        __attribute__nonnull__(1);

static void pt_ns_clone(
    ARGOUT(Parrot_Interp d),
    ARGOUT(PMC *dest_ns),
    ARGIN(Parrot_Interp s),
    ARGIN(PMC *source_ns))
        __attribute__nonnull__(1)
        __attribute__nonnull__(2)
        __attribute__nonnull__(3)
        __attribute__nonnull__(4)
        FUNC_MODIFIES(d)
        FUNC_MODIFIES(*dest_ns);

static void pt_suspend_all_for_gc(PARROT_INTERP)
        __attribute__nonnull__(1);

static void pt_suspend_one_for_gc(PARROT_INTERP)
        __attribute__nonnull__(1);

static void pt_thread_signal(NOTNULL(Parrot_Interp self), PARROT_INTERP)
        __attribute__nonnull__(1)
        __attribute__nonnull__(2);

static void pt_thread_wait(PARROT_INTERP)
        __attribute__nonnull__(1);

PARROT_CAN_RETURN_NULL
static QUEUE_ENTRY * remove_queued_suspend_gc(PARROT_INTERP)
        __attribute__nonnull__(1);

PARROT_CAN_RETURN_NULL
static void* thread_func(ARGIN_NULLOK(void *arg));

/* Don't modify between HEADERIZER BEGIN / HEADERIZER END.  Your changes will be lost. */
/* HEADERIZER END: static */

#if defined THREAD_DEBUG && THREAD_DEBUG
#  define DEBUG_ONLY(x) (x)
#else
#  define DEBUG_ONLY(x)
#endif

static int running_threads;

void Parrot_really_destroy(PARROT_INTERP, int exit_code, void *arg);

/*

=item C<static PMC * make_local_copy>

Creates a local copy of the PMC if necessary. (No copy is made if it is marked
shared.) This includes workarounds for Parrot_clone() not doing the Right Thing
with subroutines (specifically, code segments aren't preserved and it is
difficult to do so as long as Parrot_clone() depends on freezing).

=cut

*/

PARROT_CAN_RETURN_NULL
static PMC *
make_local_copy(PARROT_INTERP, ARGIN(Parrot_Interp from), ARGIN(PMC *arg))
{
    PMC            *ret_val;
    STRING * const  _sub       = interp->vtables[enum_class_Sub]->whoami;
    STRING * const  _multi_sub = interp->vtables[enum_class_MultiSub]->whoami;

    if (PMC_IS_NULL(arg)) {
        ret_val = PMCNULL;
    }
    else if (PObj_is_PMC_shared_TEST(arg)) {
        ret_val = arg;
    }
    else if (VTABLE_isa(from, arg, _multi_sub)) {
        INTVAL i = 0;
        const INTVAL n = VTABLE_elements(from, arg);
        ret_val  = pmc_new(interp, enum_class_MultiSub);

        for (i = 0; i < n; ++i) {
            PMC *const orig = VTABLE_get_pmc_keyed_int(from, arg, i);
            PMC *const copy = make_local_copy(interp, from, orig);
            VTABLE_push_pmc(interp, ret_val, copy);
        }
    }
    else if (VTABLE_isa(from, arg, _sub)) {
        /* this is a workaround for cloning subroutines not actually
         * working as one might expect mainly because the segment is
         * not correctly copied
         */
        ret_val               = Parrot_clone(interp, arg);
        PMC_sub(ret_val)->seg = PMC_sub(arg)->seg;
        /* Skip vtable overrides and methods. */
        if (PMC_sub(ret_val)->vtable_index == -1
                && !(PMC_sub(ret_val)->comp_flags & SUB_COMP_FLAG_METHOD)) {
            Parrot_store_sub_in_namespace(interp, ret_val);
        }
    }
    else {
        ret_val = Parrot_clone(interp, arg);
    }
    return ret_val;
}

/*

=item C<static Shared_gc_info * get_pool>

Gets the shared gc information.  For now this is global data; ideally it will
become something other than a static variable.  If everything uses this
function, it will be easier to change.

=cut

*/

PARROT_CAN_RETURN_NULL
static Shared_gc_info *
get_pool(PARROT_INTERP)
{
    return shared_gc_info;
}

/*

=item C<void pt_free_pool>

Frees the shared GC information.  This clears any global data when joining all
threads at parent interpreter destruction.

=cut

*/

void
pt_free_pool(PARROT_INTERP)
{
    if (shared_gc_info) {
        COND_DESTROY(shared_gc_info->gc_cond);
        PARROT_ATOMIC_INT_DESTROY(shared_gc_info->gc_block_level);
        mem_sys_free(shared_gc_info);
        shared_gc_info = NULL;
    }
}

/*

=item C<static PMC * make_local_args_copy>

Make a local copy of the corresponding array of arguments.

=cut

*/

PARROT_CAN_RETURN_NULL
static PMC *
make_local_args_copy(PARROT_INTERP, ARGIN(Parrot_Interp old_interp), ARGIN_NULLOK(PMC *args))
{
    PMC   *ret_val;
    INTVAL old_size;
    INTVAL i;

    if (PMC_IS_NULL(args))
        return PMCNULL;

    old_size = VTABLE_get_integer(old_interp, args);

    /* XXX should this be a different type? */
    ret_val = pmc_new(interp, enum_class_FixedPMCArray);
    VTABLE_set_integer_native(interp, ret_val, old_size);

    for (i = 0; i < old_size; ++i) {
        PMC * const copy = make_local_copy(interp, old_interp,
                VTABLE_get_pmc_keyed_int(old_interp, args, i));

        VTABLE_set_pmc_keyed_int(interp, ret_val, i, copy);
    }

    return ret_val;
}

/*

=item C<PMC * pt_shared_fixup>

Modifies a PMC to be sharable. Right now, reassigns the vtable to one
owned by some master interpreter, so the PMC can be safely reused
after thread death.

In the future the PMC returned might be different than the one
passed, e.g., if we need to reallocate the PMC in a different
interpreter.

=cut

*/

PARROT_CAN_RETURN_NULL
PMC *
pt_shared_fixup(PARROT_INTERP, ARGMOD(PMC *pmc))
{
    /* TODO this will need to change for thread pools
     * XXX should we have a separate interpreter for this?
     */
    INTVAL        type_num;
    Parrot_Interp master = interpreter_array[0];
    const int     is_ro  = pmc->vtable->flags & VTABLE_IS_READONLY_FLAG;

    /* This lock is paired with one in objects.c. It is necessary to protect
     * against the master interpreter adding classes and consequently
     * resizing its classname->type_id hashtable and/or expanding its vtable
     * array.
     * TODO investigate if a read-write lock results in substantially
     * better performance.
     */
    LOCK_INTERPRETER(master);
    type_num = pmc->vtable->base_type;

    if (type_num == enum_type_undef) {
        UNLOCK_INTERPRETER(master);
        real_exception(interp, NULL, 1, "pt_shared_fixup: unsharable type");
    }

    pmc->vtable = master->vtables[type_num];

    UNLOCK_INTERPRETER(master);

    if (is_ro)
        pmc->vtable = pmc->vtable->ro_variant_vtable;

    add_pmc_sync(interp, pmc);

    PObj_is_PMC_shared_SET(pmc);

    /* make sure metadata doesn't go away unexpectedly */
    if (PMC_metadata(pmc))
        PMC_metadata(pmc) = pt_shared_fixup(interp, PMC_metadata(pmc));

    return pmc;
}

/*

=item C<static void pt_thread_signal>

Wakes up an C<interp> which should have called pt_thread_wait().

=cut

*/

static void
pt_thread_signal(NOTNULL(Parrot_Interp self), PARROT_INTERP)
{
    COND_SIGNAL(interp->thread_data->interp_cond);
}

/*

=item C<void pt_thread_wait_with>

Waits for this interpreter to be signalled through its condition variable,
dealing properly with GC issues. C<*mutex> is assumed locked on entry and
will be locked on exit from this function. If a GC run occurs in the middle of
this function, then a spurious wakeup may occur.

=cut

*/

void
pt_thread_wait_with(PARROT_INTERP, ARGMOD(Parrot_mutex *mutex))
{
    LOCK(interpreter_array_mutex);
    if (interp->thread_data->state & THREAD_STATE_SUSPEND_GC_REQUESTED) {
        interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;
        UNLOCK(interpreter_array_mutex);
        UNLOCK(*mutex);

        pt_suspend_self_for_gc(interp);

        LOCK(*mutex);
        /* since we unlocked the mutex something bad may have occured */
        return;
    }

    interp->thread_data->state |= THREAD_STATE_GC_WAKEUP;

    UNLOCK(interpreter_array_mutex);
    COND_WAIT(interp->thread_data->interp_cond, *mutex);
    LOCK(interpreter_array_mutex);

    interp->thread_data->state &= ~THREAD_STATE_GC_WAKEUP;

    if (interp->thread_data->state & THREAD_STATE_SUSPENDED_GC) {
        UNLOCK(*mutex);
        /* XXX loop needed? */
        do {
            UNLOCK(interpreter_array_mutex);
            /* fprintf(stderr, "%p: woken up, doing GC run\n", interp); */
            pt_suspend_self_for_gc(interp);
            LOCK(interpreter_array_mutex);
        } while (interp->thread_data->state & THREAD_STATE_SUSPENDED_GC);

        UNLOCK(interpreter_array_mutex);
        LOCK(*mutex);
    }
    else {
        UNLOCK(interpreter_array_mutex);
    }
}

/*

=item C<static void pt_thread_wait>

Waits for a signal, handling GC matters correctly.  C<interpreter_array_mutex>
is assumed held. Spurious wakeups may occur.

=cut

*/

static void
pt_thread_wait(PARROT_INTERP)
{
    if (interp->thread_data->state & THREAD_STATE_SUSPEND_GC_REQUESTED) {
        interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;
        /* fprintf(stderr, "%p: pt_thread_wait, before sleep, doing GC run\n",
         * interp); */

        UNLOCK(interpreter_array_mutex);
        pt_suspend_self_for_gc(interp);
        LOCK(interpreter_array_mutex);

        /* while we were GCing, whatever we were waiting on might have
         * changed */
        return;
    }

    interp->thread_data->state |= THREAD_STATE_GC_WAKEUP;

    COND_WAIT(interp->thread_data->interp_cond, interpreter_array_mutex);

    interp->thread_data->state &= ~THREAD_STATE_GC_WAKEUP;

    while (interp->thread_data->state & THREAD_STATE_SUSPENDED_GC) {
        UNLOCK(interpreter_array_mutex);
        /* fprintf(stderr, "%p: woken up, doing GC run\n", interp); */
        pt_suspend_self_for_gc(interp);
        LOCK(interpreter_array_mutex);
    }
}


/*

=item C<static void* thread_func>

The actual thread function.

=cut

*/

PARROT_CAN_RETURN_NULL
static void*
thread_func(ARGIN_NULLOK(void *arg))
{
    Parrot_exception exp;
    int              lo_var_ptr;
    UINTVAL          tid;
    PMC             *sub;
    PMC             *sub_arg;
    PMC * const      self    = (PMC*) arg;
    PMC             *ret_val = NULL;
    Parrot_Interp    interp  = (Parrot_Interp)PMC_data(self);

    Parrot_block_GC_mark(interp);
    Parrot_block_GC_sweep(interp);

    /* need to set it here because argument passing can trigger GC */
    interp->lo_var_ptr = &lo_var_ptr;
    sub                = (PMC *)PMC_struct_val(self);
    sub_arg            = PMC_pmc_val(self);

    if (setjmp(exp.destination)) {
        const Parrot_exception * const except = interp->exceptions;
        /* caught exception */
        /* XXX what should we really do here */
        PIO_eprintf(interp,
                    "Unhandled exception in thread with tid %d "
                    "(message=%Ss, number=%d)\n",
                    interp->thread_data->tid,
                    except->msg,
                    except->error);

        ret_val = PMCNULL;
    }
    else {
        /* run normally */
        push_new_c_exception_handler(interp, &exp);
        Parrot_unblock_GC_mark(interp);
        Parrot_unblock_GC_sweep(interp);
        ret_val = Parrot_runops_fromc_args(interp, sub, "PF", sub_arg);
    }

    /* thread is finito */
    LOCK(interpreter_array_mutex);
    DEBUG_ONLY(fprintf(stderr, "marking an thread as finished\n"));

    interp->thread_data->state |= THREAD_STATE_FINISHED;
    tid                         = interp->thread_data->tid;

    if (interp != interpreter_array[tid]) {
        UNLOCK(interpreter_array_mutex);
        PANIC(interp, "thread finished: interpreter mismatch");
    }
    if (interp->thread_data->state & THREAD_STATE_DETACHED) {
        interpreter_array[tid] = NULL;
        DEBUG_ONLY(fprintf(stderr,
            "really destroying an interpreter [exit while detached]\n"));
        Parrot_really_destroy(interp, 0, NULL);
    }
    else if (interp->thread_data->state & THREAD_STATE_JOINED) {
        pt_thread_signal(interp, interp->thread_data->joiner);
    }

    /* make sure we don't block a GC run */
    pt_gc_wakeup_check(interp);
    PARROT_ASSERT(interp->thread_data->state & THREAD_STATE_FINISHED);

    UNLOCK(interpreter_array_mutex);

    return ret_val;
}

/*

=back

=head2 Helper functions used also for running plain interpreters

=over 4

=item C<void pt_clone_code>

Copies/clones the packfile/code from interpreter C<s> to C<d>. All
resources are created in C<d>.

=cut

*/

void
pt_clone_code(Parrot_Interp d, Parrot_Interp s)
{
    Parrot_block_GC_mark(d);
    Interp_flags_SET(d, PARROT_EXTERN_CODE_FLAG);
    d->code = NULL;
    Parrot_switch_to_cs(d, s->code, 1);
    Parrot_unblock_GC_mark(d);
}

/*

=item C<static void pt_ns_clone>

Clones all globals from C<s> to C<d>.

=cut

*/

static void
pt_ns_clone(ARGOUT(Parrot_Interp d), ARGOUT(PMC *dest_ns),
            ARGIN(Parrot_Interp s), ARGIN(PMC *source_ns))
{
    PMC * const iter = VTABLE_get_iter(s, source_ns);
    const INTVAL n   = VTABLE_elements(s, source_ns);
    INTVAL i;

    for (i = 0; i < n; ++i) {
        /* XXX what if 'key' is a non-constant-pool string? */
        STRING * const key = VTABLE_shift_string(s, iter);
        PMC    * const val = VTABLE_get_pmc_keyed_str(s, source_ns, key);

        if (val->vtable->base_type == enum_class_NameSpace) {
            PMC *sub_ns = VTABLE_get_pmc_keyed_str(d, dest_ns, key);
            if (PMC_IS_NULL(sub_ns) || sub_ns->vtable->base_type !=
                    enum_class_NameSpace) {
                sub_ns = pmc_new(d, enum_class_NameSpace);
                VTABLE_set_pmc_keyed_str(d, dest_ns, key, sub_ns);
            }
            pt_ns_clone(d, sub_ns, s, val);
        }
        else {
            PMC * const dval = VTABLE_get_pmc_keyed_str(d, dest_ns, key);

            if (PMC_IS_NULL(dval)) {
                PMC * const copy = make_local_copy(d, s, val);

                /* Vtable overrides and methods were already cloned, so don't reclone them. */
                if (!(val->vtable->base_type == enum_class_Sub
                        && (PMC_sub(val)->vtable_index != -1
                        || PMC_sub(val)->comp_flags & SUB_COMP_FLAG_METHOD))) {
                    VTABLE_set_pmc_keyed_str(d, dest_ns, key, copy);
                }
            }
        }
    }
}

/*

=item C<void pt_clone_globals>

Copies the global namespace when cloning a new interpreter.

=cut

*/

void
pt_clone_globals(Parrot_Interp d, Parrot_Interp s)
{
    Parrot_block_GC_mark(d);
    pt_ns_clone(d, d->root_namespace, s, s->root_namespace);
    Parrot_unblock_GC_mark(d);
}

/*

=item C<void pt_thread_prepare_for_run>

Sets up a new thread to run.

=cut

*/

void
pt_thread_prepare_for_run(Parrot_Interp d, Parrot_Interp s)
{
    Parrot_setup_event_func_ptrs(d);
}

/*

=back

=head2 ParrotThread methods

=over 4

=cut

*/

/*

=item C<PMC * pt_transfer_sub>

Clones the sub so that it's suitable for the other interpreter.

=cut

*/

PARROT_CAN_RETURN_NULL
PMC *
pt_transfer_sub(ARGOUT(Parrot_Interp d), ARGIN(Parrot_Interp s), ARGIN(PMC *sub))
{
#if defined THREAD_DEBUG && THREAD_DEBUG
    PIO_eprintf(s, "copying over subroutine [%Ss]\n",
        Parrot_full_sub_name(s, sub));
#endif
    return make_local_copy(d, s, sub);
}

/*

=item C<int pt_thread_run>

Runs the C<*sub> PMC in a separate thread using the interpreter in
C<*dest_interp>.

C<arg> should be an array of arguments for the subroutine.

=cut

*/

int
pt_thread_run(PARROT_INTERP, ARGOUT(PMC *dest_interp), ARGIN(PMC *sub), ARGIN_NULLOK(PMC *arg))
{
    PMC *old_dest_interp;
    PMC *parent;
    Interp * const interpreter = (Parrot_Interp)PMC_data(dest_interp);

    Parrot_block_GC_sweep(interpreter);
    Parrot_block_GC_mark(interpreter);
    Parrot_block_GC_sweep(interp);
    Parrot_block_GC_mark(interp);

    /* make a copy of the ParrotThread PMC so we can use it
     * to hold parameters to the new thread without it being
     * garbage collected or otherwise changed by the parent thread.
     * Also so the new thread's getinterp doesn't return an object
     * owned by the wrong interpreter -- which would be very bad
     * if the parent is destroyed before the child.
     * XXX FIXME move this elsewhere? at least the set_pmc_keyed_int
     */
    old_dest_interp = dest_interp;
    dest_interp     = pmc_new_noinit(interpreter, enum_class_ParrotThread);

    /* so it's not accidentally deleted */
    PMC_data(old_dest_interp) = NULL;
    PMC_data(dest_interp)     = interpreter;

    VTABLE_set_pmc_keyed_int(interpreter, interpreter->iglobals,
        (INTVAL) IGLOBALS_INTERPRETER, dest_interp);

    parent = VTABLE_get_pmc_keyed_int(interp, interp->iglobals,
                IGLOBALS_INTERPRETER);

    /*
     * TODO check if thread flags are consistent
     */
    if (interp->flags & PARROT_THR_COPY_INTERP)
        clone_interpreter(interpreter, (Parrot_Interp)PMC_data(parent), PARROT_CLONE_DEFAULT);
    /*
     * TODO thread pools
     */

    pt_thread_prepare_for_run(interpreter, interp);

    PMC_struct_val(dest_interp) = pt_transfer_sub(interpreter, interp, sub);
    PMC_pmc_val(dest_interp)    = make_local_args_copy(interpreter, interp, arg);

    /*
     * set regs according to pdd03
     */
    interpreter->current_object = dest_interp;
    /*
     * create a joinable thread
     */
    interpreter->thread_data->state = THREAD_STATE_JOINABLE;

    Parrot_unblock_GC_mark(interpreter);
    Parrot_unblock_GC_sweep(interpreter);
    Parrot_unblock_GC_mark(interp);
    Parrot_unblock_GC_sweep(interp);

    THREAD_CREATE_JOINABLE(interpreter->thread_data->thread,
            thread_func, dest_interp);

    /* check for pending GC */
    LOCK(interpreter_array_mutex);
    if (interp->thread_data->state & THREAD_STATE_SUSPEND_GC_REQUESTED)
        pt_suspend_one_for_gc(interpreter);

    UNLOCK(interpreter_array_mutex);
    return 0;
}

/*

=item C<int pt_thread_run_1>

Runs a thread that shares nothing and does not communicate with the other
interpreter.

=cut

*/

int
pt_thread_run_1(PARROT_INTERP, ARGOUT(PMC* dest_interp), ARGIN(PMC* sub), ARGIN(PMC *arg))
{
    interp->flags |= PARROT_THR_TYPE_1;
    return pt_thread_run(interp, dest_interp, sub, arg);
}

/*

=item C<int pt_thread_run_2>

Runs an interpreter in a thread with no shared variables, but which
communicates by sending messages.

=cut

*/

int
pt_thread_run_2(PARROT_INTERP, ARGOUT(PMC* dest_interp), ARGIN(PMC* sub), ARGIN(PMC *arg))
{
    interp->flags |= PARROT_THR_TYPE_2;
    return pt_thread_run(interp, dest_interp, sub, arg);
}

/*

=item C<int pt_thread_run_3>

Runs an interpreter in a thread, allowing shared variables and using a thread
pool.

=cut

*/

int
pt_thread_run_3(PARROT_INTERP, ARGOUT(PMC* dest_interp), ARGIN(PMC* sub), ARGIN(PMC *arg))
{
    interp->flags |= PARROT_THR_TYPE_3;
    return pt_thread_run(interp, dest_interp, sub, arg);
}

/*

=item C<void pt_thread_yield>

Relinquishes hold on the processor.

=cut

*/

void
pt_thread_yield(void)
{
    YIELD;
}

/*

=item C<static Parrot_Interp pt_check_tid>

Helper function. Checks if the given thread ID is valid. The caller holds the
mutex.  Returns the interpreter for C<tid>.

=cut

*/

static Parrot_Interp
pt_check_tid(UINTVAL tid, ARGIN(const char *from))
{
    if (tid >= n_interpreters) {
        UNLOCK(interpreter_array_mutex);
        internal_exception(1, "%s: illegal thread tid %d", from, tid);
    }
    if (tid == 0) {
        UNLOCK(interpreter_array_mutex);
        internal_exception(1, "%s: illegal thread tid %d (main)", from, tid);
    }
    if (!interpreter_array[tid]) {
        UNLOCK(interpreter_array_mutex);
        internal_exception(1, "%s: illegal thread tid %d - empty", from, tid);
    }
    return interpreter_array[tid];
}

/*

=item C<static void mutex_unlock>

Unlocks the mutex C<*arg>.

=cut

*/

static void
mutex_unlock(ARGMOD(void *arg))
{
    UNLOCK(*(Parrot_mutex *) arg);
}

/*

=item C<static int is_suspended_for_gc>

Returns true iff C<interp> is suspended for a global GC run.  Be sure to hold
C<interpreter_array_mutex>.

=cut

*/

PARROT_WARN_UNUSED_RESULT
static int
is_suspended_for_gc(PARROT_INTERP)
{
    if (!interp)
        return 1;
    else if (interp->thread_data->wants_shared_gc)
        return 1;
    else if (interp->thread_data->state & THREAD_STATE_SUSPENDED_GC)
        return 1;
    else if ((interp->thread_data->state & THREAD_STATE_FINISHED) ||
               (interp->thread_data->state & THREAD_STATE_NOT_STARTED))
        return 1;
    else
        return 0;
}

/*

=item C<static QUEUE_ENTRY * remove_queued_suspend_gc>

Removes an event requesting that the interpreter suspend itself for a
garbage-collection run from the event queue.

=cut

*/

PARROT_CAN_RETURN_NULL
static QUEUE_ENTRY *
remove_queued_suspend_gc(PARROT_INTERP)
{
    parrot_event *ev    = NULL;
    QUEUE        *queue = interp->task_queue;
    QUEUE_ENTRY  *prev  = NULL;
    QUEUE_ENTRY  *cur;

    queue_lock(queue);
    cur = queue->head;

    while (cur) {
        ev   = (parrot_event *)cur->data;

        if (ev->type == EVENT_TYPE_SUSPEND_FOR_GC)
            break;

        prev = cur;
        cur  = cur->next;
    }

    if (cur) {
        if (prev)
            prev->next  = cur->next;
        else
            queue->head = cur->next;

        if (cur == queue->tail)
            queue->tail = prev;

        if (cur == queue->head)
            queue->head = cur->next;

        mem_sys_free(ev);
        mem_sys_free(cur);
        cur = NULL;
        DEBUG_ONLY(fprintf(stderr, "%p: remove_queued_suspend_gc: got one\n", interp));
    }

    queue_unlock(queue);
    return cur;
}

/*

=item C<static int pt_gc_count_threads>

Returns the number of active threads in the system (running or suspended).  Be
sure to hold C<interpreter_array_mutex>.

=cut

*/

static int
pt_gc_count_threads(PARROT_INTERP)
{
    UINTVAL i;
    int     count = 0;

    for (i = 0; i < n_interpreters; ++i) {
        Parrot_Interp cur;
        cur = interpreter_array[i];
        if (!cur)
            continue;
        if (cur->thread_data->state & (THREAD_STATE_NOT_STARTED |
                                       THREAD_STATE_FINISHED))
            continue;
        ++count;
    }
    DEBUG_ONLY(fprintf(stderr, "found %d threads\n", count));
    return count;
}

/*

=item C<static void pt_gc_wait_for_stage>

Waits until all threads have reached the desired stage.  Takes an interpreter,
starting stage and ending stage as arguments.  Updates the thread information.
Used in C<pt_DOD_start_mark> and C<pt_DOD_stop_mark>.

=cut

*/

static void
pt_gc_wait_for_stage(PARROT_INTERP, thread_gc_stage_enum from_stage,
            thread_gc_stage_enum to_stage)
{
    Shared_gc_info * const info = shared_gc_info;
    int             thread_count;

    DEBUG_ONLY(fprintf(stderr, "%p: gc_wait_for_stage: %d->%d\n", interp, from_stage, to_stage));

    /* XXX well-timed thread death can mess this up */
    LOCK(interpreter_array_mutex);

    DEBUG_ONLY(fprintf(stderr, "%p: got lock\n", interp));
    thread_count = pt_gc_count_threads(interp);

    PARROT_ASSERT(info->gc_stage == from_stage);
    PARROT_ASSERT(!(interp->thread_data->state & THREAD_STATE_NOT_STARTED));
    PARROT_ASSERT(!(interp->thread_data->state & THREAD_STATE_FINISHED));

    if (from_stage == 0)
        PARROT_ASSERT(interp->thread_data->state & THREAD_STATE_SUSPENDED_GC);
    else
        PARROT_ASSERT(!(interp->thread_data->state & THREAD_STATE_SUSPENDED_GC));

    ++info->num_reached;

    DEBUG_ONLY(fprintf(stderr, "%p: gc_wait_for_stage: got %d\n", interp, info->num_reached));

    if (info->num_reached == thread_count) {
        info->gc_stage    = to_stage;
        info->num_reached = 0;
        COND_BROADCAST(info->gc_cond);
    }
    else {
        do {
            COND_WAIT(info->gc_cond, interpreter_array_mutex);
        } while (info->gc_stage != to_stage);
    }
    UNLOCK(interpreter_array_mutex);
}


/*

=item C<static void pt_gc_wakeup_check>

Checks if it's necessary to wake threads to perform garbage collection.  This
is called after thread death.  Be sure to hold C<interpreter_array_mutex>.

=cut

*/

static void
pt_gc_wakeup_check(PARROT_INTERP)
{
    Shared_gc_info * const info = shared_gc_info;
    int             thread_count;

    /* XXX: maybe a little hack; see RT #49532 */
    if (!info)
        return;

    thread_count = pt_gc_count_threads(interp);

    if (info->num_reached == thread_count) {
        PARROT_ASSERT(info->gc_stage == THREAD_GC_STAGE_NONE);
        info->gc_stage    = THREAD_GC_STAGE_MARK;
        info->num_reached = 0;
        COND_BROADCAST(info->gc_cond);
    }
}

/*

=item C<static void pt_suspend_one_for_gc>

Suspends a single interpreter for GC.  Be sure to hold
C<interpreter_array_mutex>.

=cut

*/

static void
pt_suspend_one_for_gc(PARROT_INTERP)
{
    DEBUG_ONLY(fprintf(stderr, "suspend one: %p\n", interp));
    if (is_suspended_for_gc(interp)) {
        DEBUG_ONLY(fprintf(stderr, "ignoring already suspended\n"));
        return;
    }

    if (interp->thread_data->state & THREAD_STATE_GC_WAKEUP) {
        DEBUG_ONLY(fprintf(stderr, "just waking it up\n"));
        interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;
        COND_SIGNAL(interp->thread_data->interp_cond);
    }
    else {
        DEBUG_ONLY(fprintf(stderr, "queuing event\n"));
        interp->thread_data->state |= THREAD_STATE_SUSPEND_GC_REQUESTED;
        Parrot_cx_request_suspend_for_gc(interp);
    }
}

/*

=item C<static void pt_suspend_all_for_gc>

Notifies all threads to perform a GC run.

=cut

*/

static void
pt_suspend_all_for_gc(PARROT_INTERP)
{
    UINTVAL i;

    DEBUG_ONLY(fprintf(stderr, "suspend_all_for_gc [interp=%p]\n", interp));

    LOCK(interpreter_array_mutex);
    interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;

    if (interp->thread_data->state & THREAD_STATE_SUSPEND_GC_REQUESTED) {
        DEBUG_ONLY(fprintf(stderr, "found while suspending all\n"));
        Parrot_cx_delete_suspend_for_gc(interp);
        interp->thread_data->state &= ~THREAD_STATE_SUSPEND_GC_REQUESTED;
        UNLOCK(interpreter_array_mutex);
        return;
    }

#if 0
    for (i = 0; i < n_interpreters; ++i) {
        Parrot_Interp other_interp;
        other_interp = interpreter_array[i];
        if (!other_interp)
            continue;

        if (is_suspended_for_gc(other_interp) &&
            other_interp != interp &&
            (other_interp->thread_data->state & THREAD_STATE_SUSPENDED_GC))
        {
            PMC *successp;
            /* this means that someone else already got this far,
             * so we have a suspend event in our queue to ignore
             */
            /* XXX still reachable? */
            DEBUG_ONLY(fprintf(stderr, "apparently someone else is doing it [%p]\n", other_interp));
            fprintf(stderr, "??? found later (%p)\n", other_interp);
            successp = Parrot_cx_delete_suspend_for_gc(interp);
            PARROT_ASSERT(successp);
            UNLOCK(interpreter_array_mutex);
            return;
        }
    }
#endif

    /* now send all the non-suspended threads to suspend for GC */
    for (i = 0; i < n_interpreters; ++i) {
        Parrot_Interp other_interp = interpreter_array[i];

        if (interp == other_interp)
            continue;

        if (is_suspended_for_gc(other_interp))
            continue;

        pt_suspend_one_for_gc(other_interp);
    }
    UNLOCK(interpreter_array_mutex);
}

/*

=item C<void pt_suspend_self_for_gc>

Suspends this thread for a full GC run.

XXX FIXME -- if GC is blocked, we need to do a GC run as soon
as it becomes unblocked.

=cut

*/

void
pt_suspend_self_for_gc(PARROT_INTERP)
{
    PARROT_ASSERT(interp);
    PARROT_ASSERT(!interp->arena_base->DOD_block_level);
    DEBUG_ONLY(fprintf(stderr, "%p: suspend_self_for_gc\n", interp));
    /* since we are modifying our own state, we need to lock
     * the interpreter_array_mutex.
     */
    LOCK(interpreter_array_mutex);
    DEBUG_ONLY(fprintf(stderr, "%p: got lock\n", interp));

    PARROT_ASSERT(interp->thread_data->state &
            (THREAD_STATE_SUSPEND_GC_REQUESTED | THREAD_STATE_SUSPENDED_GC));

    if (interp->thread_data->state & THREAD_STATE_SUSPEND_GC_REQUESTED) {
        DEBUG_ONLY(fprintf(stderr, "remove queued request\n"));
        while (!PMC_IS_NULL(Parrot_cx_delete_suspend_for_gc(interp)));
        DEBUG_ONLY(fprintf(stderr, "removed all queued requests\n"));
        interp->thread_data->state &= ~THREAD_STATE_SUSPEND_GC_REQUESTED;
    }
    if (!(interp->thread_data->state & THREAD_STATE_SUSPENDED_GC)) {
        interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;
    }
    else {
        DEBUG_ONLY(fprintf(stderr, "no need to set suspended\n"));
    }
    UNLOCK(interpreter_array_mutex);

    /* mark and sweep our world -- later callbacks will keep
     * it sync'd
     */
    Parrot_dod_ms_run(interp, GC_trace_stack_FLAG);

    PARROT_ASSERT(!(interp->thread_data->state & THREAD_STATE_SUSPENDED_GC));
}

/*

=item C<PMC* pt_thread_join>

Joins (by waiting for) a joinable thread.

=cut

*/

PARROT_CAN_RETURN_NULL
PMC*
pt_thread_join(NOTNULL(Parrot_Interp parent), UINTVAL tid)
{
    int           state;
    Parrot_Interp interp;

    LOCK(interpreter_array_mutex);

    interp = pt_check_tid(tid, "join");

    if (interp == parent)
        do_panic(parent, "Can't join self", __FILE__, __LINE__);

    if ((!(interp->thread_data->state & (THREAD_STATE_DETACHED
            | THREAD_STATE_JOINED)) &&
         !(interp->thread_data->state & THREAD_STATE_NOT_STARTED)) ||
            interp->thread_data->state == THREAD_STATE_FINISHED) {
        void *raw_retval = NULL;
        PMC *retval;

        interp->thread_data->state |= THREAD_STATE_JOINED;

        while (!(interp->thread_data->state & THREAD_STATE_FINISHED)) {
            interp->thread_data->joiner = parent;
            pt_thread_wait(parent);
        }

        UNLOCK(interpreter_array_mutex);
        JOIN(interp->thread_data->thread, raw_retval);

        retval = (PMC *)raw_retval;
        /*
         * we need to push a cleanup handler here: if cloning
         * of the retval fails (e.g. it's a NULLPMC) this lock
         * isn't released until eternity or someone hits ^C
         *
         * TODO This is needed for all places holding a lock for
         *      non-trivial tasks
         *      -leo
         * TODO remove that and replace it with proper exception
         *      handling, so that a failing clone in the parent
         *      just stops that thread
         *      -leo
         */
        LOCK(interpreter_array_mutex);
        CLEANUP_PUSH(mutex_unlock, &interpreter_array_mutex);

        if (retval) {
            PMC *parent_ret;
            /*
             * clone the PMC into caller, if its not a shared PMC
             * the PMC is not in the parents root set nor in the
             * stack so block DOD during clone
             * XXX should probably acquire the parent's interpreter mutex
             */
            Parrot_block_GC_mark(parent);
            parent_ret = make_local_copy(parent, interp, retval);

            /* this PMC is living only in the stack of this currently
             * dying interpreter, so register it in parent's DOD registry
             * XXX is this still needed?
             */
            dod_register_pmc(parent, parent_ret);
            Parrot_unblock_GC_mark(parent);
            retval = parent_ret;
        }
        else {
            retval = PMCNULL;
        }
        interpreter_array[tid] = NULL;
        running_threads--;

        DEBUG_ONLY(fprintf(stderr, "destroying an interpreter [join]\n"));
        if (Interp_debug_TEST(parent, PARROT_THREAD_DEBUG_FLAG))
            fprintf(stderr, "running threads %d\n", running_threads);

        /* reparent it so memory pool merging works */
        interp->parent_interpreter = parent;
        Parrot_really_destroy(interp, 0, NULL);

        CLEANUP_POP(1);
        /*
         * interpreter destruction is done - unregister the return
         * value, caller gets it now
         */
        if (retval)
            dod_unregister_pmc(parent, retval);

        return retval;
    }
    /*
     * when here thread was in wrong state
     */
    state = interp->thread_data->state;
    UNLOCK(interpreter_array_mutex);
    real_exception(interp, NULL, 1, "join: illegal thread state %d tid %d",
            state, tid);
}

/*

=item C<void pt_join_threads>

Possibly waits for other running threads. This is called when destroying
C<interp>.

=cut

*/

void
pt_join_threads(PARROT_INTERP)
{
    size_t          i;
    pt_free_pool(interp);

    /* if no threads were started - fine */
    LOCK(interpreter_array_mutex);
    if (n_interpreters <= 1) {
        n_interpreters = 0;
        UNLOCK(interpreter_array_mutex);
        return;
    }

    /* only the first interpreter waits for other threads */
    if (interp != interpreter_array[0]) {
        UNLOCK(interpreter_array_mutex);
        return;
    }

    for (i = 1; i < n_interpreters; ++i) {
        Parrot_Interp thread_interp = interpreter_array[i];
        if (thread_interp == NULL)
            continue;
        if (thread_interp->thread_data->state == THREAD_STATE_JOINABLE ||
            (thread_interp->thread_data->state & THREAD_STATE_FINISHED)) {

            void *retval = NULL;
            thread_interp->thread_data->state |= THREAD_STATE_JOINED;
            UNLOCK(interpreter_array_mutex);
            JOIN(thread_interp->thread_data->thread, retval);
            LOCK(interpreter_array_mutex);
        }
    }
    UNLOCK(interpreter_array_mutex);
    return;
}

/*

=item C<static Parrot_Interp detach>

Helper for detach and kill.

Returns the interpreter, if it didn't finish yet.

=cut

*/

static Parrot_Interp
detach(UINTVAL tid)
{
    Parrot_Interp interp;

    LOCK(interpreter_array_mutex);
    interp = pt_check_tid(tid, "detach");
    /*
     * if interpreter is joinable, we detach em
     */
    if (interp->thread_data->state == THREAD_STATE_JOINABLE ||
            interp->thread_data->state == THREAD_STATE_FINISHED) {
        DETACH(interp->thread_data->thread);
        interp->thread_data->state |= THREAD_STATE_DETACHED;

        if (interp->thread_data->state & THREAD_STATE_FINISHED) {
            interpreter_array[tid] = NULL;
            DEBUG_ONLY(fprintf(stderr, "destroying an interpreter [detach]\n"));
            Parrot_really_destroy(interp, 0, NULL);
            interp = NULL;
        }
    }
    UNLOCK(interpreter_array_mutex);
    return interp;
}

/*

=item C<void pt_thread_detach>

Detaches the thread, making it non-joinable.

=cut

*/

void
pt_thread_detach(UINTVAL tid)
{
    (void) detach(tid);
}

/*

=item C<void pt_thread_kill>

Kills the thread.

=cut

*/

void
pt_thread_kill(UINTVAL tid)
{
    PARROT_INTERP = detach(tid);

    /* schedule a terminate event for that interpreter */
    if (interp)
        Parrot_cx_runloop_end(interp);
}

/*

=back

=head2 Threaded interpreter book-keeping

=over 4

=item C<void pt_add_to_interpreters>

Stores the given interpreter in the array of all interpreters.  Be sure to hold
C<interpreter_array_mutex>.

=cut

*/

void
pt_add_to_interpreters(PARROT_INTERP, ARGIN_NULLOK(Parrot_Interp new_interp))
{
    size_t i;
    DEBUG_ONLY(fprintf(stderr, "interp = %p\n", interp));

    if (!new_interp) {
        /*
         * Create an entry for the very first interpreter, event
         * handling needs it
         */
        PARROT_ASSERT(!interpreter_array);
        PARROT_ASSERT(n_interpreters == 0);

        interpreter_array    = mem_allocate_typed(Interp *);
        interpreter_array[0] = interp;
        n_interpreters       = 1;

        shared_gc_info = (Shared_gc_info *)mem_sys_allocate_zeroed(sizeof (*shared_gc_info));
        COND_INIT(shared_gc_info->gc_cond);
        PARROT_ATOMIC_INT_INIT(shared_gc_info->gc_block_level);
        PARROT_ATOMIC_INT_SET(shared_gc_info->gc_block_level, 0);

        /* XXX try to defer this until later */
        PARROT_ASSERT(interp == interpreter_array[0]);
        interp->thread_data      = mem_allocate_zeroed_typed(Thread_data);
        INTERPRETER_LOCK_INIT(interp);
        interp->thread_data->tid = 0;

        return;
    }


    new_interp->thread_data = mem_allocate_zeroed_typed(Thread_data);
    INTERPRETER_LOCK_INIT(new_interp);
    running_threads++;
    if (Interp_debug_TEST(interp, PARROT_THREAD_DEBUG_FLAG))
        fprintf(stderr, "running threads %d\n", running_threads);

    /* look for an empty slot */
    for (i = 0; i < n_interpreters; ++i) {
        if (interpreter_array[i] == NULL) {
            interpreter_array[i] = new_interp;
            new_interp->thread_data->tid   = i;
            new_interp->thread_data->state = THREAD_STATE_NOT_STARTED;
            return;
        }
    }

    /* need to resize */
    interpreter_array = (Interp **)mem_sys_realloc(interpreter_array,
            (n_interpreters + 1) * sizeof (Interp *));

    interpreter_array[n_interpreters] = new_interp;
    new_interp->thread_data->tid      = n_interpreters;
    new_interp->thread_data->state    = THREAD_STATE_NOT_STARTED;

    ++n_interpreters;
}

/*

=back

=head2 DOD Synchronization Functions

=over 4

=item C<void pt_DOD_start_mark>

Record that the mark phase of DOD is about to begin. In the presence of shared
PMCs, we can only run one DOD run at a time because C<< PMC->next_for_GC >> may
be changed.

C<flags> are the DOD flags. We check if we need to collect shared objects or
not.

TODO - Have a count of shared PMCs and check it during DOD.

TODO - Evaluate if a interpreter lock is cheaper when C<dod_mark_ptr> is
updated.

=cut

*/

void
pt_DOD_start_mark(PARROT_INTERP)
{
    Shared_gc_info *info;
    int             block_level;

    DEBUG_ONLY(fprintf(stderr, "%p: pt_DOD_start_mark\n", interp));
    /* if no other threads are running, we are safe */
    if (!running_threads)
        return;

    info = get_pool(interp);
    PARROT_ATOMIC_INT_GET(block_level, info->gc_block_level);

    DEBUG_ONLY(fprintf(stderr, "start threaded mark\n"));
    /*
     * TODO now check, if we are the owner of a shared memory pool
     * if yes:
     * - suspend all other threads by sending them a suspend event
     *   (or put a LOCK around updating the mark pointers)
     * - return and continue the mark phase
     * - then s. comments below
     */
    LOCK(interpreter_array_mutex);
    if (interp->thread_data->state & THREAD_STATE_SUSPENDED_GC) {
        PARROT_ASSERT(!(interp->thread_data->state &
                 THREAD_STATE_SUSPEND_GC_REQUESTED));
        DEBUG_ONLY(fprintf(stderr, "already suspended...\n"));
        UNLOCK(interpreter_array_mutex);
    }
    else if (block_level) {
        /* unthreaded collection */
        DEBUG_ONLY(fprintf(stderr, "... but blocked\n"));

        /* holding the lock */
        return;
    }
    else if (interp->thread_data->state &
               THREAD_STATE_SUSPEND_GC_REQUESTED) {
        while (!PMC_IS_NULL(Parrot_cx_delete_suspend_for_gc(interp)));

        interp->thread_data->state &= ~THREAD_STATE_SUSPEND_GC_REQUESTED;
        interp->thread_data->state |= THREAD_STATE_SUSPENDED_GC;

        DEBUG_ONLY(fprintf(stderr, "%p: detected request\n", interp));
        UNLOCK(interpreter_array_mutex);
    }
    else {
        /* we need to stop the world */
        DEBUG_ONLY(fprintf(stderr, "stop the world\n"));
        UNLOCK(interpreter_array_mutex);

        pt_suspend_all_for_gc(interp);
    }

    DEBUG_ONLY(fprintf(stderr, "%p: wait for stage\n", interp));
    pt_gc_wait_for_stage(interp, THREAD_GC_STAGE_NONE, THREAD_GC_STAGE_MARK);

    DEBUG_ONLY(fprintf(stderr, "actually mark\n"));
    /*
     * we can't allow parallel running DODs both would mess with shared PMCs
     * next_for_GC pointers
     */
    LOCK(interpreter_array_mutex);
    DEBUG_ONLY(fprintf(stderr, "got marking lock\n"));
}

/*

=item C<void pt_DOD_mark_root_finished>

Records that DOD has finished for the root set.  UNIMPLEMENTED

=cut

*/

void
pt_DOD_mark_root_finished(PARROT_INTERP)
{
    if (!running_threads)
        return;
    /*
     * TODO now check, if we are the owner of a shared memory pool
     * if yes:
     * - now run DOD_mark on all members of our pool
     * - if all shared PMCs are marked by all threads then
     *   - we can continue to free unused objects
     */
}

/*

=item C<void pt_DOD_stop_mark>

Records that the mark phase of DOD has completed.

=cut

*/

void
pt_DOD_stop_mark(PARROT_INTERP)
{
    if (!running_threads)
        return;
    /*
     * normal operation can continue now
     *   - other threads may or not free unused objects then,
     *     depending on their resource statistics
     */
    if (!(interp->thread_data->state & THREAD_STATE_SUSPENDED_GC)) {
        UNLOCK(interpreter_array_mutex);
        return;
    }

    PARROT_ASSERT(!(interp->thread_data->state &
             THREAD_STATE_SUSPEND_GC_REQUESTED));
    interp->thread_data->state &= ~THREAD_STATE_SUSPENDED_GC;

    while (!PMC_IS_NULL(Parrot_cx_delete_suspend_for_gc(interp))) {
        /* XXX FIXME make this message never trigger */
        fprintf(stderr, "%p: extraneous suspend_gc event\n", (void *)interp);
    }

    DEBUG_ONLY(fprintf(stderr, "%p: unlock\n", interp));
    UNLOCK(interpreter_array_mutex);
    DEBUG_ONLY(fprintf(stderr, "wait to sweep\n"));

    pt_gc_wait_for_stage(interp, THREAD_GC_STAGE_MARK, THREAD_GC_STAGE_SWEEP);
}

/*

=item C<void Parrot_shared_DOD_block>

Blocks stop-the-world DOD runs.

=cut

*/

PARROT_API
void
Parrot_shared_DOD_block(PARROT_INTERP)
{
    Shared_gc_info * const info = get_pool(interp);

    if (info) {
        int level;
        PARROT_ATOMIC_INT_INC(level, info->gc_block_level);
        PARROT_ASSERT(level > 0);
    }
}

/*

=item C<void Parrot_shared_DOD_unblock>

Unblocks stop-the-world DOD runs.

=cut

*/

PARROT_API
void
Parrot_shared_DOD_unblock(PARROT_INTERP)
{
    Shared_gc_info * const info = get_pool(interp);
    if (info) {
        int level;
        PARROT_ATOMIC_INT_DEC(level, info->gc_block_level);
        PARROT_ASSERT(level >= 0);
    }
}

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