<commit_before>#include <ross.h>

/**
 * \brief Reset the event bitfield prior to entering the event handler
 *  post-reverse - reset the bitfield so that a potential re-running of the
 *  event is presented with a consistent bitfield state
 *  NOTE: the size checks are to better support the experimental reverse
 *  computation compiler, which can use a larger bitfield.
 *  Courtesy of <NAME>
 */
static inline void reset_bitfields(tw_event *revent)
{
    if (sizeof(revent->cv) == sizeof(uint32_t)){
        *(uint32_t*)&revent->cv = 0;
    }
    else if (sizeof(revent->cv) == sizeof(uint64_t)){
        *(uint64_t*)&revent->cv = 0;
    }
    else{
        memset(&revent->cv, 0, sizeof(revent->cv));
    }
}

/* 
 * Get all events out of my event queue and spin them out into
 * the priority queue so they can be processed in time stamp
 * order.  
 */
static void tw_sched_event_q(tw_pe * me) {
    tw_clock     start;
    tw_kp       *dest_kp;
    tw_event    *cev;
    tw_event    *nev;

    while (me->event_q.size) {
        cev = tw_eventq_pop_list(&me->event_q);

        for (; cev; cev = nev) {
            nev = cev->next;

            if(!cev->state.owner || cev->state.owner == TW_pe_free_q) {
                tw_error(TW_LOC, "no owner!");
            }
            if (cev->state.cancel_q) {
                cev->state.owner = TW_pe_anti_msg;
                cev->next = cev->prev = NULL;
                continue;
            }

            switch (cev->state.owner) {
                case TW_pe_event_q:
                    dest_kp = cev->dest_lp->kp;

                    if (dest_kp->last_time > cev->recv_ts) {
                        /* cev is a straggler message which has arrived
                        * after we processed events occuring after it.
                        * We need to jump back to before cev's timestamp.
                        */
                        start = tw_clock_read();
                        tw_kp_rollback_to(dest_kp, cev->recv_ts);
                        me->stats.s_rollback += tw_clock_read() - start;
                    }
                    start = tw_clock_read();
                    tw_pq_enqueue(me->pq, cev);
                    me->stats.s_pq += tw_clock_read() - start;
                    break;

                default:
                    tw_error(TW_LOC, "Event in event_q, but owner %d not recognized", cev->state.owner);
            }
        }
    }
}

/*
 * OPT: need to link events into canq in reverse order so 
 *      that when we rollback the 1st event, we should not
 *  need to do any further rollbacks.
 */
static void tw_sched_cancel_q(tw_pe * me) {
    tw_clock     start;
    tw_event    *cev;
    tw_event    *nev;

    start = tw_clock_read();
    while (me->cancel_q) {
        cev = me->cancel_q;
        me->cancel_q = NULL;

        for (; cev; cev = nev) {
            nev = cev->cancel_next;

            if (!cev->state.cancel_q) {
                tw_error(TW_LOC, "No cancel_q bit on event in cancel_q");
            }

            if(!cev->state.owner || cev->state.owner == TW_pe_free_q) {
                tw_error(TW_LOC, "Cancelled event, no owner!");
            }

            switch (cev->state.owner) {
                case TW_pe_event_q:
                    /* This event hasn't been added to our pq yet and we
                    * have not officially received it yet either.  We'll
                    * do the actual free of this event when we receive it
                    * as we spin out the event_q chain.
                    */
                    tw_eventq_delete_any(&me->event_q, cev);

                    tw_event_free(me, cev);
                    break;

                case TW_pe_anti_msg:
                    tw_event_free(me, cev);
                    break;

                case TW_pe_pq:
                    /* Event was not cancelled directly from the event_q
                    * because the cancel message came after we popped it
                    * out of that queue but before we could process it.
                    */
                    tw_pq_delete_any(me->pq, cev);
                    tw_event_free(me, cev);
                    break;

                case TW_kp_pevent_q:
                    /* The event was already processed. 
                    * SECONDARY ROLLBACK
                    */
                    tw_kp_rollback_event(cev);
                    tw_event_free(me, cev);
                    break;

                default:
                    tw_error(TW_LOC, "Event in cancel_q, but owner %d not recognized", cev->state.owner);
            }
        }
    }

    me->stats.s_cancel_q += tw_clock_read() - start;
}

static void tw_sched_batch(tw_pe * me) {
    /* Number of consecutive times we gave up because there were no free event buffers. */
    static int no_free_event_buffers = 0;
    static int warned_no_free_event_buffers = 0;

    tw_clock     start;
    unsigned int     msg_i;

    /* Process g_tw_mblock events, or until the PQ is empty
    * (whichever comes first). 
    */
    for (msg_i = g_tw_mblock; msg_i; msg_i--) {
        tw_event *cev;
        tw_lp *clp;
        tw_kp *ckp;

        /* OUT OF FREE EVENT BUFFERS.  BAD.
        * Go do fossil collect immediately.
        */
        if (me->free_q.size <= g_tw_gvt_threshold) {
            /* Suggested by <NAME> */
            if(++no_free_event_buffers > 10 && !warned_no_free_event_buffers) {
                printf("WARNING: No free event buffers.  Try increasing memory via the --extramem option.\n");
                warned_no_free_event_buffers = 1;
            }
            tw_gvt_force_update(me);
            break;
        }<commit_msg>Closes #22.  I opened a new issue for the other request.<commit_after>#include <ross.h>

/**
 * \brief Reset the event bitfield prior to entering the event handler
 *  post-reverse - reset the bitfield so that a potential re-running of the
 *  event is presented with a consistent bitfield state
 *  NOTE: the size checks are to better support the experimental reverse
 *  computation compiler, which can use a larger bitfield.
 *  Courtesy of <NAME>
 */
static inline void reset_bitfields(tw_event *revent)
{
    if (sizeof(revent->cv) == sizeof(uint32_t)){
        *(uint32_t*)&revent->cv = 0;
    }
    else if (sizeof(revent->cv) == sizeof(uint64_t)){
        *(uint64_t*)&revent->cv = 0;
    }
    else{
        memset(&revent->cv, 0, sizeof(revent->cv));
    }
}

/* 
 * Get all events out of my event queue and spin them out into
 * the priority queue so they can be processed in time stamp
 * order.  
 */
static void tw_sched_event_q(tw_pe * me) {
    tw_clock     start;
    tw_kp       *dest_kp;
    tw_event    *cev;
    tw_event    *nev;

    while (me->event_q.size) {
        cev = tw_eventq_pop_list(&me->event_q);

        for (; cev; cev = nev) {
            nev = cev->next;

            if(!cev->state.owner || cev->state.owner == TW_pe_free_q) {
                tw_error(TW_LOC, "no owner!");
            }
            if (cev->state.cancel_q) {
                cev->state.owner = TW_pe_anti_msg;
                cev->next = cev->prev = NULL;
                continue;
            }

            switch (cev->state.owner) {
                case TW_pe_event_q:
                    dest_kp = cev->dest_lp->kp;

                    if (dest_kp->last_time > cev->recv_ts) {
                        /* cev is a straggler message which has arrived
                        * after we processed events occuring after it.
                        * We need to jump back to before cev's timestamp.
                        */
                        start = tw_clock_read();
                        tw_kp_rollback_to(dest_kp, cev->recv_ts);
                        me->stats.s_rollback += tw_clock_read() - start;
                    }
                    start = tw_clock_read();
                    tw_pq_enqueue(me->pq, cev);
                    me->stats.s_pq += tw_clock_read() - start;
                    break;

                default:
                    tw_error(TW_LOC, "Event in event_q, but owner %d not recognized", cev->state.owner);
            }
        }
    }
}

/*
 * OPT: need to link events into canq in reverse order so 
 *      that when we rollback the 1st event, we should not
 *  need to do any further rollbacks.
 */
static void tw_sched_cancel_q(tw_pe * me) {
    tw_clock     start;
    tw_event    *cev;
    tw_event    *nev;

    start = tw_clock_read();
    while (me->cancel_q) {
        cev = me->cancel_q;
        me->cancel_q = NULL;

        for (; cev; cev = nev) {
            nev = cev->cancel_next;

            if (!cev->state.cancel_q) {
                tw_error(TW_LOC, "No cancel_q bit on event in cancel_q");
            }

            if(!cev->state.owner || cev->state.owner == TW_pe_free_q) {
                tw_error(TW_LOC, "Cancelled event, no owner!");
            }

            switch (cev->state.owner) {
                case TW_pe_event_q:
                    /* This event hasn't been added to our pq yet and we
                    * have not officially received it yet either.  We'll
                    * do the actual free of this event when we receive it
                    * as we spin out the event_q chain.
                    */
                    tw_eventq_delete_any(&me->event_q, cev);

                    tw_event_free(me, cev);
                    break;

                case TW_pe_anti_msg:
                    tw_event_free(me, cev);
                    break;

                case TW_pe_pq:
                    /* Event was not cancelled directly from the event_q
                    * because the cancel message came after we popped it
                    * out of that queue but before we could process it.
                    */
                    tw_pq_delete_any(me->pq, cev);
                    tw_event_free(me, cev);
                    break;

                case TW_kp_pevent_q:
                    /* The event was already processed. 
                    * SECONDARY ROLLBACK
                    */
                    tw_kp_rollback_event(cev);
                    tw_event_free(me, cev);
                    break;

                default:
                    tw_error(TW_LOC, "Event in cancel_q, but owner %d not recognized", cev->state.owner);
            }
        }
    }

    me->stats.s_cancel_q += tw_clock_read() - start;
}

static void tw_sched_batch(tw_pe * me) {
    /* Number of consecutive times we gave up because there were no free event buffers. */
    static int no_free_event_buffers = 0;
    static int warned_no_free_event_buffers = 0;

    tw_clock     start;
    unsigned int     msg_i;

    /* Process g_tw_mblock events, or until the PQ is empty
    * (whichever comes first). 
    */
    for (msg_i = g_tw_mblock; msg_i; msg_i--) {
        tw_event *cev;
        tw_lp *clp;
        tw_kp *ckp;

        /* OUT OF FREE EVENT BUFFERS.  BAD.
        * Go do fossil collect immediately.
        */
        if (me->free_q.size <= g_tw_gvt_threshold) {
            /* Suggested by <NAME> */
            if(++no_free_event_buffers > 10 && !warned_no_free_event_buffers) {
                fprintf(stderr, "WARNING: No free event buffers.  Try increasing memory via the --extramem option.\n");
                warned_no_free_event_buffers = 1;
            }
            tw_gvt_force_update(me);
            break;
        }
{'fraction_non_alphanumeric': 0.09229155742003146, 'fraction_numerical': 0.003146303093864709, 'mean_word_length': 1.4613895461389546, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '455812', 'n_tokens_mistral': 3599, 'n_tokens_neox': 3345, 'n_words': 941}