Function prepareThreads

numexpr3/interpreter.cpp:144–229

This function prepares the pointers towards the temporary memory block, and the NumExprObjects used by threads. Replacement for get_temps_space() and NumExprObject_copy_threadsafe().

// the NumExprObjects used by threads.
// Replacement for get_temps_space() and NumExprObject_copy_threadsafe().
int 
prepareThreads( NumExprObject* self, NpyIter *iter, int *pc_error, char **errorMessage ) {
    // Variables
    int I = 0, R = 0; 
    npy_intp memOffset = 0, taskFactor, numBlocks;

    // Setup error tracking
    gs.ret_code = 0;
    gs.pc_error = pc_error;
    gs.errorMessage = errorMessage;

    // printf( "prepareThreads #1\n");
    // Stuff from vm_engine_iter_parallel:
    NpyIter_GetIterIndexRange(iter, &gs.start, &gs.vlen);
    
    // Try to make it so each thread gets 16 tasks.  This is a compromise
    // between 1 task per thread and one block per task.
    // RAM: would be nice to benchmark this sort of assumption.
    taskFactor = TASKS_PER_THREAD*BLOCK_SIZE1*gs.n_thread;
    numBlocks = (gs.vlen - gs.start + taskFactor - 1) /
                            taskFactor;
    gs.task_size = numBlocks * BLOCK_SIZE1; // Note this is much bigger than BLOCK_SIZE1

    
    // Ensure the temporary memory storage is big enough
    // printf("Original: gs.tempSize = %d, gs.tempArena = %p\n", gs.tempSize, gs.tempArena );
    if( BLOCK_SIZE1 * gs.n_thread * self->total_temp_itemsize >  gs.tempSize ) {
        // printf( "temp size too small, resizing #1\n");
        numexpr_set_tempsize( BLOCK_SIZE1 * gs.n_thread * self->total_temp_itemsize );
        // printf("Engorged: gs.tempSize = %d, gs.tempArena = %p\n", gs.tempSize, gs.tempArena );
    }

    // `Do` part of the `for` loop before, use `self` for the first thread.
    gs.iter[0] = iter;
    gs.params[0] = *self;
    gs.params[0].registers = (NumExprReg *)gs.registerArena;
    memcpy( gs.params[0].registers, self->registers, self->n_reg*sizeof(NumExprReg) );

    // Setup temporaries memory pointers for the first NumExprObject
    // printf( "self->n_reg = %d, gs.tempArena = %p\n", self->n_reg, gs.tempArena );
    for( R=0; R < self->n_reg; R++ ) {
        if( self->registers[R].kind != KIND_TEMP ) continue;

        // printf( "reg#%d has kind %d and itemsize %d\n", R, self->registers[R].kind, (int)self->registers[R].itemsize );
        gs.params[0].registers[R].mem = gs.tempArena + memOffset;
        // printf( "    thread #0, reg #%d, points to %p\n", R, gs.params[0].registers[R].mem );
        memOffset += BLOCK_SIZE1 * self->registers[R].itemsize;
    }

    // Make copies of iterators and NumExprObjects for each additional thread
    for( I = 1; I < gs.n_thread; ++I ) {
        // TODO: add reduce_iter when you parallelize reductions
        gs.iter[I] = NpyIter_Copy(iter);
        if (gs.iter[I] == NULL) {
            // Error: deallocate all iterators and return an error code.
            --I;
            // Only the original iterator should ever be called, and not copies.
            NpyIter_Deallocate(gs.iter[0]);