Координация нескольких потоков после N исполнений

Question

У меня есть многопоточный процесс Java/Groovy, который я хочу «синхронизировать» после выполнения N: - общий счетчик уменьшается после каждого выполнения потока - целью является сброс счетчика один раз он достигает 0, пока ни одна другая нить не обращается к нему.

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

public static void main(String[] args) { 
    AtomicInteger counter = new AtomicInteger(Decrementer.MAX_SIZE) 
    ReadWriteLock lock = new ReentrantReadWriteLock() 

    for (int i = 1; i <= 10; i++) { 
     Decrementer d = new Decrementer(counter, lock) 
     new Thread(d).start() 
    } 
} 

public class Decrementer implements Runnable { 

    public final static int MAX_SIZE = 5 
    private ReadWriteLock lock 
    private AtomicInteger counter 

    public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
     this.counter = counter 
     this.lock = lock 
    } 

    public void run() { 
     while (true) { 
      try{ 
       lock.readLock().lock() 
       int current = this.counter.decrementAndGet() 
       System.out.println(Thread.currentThread().getName() + " at counter " + current) 
       Thread.sleep(762) 
      } finally { 
       lock.readLock().unlock() 
      } 

      try { 
       lock.writeLock().lock() 
       int current = this.counter.get() 
       if (current <= 0) { 
        this.counter.set(Decrementer.MAX_SIZE) 
        System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE) 
        Thread.sleep(4217) 
       } 
      } finally { 
       lock.writeLock().unlock() 
      } 
     } 
    } 
} 

которые дают следующий странный результат (с отрицательным значением счетчика), вероятно, из-за отсутствующей «синхронизированной» Проверку значения AtomicInteger.

Thread-3 at counter 2 
Thread-2 at counter 4 
Thread-1 at counter 3 
Thread-4 at counter 1 
Thread-5 at counter 0 
Thread-6 at counter -1 
Thread-7 at counter -2 
Thread-8 at counter -3 
Thread-9 at counter -4 
Thread-10 at counter -5 
Thread-2 reset -5 to 5 
Thread-3 at counter 4 
Thread-4 at counter 2 
Thread-2 at counter 3 
Thread-1 at counter 1 
Thread-5 at counter -3 
Thread-10 at counter -4 
Thread-7 at counter -1 
Thread-6 at counter -2 
Thread-8 at counter 0 
Thread-9 at counter -5 
Thread-9 reset -5 to 5 

Я также видел CountDownLatch и CyclicBarrier классов, но моя цель состоит в том, чтобы не синхронизировать все темы, но для обеспечения сброс счетчика атомный и исключая других модификации других потоков.

Вы видите какую-либо очевидную проблему параллелизма, которую я пропустил в своем коде?

Answer 1

Ваш замок наоборот. Вы приобретаете блокировку чтения, а затем делаете «писать». Это должно работать:

public static void main(String[] args) { 
    AtomicInteger counter = new AtomicInteger(Decrementer.MAX_SIZE); 
    ReadWriteLock lock = new ReentrantReadWriteLock(); 

    for (int i = 1; i <= 10; i++) { 
     Decrementer d = new Decrementer(counter, lock); 
     new Thread(d).start(); 
    } 
} 

static public class Decrementer implements Runnable { 

    public final static int MAX_SIZE = 5; 
    private ReadWriteLock lock; 
    private AtomicInteger counter; 

    public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
     this.counter = counter; 
     this.lock = lock; 
    } 

    public void run() { 
     while (true) { 
      try { 
       lock.writeLock().lock(); 
       int current = this.counter.decrementAndGet(); 
       System.out.println(Thread.currentThread().getName() + " at counter " + current); 
       if (current <= 0) { 
        this.counter.set(Decrementer.MAX_SIZE); 
        System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE); 
        Thread.sleep(4217); 
       } 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } finally { 
       lock.writeLock().unlock(); 
      } 
     } 
    } 
} 

Answer 2

на основе блокировок записи приобретают комментарий, здесь рабочий раствор:

public class Decrementer implements Runnable { 

public final static int MAX_SIZE = 20; 
private ReadWriteLock lock; 
private AtomicInteger counter; 

public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
    this.counter = counter; 
    this.lock = lock; 
} 

public void run() { 

    while (true) { 
     int current 
     try { 
      lock.writeLock().lock(); 
      if (this.counter.compareAndSet(0, Decrementer.MAX_SIZE)) { 
       current = 0; 
       flushWork(current); 
      } else { 
       current = counter.decrementAndGet(); 
       System.out.println(Thread.currentThread().getName() + " decrement " + current); 
      } 
     } finally { 
      lock.writeLock().unlock(); 
     } 

     try { 
      lock.readLock().lock(); 
      doWork(current); 
     } finally { 
      lock.readLock().unlock(); 
     } 
    } 
} 

private void flushWork(int current) { 
    System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE); 
    System.out.println('------------------------------------------------------------------------------------'); 
    Thread.sleep(4217); 
} 

private void doWork(int current) { 
    System.out.println(Thread.currentThread().getName() + " at counter " + current); 
    Thread.sleep(62); 
} 
}