3
\$\begingroup\$

I've been working on a non-blocking IO library, and wanted to get some feedback on my generic event loop class. The goal is that this class will manage select() calls as well as other "events" which must occur on the same thread that the IO is processing.

I'm hoping that this can scale to a relatively large size of simultaneous connections, though I don't actually have a target metric as I just am over-engineering a simple telnet library ;-)

The complete source, including unit tests, is available on github net.virtualinfinity.nio

package net.virtualinfinity.nio;

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Provides the functionality of an event loop that can listen to {@link SelectableChannel}, as well as execute events
 * at some point in the future.
 *
 * The { @link EventLoop#run() } method does the work.  This class is thread safe.
 *
 * @author <a href='mailto:Daniel@coloraura.com'>Daniel Pitts</a>
 */
public class EventLoop implements Closeable {
    private final Selector selector;
    private final ExceptionHandler<IOException> handler;
    private final Queue<Event> events = new PriorityQueue<>();
    private volatile boolean running;

    private EventLoop(Selector selector, ExceptionHandler<IOException> handler) {
        this.selector = selector;
        this.handler = handler == null ? (key, e) -> { throw e; } : handler;
    }

    /**
     * Creates an EventLoop with the given exception handler.  If the handler is null, the default exception handler is
     * used, which will re-through the exception.  This is generally not the best behavior, and a more suitable exception
     * handler should be installed that is specific to your use.
     *
     * @param handler the exception handler, or null to use the default handler.
     *
     * @throws IOException if there is an error opening a selector.
     */
    public EventLoop(ExceptionHandler<IOException> handler) throws IOException {
        this(Selector.open(), handler);
    }

    /**
     * Creates an EventLoop instance with the default exception handler.
     *
     * @throws IOException if there is an error opening a selector.
     *
     * @see #EventLoop(ExceptionHandler)
     */
    public EventLoop() throws IOException {
        this(Selector.open(), null);
    }

    /**
     * Runs the event loop, dispatching events and listening to {@link SelectableChannel}
     * @throws IOException
     */
    public void run() throws IOException {
        synchronized (this) {
            if (running) {
                throw new IllegalStateException("Event loop is already running, and is not thread safe");
            }
            running = true;
        }
        try {
            while (running) {
                running = doSelect(timeout(executePendingEvents()));
            }
        } finally {
            synchronized (this) {
                running = false;
            }
        }
    }

    private boolean doSelect(long timeout) throws IOException {
        if (!selector.isOpen()) {
            return false;
        }
        select(timeout);
        executeSelected();

        return selector.isOpen();
    }

    private long timeout(Event nextEvent) {
        return nextEvent != null ? nextEvent.timeRemaining(TimeUnit.MILLISECONDS) : 0;
    }

    /**
     * Calls select on the selector, delegating exception management to the exception handler.
     *
     * @param timeout the timeout parameter to the {@link Selector#select(long)} call.
     *
     * @throws IOException if there is an exception thrown by the exception handler.
     */
    private void select(long timeout) throws IOException {
        try {
            selector.select(timeout);
        } catch (IOException e) {
            handler.handleException(null, e);
        }
    }

    /**
     * Loops through all the selected keys, and executes there Runnable or selected methods.
     * This method delegates exception management to the exception handler.
     *
     * @throws IOException if there is an exception thrown by the exception handler.
     */
    private void executeSelected() throws IOException {
        for (final Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); iterator.hasNext(); ) {
            final SelectionKey key = iterator.next();
            try {
                final Object attachment = key.attachment();
                if (attachment instanceof SelectionKeyHandler) {
                    ((SelectionKeyHandler) attachment).selected();
                } else if (attachment instanceof Runnable) {
                    ((Runnable)attachment).run();
                }
                iterator.remove();
            } catch (IOException e) {
                handler.handleException(key, e);
            }
        }
    }

    /**
     * Dispatches an due events, and returns the timeout until the next event.
     *
     * @return the time in milliseconds until the next event, or 0 if there are no events.
     */
    private Event executePendingEvents() {
        final Collection<Runnable> toRun = new ArrayList<>();
        final Event nextEvent = getReadyToRun(toRun);
        toRun.forEach(Runnable::run);
        return nextEvent;
    }

    private Event getReadyToRun(Collection<Runnable> toRun) {
        synchronized (events) {
            while (events.peek() != null) {
                final Event nextEvent = events.peek();
                final long nextEventTime = nextEvent.timeRemaining(TimeUnit.MILLISECONDS);
                if (nextEventTime > 0) {
                    return nextEvent;
                }
                // It's ready to run, so run it outside of the synchronized block.
                toRun.add(events.poll());
            }
        }

        return null;
    }

    /**
     * Enqueue the given runnable at some time in the future. The Runnable will be executed from the
     * thread that called {@link #run()} on this object.
     *
     * @param runnable the runnable to execute on the event thread.
     */
    public void invokeLater(Runnable runnable) {
        invokeAfter(runnable, 0, TimeUnit.NANOSECONDS);
    }

    /**
     * Enqueue the given runnable to run after a specific point of time.  The Runnable will be executed from the
     * thread that called {@link #run()} on this object.
     *
     * @param runnable the command to run
     * @param absoluteTime the earliest time to run it.
     */
    public void invokeAfter(Runnable runnable, Date absoluteTime) {
        invokeAfter(runnable, absoluteTime.getTime() - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    /**
     * Enqueue the given runnable to run after a specific point of time.  The Runnable will be executed from the
     * thread that called {@link #run()} on this object.
     *
     * @param runnable the command to run
     * @param timeInFuture the amount of time in the future
     * @param timeInFutureUnit the unit that the timeInFuture value is of.
     */
    public void invokeAfter(Runnable runnable, long timeInFuture, TimeUnit timeInFutureUnit) {
        final Event e = new Event(System.nanoTime() + TimeUnit.NANOSECONDS.convert(timeInFuture, timeInFutureUnit), runnable);
        synchronized (events) {
            events.add(e);
            selector.wakeup();
        }
    }

    /**
     * Registers handlers that will be invoked when the channel is selected.  The handler will be invoked from the event
     * loop.
     *
     * @param channel The channel.
     * @param handlers The handlers for the selected channel.
     *
     * @throws ClosedChannelException
     */
    public void registerHandler(SelectableChannel channel, SelectionKeyActions handlers) throws ClosedChannelException {
      handlers.setSelectionKey(doRegister(channel, handlers.interestOps(), handlers));
    }

    /**
     * Registers handlers that will be invoked when the channel is selected.  The handler will be invoked from the event
     * loop.
     *
     * @param channel The channel of interest
     * @param ops The valid operations. {@link SelectionKey}
     * @param handler The handler for the selected channel.
     *
     * @throws ClosedChannelException
     *
     * @see SelectableChannel#register(Selector, int)
     */
    public void registerHandler(SelectableChannel channel, int ops, SelectionKeyHandler handler) throws ClosedChannelException {
        doRegister(channel, ops, handler);
    }

    /**
     * Registers the channel with the selector.
     *
     * @param channel the channel to register
     * @param ops the interestOps.
     * @param handler the handler.
     *
     * @return The corresponding SelectionKey.
     *
     * @throws ClosedChannelException if the channel is closed.
     */
    private SelectionKey doRegister(SelectableChannel channel, int ops, SelectionKeyHandler handler) throws ClosedChannelException {
        return channel.register(selector, ops, handler);
    }

    /**
     * Closes the selector, causing the event loop to terminate.
     */
    @Override
    public void close() throws IOException {
        selector.close();
    }

    /**
     * Priority queue event item.
     */
    private static class Event implements Comparable<Event>, Runnable {
        private final long desiredTimeNanos;
        private final Runnable handler;

        public Event(long desiredTimeNanos, Runnable handler) {
            this.desiredTimeNanos = desiredTimeNanos;
            this.handler = handler;
        }

        @Override
        public int compareTo(Event event) {
            return Long.compare(timeRemainingNanos(), event.timeRemainingNanos());
        }

        @Override
        public void run() {
            handler.run();
        }

        public long timeRemaining(TimeUnit time) {
            return time.convert(timeRemainingNanos(), TimeUnit.NANOSECONDS);
        }

        public long timeRemainingNanos() {
            return desiredTimeNanos - System.nanoTime();
        }
    }
}
\$\endgroup\$
3
\$\begingroup\$

Just some small nitpicks:

public void registerHandler(SelectableChannel channel, SelectionKeyActions handlers) throws ClosedChannelException {
  handlers.setSelectionKey(doRegister(channel, handlers.interestOps(), handlers));
}

The second line is only two-spaced. Make sure that your spacing is consistent through your code.

Don't import the whole package:

import java.util.*;

Even though it saves a lot of lines and typing, it's at the cost of performance and readability. Import each class individually.

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.PriorityQueue;
import java.util.Queue;

Also, I really don't like the sight of this:

Scrollbar

The conventions say that each line is a maximum of 80 characters. You exceed that many times, especially because of a large amount of arguments to a method, or calling a method with a lot of arguments, or maybe some other reason. After formatting:

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

/**
 * Provides the functionality of an event loop that can listen to
 * {@link SelectableChannel}, as well as execute events at some point in the
 * future.
 *
 * The { @link EventLoop#run() } method does the work. This class is thread
 * safe.
 *
 * @author <a href='mailto:Daniel@coloraura.com'>Daniel Pitts</a>
 */
public class EventLoop implements Closeable {
    private final Selector selector;
    private final ExceptionHandler<IOException> handler;
    private final Queue<Event> events = new PriorityQueue<>();
    private volatile boolean running;

    private EventLoop(Selector selector, ExceptionHandler<IOException> handler) {
        this.selector = selector;
        this.handler = handler == null ? (key, e) -> {
            throw e;
        } : handler;
    }

    /**
     * Creates an EventLoop with the given exception handler. If the handler is
     * null, the default exception handler is used, which will re-through the
     * exception. This is generally not the best behavior, and a more suitable
     * exception handler should be installed that is specific to your use.
     *
     * @param handler
     *            the exception handler, or null to use the default handler.
     *
     * @throws IOException
     *             if there is an error opening a selector.
     */
    public EventLoop(ExceptionHandler<IOException> handler) throws IOException {
        this(Selector.open(), handler);
    }

    /**
     * Creates an EventLoop instance with the default exception handler.
     *
     * @throws IOException
     *             if there is an error opening a selector.
     *
     * @see #EventLoop(ExceptionHandler)
     */
    public EventLoop() throws IOException {
        this(Selector.open(), null);
    }

    /**
     * Runs the event loop, dispatching events and listening to
     * {@link SelectableChannel}
     * 
     * @throws IOException
     */
    public void run() throws IOException {
        synchronized (this) {
            if (running) {
                throw new IllegalStateException(
                        "Event loop is already running, and is not thread safe");
            }
            running = true;
        }
        try {
            while (running) {
                running = doSelect(timeout(executePendingEvents()));
            }
        } finally {
            synchronized (this) {
                running = false;
            }
        }
    }

    private boolean doSelect(long timeout) throws IOException {
        if (!selector.isOpen()) {
            return false;
        }
        select(timeout);
        executeSelected();

        return selector.isOpen();
    }

    private long timeout(Event nextEvent) {
        return nextEvent != null ? nextEvent
                .timeRemaining(TimeUnit.MILLISECONDS) : 0;
    }

    /**
     * Calls select on the selector, delegating exception management to the
     * exception handler.
     *
     * @param timeout
     *            the timeout parameter to the {@link Selector#select(long)}
     *            call.
     *
     * @throws IOException
     *             if there is an exception thrown by the exception handler.
     */
    private void select(long timeout) throws IOException {
        try {
            selector.select(timeout);
        } catch (IOException e) {
            handler.handleException(null, e);
        }
    }

    /**
     * Loops through all the selected keys, and executes there Runnable or
     * selected methods. This method delegates exception management to the
     * exception handler.
     *
     * @throws IOException
     *             if there is an exception thrown by the exception handler.
     */
    private void executeSelected() throws IOException {
        for (final Iterator<SelectionKey> iterator = selector.selectedKeys()
                .iterator(); iterator.hasNext();) {
            final SelectionKey key = iterator.next();
            try {
                final Object attachment = key.attachment();
                if (attachment instanceof SelectionKeyHandler) {
                    ((SelectionKeyHandler) attachment).selected();
                } else if (attachment instanceof Runnable) {
                    ((Runnable) attachment).run();
                }
                iterator.remove();
            } catch (IOException e) {
                handler.handleException(key, e);
            }
        }
    }

    /**
     * Dispatches an due events, and returns the timeout until the next event.
     *
     * @return the time in milliseconds until the next event, or 0 if there are
     *         no events.
     */
    private Event executePendingEvents() {
        final Collection<Runnable> toRun = new ArrayList<>();
        final Event nextEvent = getReadyToRun(toRun);
        toRun.forEach(Runnable::run);
        return nextEvent;
    }

    private Event getReadyToRun(Collection<Runnable> toRun) {
        synchronized (events) {
            while (events.peek() != null) {
                final Event nextEvent = events.peek();
                final long nextEventTime = nextEvent
                        .timeRemaining(TimeUnit.MILLISECONDS);
                if (nextEventTime > 0) {
                    return nextEvent;
                }
                // It's ready to run, so run it outside of the synchronized
                // block.
                toRun.add(events.poll());
            }
        }

        return null;
    }

    /**
     * Enqueue the given runnable at some time in the future. The Runnable will
     * be executed from the thread that called {@link #run()} on this object.
     *
     * @param runnable
     *            the runnable to execute on the event thread.
     */
    public void invokeLater(Runnable runnable) {
        invokeAfter(runnable, 0, TimeUnit.NANOSECONDS);
    }

    /**
     * Enqueue the given runnable to run after a specific point of time. The
     * Runnable will be executed from the thread that called {@link #run()} on
     * this object.
     *
     * @param runnable
     *            the command to run
     * @param absoluteTime
     *            the earliest time to run it.
     */
    public void invokeAfter(Runnable runnable, Date absoluteTime) {
        invokeAfter(runnable,
                absoluteTime.getTime() - System.currentTimeMillis(),
                TimeUnit.MILLISECONDS);
    }

    /**
     * Enqueue the given runnable to run after a specific point of time. The
     * Runnable will be executed from the thread that called {@link #run()} on
     * this object.
     *
     * @param runnable
     *            the command to run
     * @param timeInFuture
     *            the amount of time in the future
     * @param timeInFutureUnit
     *            the unit that the timeInFuture value is of.
     */
    public void invokeAfter(Runnable runnable, long timeInFuture,
            TimeUnit timeInFutureUnit) {
        final Event e = new Event(System.nanoTime()
                + TimeUnit.NANOSECONDS.convert(timeInFuture, timeInFutureUnit),
                runnable);
        synchronized (events) {
            events.add(e);
            selector.wakeup();
        }
    }

    /**
     * Registers handlers that will be invoked when the channel is selected. The
     * handler will be invoked from the event loop.
     *
     * @param channel
     *            The channel.
     * @param handlers
     *            The handlers for the selected channel.
     *
     * @throws ClosedChannelException
     */
    public void registerHandler(SelectableChannel channel,
            SelectionKeyActions handlers) throws ClosedChannelException {
        handlers.setSelectionKey(doRegister(channel, handlers.interestOps(),
                handlers));
    }

    /**
     * Registers handlers that will be invoked when the channel is selected. The
     * handler will be invoked from the event loop.
     *
     * @param channel
     *            The channel of interest
     * @param ops
     *            The valid operations. {@link SelectionKey}
     * @param handler
     *            The handler for the selected channel.
     *
     * @throws ClosedChannelException
     *
     * @see SelectableChannel#register(Selector, int)
     */
    public void registerHandler(SelectableChannel channel, int ops,
            SelectionKeyHandler handler) throws ClosedChannelException {
        doRegister(channel, ops, handler);
    }

    /**
     * Registers the channel with the selector.
     *
     * @param channel
     *            the channel to register
     * @param ops
     *            the interestOps.
     * @param handler
     *            the handler.
     *
     * @return The corresponding SelectionKey.
     *
     * @throws ClosedChannelException
     *             if the channel is closed.
     */
    private SelectionKey doRegister(SelectableChannel channel, int ops,
            SelectionKeyHandler handler) throws ClosedChannelException {
        return channel.register(selector, ops, handler);
    }

    /**
     * Closes the selector, causing the event loop to terminate.
     */
    @Override
    public void close() throws IOException {
        selector.close();
    }

    /**
     * Priority queue event item.
     */
    private static class Event implements Comparable<Event>, Runnable {
        private final long desiredTimeNanos;
        private final Runnable handler;

        public Event(long desiredTimeNanos, Runnable handler) {
            this.desiredTimeNanos = desiredTimeNanos;
            this.handler = handler;
        }

        @Override
        public int compareTo(Event event) {
            return Long.compare(timeRemainingNanos(),
                    event.timeRemainingNanos());
        }

        @Override
        public void run() {
            handler.run();
        }

        public long timeRemaining(TimeUnit time) {
            return time.convert(timeRemainingNanos(), TimeUnit.NANOSECONDS);
        }

        public long timeRemainingNanos() {
            return desiredTimeNanos - System.nanoTime();
        }
    }
}
\$\endgroup\$

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.