Event-driven programming and concurrency are not always thought of as relating to the same problem, namely, scaling up system throughput. In a recent tutorial, Towards a Timely, Well-Balanced, Event-Driven Architecture, Lorenzo Puccetti illustrates how event-based programming and concurrency can be used to effectively scale a system. Noting that:

If we could assume that any type of event is consumed as fast as it can be produced, there would be no need to separate "event submission" from "event handling." In fact, it is the inability to recognize this need that leads to bottlenecks that become only visible when the system is tested under a heavy load.

In order to provide such a scalable solution, Puccetti proposes:

An implementation [that] translates calls from dispatchers into timely, well-regulated calls to the receiver, so that if the dispatcher's push method is called several times in quick succession, it translates into just one call to the receiver. We call this implementation the TimelyDispatcher...

An embedded thread is responsible for monitoring the activity of the push method. When the method is invoked, we register the time of the call and update a thread-safe queue. We use a LinkedBlockingQueue so that as items are added to it, the embedded thread can take them off and store them into a private (thread-confined) list. This list's lifecycle is add-add-add-... , snapshot (toArray), and empty...

The embedded thread encapsulates the logic that determines when to invoke the receiver's received method by checking on the current nano time against the values stored in the atomic timeLastAction and thread-private timeLastNotification variables.

Puccetti notes this event-based approach differs from a more traditional consumer-producer pattern:

In the Producer/Consumer pattern, events represent self-contained units of work (tasks) that can be tackled by any consumer. In its simplest forms, scalability is achieved by adding more consumers. The developer's task is therefore to identify highly homogeneous, parallelizable activities. On the other hand, in the dispatcher/receiver framework, events represent the trigger for generally expensive activities to be performed. The time to perform those activities cannot be reduced and the separation between event production and consumption is there to prevent requests from executing more activities than the JVM would be able to sustain. In a way, the two approaches are orthogonal to each other, and in some circumstances it is also possible to combine them together--for example, the same class could act as receiver and producer at the same time.

What do you think of Puccetti's approach to concurrent event handling?

public void push(T event) throws NullPointerException

> public void push(T event) throws NullPointerException
> 

public void push(EventIterator it)

public interface Dispatcher<T>
{
    public void push(T event); 
    public void push(ArrayList<T> events); 
}
 
public interface Receiver<T>
{
   public void received(ArrayList<T> events);
}

> public interface Dispatcher<T>
> {
>     public void push(T event); 
>     public void push(ArrayList<T> events); 
> }
> 
> public interface Receiver<T>
> {
>    public void received(ArrayList<T> events);
> }
> 

java version "1.6.0"
Java(TM) SE Runtime Environment (build 1.6.0-b105)
Java HotSpot(TM) Client VM (build 1.6.0-b105, mixed mode)

AMD Turion 64 Mobile 1.99 GHz
Windows XP SP2

import java.io.*;
import java.util.*;
 
public class Test
{
    public static final int INNER = 1000000;
    
    static final Time time = new ConcreteTime();
    static final ConcreteTime concrete = new ConcreteTime();
 
    public static void main(String[] args) throws IOException
    {
        long timeA = 0;
        long timeB = 0;
        int tests = Integer.valueOf(args[0]);
 
        for (int i = 0; i < tests; i++) {
            if (i % 2 == 0) {
                timeA += test('A');
                timeB += test('B'); 
            } else {
                timeB += test('B');
                timeA += test('A');
            }
        }
 
        System.err.println("A: " + timeA);
        System.err.println("B: " + timeB);
 
        long diff = timeA - timeB;
        double diffPerTest = ((double) diff) / (INNER * tests);
 
        System.err.println("difference: " + diff + " nanoseconds");
        System.err.println("difference per test: " + diffPerTest + " nanoseconds");        
        System.err.println("percent: " + (((double) diff / (double) timeB)) * 100);
    }
 
    static long test(char test) throws IOException {
        long time = System.nanoTime();        
        long l = 0;
 
        for (int j = 0; j < INNER; j++) {
            switch (test) {
                case 'A':
                    l += testA();
                    break;
                case 'B':
                    l += testB();
                    break;
            }
        }
 
        System.out.println(l);
 
        time = System.nanoTime() - time;
 
        if (time < 10) throw new RuntimeException("INNER too small");
 
        return time;
    }
 
    public static long testA() throws IOException
    {
        return time.getTime();
    }
 
    public static long testB() throws IOException
    {
        return concrete.getTime();
    }
}
 
interface Time
{
   long getTime();
}
 
class ConcreteTime implements Time
{
   public long getTime()
   {
       return System.currentTimeMillis();
   }
}