channel migration2.java

// The problem of associating tasks to channels could be solved by
// having the developer tell us, which channel a task belongs to
// this would make filtering the tasks on migration very easy.
// The downside is that it is subject to user errors.
//
// null if it doesn't belong to any particular channel
execute(ch, new Runnable() {
    @Override
    public void run() {
        ch.flush();
    }
});

public class NioEventLoop {
    private ChannelMatcher matcher;
    private NioEventLoop destination;
    private boolean migrateFrom;
    private volatile boolean migrateTo;
    private volatile boolean destinationFinished;
    private volatile Thread thread;
    
    private static class ChannelTask implements Runnable {
        ChannelTask(Channel ch, Runnable task) {
            ...
        }
    
        // returns the channel this task is associated with
        Channel channel();
    
        public void run() {
            task.run();
        }
    }
    
    @Override
    public void execute(Runnable task) {
        execute(null, task);
    }
    
    @Override
    public void execute(Channel ch, Runnable task) {
        ...
        addTask(new ChannelTask(ch, task));
    }
    
    public boolean reregister(NioEventLoop destination, ChannelMatcher matcher) {
        if (migrateFrom || migrateTo)
            return false;
            
        this.migrateFrom = destination.migrateTo = true;
        this.matcher = matcher;
        this.destination = destination;
        return true;
    }
    
    @Override
    public void run() {
        // by having to set the thread variable anyway,
        // the memory barrier comes for free
        thread = Thread.currentThread();
        
        try {
            select();
            runAllTasks();
        } finally {
            thread = null;
            
            if (migrateFrom) {
                migrateChannels0();
            } else if (migrateTo) {
                destinationFinished = true;
            } else {
                executor.execute(this);
            }
        }
    }
    
    private void migrateChannels0() {
        /*
         * 1) filter channels & cancel the corresponding selection keys.
              If the deregister(...) method is reintroduced, call deregister
              instead of cancel(..).
         * 3) spin lock on destination.destinationFinished
         *    to wait for the destination event loop to finish.
         * 4) register the channels to destination selector
         * 5) merge source tasks into destination taskQueues
         * 6) reset fields used for the migration 
         * 7) resubmit both event loops to the execution pool
         */
    }
}