LimitTheMaximumConcurrency

# Limit the Maximum Concurrency

Define a Semaphore

Define a Semaphore

LimitTheMaximumConcurrency_1DefineASemaphore
Use Semaphores Across Different Tasks

Use Semaphores Across Different Tasks

LimitTheMaximumConcurrency_1UseSemaphoresAcrossDifferentTasks
Define a Conflict Graph

Define a Conflict Graph

LimitTheMaximumConcurrency_1DefineAConflictGraph
Reset a Semaphore

Reset a Semaphore

LimitTheMaximumConcurrency_1ResetASemaphore
Understand the Limitation of Semaphores

Understand the Limitation of Semaphores

LimitTheMaximumConcurrency_1UnderstandTheLimitationOfSemaphores

This chapters discusses how to limit the concurrency or the maximum number of workers in your Taskflow applications.

# Define a Semaphore

Taskflow provides a mechanism, tf::Semaphore, for you to limit the maximum concurrency in a section of tasks. You can let a task acquire/release one or multiple semaphores before/after executing its work. A task can acquire and release a semaphore, or just acquire or just release it. A tf::Semaphore object starts with an initial value. As long as that value is above 0, tasks can acquire the semaphore and do their work. If the value is 0 or less, a task trying to acquire the semaphore will not run but goes to a waiting list of that semaphore. When the semaphore is released by another task, it reschedules all tasks on that waiting list.

    tf::Executorexecutor(8);//createanexecutorof8workers
    tf::Taskflowtaskflow;

    tf::Semaphoresemaphore(1);//createasemaphorewithinitialvalueof1

    std::vector<tf::Task>tasks{
    taskflow.emplace([](){std::cout<<"A"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"B"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"C"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"D"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"E"<<std::endl;})
    };

    for(auto&task:tasks){//eachtaskacquiresandreleasethesemaphore
    task.acquire(semaphore);
    task.release(semaphore);
    }

    executor.run(taskflow).wait();

The above example creates five tasks with no dependencies between them. Under normal circumstances, the five tasks would be executed concurrently. However, this example has a semaphore with initial value of 1, and all tasks need to acquire that semaphore before running and release that semaphore after they are done. This organization limits the number of concurrently running tasks to only one. One possible output is shown below:

    {.shell-session}
    #theoutputisasequentialchainoffivetasks
    A
    B
    E
    D
    C

It is your responsibility to ensure the semaphore stays alive during the execution of tasks that acquire and release it. The executor and taskflow do not manage lifetime of any semaphores.

For the same example above, we can limit the semaphore concurrency to another value different from 1, say 3, which will limit only three workers to run the five tasks, `A`, `B`, `C`, `D`, and `E`.

    tf::Executorexecutor(8);//createanexecutorof8workers
    tf::Taskflowtaskflow;

    tf::Semaphoresemaphore(3);//createasemaphorewithinitialvalueof3

    std::vector<tf::Task>tasks{
    taskflow.emplace([](){std::cout<<"A"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"B"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"C"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"D"<<std::endl;}),
    taskflow.emplace([](){std::cout<<"E"<<std::endl;})
    };

    for(auto&task:tasks){//eachtaskacquiresandreleasethesemaphore
    task.acquire(semaphore);
    task.release(semaphore);
    }

    executor.run(taskflow).wait();

    {.shell-session}
    #Onepossibleoutput:A,B,andCrunconcurrently,DandErunconcurrently
    ABC
    ED

Semaphores are powerful for limiting the maximum concurrency of not only a section of tasks but also different sections of tasks. Specifically, you can have one task acquire a semaphore and have another task release that semaphore to impose concurrency on subgraphs of tasks. The following example serializes the execution of five pairs of tasks using a semaphore rather than explicit dependencies.

    tf::Executorexecutor(4);//createsanexecutorof4workers
    tf::Taskflowtaskflow;
    tf::Semaphoresemaphore(1);

    intN=5;
    intcounter=0;//non-atomicintegercounter

    for(inti=0;i<N;i++){
    tf::Taskf=taskflow.emplace([&](){counter++;})
    .name("from-"s+std::to_string(i));
    tf::Taskt=taskflow.emplace([&](){counter++;})
    .name("to-"s+std::to_string(i));
    f.precede(t);
    f.acquire(semaphore);
    t.release(semaphore);
    }

    executor.run(taskflow).wait();

    assert(counter==2*N);

Without semaphores, each pair of tasks, e.g., `from-0 -> to-0`, will run independently and concurrently. However, the program forces each `from` task to acquire the semaphore before running its work and not to release it until its paired `to` task is done. This constraint forces each pair of tasks to run sequentially, while the order of which pair runs first is up to the scheduler.

# Use Semaphores Across Different Tasks

You can use semaphores to limit the concurrency across different sections of taskflow graphs. When you submit multiple taskflows to an executor, the executor view them as a bag of dependent tasks. It does not matter which task in which taskflow graph acquires or releases a semaphore.

    tf::Executorexecutor(8);//createanexecutorof8workers
    tf::Taskflowtaskflow1;
    tf::Taskflowtaskflow2;

    tf::Semaphoresemaphore(1);//createasemaphorewithinitialvalueof1

    taskflow1.emplace([](){std::cout<<"taskintaskflow1";})
    .acquire(semaphore)
    .release(semaphore);

    taskflow2.emplace([](){std::cout<<"taskintaskflow2";})
    .acquire(semaphore)
    .release(semaphore);

    executor.run(taskflow1);
    executor.run(taskflow2);
    executor.wait_for_all();

The above examples creates one task from each taskflow and submits the two taskflows to the executor. Again, under normal circumstances, the two tasks can run concurrently, but the semaphore restricts one worker to run the two task sequentially in arbitrary order.

# Define a Conflict Graph

One important application of tf::Semaphore is conflict-aware scheduling using a conflict graph. A conflict graph is a undirected graph where each vertex represents a task and each edge represents a conflict between a pair of tasks. When a task conflicts with another task, they cannot run together. Consider the conflict graph below, task `A` conflicts with task `B` and task `C` (and vice versa), meaning that `A` cannot run together with `B` and `C` whereas `B` and `C` can run together.

We can create one semaphore of one concurrency for each edge in the conflict graph and let the two tasks of that edge acquire the semaphore. This organization forces the two tasks to not run concurrently.

    tf::Executorexecutor;
    tf::Taskflowtaskflow;

    tf::Semaphoreconflict_AB(1);
    tf::Semaphoreconflict_AC(1);

    tf::TaskA=taskflow.emplace([](){std::cout<<"A"<<std::endl;});
    tf::TaskB=taskflow.emplace([](){std::cout<<"B"<<std::endl;});
    tf::TaskC=taskflow.emplace([](){std::cout<<"C"<<std::endl;});

    //describetheconflictbetweenAandB
    A.acquire(conflict_AB).release(conflict_AB);
    B.acquire(conflict_AB).release(conflict_AB);

    //describetheconflictbetweenAandC
    A.acquire(conflict_AC).release(conflict_AC);
    C.acquire(conflict_AC).release(conflict_AC);

    executor.run(taskflow).wait();

    {.shell-session}
    #Onepossibleoutput:BandCrunconcurrentlyafterA
    A
    BC

A task can acquire and release multiple semaphores. When the executor runs a task, it will try to acquire all semaphores needed by that task. When the executor finishes that task, it will release all acquired semaphores by that task.

# Reset a Semaphore

You can reset a semaphore to its initial state using tf::Semaphore::reset(), or set a new maximum value with tf::Semaphore::reset(size_t new_max_value). The method tf::Semaphore::value() allows you to query the current value of the semaphore, which represents the number of available acquisitions.

    tf::Semaphoresemaphore(4);
    assert(semaphore.value()==4&&semaphore.max_value()==4);

    //resetthesemaphoretoanewvalue
    semaphore.reset(11);
    assert(semaphore.value()==11&&semaphore.max_value()==11);

When a semaphore is acquired more times than its maximum value, an exception will be thrown.

# Understand the Limitation of Semaphores

Currently, tf::Semaphore has limited support for exception handling and taskflow cancellation. If a task throws an exception or the taskflow is canceled, subsequent acquire and release operations on the semaphore may result in undefined behavior. To ensure correct behavior, you should call tf::Semaphore::reset before reusing the semaphore in the next run. For instance, in the code below, when task `B` throws an exception, the executor will cancel the execution of the taskflow. That is, tasks `C` and `D` will not run, and thus no task will release the acquired semaphore. To resolve this situation, we must reset the semaphore to a clean state for the next run.

    tf::Executorexecutor;
    tf::Taskflowtaskflow;
    tf::Semaphoresemaphore(1);

    tf::TaskA=taskflow.emplace([](){});
    tf::TaskB=taskflow.emplace([](){throwstd::runtime_error("exception");});
    tf::TaskC=taskflow.emplace([](){});
    tf::TaskD=taskflow.emplace([](){});
    A.precede(B);
    B.precede(C);
    C.precede(D);

    A.acquire(semaphore);
    D.release(semaphore);

    //currentsemaphorehasavalueof1
    assert(semaphore.value()==1);

    //whenBthrowstheexception,Dwillnotrunandthussemaphoreisnotreleased
    try{
    executor.run(taskflow).get();
    }
    catch(std::runtime_error&e){
    std::cout<<e.what()<<std::endl;
    }

    //sinceAacquiredthesemaphore,itsvalueis0
    assert(semaphore.value()==0);

    //resetthesemaphoretoacleanstatebeforerunningthetaskflowagain
    semaphore.reset();
    assert(semaphore.value()==1);

    executor.run(taskflow).get();