elem *cilkred_map::insert_no_rehash(__cilkrts_worker           *w,
				    void                       *key,
				    __cilkrts_hyperobject_base *hb,
                                    void                       *val)
{

#if REDPAR_DEBUG >= 2
    fprintf(stderr, "[W=%d, desc=insert_no_rehash, this_map=%p]\n",
	    w->self, this);
    verify_current_wkr(w);
#endif
    
    CILK_ASSERT((w == 0 && g == 0) || w->g == g);
    CILK_ASSERT(key != 0);
    CILK_ASSERT(val != 0);
	    
    elem *el = grow(w, &(buckets[hashfun(this, key)]));

#if REDPAR_DEBUG >= 3
    fprintf(stderr, "[W=%d, this=%p, inserting key=%p, val=%p, el = %p]\n",
	    w->self, this, key, val, el);
#endif

    el->key = key;
    el->hb  = hb;
    el->val = val;
    ++nelem;

    return el;
}

struct os_mutex *__cilkrts_os_mutex_create(void)
{
    int status;
    struct os_mutex *mutex = (struct os_mutex *)malloc(sizeof(struct os_mutex));
    pthread_mutexattr_t attr;

    ITT_SYNC_CREATE(mutex, "OS Mutex");

    if (!mutex) {
        if (static_mutex_used) {
            __cilkrts_bug("Cilk RTS library initialization failed");
        } else {
            static_mutex_used = 1;
            mutex = &static_mutex;
        }
    }

    status = pthread_mutexattr_init(&attr);
    CILK_ASSERT (status == 0);
#if defined DEBUG || CILK_LIB_DEBUG 
#ifdef PTHREAD_MUTEX_ERRORCHECK
    status = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
#else
    status = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK_NP);
#endif
    CILK_ASSERT (status == 0);
#endif
    status = pthread_mutex_init (&mutex->mutex, &attr);
    CILK_ASSERT (status == 0);
    pthread_mutexattr_destroy(&attr);

    return mutex;
}

static Closure *deque_xtract_bottom(CilkWorkerState *const ws, int pn)
{
     Closure *cl;

     deque_assert_ownership(ws, pn);

     cl = USE_PARAMETER(deques)[pn].bottom;
     if (cl) {
	  CILK_ASSERT(ws, cl->owner_ready_deque == pn);
	  USE_PARAMETER(deques)[pn].bottom = cl->prev_ready;
	  if (cl == USE_PARAMETER(deques)[pn].top) {
	       CILK_ASSERT(ws, cl->prev_ready == (Closure *) NULL);
	       USE_PARAMETER(deques)[pn].top = (Closure *) NULL;
	  } else {
	       CILK_ASSERT(ws, cl->prev_ready);
	       (cl->prev_ready)->next_ready = (Closure *) NULL;
	  }

	  WHEN_CILK_DEBUG(cl->owner_ready_deque = NOBODY);
     } else {
	  CILK_ASSERT(ws, USE_PARAMETER(deques)[pn].top == (Closure *) NULL);
     }
     
     return cl;
}

void cilk_fiber::reset_state(cilk_fiber_proc start_proc)
{
    // Setup the fiber and return.
    this->m_start_proc = start_proc;
    
    CILK_ASSERT(!this->is_resumable());
    CILK_ASSERT(NULL == this->m_pending_remove_ref);
    CILK_ASSERT(NULL == this->m_pending_pool);
}

/**
 * Record data for a successful steal.
 *
 * The pedigree for a STEAL record is the pedigree of the stolen frame.
 *
 * @note It's assumed that replay_record_steal() has already checked that we're
 * recording a log and that the record/replay functionality has not been
 * compiled out.
 *
 * @param w The worker stealing a frame.
 * @param victim_id The ID of the worker which had it's frame stolen.
 */
void replay_record_steal_internal(__cilkrts_worker *w, int32_t victim_id)
{
    // Follow the pedigree chain using worker's stack frame
    CILK_ASSERT(w->l->next_frame_ff);
    CILK_ASSERT(w->l->next_frame_ff->call_stack);

    // Record steal: STEAL pedigree victim_id thief_id
    write_to_replay_log (w, PED_TYPE_STR_STEAL,
                         &(w->l->next_frame_ff->call_stack->parent_pedigree),
                         victim_id);
}

CILK_EXPORT
void __cilkrts_hyper_create(__cilkrts_hyperobject_base *hb)
{
    // This function registers the specified hyperobject in the current
    // reducer map and registers the initial value of the hyperobject as the
    // leftmost view of the reducer.
    __cilkrts_worker *w = __cilkrts_get_tls_worker();
    if (! w) {
        // If there is no worker, then there is nothing to do: The iniitial
        // value will automatically be used as the left-most view when we
        // enter Cilk.
        return;
    }

    // Disable Cilkscreen for the duration of this call.  The destructor for
    // this class will re-enable Cilkscreen when the method returns.  This
    // will prevent Cilkscreen from reporting apparent races in reducers
    DisableCilkscreen x;

    void* key = get_hyperobject_key(hb);
    void* view = get_leftmost_view(key);
    cilkred_map *h = w->reducer_map;

    if (__builtin_expect(!h, 0)) {
	h = install_new_reducer_map(w);
#if REDPAR_DEBUG >= 2
	fprintf(stderr, "[W=%d, hb=%p, hyper_create, isntalled new map %p, view=%p]\n",
		w->self, hb, h, view);
#endif
    }

    /* Must not exist. */
    CILK_ASSERT(h->lookup(key) == NULL);

#if REDPAR_DEBUG >= 3
    verify_current_wkr(w);
    fprintf(stderr, "[W=%d, hb=%p, lookup in map %p of view %p, should be null]\n",
	    w->self, hb, h, view);
    fprintf(stderr, "W=%d, h=%p, inserting key %p, view%p\n",
	    w->self,
	    h,
	    &(hb->__c_monoid),
	    view);
#endif    

    if (h->merging)
        __cilkrts_bug("User error: hyperobject used by another hyperobject");

    CILK_ASSERT(w->reducer_map == h);
    // The address of the leftmost value is the same as the key for lookup.
    (void) h->rehash_and_insert(w, view, hb, view);
}

void cilkred_map::make_buckets(__cilkrts_worker *w, 
                               size_t            new_nbuckets)
{     
    nbuckets = new_nbuckets;

    CILK_ASSERT(is_power_of_2(nbuckets));
#if defined __GNUC__ && defined __ICC 
    /* bug workaround -- suppress calls to _intel_fast_memset */
    bucket *volatile*new_buckets = (bucket *volatile*)
#else
    bucket **new_buckets = (bucket **)
#endif
        __cilkrts_frame_malloc(w, nbuckets * sizeof(*(buckets)));

#if REDPAR_DEBUG >= 1
    fprintf(stderr, "W=%d, desc=make_buckets, new_buckets=%p, new_nbuckets=%zd\n",
	    w->self, new_buckets, new_nbuckets);
#endif

    for (size_t i = 0; i < new_nbuckets; ++i)
        new_buckets[i] = 0;
#if defined __GNUC__ && defined __ICC 
    buckets = (bucket **)new_buckets;
#else
    buckets = new_buckets;
#endif
    nelem = 0;
}

void __cilkrts_save_exception_state(__cilkrts_worker *w, full_frame *ff)
{
    save_exception_info(w, __cxa_get_globals(), 0, false, "undo-detach");
    CILK_ASSERT(NULL == ff->pending_exception);
    ff->pending_exception = w->l->pending_exception;
    w->l->pending_exception = NULL;    
}

void cilk_fiber::suspend_self_and_resume_other(cilk_fiber* other)
{
#if FIBER_DEBUG >=1
    fprintf(stderr, "suspend_self_and_resume_other: self =%p, other=%p [owner=%p, resume_sf=%p]\n",
            this, other, other->owner, other->resume_sf);
#endif

    // Decrement my reference count (to suspend)
    // Increment other's count (to resume)
    // Suspended fiber should have a reference count of at least 1.  (It is not in a pool).
    this->dec_ref_count();
    other->inc_ref_count();
    this->assert_ref_count_at_least(1);

    // Pass along my owner.
    other->owner = this->owner;
    this->owner  = NULL;

    // Change this fiber to resumable.
    CILK_ASSERT(!this->is_resumable());
    this->set_resumable(true);

    // Normally, I'd assert other->is_resumable().  But this flag may
    // be false the first time we try to "resume" a fiber.
    cilk_fiber_sysdep* self = this->sysdep();
    self->suspend_self_and_resume_other_sysdep(other->sysdep());

    // HAVE RESUMED EXECUTION
    // When we come back here, we should have at least two references:
    // one for the fiber being allocated / out of a pool, and one for it being active.
    this->assert_ref_count_at_least(2);
}

void cilk_fiber_pool_destroy(cilk_fiber_pool* pool)
{
    CILK_ASSERT(cilk_fiber_pool_sanity_check(pool, "pool_destroy"));

    // Lock my own pool, if I need to.
    if (pool->lock) {
        spin_mutex_lock(pool->lock);
    }

    // Give any remaining fibers to parent pool.
    if (pool->parent) {
        cilk_fiber_pool_move_fibers_to_parent_pool(pool, 0);
    }

    // Unlock pool.
    if (pool->lock) {
        spin_mutex_unlock(pool->lock);
    }

    // If I have any left in my pool, just free them myself.
    // This method may acquire the pool lock.
    cilk_fiber_pool_free_fibers_from_pool(pool, 0, NULL);

    // Destroy the lock if there is one.
    if (pool->lock) {
        spin_mutex_destroy(pool->lock);
    }
    __cilkrts_free(pool->fibers);
}

void cilk_fiber_pool_set_fiber_limit(cilk_fiber_pool* root_pool,
                                     unsigned max_fibers_to_allocate)
{
    // Should only set limit on root pool, not children.
    CILK_ASSERT(NULL == root_pool->parent);
    root_pool->alloc_max = max_fibers_to_allocate;
}

/*
 * Do any necessary cleanup for the logs - See record-replay.h for full
 * routine header.
 */
void replay_term(global_state_t *g)
{
    // Free memory for the record/replay log file name, if we've got one
    if (g->record_replay_file_name)
        __cilkrts_free(g->record_replay_file_name);

    // Per-worker cleanup
    for(int i = 0; i < g->total_workers; ++i)
    {
        __cilkrts_worker *w = g->workers[i];

        // Close the log files, if we've opened them
        if(w->l->record_replay_fptr)
            fclose(w->l->record_replay_fptr);

        if (w->l->replay_list_root)
        {
            // We should have consumed the entire list
            CILK_ASSERT(ped_type_last == w->l->replay_list_entry->m_type);

            replay_entry_t *entry = w->l->replay_list_root;
            while (ped_type_last != entry->m_type)
            {
                // Free the pedigree memory for each entry
                entry->unload();
                entry++;
            }
            __cilkrts_free(w->l->replay_list_root);
            w->l->replay_list_root = NULL;
            w->l->replay_list_entry = NULL;
        }
    }
}

void __cilkrts_os_mutex_unlock(struct os_mutex *p)
{
    int status;
    ITT_SYNC_RELEASING(p);
    status = pthread_mutex_unlock (&p->mutex);
    CILK_ASSERT(status == 0);
}

__CILKRTS_BEGIN_EXTERN_C

/**
 * Walk the pedigree and generate a string representation with underscores
 * between terms.  Currently does a recursive walk to generate a forward
 * pedigree.
 *
 * @param p The buffer that is to be filled.  Assumed to be PEDIGREE_BUFF_SIZE
 * characters long
 * @param pnode The initial pedigree term to be written.
 *
 * @return A pointer into the pedigree string buffer after a term has been
 * written.
 */
static
char * walk_pedigree_nodes(char *p, const __cilkrts_pedigree *pnode)
{
    CILK_ASSERT(pnode);
    if (pnode->parent)
    {
        p = walk_pedigree_nodes(p, pnode->parent);
        p += cilk_snprintf_s(p, PEDIGREE_BUFF_SIZE, "%s", (char *)"_");
    }
    return p + cilk_snprintf_l(p, PEDIGREE_BUFF_SIZE, "%" PRIu64, pnode->rank);
}

/* Destroy a reducer map.  The map must have been allocated
   from the worker's global context and should have been
   allocated from the same worker. */
void __cilkrts_destroy_reducer_map(__cilkrts_worker *w, cilkred_map *h)
{
    CILK_ASSERT((w == 0 && h->g == 0) || w->g == h->g);
    verify_current_wkr(w);

    /* the reducer map is allowed to contain el->val == NULL here (and
       only here).  We set el->val == NULL only when we know that the
       map will be destroyed immediately afterwards. */
    DBG h->check(/*allow_null_val=*/true);

    bucket *b;
    size_t i;

    for (i = 0; i < h->nbuckets; ++i) {
        b = h->buckets[i];
        if (b) {
            elem *el;
            for (el = b->el; el->key; ++el) {
                if (el->val)
                    el->destroy();
            }
        }
    }

    free_buckets(w, h->buckets, h->nbuckets);

#if REDPAR_DEBUG >= 1
    fprintf(stderr, "W=%d, destroy_red_map, freeing map h=%p, size=%zd\n",
	    w->self, h, sizeof(*h));
#endif
    
    __cilkrts_frame_free(w, h, sizeof(*h));
}

void __cilkrts_mutex_lock(__cilkrts_worker *w, struct mutex *m)
{
    int count;
    const int maxspin = 1000; /* SWAG */

    NOTE_INTERVAL(w, INTERVAL_MUTEX_LOCK);
    if (!TRY_ACQUIRE(m)) {
        START_INTERVAL(w, INTERVAL_MUTEX_LOCK_SPINNING);
        count = 0;
        do {
            do {
                __cilkrts_short_pause();
                if (++count >= maxspin) {
                    STOP_INTERVAL(w, INTERVAL_MUTEX_LOCK_SPINNING);
                    START_INTERVAL(w, INTERVAL_MUTEX_LOCK_YIELDING);
                    /* let the OS reschedule every once in a while */
                    __cilkrts_yield();
                    STOP_INTERVAL(w, INTERVAL_MUTEX_LOCK_YIELDING);
                    START_INTERVAL(w, INTERVAL_MUTEX_LOCK_SPINNING);
                    count = 0;
                }
            } while (m->lock != 0);
        } while (!TRY_ACQUIRE(m));
        STOP_INTERVAL(w, INTERVAL_MUTEX_LOCK_SPINNING);
    }

    CILK_ASSERT(m->owner == 0);
    m->owner = w;
}

CILK_EXPORT void* __CILKRTS_STRAND_PURE(
    __cilkrts_hyper_lookup(__cilkrts_hyperobject_base *hb))
{
    __cilkrts_worker* w = __cilkrts_get_tls_worker_fast();
    void* key = get_leftmost_view(hb);
    if (! w)
        return get_leftmost_view(key);

    // Disable Cilkscreen for the duration of this call.  This will
    // prevent Cilkscreen from reporting apparent races in reducers
    DisableCilkscreen dguard;

    if (__builtin_expect(w->g->force_reduce, 0))
        __cilkrts_promote_own_deque(w);
    cilkred_map* h = w->reducer_map;

    if (__builtin_expect(!h, 0)) {
	h = install_new_reducer_map(w);
    }

    if (h->merging)
        __cilkrts_bug("User error: hyperobject used by another hyperobject");
    elem* el = h->lookup(key);
    if (! el) {
        /* lookup failed; insert a new default element */
        void *rep;

        {
            /* re-enable cilkscreen while calling the constructor */
            EnableCilkscreen eguard;
            if (h->is_leftmost)
            {
                // This special case is called only if the reducer was not
                // registered using __cilkrts_hyper_create, e.g., if this is a
                // C reducer in global scope or if there is no bound worker.
                rep = get_leftmost_view(key);
            }
            else
            {
                rep = hb->__c_monoid.allocate_fn((void*)hb,
						 hb->__view_size);
                // TBD: Handle exception on identity function
                hb->__c_monoid.identity_fn((void*)hb, rep);
            }
        }

#if REDPAR_DEBUG >= 3
	fprintf(stderr, "W=%d, h=%p, inserting key %p, val%p\n",
		w->self,
		h,
		&(hb->__c_monoid),
		rep);
	CILK_ASSERT(w->reducer_map == h);
#endif
        el = h->rehash_and_insert(w, key, hb, rep);
    }

    return el->val;
}

/**
 * Advance to the next log entry from a SYNC record.  Consume the current
 * SYNC record on this worker and advance to the next one.
 *
 * @note It's assumed that replay_advance_from_sync() has already returned if
 * we're not replaying a log, or if record/replay functionality has been
 * compiled out.
 *
 * @param w The worker whose replay log we're advancing.
 */
void replay_advance_from_sync_internal (__cilkrts_worker *w)
{
    // The current replay entry must be a SYNC
    CILK_ASSERT(ped_type_sync == w->l->replay_list_entry->m_type);

    // Advance to the next entry
    w->l->replay_list_entry = w->l->replay_list_entry->next_entry();
}

static void deque_add_bottom(CilkWorkerState *const ws, Closure *cl, int pn)
{
     deque_assert_ownership(ws, pn);
     CILK_ASSERT(ws, cl->owner_ready_deque == NOBODY);

     cl->prev_ready = USE_PARAMETER(deques)[pn].bottom;
     cl->next_ready = (Closure *)NULL;
     USE_PARAMETER(deques)[pn].bottom = cl;
     WHEN_CILK_DEBUG(cl->owner_ready_deque = pn);
     
     if (USE_PARAMETER(deques)[pn].top) {
	  CILK_ASSERT(ws, cl->prev_ready);
	  (cl->prev_ready)->next_ready = cl;
     } else {
	  USE_PARAMETER(deques)[pn].top = cl;
     }
}

void __cilkrts_note_interval(__cilkrts_worker *w, enum interval i)
{
    if (w) {
        statistics *s = w->l->stats;
        CILK_ASSERT(s->start[i] == INVALID_START);
        s->count[i]++;
    }
}