Reducing contention/locking in a job system involving read/write dependencies?

Question

I have a job system that has two different types of dependencies per-job:

• Jobs can rely on other jobs to be completed first, but this is optional
• Jobs have read/write dependencies (shared/exclusive) that are used to restrict which jobs run at the same time. This is currently represented as a series of type hashes for each read-only and each writable C++ type

The C++-flavoured psuedo-code shows the algorithm being used. My main issue is that when we want to check if a job can run (after its optional dependent jobs have completed), we need to lock around the list of currently in-use dependencies, and again when a job finishes so that job can remove its own dependencies from the active list, unblocking other jobs.

I'm wondering if it's possible to make this atomic; like representing each dependency type as a bit and trying to hit it in one instruction, but with the mix of shared/exclusive types, I don't know how you can represent that without a mix of different bitwise ops.

OnJobFinished(job)
{
    lock(dependenciesMtx)
    PopJobDependencies(job)  // Clear this job's shared/exclusive dependencies
    unlock(dependenciesMtx)
    
    for connection in job.connections
        connection.numBlocking--;
        if connection.numBlocking == 0
            contendingQueue.push(connection)
            
    ProcessContendingJobs()
}

Job.Process()
{
    do actual job work
    
    // ...
    
    OnJobFinished(this)
}

LaunchJob(job)
{
    Launch job fiber -> job.Process(...) // this puts job.Process on another thread/fiber and is non-blocking
}

bool CanRunNode(job)
{
    if job.exclusiveDeps in (currentDeps.shared + currentDeps.exclusive)
        return false
        
    if job.sharedDeps in currentDeps.exclusive
        return false
        
    return true
}

ProcessContendingJobs()
{
    while (job = contendingQueue.pop())
        lock(dependenciesMtx)
        
        if CanRunNode(job)
            PushJobDependencies(job) // Store this job's shared/exclusive dependencies
            LaunchJob(job)
        else
            jobsToRepush.push(job)
        
        unlock(dependenciesMtx)
        
    // we couldn't execute them this time, so re-add to queue for later
    for job in jobsToRepush
        contendingQueue.push(job)
}

void ProcessJobs(jobs) <<<<< Start here
{
    // Jobs have optional explicit dependencies on other jobs (job A -> 'connections' -> job B),
    // and the dependee knows how many jobs are blocking it from executing.
    // They also have dependencies on specific types (as shared deps [read] and exclusive deps [write])

    // contendingQueue is a concurrent queue (via Intel TBB)

    // Get at least some work in the queue before we start
    for job in jobs
        if job.numBlocking == 0
            contendingQueue.push(job)
    
    ProcessContendingJobs()
    
    //...
    
    // Wait for all work to finish
}

There are a number of issues here;

1. The contention on the dependencies mutex, as stated
2. We have to pop a job from the queue to check if that job can run, so it's often the case that it can't be and we have to re-push it. This is a potential source of churn and is a waste of work
3. Should a queue processor in this context push the job right back into the queue it's looping over, potentially looping constantly until a runnable job is found? Also seems wasteful.

EDIT: Some ideas I thought of (though unrelated to current answer and won't change it):

• Switch to a read-write lock, where the dependency check acquires only a read lock, and if the job can run, we upgrade to a write lock and push its dependencies. Bit tricky since the fiber library I use has no shared_mutex equivalent, but can be written.
• Keep the lock in OnJobFinished locked and call a modified ProcessContendingJobs that expects the lock to already be acquired and simply unlocks after the while loop completes. Less-frequent lock/unlock but holds the lock for longer.
• Similarly, if not the above, then change the while loop to keep a lock before iteration and unlock after the loop is finished. Similar issue to the above; not sure if it's better to reduce number of lock/unlock calls at the expense of making the lock duration longer.

Answer 1

Yes, this is probably doable.

First, let's ignore the shared vs exclusive access problem, and assume that all accesses are exclusive.

For each dependency, consider an atomic boolean representing whether it is in use. We impose a total order over dependencies, i.e. they form an array/vector. This means that a job can declare its dependencies via an ordered list of dependency indexes.

This makes it possible for the job to atomically acquire each dependency in turn. If all jobs acquire dependencies in the same order, then deadlocks are prevented (this is why we imposed a total order over dependencies). If one dependency is already in-use, the previously acquired dependencies are released again.

The essential primitive for such atomic operations is a variable.compare_exchange(expected, new) method that atomically sets the variable to the new value, but only if it currently contains the expected old value.

Essentially, we would replace this code fragment

lock(dependenciesMtx)
        
if CanRunNode(job)
    PushJobDependencies(job) // Store this job's shared/exclusive dependencies
    LaunchJob(job)
else
    jobsToRepush.push(job)        

unlock(dependenciesMtx)

with something like:

std::vector<std::atomic<bool>> available_deps = ...;

int dep = 0;
bool ok = true;

// acquire the dependencies in order
for (; dep < job.deps.size(); dep++) {
   ok &= available_deps[job.deps[dep]].compare_exchange(false, true);
   if (!ok) break;
}

if (ok) {
  LaunchJob(job);
} else {
  // release already-acquired dependencies, must use reverse order
  for(; dep >= 0; dep--)
    assert(available_deps[jobs.dep[dep]].compare_exchange(true, false));

  jobsToRepush.push(job);
}

This approach can be extended to shared dependencies by replacing the atomic bool with an atomic unsigned integer that counts jobs that have access. We would then have three classes of values:

• 0: no one is currently accessing the dependency
• n (any other value): there are n jobs with shared access to this dependency
• -1 (or some maximum value): someone holds exclusive access

Acquiring shared access would then involve incrementing the current count, while making sure that the max value won't be reached.

Acquiring exclusive access would then involve a dependency.compare_exchange(0, -1).

It is not a given though that atomics will perform better.

First of all, atomics have different “memory orders” that you can select per-operation. C++ defaults to the most restrictive and least performant seq_cst memory order, which might be emulated on some platforms with the help of locks (which you're trying to avoid). You might be able to get greatly improved performance if you are sure that you can get by with more relaxed memory orders such as acq_rel.

Second, mutexes are not necessarily slow in practice. This depends on lock contention, and on the chosen memory orderings. An atomics-based approach is probably going to fare better if jobs have dependency sets with little overlap. But this is definitely worth benchmarking.