Function vm_engine_iter_parallel

numexpr3/interpreter.cpp:342–393

Parallel iterator version of VM engine This function fills out the global state and then unlocks the mutexes used to control the threads in module.cpp::th_worker prepareThreads() must be called beforehand (or else `self` could not be const).

// used to control the threads in module.cpp::th_worker
// prepareThreads() must be called beforehand (or else `self` could not be const).
static int
vm_engine_iter_parallel(NpyIter *iter, const NumExprObject *self,
                        bool need_output_buffering, int *pc_error,
                        char **errorMessage)
{
    if (errorMessage == NULL) return -1;

    // Threads are prepared for execution in prepareThreads()
    BENCH_TIME(8);
    Py_BEGIN_ALLOW_THREADS;
    DIFF_TIME(8);

    BENCH_TIME(4);
    // Synchronization point for all threads (wait for initialization)
    pthread_mutex_lock(&gs.count_threads_mutex);
    BENCH_RANGE(150, gs.n_thread);
    if (gs.count_threads < gs.n_thread) {
        gs.count_threads++; 
        // printf( "Main thread init %d\n", gs.count_threads );
        do {
            pthread_cond_wait(&gs.count_threads_cv, &gs.count_threads_mutex);
        } while (gs.barrier_passed == BARRIER_HALT);
    }
    else {
        gs.barrier_passed = BARRIER_PASS;
        pthread_cond_broadcast(&gs.count_threads_cv);
    }
    pthread_mutex_unlock(&gs.count_threads_mutex);
    DIFF_TIME(4);
    BENCH_TIME(5);
    // Synchronization point for all threads (wait for finalization)
    pthread_mutex_lock(&gs.count_threads_mutex);
    if (gs.count_threads > 0) {
        gs.count_threads--; 
        // printf( "Main thread finalize %d\n", gs.count_threads );
        do {
            pthread_cond_wait(&gs.count_threads_cv, &gs.count_threads_mutex);
        } while (gs.barrier_passed == BARRIER_PASS);
    }
    else {
        gs.barrier_passed = BARRIER_HALT;
        pthread_cond_broadcast(&gs.count_threads_cv);
    }
    pthread_mutex_unlock(&gs.count_threads_mutex);
    DIFF_TIME(5);

    BENCH_TIME(9);
    Py_END_ALLOW_THREADS;
    DIFF_TIME(9);
   
    return gs.ret_code;
}

static int
run_interpreter(NumExprObject *self, NpyIter *iter, NpyIter *reduce_iter,