<commit_before>#include <cphvb.h>
#include <cphvb_compute.h>
#include "dispatch.h"
#include <assert.h>
#include <pthread.h>

//The data type of cphvb_array extended with temporary info.
typedef struct
{
    CPHVB_ARRAY_HEAD
    //Saved original values
    cphvb_index      org_start;
    cphvb_index      org_shape[CPHVB_MAXDIM];

} dispatch_ary;

//Returns the offset based on the current block.
inline cphvb_intp get_offset(cphvb_intp block, cphvb_instruction *inst,
                             cphvb_intp nblocks)
{
    if(block == 0)
        return 0;

    //We compute the offset based on the output operand.
    dispatch_ary *ary = (dispatch_ary*) inst->operand[0];

    return ary->org_shape[0] / nblocks * block + //Whole blocks
           ary->org_shape[0] % nblocks;//The reminder.
}

//Returns the shape based on the current block.
inline cphvb_intp get_shape(cphvb_intp block, cphvb_instruction *inst,
                            cphvb_intp nblocks)
{
    dispatch_ary *ary = (dispatch_ary*) inst->operand[0];

    //We block over the most significant dimension
    //and the first block gets the reminder.
    if(block == 0)
    {
        return ary->org_shape[0] / nblocks +
               ary->org_shape[0] % nblocks;
    }
    else
    {
        return ary->org_shape[0] / nblocks;
    }
}

//Shared thread variables.
static pthread_t thd_ids[32];
static pthread_cond_t  thd_cond_wait     = PTHREAD_COND_INITIALIZER;
static pthread_cond_t  thd_cond_finished = PTHREAD_COND_INITIALIZER;
static pthread_mutex_t thd_mutex         = PTHREAD_MUTEX_INITIALIZER;
static int thd_wait         = 1;//0==start,1==wait
static int thd_finished[2]  = {0,0};//Whan it reaches nthd we are finished.
static int thd_exit         = 0;//1==exit,0=running
static cphvb_intp thd_nblocks;
static cphvb_intp thd_nthds;
static cphvb_intp thd_size;
static cphvb_instruction** thd_inst_bundle;
static computeloop *thd_traverses;
static int main_finished_idx = 0;//Finished index for main.

//The thread function.
static void *thd_do(void *msg)
{
    cphvb_intp myid = ((cphvb_intp)msg);
    int finished_idx = 0;//Finished index (swapped between iterations).
    while(1)
    {
        //Wait for start signal.
        pthread_mutex_lock(&thd_mutex);
        while(thd_wait)
            pthread_cond_wait(&thd_cond_wait,&thd_mutex);
        pthread_mutex_unlock(&thd_mutex);

        if(thd_exit)//Check if we should exit.
            break;

        cphvb_intp nthds = thd_nthds;
        cphvb_intp size = thd_size;
        cphvb_intp nblocks = thd_nblocks;
        cphvb_instruction** inst_bundle = thd_inst_bundle;
        computeloop *traverses = thd_traverses;

        if(nthds > nblocks)
            nblocks = nthds;//Minimum one block per thread.

        //Check if this thread has any work to do.
        if(myid < nthds)
        {
            //We will not block a single instruction into more blocks
            //than threads.
            if(thd_size == 1)
                if(nblocks > nthds)
                    nblocks = nthds;

            cphvb_intp length = nblocks / nthds; // Find this thread's length of work.
            cphvb_intp start = myid * length;    // Find this thread's start block.
            if(myid == nthds-1)
                length += nblocks % nthds;       // The last thread gets the reminder.

            //Clone the instruction and make new views of all operands.
            cphvb_instruction thd_inst[CPHVB_MAX_NO_INST];
            cphvb_array ary_stack[CPHVB_MAX_NO_INST*CPHVB_MAX_NO_OPERANDS];
            cphvb_intp ary_stack_count=0;
            for(cphvb_intp j=0; j<size; ++j)
            {
                thd_inst[j] = *inst_bundle[j];
                cphvb_intp nops = cphvb_operands(inst_bundle[j]->opcode);
                for(cphvb_intp i=0; i<nops; ++i)
                {
                    cphvb_array *ary_org = inst_bundle[j]->operand[i];
                    cphvb_array *ary = &ary_stack[ary_stack_count++];
                    *ary = *ary_org;

                    if(ary_org->base == NULL)//base array
                    {
                        ary->base = ary_org;
                    }

                    thd_inst[j].operand[i] = ary;//Save the operand.
                 }
            }

            //Handle one block at a time.
            for(cphvb_intp b=start; b<start+length; ++b)
            {
                //Update the operands to match the current block.
                for(cphvb_intp j=0; j<size; ++j)
                {
                    cphvb_instruction *inst = &thd_inst[j];
                    cphvb_intp nops = cphvb_operands(inst->opcode);
                    for(cphvb_intp i=0; i<nops; ++i)
                    {
                        dispatch_ary *ary = (dispatch_ary*) inst->operand[i];
                        ary->start = ary->org_start + ary->stride[0] *
                                     get_offset(b, inst, nblocks);
                        ary->shape[0] = get_shape(b, inst, nblocks);
                    }
                }

                //Dispatch a block.
                for(cphvb_intp j=0; j<size; ++j)
                {
                    cphvb_instruction *inst = &thd_inst[j];
                    if(inst->operand[0]->shape[0] <= 0)
                        break;//Nothing to do.

                    inst->status = traverses[j](inst);
                    if(inst->status != CPHVB_SUCCESS)
                    {
                        //ret = CPHVB_PARTIAL_SUCCESS;
                        start = length;//Force a complete exit.
                        break;
                    }
                }
            }
        }

        //Signal that we are finished.<commit_msg>Changed traversal call to match updated travelsal function.<commit_after>#include <cphvb.h>
#include <cphvb_compute.h>
#include "dispatch.h"
#include <assert.h>
#include <pthread.h>

//The data type of cphvb_array extended with temporary info.
typedef struct
{
    CPHVB_ARRAY_HEAD
    //Saved original values
    cphvb_index      org_start;
    cphvb_index      org_shape[CPHVB_MAXDIM];

} dispatch_ary;

//Returns the offset based on the current block.
inline cphvb_intp get_offset(cphvb_intp block, cphvb_instruction *inst,
                             cphvb_intp nblocks)
{
    if(block == 0)
        return 0;

    //We compute the offset based on the output operand.
    dispatch_ary *ary = (dispatch_ary*) inst->operand[0];

    return ary->org_shape[0] / nblocks * block + //Whole blocks
           ary->org_shape[0] % nblocks;//The reminder.
}

//Returns the shape based on the current block.
inline cphvb_intp get_shape(cphvb_intp block, cphvb_instruction *inst,
                            cphvb_intp nblocks)
{
    dispatch_ary *ary = (dispatch_ary*) inst->operand[0];

    //We block over the most significant dimension
    //and the first block gets the reminder.
    if(block == 0)
    {
        return ary->org_shape[0] / nblocks +
               ary->org_shape[0] % nblocks;
    }
    else
    {
        return ary->org_shape[0] / nblocks;
    }
}

//Shared thread variables.
static pthread_t thd_ids[32];
static pthread_cond_t  thd_cond_wait     = PTHREAD_COND_INITIALIZER;
static pthread_cond_t  thd_cond_finished = PTHREAD_COND_INITIALIZER;
static pthread_mutex_t thd_mutex         = PTHREAD_MUTEX_INITIALIZER;
static int thd_wait         = 1;//0==start,1==wait
static int thd_finished[2]  = {0,0};//Whan it reaches nthd we are finished.
static int thd_exit         = 0;//1==exit,0=running
static cphvb_intp thd_nblocks;
static cphvb_intp thd_nthds;
static cphvb_intp thd_size;
static cphvb_instruction** thd_inst_bundle;
static computeloop *thd_traverses;
static int main_finished_idx = 0;//Finished index for main.

//The thread function.
static void *thd_do(void *msg)
{
    cphvb_intp myid = ((cphvb_intp)msg);
    int finished_idx = 0;//Finished index (swapped between iterations).
    while(1)
    {
        //Wait for start signal.
        pthread_mutex_lock(&thd_mutex);
        while(thd_wait)
            pthread_cond_wait(&thd_cond_wait,&thd_mutex);
        pthread_mutex_unlock(&thd_mutex);

        if(thd_exit)//Check if we should exit.
            break;

        cphvb_intp nthds = thd_nthds;
        cphvb_intp size = thd_size;
        cphvb_intp nblocks = thd_nblocks;
        cphvb_instruction** inst_bundle = thd_inst_bundle;
        computeloop *traverses = thd_traverses;

        if(nthds > nblocks)
            nblocks = nthds;//Minimum one block per thread.

        //Check if this thread has any work to do.
        if(myid < nthds)
        {
            //We will not block a single instruction into more blocks
            //than threads.
            if(thd_size == 1)
                if(nblocks > nthds)
                    nblocks = nthds;

            cphvb_intp length = nblocks / nthds; // Find this thread's length of work.
            cphvb_intp start = myid * length;    // Find this thread's start block.
            if(myid == nthds-1)
                length += nblocks % nthds;       // The last thread gets the reminder.

            //Clone the instruction and make new views of all operands.
            cphvb_instruction thd_inst[CPHVB_MAX_NO_INST];
            cphvb_array ary_stack[CPHVB_MAX_NO_INST*CPHVB_MAX_NO_OPERANDS];
            cphvb_intp ary_stack_count=0;
            for(cphvb_intp j=0; j<size; ++j)
            {
                thd_inst[j] = *inst_bundle[j];
                cphvb_intp nops = cphvb_operands(inst_bundle[j]->opcode);
                for(cphvb_intp i=0; i<nops; ++i)
                {
                    cphvb_array *ary_org = inst_bundle[j]->operand[i];
                    cphvb_array *ary = &ary_stack[ary_stack_count++];
                    *ary = *ary_org;

                    if(ary_org->base == NULL)//base array
                    {
                        ary->base = ary_org;
                    }

                    thd_inst[j].operand[i] = ary;//Save the operand.
                 }
            }

            //Handle one block at a time.
            for(cphvb_intp b=start; b<start+length; ++b)
            {
                //Update the operands to match the current block.
                for(cphvb_intp j=0; j<size; ++j)
                {
                    cphvb_instruction *inst = &thd_inst[j];
                    cphvb_intp nops = cphvb_operands(inst->opcode);
                    for(cphvb_intp i=0; i<nops; ++i)
                    {
                        dispatch_ary *ary = (dispatch_ary*) inst->operand[i];
                        ary->start = ary->org_start + ary->stride[0] *
                                     get_offset(b, inst, nblocks);
                        ary->shape[0] = get_shape(b, inst, nblocks);
                    }
                }

                //Dispatch a block.
                for(cphvb_intp j=0; j<size; ++j)
                {
                    cphvb_instruction *inst = &thd_inst[j];
                    if(inst->operand[0]->shape[0] <= 0)
                        break;//Nothing to do.

                    inst->status = traverses[j](inst, 0, 0 );
                    if(inst->status != CPHVB_SUCCESS)
                    {
                        //ret = CPHVB_PARTIAL_SUCCESS;
                        start = length;//Force a complete exit.
                        break;
                    }
                }
            }
        }

        //Signal that we are finished.
{'fraction_non_alphanumeric': 0.09526725570816201, 'fraction_numerical': 0.006648587175225265, 'mean_word_length': 1.6962264150943396, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3995560', 'n_tokens_mistral': 3779, 'n_tokens_neox': 3616, 'n_words': 834}