Ordering messages from multiple consumers without using synchronized

[Fragle]

My first post here. Hi everyone.

I've got a little challenge for those into concurrent programming. Sometimes I'll find a way to use conccurent collections or atomics to improve concurrency but I've come across a problem where I can't and I was hoping to get an opinion from those much better at it than me.

I want to achieve the equivalent of this but without using synchornized on the accept method
Thanks.

public class SynchronizedSignalOrderer implements SignalReciever {
 
  private long nextSequenceNumber;
  private Map<Long,Signal<?>> delayed = new HashMap<Long, Signal<?>>();
  private final SignalReciever delegate;
 
  public SynchronizedSignalOrderer(SignalReciever delegate, long startSequenceNumber) {
    this.delegate=delegate;
    this.nextSequenceNumber = startSequenceNumber;
  }
 
 @Override
  public synchronized void accept(Signal<?> signal) {
    if (nextSequenceNumber == signal.sequenceNumber()) {
      forward(signal);
      nextSequenceNumber++;
      catchUp();
    } else {
      delay(signal);
    }
 }
 
  private void delay(Signal<?> signal) {
    delayed.put(signal.sequenceNumber(), signal);
  }
 
  private void catchUp() {
    Signal<?> signal = delayed.remove(nextSequenceNumber);
    while(signal != null) {
      forward(signal);
      nextSequenceNumber++;
      signal = delayed.remove(nextSequenceNumber);
    }
  }
 
  private final void forward(Signal<?> signal) {
    delegate.accept(signal);
  }
}

[Norm]

Can you explain why do you not want to use synchronized methods?

Can you edit your code and wrap it in code tags for better readability?
See: BB Code List - Java Programming Forums
or use Go Advanced button and the #icon

[Norm]

Do you have a multithreaded test script to test this code?

[Fragle]

Ok. Done that now (still learning how the forum works!).

I'm trying to avoid the synchronized method because it's not offering much concurreny. There are many calling threads and I don't want them queueing up behind each other just to pop something into the 'queue'.

initially I tried the following

public class ConcurrentSignalOrderer implements SignalReciever {
 
	private final ConcurrentMap<Long,Signal<?>> delayed = new ConcurrentHashMap<Long, Signal<?>>();
	private final AtomicLong atomicSequenceNumber;
	private final SignalReciever delegate;
 
	public ConcurrentSignalOrderer(SignalReciever delegate, long startSequenceNumber) {
		this.delegate = delegate;
		this.atomicSequenceNumber = new AtomicLong(startSequenceNumber);
	}
 
	@Override
	public void accept(Signal<?> signal) {
		long nextSequenceNumber;
		nextSequenceNumber = atomicSequenceNumber.get();
		if (nextSequenceNumber != signal.sequenceNumber()) {
			delay(signal);
		} else {
			forward(signal);
			catchUp(nextSequenceNumber);
		}
	}
 
	private void delay(Signal<?> signal) {
		delayed.put(signal.sequenceNumber(), signal);
	}
 
	private void catchUp(long nextSequenceNumber) {
		nextSequenceNumber++;
		Signal<?> signal = delayed.remove(nextSequenceNumber);
		while(signal != null) {
			forward(signal);
			nextSequenceNumber++;
			signal = delayed.remove(nextSequenceNumber);
		}
		atomicSequenceNumber.set(nextSequenceNumber);
	}
 
  private final void forward(Signal<?> signal) {
    delegate.accept(signal);
  }

This has problems. If a two thread both check the current sequnece number at the same time, one them matching it and the other not, it's possible that item placed in delayed map will never be checked resulting in items will just infinitly building up in delayed. Having understood the proble, here was my next attempt.

public class ConcurrentSignalOrderer implements SignalReciever {
 
	private final ConcurrentMap<Long,Signal<?>> delayed = new ConcurrentHashMap<Long, Signal<?>>();
	private final AtomicLong atomicSequenceNumber;
	private final SignalReciever delegate;
 
	public ConcurrentSignalOrderer(SignalReciever delegate, long startSequenceNumber) {
		this.delegate = delegate;
		this.atomicSequenceNumber = new AtomicLong(startSequenceNumber);
	}
 
	@Override
	public void accept(Signal<?> signal) {
		long nextSequenceNumber;
		nextSequenceNumber = atomicSequenceNumber.get();
		if (nextSequenceNumber != signal.sequenceNumber()) {
			delay(signal);
		} else {
			forward(signal);
			catchUp(nextSequenceNumber);
		}
		checkCurrent();	
}
 
	private void checkCurrent() {
		long nextSequenceNumber = atomicSequenceNumber.get();
		Signal<?> signal = delayed.remove(nextSequenceNumber);
		if (signal != null) {
			forward(signal);
			catchUp(nextSequenceNumber);
		}
	}
 
	private void delay(Signal<?> signal) {
		delayed.put(signal.sequenceNumber(), signal);
	}
 
	private void catchUp(long nextSequenceNumber) {
		nextSequenceNumber++;
		Signal<?> signal = delayed.remove(nextSequenceNumber);
		while(signal != null) {
			forward(signal);
			nextSequenceNumber++;
			signal = delayed.remove(nextSequenceNumber);
		}
		atomicSequenceNumber.set(nextSequenceNumber);
	}
 
  private final void forward(Signal<?> signal) {
    delegate.accept(signal);
  }

This is better becuase the additional checks ensure that items are always evaluated but there is another problem here. It's possible for the 'last' thread to leavea the code with items on the delayed queue, which are ready to be delivered, but this won't happen until another thread enters the code... which may never happen. So at this stage I start thinking about a seperate thread to monitor the delayed queue and I realise I'm moving away from an elegant solutions. Also the whole checkCurrent() things doesn't feel right. So here I am asking if there is an elegant way.

(I'll reply with my test cases in another post)

[Fragle]

Here are my test cases:

I don't think they are two great. I find it quite challenging to write good multi-threaded test cases.

This was was not too bad for my first attempt at the concurrent implementation but the 2nd attempt leaves such a small window for error that it's difficult to catch.

package threading;
 
 
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
 
import org.junit.Test;
 
 
public class SynchronizedSignalOrdererTest {
 
	private CountDownLatch readyLatch;
	private CountDownLatch startLatch;
	private CountDownLatch doneLatch;
 
	class SignalProducer extends Thread {
 
		private final long start;
		private final long end;
		private final SignalReciever reciever;
 
		public SignalProducer(long start, long end, SignalReciever reciever) {
			this.start = start;
			this.end = end;
			this.reciever = reciever;
		}
 
		@Override
		public void run() {
			readyLatch.countDown();
			try {
				startLatch.await();
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				throw new RuntimeException(e);
			}
			for (long i=start;i<=end;i++) {
				reciever.accept(new Signal<String>(i, null));
			}
			doneLatch.countDown();
 
		}
	}
 
	public static class SignalRecorder implements SignalReciever {
 
		private List<Signal<?>> signals = new ArrayList<Signal<?>>();
 
		@Override
		public synchronized void accept(Signal<?> signal) {
			signals.add(signal);
		}
 
		public List<Signal<?>> getSignals() {
			return signals;
		}
 
	}
 
	@Test
	public void testOrdering() {
		SignalRecorder recorder = new SignalRecorder();
		ConcurrentSignalOrderer orderer = new ConcurrentSignalOrderer(recorder,0);
		long[] order = new long[]{0,5,2,1,3,7,4,6,9,8};
		for (long next : order) {
			orderer.accept(new Signal<Object>(next, null));
		}
		Long[] recordedOrder = new Long[recorder.getSignals().size()];
		int i = 0;
		for (Signal<?> signal : recorder.getSignals()) {
			recordedOrder[i++] = signal.sequenceNumber();
		}
		assertArrayEquals(new Long[]{0L,1L,2L,3L,4L,5L,6L,7L,8L,9L},recordedOrder);
	}
 
 
	@Test
	public void testSynchronizationTestCase() throws InterruptedException {
		SignalRecorder recorder = new SignalRecorder();
		int producerCount = 20;
		int dataCount = 1000;
		runProducers(recorder, producerCount, dataCount);
 
		List<Signal<?>> recorded = recorder.getSignals();
		assertEquals(producerCount*dataCount,recorded.size());
		boolean failed = false;
		for (int i=1;i<recorded.size();i++) {
			if ((recorded.get(i-1).sequenceNumber()+1 != recorded.get(i).sequenceNumber())) {
				failed = true;
			}
		}
		assertTrue(failed);
	}
 
	@Test
	public void testConcurrentOrdered() throws InterruptedException {
		SignalRecorder recorder = new SignalRecorder();
		int producerCount = 50;
		int dataCount = 2;
		runProducers(new ConcurrentSignalOrderer(recorder,0), producerCount, dataCount);
 
		List<Signal<?>> recorded = recorder.getSignals();
		assertEquals(producerCount*dataCount,recorded.size());
		for (int i=1;i<recorded.size();i++) {
			assertEquals(recorded.get(i-1).sequenceNumber()+1,recorded.get(i).sequenceNumber());
		}
	}
 
	@Test
	public void testSynchronizedOrdered() throws InterruptedException {
		SignalRecorder recorder = new SignalRecorder();
		int producerCount = 50;
		int dataCount = 2;
		runProducers(new SynchronizedSignalOrderer(recorder,0), producerCount, dataCount);
 
		List<Signal<?>> recorded = recorder.getSignals();
		assertEquals(producerCount*dataCount,recorded.size());
		for (int i=1;i<recorded.size();i++) {
			assertEquals(recorded.get(i-1).sequenceNumber()+1,recorded.get(i).sequenceNumber());
		}
	}
 
	private void runProducers(SignalReciever reciever, int producerCount, int dataCount) throws InterruptedException {
		readyLatch = new CountDownLatch(producerCount);
		startLatch = new CountDownLatch(1);
		doneLatch = new CountDownLatch(producerCount);
 
		for(int i=0;i<producerCount;i++) {
			SignalProducer producer = new SignalProducer(i*dataCount,i*dataCount+dataCount-1,reciever);
			producer.start();
		}
		readyLatch.await();
		startLatch.countDown();
		doneLatch.await();
 
	}
}

Here is some additional code need to make everything compile.

package threading;
 
public class Signal<T> {
	private final long sequenceNumber;
	private final T payload;
 
	public Signal(long sequenceNumber, T payload) {
		super();
		this.sequenceNumber = sequenceNumber;
		this.payload = payload;
	}
	long sequenceNumber() {
		return sequenceNumber;
	}
	T payload() {
		return payload;
	}
}

package threading;
 
public interface SignalReciever {
	void accept(Signal<?> signal);
}

[Norm]

Do you have a multithreaded test script to test this code?
By that I mean: code with a main that will execute and test your code.

[Fragle]

Sorry, I pasted the wrong code into that last post. I've fixed it now. It's a JUnit4 test case. I assume that's OK, I don't tend to use main methods for test code.

[Norm]

Ok, I'll have to leave it then. I only use the main method for my testing.

[Fragle]

I'd be disapointed to lose your input. If you add the following you can run the Junits from a main method. It any test fails you'll get an acception. You might need remove the @Test annotations from SynchronizedSignalOrderer.java

package threading;
 
public class Test {
	public static void main(String[] args) throws InterruptedException {
		SynchronizedSignalOrdererTest test = new SynchronizedSignalOrdererTest();
		test.testSynchronizedOrdered();
		test.testConcurrentOrdered();
 
	}
}

[Norm]

Because of the way I work with forum problems (I put all the classes in one file with inner classes)
I'm not sure if I will be able to put together a program that will compile. I currently get 27 compiler errors with this code.

I got it to compile and execute. It executes very quickly and produces no output. I dummied out all of the assert methods

[Fragle]

Okay. I just checked and it works as long as you need junit4 in the classpath. If you don't have that don't worry

As I explained I am already aware of the flaws in the code, and the test won't always catch the window where race conditions occur so I'm not sure they are that helpful. I was just looking for some input on the design of the code.

Thanks anyway.

[Norm]

Good luck.

[Sean4u]

Why aren't you using PriorityBlockingQueue?

[Fragle]

Could be Interesting. I can't see how I'd use it though. My understanding is that it ensures you take the items with the highest priority, but I wouldn't think it'd preserve order for me.

I suppose I could use it to order the items that are being delays (if I used a revserse order comparator on sequence number), but I'm not sure all the ordering the queue does is going to give me much benefit and the concurrent map.

Or where you thinking of using it some other way?

[Sean4u]

That's exactly how I imagined you would use it. The greatest benefit would be simplifying your concurrent side, since all your sources would be adding Signals using put() - end of story. If they're separate threads, they should be allowed to return quickly to whatever it is they're doing while whatever it is that's consuming the ordered Signals should be doing the bulk of the consuming work in its own thread. PBQ would give you that break between put() and take().

[Fragle]

I can see how that would work if I had a consumer-producer model where a consuming thread was given the responsibility of retrieving in sequence. That's not what I have here though. If the sequencing allows it, the calling thread continues through to 'execute' the signal with very little latency.

Introducing a seperate consuming thread would introduce latency. Also to achieve the same level of concurrency in I'd need multiple consumers so I'd double the number of threads in the application.

[Sean4u]

the calling thread continues through to 'execute' the signal with very little latency.

That's not what you've implemented. What you've implemented doesn't really fit a producer / consumer model. Your producer threads are occasionally tied up 'catching up' the out-of-order signals that other producer threads have placed in the orderer.

Introducing a seperate consuming thread

You wouldn't be - your SignalReciever (sic) is presumably idle while it's waiting for one of the producer threads to feed it Signals.

would introduce latency

There is obviously some latency incurred by data crossing thread contexts, but your current solution ties up one of your producers - that's not being done for free.

Also to achieve the same level of concurrency

As what? You're serialising data from N sources. Ideally you'd have N+1 threads. Serialisation is ... serial - it's single-threaded. Your SignalReciever is a separate thread which must currently be waiting for one of the producers to 'push' its consumption.

I'd need multiple consumers so I'd double the number of threads in the application.

Uh ... no.

At the end of the day this is your implementation - but you did ask. If you are really concerned about latency, you should have a high-throughput, long-lived test plan. What kind of throughput over what duration are you expecting? If this is for anything other than a school assignment and you're not a concurrency god (I don't know any), you should test a few alternative implementations.

[Fragle]

So I agree with most of what you're saying and I'm not trying to defend the current code I have. It has a hole in it which I don't know how to fix and probably more I don't know about. I think I've done a bad job of explaining the problem I am trying to solve so I will try to make it clearer.

A diagram will probably help.
SignalProcessing.jpg

So I'm showing only 3 processing 'streams' here but imagine there could be many more. On the left we have data arriving on sockets, the coloured circles. The square boxes are instances of classes and the arrows indicate method calls.

There are 3 key design objectives directing the design.
1) Latency - the time t that it takes from when data arrives at the socket to making it available to all signal processors should be a small as possible.
2) The order in which messages arrive on any of the streams should be preserved as accurately as possible.
3) All Signal processors must recieve exactly the same signals and in exectly the same order.

It's also worth highlighting that the most of the time there will not be much concurrency but that latency is still very imporant.

So my reasoning so far, and please tell me if I'm mistaken, is that to try and reduce the time t to a minumum it is best to have a single thread perform the read, convert, sequence and splitting. Any handover between threads and synchronization is a cost which I want to keep to a mininum. I also need to avoid blocking any of the threads becuase they really should get back to reading from the socket so that a sequence number can be assigned to the data as quickly as possible. This gives us the best chance of preserving the real-time order in which data arrives.

I have been testing the code which has lead to the reasoning above and so far the least latency comes from simply synchronzing the accept method on the sequencer. Plus I know it be logically correct.

 @Override
  public synchronized void accept(Signal<?> signal) {
    if (nextSequenceNumber == signal.sequenceNumber()) {
      forward(signal);
      nextSequenceNumber++;
      catchUp();
    } else {
      delay(signal);
    }
 }

The latancy appears to shoot up when I add any sort of queue which I beleive is due to the hand-off between threads.

So really we should forget my shabby attempt at a concurrent sequence and my question for the forum is: Is there a more performant way than using a synchronized method. Maybe there is a better way of organizing the logic?

[Sean4u]

Your best solution for preserving the order of arriving data would be to use the same socket - is that not possible? Using separate threads and sockets - unless you have geological time passing between signals arriving - is a great way to disrupt the order of arriving signals. Is there no way to establish a sequence earlier - you can't use a reference clock at the signal sources, so that they create Signal packets with time stamps? That would probably work out to be no worse than many-threads-listening, and it might give you a physical time-ordering rather than a rather arbitrary time-order-data-arrived-someplace-on-my-network.

Where are the slow network paths in this? Some solutions are better or worse depending on which parts of your logic diagram are on the same host and which are divided by comparatively high-latency networking. Speculating on which solutions are best in the absence of that kind of information is like playing Twenty Questions.

simply synchronzing the accept method

I have a feeling that somewhere in the API docs or the JLS is an explanation of what happens to waiting threads when the first thread exits the synchronized method and the explanation is that the choice of which blocked thread next gets access to the synchronized method is not first-come-first-served. If what you say about there being little concurrency in your application is true, then you may simply be getting lucky with a synchronized method and your Signal order is preserved by your data being - like you say - not very concurrent. If a lot of data is arriving at about the same time, and your goal is to preserve the order of arriving data, then using a synchronized Java method might be getting you something else altogether. Doesn't 'starvation' refer to the situation where a thread might wait a long time before getting access to the synchronized method?

[Fragle]

I beleive I'm still failing to get the problem across. I'm not saying there is little concurrency. I'm saying that the most optimal and correct I have found is to use the synchronized method approach and avoid any handoff. What I'm asking is whether there is a better. It's not a newbie question.

I know how synchronized works and if you look at the code I've posted I am not relying on the wait order at all. It's not getting lucky. Its the more concurrent code (which uses an atomic and concurrent hashmap) which is getting lucky 99.99% of the time. Using that code the flow performs much faster under heavy concurrent load but tests are showing there is this tiny window where I can be left with an item in the queue when the processing comes to an end.

I did think about time stamping but that creates a problem later on when you try to serialize beceause of the gaps in timestamps. If the sequencer recieves a message with a timestamp T, and then recieves a messge with timestamp T+10, it doesn't know if this message has overtaken another message with timestamp T+5. I beleive the atomic long is doing a very good job of assigning sequencial numbers (with no gaps) and the concurrency is great so that part of the design looks good for now.

The are no network paths in the flow I'm trying to optimize. The problem is all in a single process.