2
\$\begingroup\$

I posted this on SO and would appreciate any ideas on it's stability.

Much like the old double-buffering algorithm, this uses a List<T> to which are added entries. You can also take the list at any time, it will be atomically replaced with a new empty one.

Both adding and removing is thread safe. You can add from many threads and remove from many threads.

Most of the code is test code - everything after the //TESTING comment. My test consists of creating a number of producer and consumer threads and passing a known number of Widgets through. If all sent items appear on the other side correctly without loss or duplication my test passes.

Can anyone think of more tests?

Has anyone any general thoughts on the code?

public class DoubleBufferedList<T> {
  // Atomic reference so I can atomically swap it through.
  // Mark = true means I am adding to it so momentarily unavailable for iteration.
  private AtomicMarkableReference<List<T>> list = new AtomicMarkableReference<>(newList(), false);

  // Factory method to create a new list - may be best to abstract this.
  protected List<T> newList() {
    return new ArrayList<>();
  }

  // Get and replace the current list.
  public List<T> get() {
    // Atomically grab and replace the list with an empty one.
    List<T> empty = newList();
    List<T> it;
    // Replace an unmarked list with an empty one.
    if (!list.compareAndSet(it = list.getReference(), empty, false, false)) {
      // Failed to replace! 
      // It is probably marked as being appended to but may have been replaced by another thread.
      // Return empty and come back again soon.
      return Collections.<T>emptyList();
    }
    // Successfully replaced an unmarked list with an empty list!
    return it;
  }

  // Grab and lock the list in preparation for append.
  private List<T> grab() {
    List<T> it;
    // We cannot fail so spin on get and mark.
    while (!list.compareAndSet(it = list.getReference(), it, false, true)) {
      // Spin on mark - waiting for another grabber to release (which it must).
    }
    return it;
  }

  // Release the list.
  private void release(List<T> it) {
    // Unmark it - should this be a compareAndSet(it, it, true, false)?
    if (!list.attemptMark(it, false)) {
      // Should never fail because once marked it will not be replaced.
      throw new IllegalMonitorStateException("It changed while we were adding to it!");
    }
  }

  // Add an entry to the list.
  public void add(T entry) {
    List<T> it = grab();
    try {
      // Successfully marked! Add my new entry.
      it.add(entry);
    } finally {
      // Always release after a grab.
      release(it);
    }
  }

  // Add many entries to the list.
  public void add(List<T> entries) {
    List<T> it = grab();
    try {
      // Successfully marked! Add my new entries.
      it.addAll(entries);
    } finally {
      // Always release after a grab.
      release(it);
    }
  }

  // Add a number of entries.
  public void add(T... entries) {
    // Make a list of them.
    add(Arrays.asList(entries));
  }
  // TESTING.
  // How many testers to run.
  static final int N = 10;
  // The next one we're waiting for.
  static final AtomicInteger[] seen = new AtomicInteger[N];
  // The ones that arrived out of order.
  static final Set<Widget>[] queued = new ConcurrentSkipListSet[N];

  static {
    // Populate the arrays.
    for (int i = 0; i < N; i++) {
      seen[i] = new AtomicInteger();
      queued[i] = new ConcurrentSkipListSet();
    }
  }

  // Thing that is produced and consumed.
  private static class Widget implements Comparable<Widget> {
    // Who produced it.
    public final int producer;
    // Its sequence number.
    public final int sequence;

    public Widget(int producer, int sequence) {
      this.producer = producer;
      this.sequence = sequence;
    }

    @Override
    public String toString() {
      return producer + "\t" + sequence;
    }

    @Override
    public int compareTo(Widget o) {
      // Sort on producer
      int diff = Integer.compare(producer, o.producer);
      if (diff == 0) {
        // And then sequence
        diff = Integer.compare(sequence, o.sequence);
      }
      return diff;
    }
  }

  // Produces Widgets and feeds them to the supplied DoubleBufferedList.
  private static class TestProducer implements Runnable {
    // The list to feed.
    final DoubleBufferedList<Widget> list;
    // My ID
    final int id;
    // The sequence we're at
    int sequence = 0;
    // Set this at true to stop me.
    public volatile boolean stop = false;

    public TestProducer(DoubleBufferedList<Widget> list, int id) {
      this.list = list;
      this.id = id;
    }

    @Override
    public void run() {
      // Just pump the list.
      while (!stop) {
        list.add(new Widget(id, sequence++));
      }
    }
  }

  // Consumes Widgets from the suplied DoubleBufferedList
  private static class TestConsumer implements Runnable {
    // The list to bleed.
    final DoubleBufferedList<Widget> list;
    // My ID
    final int id;
    // Set this at true to stop me.
    public volatile boolean stop = false;

    public TestConsumer(DoubleBufferedList<Widget> list, int id) {
      this.list = list;
      this.id = id;
    }

    @Override
    public void run() {
      // The list I am working on.
      List<Widget> l = list.get();
      // Stop when stop == true && list is empty
      while (!(stop && l.isEmpty())) {
        // Record all items in list as arrived.
        arrived(l);
        // Grab another list.
        l = list.get();
      }
    }

    private void arrived(List<Widget> l) {
      for (Widget w : l) {
        // Mark each one as arrived.
        arrived(w);
      }
    }

    // A Widget has arrived.
    private static void arrived(Widget w) {
      // Which one is it?
      AtomicInteger n = seen[w.producer];
      // Don't allow multi-access to the same producer data or we'll end up confused.
      synchronized (n) {
        // Is it the next to be seen?
        if (n.compareAndSet(w.sequence, w.sequence + 1)) {
          // It was the one we were waiting for! See if any of the ones in the queue can now be consumed.
          for (Iterator<Widget> i = queued[w.producer].iterator(); i.hasNext();) {
            Widget it = i.next();
            // Is it in sequence?
            if (n.compareAndSet(it.sequence, it.sequence + 1)) {
              // Done with that one too now!
              i.remove();
            } else {
              // Found a gap! Stop now.
              break;
            }
          }
        } else {
          // Out of sequence - Queue it.
          queued[w.producer].add(w);
        }
      }
    }
  }

  // Main tester
  public static void main(String args[]) {
    try {
      System.out.println("DoubleBufferedList:Test");
      // Create my test buffer.
      DoubleBufferedList<Widget> list = new DoubleBufferedList<>();
      // All running threads - Producers then Consumers.
      List<Thread> running = new LinkedList<>();
      // Start some producer tests.
      List<TestProducer> producers = new ArrayList<>();
      for (int i = 0; i < N; i++) {
        TestProducer producer = new TestProducer(list, i);
        Thread t = new Thread(producer);
        t.setName("Producer " + i);
        t.start();
        producers.add(producer);
        running.add(t);
      }

      // Start the same number of consumers (could do less or more if we wanted to).
      List<TestConsumer> consumers = new ArrayList<>();
      for (int i = 0; i < N; i++) {
        TestConsumer consumer = new TestConsumer(list, i);
        Thread t = new Thread(consumer);
        t.setName("Consumer " + i);
        t.start();
        consumers.add(consumer);
        running.add(t);
      }
      // Wait for a while.
      Thread.sleep(5000);
      // Close down all.
      for (TestProducer p : producers) {
        p.stop = true;
      }
      for (TestConsumer c : consumers) {
        c.stop = true;
      }
      // Wait for all to stop.
      for (Thread t : running) {
        System.out.println("Joining " + t.getName());
        t.join();
      }
      // What results did we get?
      int totalMessages = 0;
      for (int i = 0; i < N; i++) {
        // How far did the producer get?
        int gotTo = producers.get(i).sequence;
        // The consumer's state
        int seenTo = seen[i].get();
        totalMessages += seenTo;
        Set<Widget> queue = queued[i];
        if (seenTo == gotTo && queue.isEmpty()) {
          System.out.println("Producer " + i + " ok.");
        } else {
          // Different set consumed as produced!
          System.out.println("Producer " + i + " Failed: gotTo=" + gotTo + " seenTo=" + seenTo + " queued=" + queue);
        }
      }
      System.out.println("Total messages " + totalMessages);

    } catch (InterruptedException ex) {
      ex.printStackTrace();
    }
  }
}
\$\endgroup\$
  • \$\begingroup\$ What is the specification of a Double Buffered List? I did not find a clear reference for this. \$\endgroup\$ – tb- Feb 28 '13 at 19:01
  • \$\begingroup\$ It's a list you can add to from multiple threads and at any time remove whatever has been collected, atomically replacing it with a fresh one. It is a list with double-buffer like features. Thank you for your interest. \$\endgroup\$ – OldCurmudgeon Feb 28 '13 at 19:05
2
\$\begingroup\$

Old question, new review.... I think, in general, I don't believe you are actually building anything better than what can be done with regular locking..... let me rephrase that to: the risk of spin-looping during an add is a significant drawback over regularly locked systems.

So, while AtomicReferences may in fact be (marginally) faster than a ReentrantLock or a synchronization lock, the CPU cost of a spin while waiting for a 'slow' add(T....values) may in fact be significant.

Another problem is that you may in fact be over-supplying new List instances. For example, you have to create a new List<T>() for every get(), but, if the current list is empty this is a waste. Also, it's a waste if the buffer is locked. This may add up to quite a lot of waste.

While talking about this area of code, to get around Generic warnings, you should stop using Collections.EMPTY_LIST and instead use the type-safe and generics-friendly Collections.emptyList() (See the Javadoc).

So, as an academic exercise, I think your code is reasonable, but unnecessarily cumbersome. I think a simpler lock will suffice, produce better looking code, and provides functionality that you may not even consider now (like a blocking get() that only returns when data is available)

Bottom line is that I don't believe your solution will have a noticable gain over something more simple/traditional. Certainly your solution has a number of complicated and esoteric concepts.... the Mark-flag is challenging, and the spin-lock-wait process is inefficient.

So consider this alternative:

private List<T> mylist = newList();
private final ReentrantLock mylock = new ReentrantLock();

// Factory method to create a new list - may be best to abstract this.
protected List<T> newList() {
    return new ArrayList<>();
}

// Get and replace the current list.
public List<T> get() {
    // Atomically grab and replace the list with an empty one.
    mylock.lock();
    try {
        if (mylist.isEmpty()) {
            return Collections.emptyList();
        }
        final List<T> ret = mylist;
        mylist = newList();
        return ret;
    } finally {
        mylock.unlock();
    }
}

// Add an entry to the list.
public void add(T entry) {
    mylock.lock();
    try {
        mylist.add(entry);
    } finally {
        mylock.unlock();
    }
}

// Add many entries to the list.
public void add(List<T> entries) {
    // Atomically grab and replace the list with an empty one.
    mylock.lock();
    try {
        mylist.addAll(entries);
    } finally {
        mylock.unlock();
    }
}

// Add a number of entries.
public void add(T... entries) {
    // Make a list of them.
    add(Arrays.asList(entries));
}

The above solution can be extended easily (in ways the AtomicMarkedReference cannot), for example, creating a Condition on the Lock would allow you to have a blocking get(), and other neat tricks.

\$\endgroup\$
  • \$\begingroup\$ Thankyou - I like your alternative and you make a good case for it's simplicity. I would move the check for empty outside the locked area but that is just me. \$\endgroup\$ – OldCurmudgeon Nov 21 '13 at 9:03
  • \$\begingroup\$ ... emptyList() fixed. \$\endgroup\$ – OldCurmudgeon Nov 21 '13 at 9:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.