pluggablealgorithm.java

// Hi there! I just want to let you know that I have neither compiled nor run
// this code.

EventLoopGroup workerGroup = new NioEventLoopGroup(32, new CustomEventExecutorChooser());

public interface EventExecutorChooser {
    void addExecutor(EventExecutor executor)
    MetricsProvider newMetricsProvider();
    EventExecutor next();
}

public abstract class AbstractEventExecutorChooser implements EventExecutorChooser {
    protected final CopyOnWriteArrayList<EventExecutor> executors;
    
    @Override
    public void addExecutor(EventExecutor executor) {
        executors.add(executor);
    }
    
    @Override
    public MetricsCollector newMetricsProvider() {
        // it makes sense to return a metrics collector that only collects
        // the metrics which are ultimately used by the algorithm and have
        // the rest of the operations be NOOPs.
        // I believe most EventExecutorChoosers will ultimately use only 1 or 2
        // metrics, so we should provide a factory where the user can specify
        // which metrics he wants to be collected...
        return new AllMetricsProvider();
    }
    
    @Override
    public abstract EventExecutor next();
}
    
public class NioEventLoop ... {
    NioEventLoop(..., MetricsCollector collector) {
        if (collector == null) {
            this.collector = new NoMetricsCollector();
        }
        
        this.collector = collector;
        
        ...
    }
    
    MetricsCollector metrics() {
        return collector;
    }
}

// Having the metrics object attached to the EventLoop will allow us to
// conveniently access it in all the different places required for metric 
// collection.
// e.g.
abstract class AbstractChannel {
    abstract class AbstractUnsafe {
        private void register0(...) {
            
            pipeline.fireChannelRegistered();
            eventLoop.metrics().registerChannel();
        }
    }
}

abstract class AbstractNioByteChannel {
    class NioByteUnsafe {
        public void read() {
            ...
            eventLoop.metrics().readBytes(localReadAmount);
        }
    } 
}

public class MultithreadEventExecutorGroup ... {
    
    public MultithreadEventExecutorGroup(..., EventExecutorChooser chooser) {
        ...
        
        for (int i = 0; i < nThreads; i ++) {
           children[i] = newChild(executor, args, chooser.newMetricsProvider());
            // I am not convinced that this is the best approach
            // it feels a lot like this should be put in a constructor
            // not (yet) sure how tough ... chicken egg problem?
           chooser.addExecutor(children[i]);
        }
        
        ...
    }
}
    
public interface MetricCollector {
    void registerChannel();
    void unregisterChannel();
    
    void readBytes(int readBytes);
    void writeBytes(int writtenBytes);
    
    void queueTask();
    void dequeueTask();
    ...
}

public interface MetricProvider extends MetricCollector {
    // the total number of currently registered channels
    int registeredChannels();
    // A call to registeredChannels(5) would return the number
    // of newly registered channels in the last 5 seconds
    int registeredChannels(int seconds);
    
    long bytesRead();
    long bytesRead(int seconds);
    
    long bytesWritten(); 
    long bytesWritten(int seconds);
    
    int queuedTasks();
    int queuedTasks(int seconds);
    
    int lastRead();
    int lastWrite();
    int lastTaskExecuted();
}

public class RoundRobinEventExecutorChooser extends AbstractEventExecutorChooser {
    private final AtomicInteger childIndex = new AtomicInteger();
    
    @Override
    public MetricsProvider newMetricsProvider() {
        return new NoMetricsProvider();
    }
    
    @Override
    public EventExecutor next() {
        return executors.get(Math.abs(childIndex.getAndIncrement() % executors.size()));
    }
}

// For EventExecutorChoosers that do not make use of the metrics
public final class NoMetricsCollector implements MetricsProvider {
    void registerChannel() {}
    void unregisterChannel() {}
    ...
}

public class AllMetricsCollector implements MetricsProvider {
    private int activeChannels;
    
    void registerChannel() {
        activeChannels++;
    }
    ...
}