org.apache.uima.analysis_engine.impl

## Class MultiThreadCoordination

- java.lang.Object
- - org.apache.uima.analysis_engine.impl.MultiThreadCoordination

-     public class MultiThreadCoordination
      extends Object

  ************************************************************ * * A N E X P E R I M E N T - not hooked up 10/2018 * ************************************************************ Supplies optional thread coordination when running identical pipelines in a multi-core CPU. The goal is to arrange to have threads running the same primitive(s) on separate threads at the same time, in hopes of improving locality of reference, and therefore l1/2/3 memory cache utilization Structure: 1 instance of this class per coordination group of pipelines inner classes: nbr_threads instances MultiThreadInfo, one per thread nbr_barrier instances of Barrier_info, one per barrier Inputs: list of coordination specs Where the barriers are: fully qualified AE chain, eg /key1/key2/ where key is the key name in the aggregate descriptor 2 keys: start of high-priority section (the barrier) and end of high-priority section nbr of threads to run in parallel nbr of threads to wait at barrier computed: total nbr of threads: = nbr to run in parallel + extras when coordination holdup is happening Waiting: two kinds of waiting: - low priority "fill-in" threads that activate, but go back to sleep, as needed to compensate for threads being released at barrier - barrier hold - where threads wait for cohorts waiting queues: - one per barrier - one for low-priority change of state: - when computed, threads put into queues and marked shouldWait to go to wait state - when going to wait state, another thread is signaled -- released barrier first - when arrive at barrier, if thread marked to go to wait due to previous barrier being released, -- make equivalent to having marked a different thread to go to wait due barrier being released -- two cases: --- not the last to arrive - will wait this thread, and release a low-pri thread in this case, the equivalent will be to wait another thread wait this thread release another thread (cancel out the first wait-another-thread so - just wait this thread at barrier wait, resetting the pending wait --- is the last to arrive. will wakeup the barrier threads, and wait others in this case the equivalent will be to wait another thread release the barrier threads wait n-1 low-pri threads so - wait (n-1) + 1 low-pri threads. State maintained: map - all threads running pipelines to MultiThreadInfo for that thread map - delegate key names /xxx/yyy/zzz/ to barrier start, and barrier end Insertion points in the flow: - just before and after calling primitive process - at "before": - if at barrier, suspend thread until have enough - if not at barrier, but threadinfo marked as shouldWait, then wait this thread. - at "after": - if HIGH_PRIORITY and at barrier-end, -- lower priority - if ! HIGH_PRIORITY, another HIGH_PRIORITY item, suspend thread -- whenever a thread "waits", it first attempts to wakeup a compensating one -- high priority preferred -- order in the queue - from front - at beginning of pipeline -- maintains nbr_in_process - used to detect end-state -- incr and if tracing sets thread's seq nbr - at end of pipeline -- if HIGH_PRIORITY thread, reset to low priority -- decr nbr_in_process - used to detect end-state Running and Waiting threads go into multiple LinkedBlockingDeque, (set to never block) running: (low priority running, only) used to pick next thread to wait on when need to wait waiting: (barrier or low-priority) used to pick next thread to wakeup Initial state: set first baseNbr threads to running (put in low-pri running q, not marked shouldWait). set remainder threads to waiting (put in low-pri wait q, mark as shouldWait). end-state = false; nbr_in_process = 0; currently - a race condition: it must be true that enough threads start before one gets to end-of-pipeline to prevent premature triggering of end-state eventual fix: use timer, trigger end-state when thread returns and doesn't come back in xxx. Algorithm - goal: run n threads, preferably together following a barrier - recovery from exceptions: tbd simple case: only one barrier. For this, - keep state: NORMAL, END_STATE - use a number of threads size of (n-1) + n - pools: --- low_pri_wait - initial setup: -- all threads at low-pri -- n threads active, others in low_pri_wait, added to low_pri_wait pool -- set count_of_threads_running to n - when thread hits barrier: - add to barrier, switch thread to hi-pri - if thread will wait at barrier, - decr count_of_threads_running - while (nbr active threads < n) activate a lo-pri_wait thread from low-pri pool -- if no more, go to end state: in end state, open all barriers (use reset) --- implies setting up threads at first - if thread will not wait at barrier: -- reset (releases all threads at barrier) -- update count of hi-pri running -- pick n of the not released threads to switch to shouldWait state --- account for sync, for already in this state. - when thread hits non-barrier: if ( is in shouldWait state) -- add this thread to low-pri pool (waiting) and wait it - when thread finishes a barrier-ed primitive: -- drop its priority for m barriers: - use pool size of m * (n - 1) + n (m barrier waiting n-1 threads each)

- - ### Nested Class Summary

    Nested Classes
    Modifier and Type | Class and Description
    --- | ---
    `class` | `MultiThreadCoordination.MultiThreadInfo` info per thread *

  - ### Field Summary

    Fields
    Modifier and Type | Field and Description
    --- | ---
    `static int` | `HIGH_PRIORITY`

  - ### Constructor Summary

    Constructors
    Constructor and Description
    `MultiThreadCoordination(List<String> barrier_ids, int baseNbrOfThreads, int barrierNbrOfThreads, int timeout, TimeUnit tu)` C O N S T R U C T O R * called once per coordination group, when setting it up * *

  - ### Method Summary

    All Methods Static Methods Instance Methods Concrete Methods
    Modifier and Type | Method and Description
    --- | ---
    `void` | `addThread(Thread t)` Called in different thread when setting up worker threads
    `void` | `at_call_primitive(MultiThreadCoordination.MultiThreadInfo ti, String barrier_id, int casId, int casResets)` at call primitive: if a barrier, hold up until enough cohorts, then release all if not at barrier, wait the thread if shouldWait (pending)
    `static void` | `end_of_pipeline()`
    `static void` | `start_of_pipeline()` S T A T I C M E T H O D S INTERFACE WITH PIPELINE EVENTS * * Use static weakMap with thread as key to get threadInfo * - if exists, call instance of this class's equivalent method *

    - ### Methods inherited from class java.lang.Object

      `clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

- - ### Field Detail

    - #### HIGH_PRIORITY

          public static final int HIGH_PRIORITY

  - ### Constructor Detail

    - #### MultiThreadCoordination

          public MultiThreadCoordination(List<String> barrier_ids,
                                         int baseNbrOfThreads,
                                         int barrierNbrOfThreads,
                                         int timeout,
                                         TimeUnit tu)

      C O N S T R U C T O R * called once per coordination group, when setting it up * *
      Parameters:
        `barrier_ids` - list of strings of /aaa/bbb/ * of fully-qualified key names in aggregate chain down to the primitive * but excluding the primitive name. * - 2 strings per barrier - the starting key and the ending key * - ending key can be a non-matching value (e.g. "end") * which means the barrier section extends thru to the end of the * pipeline
        `baseNbrOfThreads` - number of threads to keep active, * set to manage memory bandwidth, l1/2/3 caching *
        `barrierNbrOfThreads` - number of threads to await at a barrier before releasing * to get an affinity effect, this number should be larger than the * baseNbrOfThreads *
        `timeout` - -
        `tu` - -

  - ### Method Detail

    - #### start_of_pipeline

          public static void start_of_pipeline()

      S T A T I C M E T H O D S INTERFACE WITH PIPELINE EVENTS * * Use static weakMap with thread as key to get threadInfo * - if exists, call instance of this class's equivalent method *

    - #### end_of_pipeline

          public static void end_of_pipeline()

    - #### at_call_primitive

          public void at_call_primitive(MultiThreadCoordination.MultiThreadInfo ti,
                                        String barrier_id,
                                        int casId,
                                        int casResets)

      at call primitive: if a barrier, hold up until enough cohorts, then release all if not at barrier, wait the thread if shouldWait (pending)
      Parameters:
        `ti` - -
        `barrier_id` - -
        `casId` - -
        `casResets` - -

    - #### addThread

          public void addThread(Thread t)

      Called in different thread when setting up worker threads
      Parameters:
        `t` - the worker thread

Copyright © 2006–2021 The Apache Software Foundation. All rights reserved.