Эффективные независимые синхронизированные блоки?

Question

У меня есть сценарий, в котором в определенные моменты моей программы поток должен обновить несколько общих структур данных.Каждая структура данных может быть безопасно обновлена ​​параллельно с любой другой структурой данных, но каждая структура данных может обновляться только одним потоком за раз.Простой, наивный способ, которым я выразил это в своем коде:

synchronized updateStructure1();
synchronized updateStructure2();
// ...

Это кажется неэффективным, потому что, если несколько потоков пытаются обновить структуру 1, но ни один поток не пытается обновить структуру 2, они 'Все блокируют ожидание блокировки, которая защищает структуру 1, в то время как блокировка структуры 2 остается незамеченной.

Есть ли "стандартный" способ исправить это?Другими словами, существует ли стандартный потоковый примитив, который пытается обновить все структуры циклически, блокирует только, если все блокировки сняты, и возвращает, когда все структуры обновляются?

Это несколько языквопрос, но если это поможет, я использую язык D.

Answer 1

Если ваш язык поддерживает легкие потоки или Actors , вы всегда можете иметь обновляющий поток, который порождает новый новый поток для изменения каждого объекта, где каждый поток просто блокирует, модифицирует и разблокирует каждый объект.Затем ваш поток обновлений присоединится ко всем дочерним потокам, прежде чем вернуться.Это исключает проблему из-за графика времени выполнения, и можно свободно планировать эти дочерние потоки любым доступным способом для лучшей производительности.

Вы можете сделать это в языках с более тяжелыми потоками, но в spawn и join может быть слишком многонакладные расходы (хотя пул потоков может смягчить некоторые из них).

Answer 2

Извините, если я наивен, но разве вы не просто синхронизируете объекты, чтобы сделать проблемы независимыми?

, например

public Object lock1 = new Object; // access to resource 1 
public Object lock2 = new Object; // access to resource 2

updateStructure1() {
   synchronized( lock1 ) {
      ...
   }
}

updateStructure2() {
   synchronized( lock2 ) {
     ...
   }
}

Answer 3

Я не знаю, есть ли стандартный способ сделать это.Однако я бы реализовал это примерно так:

do
{
    if (!updatedA && mutexA.tryLock())
    {
        scope(exit) mutexA.unlock();
        updateA();
        updatedA = true;
    }

    if (!updatedB && mutexB.tryLock())
    {
        scope(exit) mutexB.unlock();
        updateB();
        updatedB = true;
    }
}
while (!(updatedA && updatedB));

Некоторое умное метапрограммирование, вероятно, могло бы сократить повторение, но я оставляю это как упражнение для вас.

Answer 4

Нет стандартного решения, а есть класс стандартных решений в зависимости от ваших потребностей.

http://en.wikipedia.org/wiki/Scheduling_algorithm

Answer 5

Я не знаю никакого "стандартного" способа сделать это, извините.Так что это ниже - просто ThreadGroup, абстрагированная Swarm -классом, которая «взламывает» список заданий до тех пор, пока все не будет сделано, в стиле циклического перебора и гарантирует, что будет использовано как можно больше потоков.Я не знаю, как это сделать без списка заданий.

Отказ от ответственности: я очень плохо знаком с D и программированием параллелизма, поэтому код довольно любительский.Я видел это больше как забавное упражнение.(Я тоже имею дело с некоторыми вещами параллелизма.) Я также понимаю, что это не совсем то, что вы ищете.Если у кого-нибудь есть указатели, я бы хотел их услышать!

import  core.thread,
        core.sync.mutex,
        std.c.stdio,
        std.stdio;

class Swarm{
    ThreadGroup group;
    Mutex mutex;
    auto numThreads = 1;
    void delegate ()[int] jobs;
    this(void delegate()[int] aJobs, int aNumThreads){
        jobs = aJobs;
        numThreads = aNumThreads;
        group = new ThreadGroup;
        mutex = new Mutex();
    }
    void runBlocking(){
        run();
        group.joinAll();
    }
    void run(){
        foreach(c;0..numThreads)
            group.create( &swarmJobs );
    }
    void swarmJobs(){
        void delegate () myJob;
        do{
            myJob = null;
            synchronized(mutex){
                if(jobs.length > 0)
                    foreach(i,job;jobs){
                        myJob = job;
                        jobs.remove(i);
                        break;
                    }
            }
            if(myJob)
                myJob();
        }while(myJob)
    }
}
class Jobs{
    void job1(){
        foreach(c;0..1000){
            foreach(j;0..2_000_000){}
            writef("1");
            fflush(core.stdc.stdio.stdout);
        }
    }
    void job2(){
        foreach(c;0..1000){
            foreach(j;0..1_000_000){}
            writef("2");
            fflush(core.stdc.stdio.stdout);
        }
    }
}
void main(){
    auto jobs = new Jobs();
    void delegate ()[int] jobsList = 
         [1:&jobs.job1,2:&jobs.job2,3:&jobs.job1,4:&jobs.job2];
    int numThreads = 2;
    auto swarm = new Swarm(jobsList,numThreads);
    swarm.runBlocking();
    writefln("end");
}

Answer 6

Насколько мне известно, нет стандартного способа сделать это, и вам придется испачкать руки.

Чтобы перефразировать ваши требования, у вас есть набор структур данных, и вам нужно работать над ними, но не в каком-то определенном порядке. Вы только хотите заблокировать ожидание в структуре данных, если все другие объекты заблокированы. Вот псевдокод, на котором я бы основал свое решение:

work = unshared list of objects that need updating
while work is not empty:
    found = false
    for each obj in work:
        try locking obj
        if successful:
            remove obj from work
            found = true
            obj.update()
            unlock obj
    if !found:
        // Everything is locked, so we have to wait
        obj = randomly pick an object from work
        remove obj from work
        lock obj
        obj.update()
        unlock obj

Обновляющий поток будет блокироваться только в том случае, если обнаружит, что все необходимые ему объекты заблокированы. Затем он должен ждать чего-то, поэтому он просто выбирает один и блокирует его. В идеале, он выберет объект, который будет разблокирован как можно раньше, но не существует простого способа сказать это.

Также возможно, что объект может стать свободным, когда программа обновления находится в цикле try , поэтому программа обновления будет пропускать ее. Но если объем выполняемой вами работы достаточно велик по сравнению с затратами на итерацию по этому циклу, ложный конфликт должен быть редким, и он будет иметь значение только в случаях чрезвычайно высокой конкуренции.