path: root/src/org/gnunet/util/Scheduler.java

/*
     This file is part of GNUnet.
     (C) 2009 Christian Grothoff (and other contributing authors)

     GNUnet is free software; you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published
     by the Free Software Foundation; either version 2, or (at your
     option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     General Public License for more details.

     You should have received a copy of the GNU General Public License
     along with GNUnet; see the file COPYING.  If not, write to the
     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
     Boston, MA 02111-1307, USA.
 */

package org.gnunet.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;

/**
 * Schedule computations using continuation passing style.
 *
 * @author Florian Dold
 */
public class Scheduler {
    private static final Logger logger = LoggerFactory
            .getLogger(Scheduler.class);

    // only valid while a task is executing
    private static TaskConfiguration activeTask = null;

    // number of tasks in the ready queue
    private static int readyCount = 0;

    // for every priority, there is a list of tasks that is definitely ready to run
    final private static ArrayList<LinkedList<TaskConfiguration>> ready = new ArrayList<LinkedList<TaskConfiguration>>
            (Priority.size);

    static {
        for (int i = 0; i < Priority.size; ++i) {
            ready.add(new LinkedList<TaskConfiguration>());
        }
    }

    private static final int EVENT_READ = 0, EVENT_WRITE = 1, EVENT_ACCEPT = 2, EVENT_CONNECT = 3;
    private static final int[] eventToInterestOp = new int[]{SelectionKey.OP_READ, SelectionKey.OP_WRITE,
            SelectionKey.OP_ACCEPT, SelectionKey.OP_CONNECT};
    private static final Reason[] eventToReason = new Reason[]{Reason.READ_READY, Reason.WRITE_READY,
            Reason.ACCEPT_READY, Reason.CONNECT_READY};


    private static Selector selector = null;


    static {
        try {
            selector = SelectorProvider.provider().openSelector();
        } catch (final IOException e) {
            // what to do here?
            logger.error("fatal: cannot create selector");
            System.exit(-1);
        }
    }


    public enum Priority {
        IDLE, BACKGROUND, DEFAULT, HIGH, UI, URGENT, SHUTDOWN;
        private static final int size = Priority.values().length;
        public static final Priority KEEP = null;
    }

    public enum Reason {
        STARTUP, SHUTDOWN, TIMEOUT, READ_READY, WRITE_READY, ACCEPT_READY, CONNECT_READY
    }

    /**
     * The context of a task that is ready to run.
     */
    public static class RunContext {
        /**
         * The reason this task has been called by the scheduler.
         */
        Set<Reason> reasons = EnumSet.noneOf(Reason.class);

        public RunContext() {
        }

        public RunContext(Set<Reason> reasons) {
            this.reasons = reasons;
        }
    }

    /**
     * A task is the basic unit of work that is managed by the scheduler.
     */
    public static interface Task {
        public void run(RunContext ctx);
    }

    /**
     * A TaskConfiguration represents a Task that will execute or has already been executed.
     */
    public static class TaskConfiguration implements Cancelable {
        private final Task task;
        private RunContext ctx = new RunContext();
        private boolean lifeness = true;
        private Priority priority;
        private final AbsoluteTime deadline;

        private ArrayList<SelectableChannel> eventChannels = null;
        private ArrayList<Integer> eventTypes = null;

        private boolean hasRun = false;
        private boolean isCanceled = false;

        /**
         * Create a TaskIdentifier.
         *
         * @param task
         */
        TaskConfiguration(RelativeTime delay, Task task) {
            this.task = task;
            this.deadline = delay.toAbsolute();
        }

        /**
         * Create a light-weight task identifier that is not registered as pending in the Scheduler,
         * used for continuations.
         *
         * @param t   task
         * @param ctx the RunContext
         */
        TaskConfiguration(Task t, RunContext ctx) {
            this.task = t;
            this.ctx = ctx;
            this.deadline = AbsoluteTime.ZERO;
            this.priority = (activeTask == null) ? Priority.DEFAULT : activeTask.priority;
            this.lifeness = true;
        }

        private void addChannelEvent(SelectableChannel channel, int eventType) {
            if (channel == null) {
                throw new AssertionError("channel must be non-null");
            }
            if (eventChannels == null) {
                eventChannels = new ArrayList<SelectableChannel>();
                eventTypes = new ArrayList<Integer>();
            }
            eventChannels.add(channel);
            eventTypes.add(eventType);

            int interestOp = eventToInterestOp[eventType];

            SelectionKey key = channel.keyFor(selector);
            if (key == null || !key.isValid()) {
                try {
                    key = channel.register(selector, interestOp, new TaskConfiguration[4]);
                } catch (ClosedChannelException e) {
                    throw new IOError(e);
                }
            } else {
                if ((key.interestOps() & interestOp) != 0) {
                    throw new AssertionError("interest op registered twice");
                }
                key.interestOps(key.interestOps() | interestOp);
            }

            TaskConfiguration[] subscribers = (TaskConfiguration[]) key.attachment();
            if (subscribers[eventType] != null) {
                throw new AssertionError("subscriber registered twice");
            }
            subscribers[eventType] = this;

            if (subscribers[EVENT_CONNECT] != null && subscribers[EVENT_READ] != null) {
                throw new AssertionError("OP_CONNECT and OP_READ are incompatible in java");
            }
        }

        private void run() {
            if (hasRun) {
                throw new AssertionError("same task ran twice");
            }
            if (isCanceled) {
                return;
            }
            TaskConfiguration old = activeTask;
            activeTask = this;
            task.run(ctx);
            hasRun = true;
            activeTask = old;
        }

        public void cancel() {
            if (isCanceled) {
                throw new AssertionError("task canceled twice");
            }
            isCanceled = true;
            pending.remove(this);
        }

        public Cancelable schedule() {
            if (priority == null) {
                if (activeTask != null) {
                    priority = activeTask.priority;
                } else {
                    priority = Priority.DEFAULT;
                }
            }
            pending.add(this);
            return this;
        }

        private void deregister() {
            if (eventChannels == null) {
                return;
            }
            for (int i = 0; i < eventChannels.size(); ++i) {
                SelectionKey key = eventChannels.get(i).keyFor(selector);
                TaskConfiguration[] subscribers = (TaskConfiguration[]) key.attachment();
                int interestOp = eventToInterestOp[eventTypes.get(i)];
                if (subscribers[eventTypes.get(i)] == null || (key.interestOps() | interestOp) == 0) {
                    throw new AssertionError("deregistering event that has not been registered");
                }
                subscribers[eventTypes.get(i)] = null;
                key.interestOps(key.interestOps() & (~interestOp));
            }
        }

        public void selectRead(SelectableChannel channel) {
            addChannelEvent(channel, EVENT_READ);
        }

        public void selectWrite(SelectableChannel channel) {
            addChannelEvent(channel, EVENT_WRITE);
        }

        public void selectConnect(SelectableChannel channel) {
            addChannelEvent(channel, EVENT_CONNECT);
        }

        public void selectAccept(SelectableChannel channel) {
            addChannelEvent(channel, EVENT_ACCEPT);
        }
    }

    // tasks that are waiting for an event, which are executed anyway after the deadline has occurred
    final private static Queue<TaskConfiguration> pending = new PriorityQueue<TaskConfiguration>(5, new Comparator
            <TaskConfiguration>() {
        @Override
        public int compare(TaskConfiguration a, TaskConfiguration b) {
            return a.deadline.compareTo(b.deadline);
        }
    });


    public static boolean getCurrentLifeness() {
        return (activeTask == null) || activeTask.lifeness;
    }

    /**
     * Run the task regardless of any prerequisites, before any other task of
     * the same priority.
     */
    public static void addContinuation(Task task,
                                       EnumSet<Reason> reasons) {
        RunContext ctx = new RunContext();
        ctx.reasons = reasons;
        queueReady(new TaskConfiguration(task, ctx));
    }

    /**
     * Schedule a new task to be run as soon as possible. The task will be run
     * with the priority of the calling task.
     *
     * @param task main function of the task
     * @return unique task identifier for the job only valid until "task" is
     *         started!
     */
    public static Cancelable add(Task task) {
        return addDelayed(RelativeTime.ZERO, task);
    }


    /**
     * Add a task to run after the specified delay.
     *
     * @param delay time to wait until running the task
     * @param task  the task to run after delay
     * @return the TaskIdentifier, can be used to cancel the task until it has been executed.
     */
    public static TaskConfiguration addDelayed(RelativeTime delay, Task task) {
        TaskConfiguration tid = new TaskConfiguration(delay, task);
        tid.schedule();
        return tid;
    }

    public static TaskConfiguration addRead(RelativeTime timeout,
                                            SelectableChannel chan, Task task) {
        TaskConfiguration tid = new TaskConfiguration(timeout, task);
        tid.addChannelEvent(chan, EVENT_READ);
        tid.schedule();
        return tid;
    }

    public static TaskConfiguration addWrite(RelativeTime timeout,
                                             SelectableChannel chan, Task task) {
        TaskConfiguration tid = new TaskConfiguration(timeout, task);
        tid.addChannelEvent(chan, EVENT_WRITE);
        tid.schedule();
        return tid;
    }

    /**
     * Check if the system is still life. Trigger disconnect if we have tasks, but
     * none of them give us lifeness.
     *
     * @return true to continue the main loop, false to exit
     */
    private static boolean checkLiveness() {
        if (readyCount > 0) {
            return true;
        }
        for (TaskConfiguration t : pending) {
            if (t.lifeness) {
                return true;
            }
        }
        // trigger shutdown if we still have pending tasks, but none of them has lifeness
        if (!pending.isEmpty()) {
            logger.debug("tasks pending but not alive -- disconnect");
            shutdown();
            return true;
        }

        return false;
    }


    /**
     * Queue a Task for execution.
     *
     * @param tid TaskIdentifier of the ready task
     */
    private static void queueReady(TaskConfiguration tid) {
        int idx = tid.priority.ordinal();
        ready.get(idx).add(tid);
        readyCount++;

        pending.remove(tid);
    }


    /**
     * Queue all tasks with expired timeout.
     *
     * @return the minimum time to wait until the next timeout expiry
     */
    private static RelativeTime handleTimeouts() {
        RelativeTime timeout = RelativeTime.FOREVER;

        // check if any timeouts occurred
        while (true) {
            TaskConfiguration t = pending.peek();
            if (t == null) {
                break;
            }
            RelativeTime remaining = t.deadline.getRemaining();
            if (remaining.getMilliseconds() <= 0) {
                t.deregister();
                t.ctx.reasons = EnumSet.of(Reason.TIMEOUT);
                queueReady(t);
            } else {
                timeout = remaining;
                break;
            }
        }
        return timeout;
    }

    private static void addSubscriberTask(Collection<TaskConfiguration> executableTasks,
                                          TaskConfiguration[] subscribers, int eventType) {
        if (subscribers[eventType] == null) {
            return;
        }
        executableTasks.add(subscribers[eventType]);
        subscribers[eventType].ctx.reasons.add(eventToReason[eventType]);
    }

    private static void handleSelect(RelativeTime timeout) {
        try {
            // selector.select(0) would block indefinitely (counter-intuitive, java's fault)
            if (timeout.getMilliseconds() == 0) {
                selector.selectNow();
            } else if (timeout.isForever()) {
                selector.select(0);
            } else {
                selector.select(timeout.getMilliseconds());
            }
        } catch (IOException e) {
            throw new IOError(e);
        }

        // we have to do this so we don't execute any task twice
        Collection<TaskConfiguration> executableTasks = new HashSet<TaskConfiguration>();
        for (SelectionKey sk : selector.selectedKeys()) {
            TaskConfiguration[] subscribers = (TaskConfiguration[]) sk.attachment();

            if (sk.isReadable()) {
                addSubscriberTask(executableTasks, subscribers, EVENT_READ);
            }
            if (sk.isWritable()) {
                addSubscriberTask(executableTasks, subscribers, EVENT_WRITE);
            }
            if (sk.isAcceptable()) {
                addSubscriberTask(executableTasks, subscribers, EVENT_ACCEPT);
            }
            if (sk.isConnectable()) {
                addSubscriberTask(executableTasks, subscribers, EVENT_CONNECT);
            }

        }
        for (TaskConfiguration tt : executableTasks) {
            // cancel subscriptions to other events, we can execute now!
            tt.deregister();
            queueReady(tt);
        }
    }


    /**
     * Initialize and run scheduler. This function will return when all tasks
     * have completed.
     */
    public static void run() {
        run(null);
    }


    /**
     * Initialize and run scheduler. This function will return when all tasks
     * have completed.
     *
     * @param initialTask the initial task to run immediately
     */
    public static void run(Task initialTask) {
        if (initialTask != null) {
            addContinuation(initialTask, EnumSet.of(Reason.STARTUP));
        }

        // the gnunet main loop
        while (checkLiveness()) {
            RelativeTime nextTimeout = handleTimeouts();
            if (nextTimeout.getMilliseconds() < 0) {
                logger.warn("negative timeout for select");
            }

            // don't select if there are no tasks; we are done!
            if (readyCount == 0 && pending.isEmpty()) {
                return;
            }

            // don't block in select if we have tasks ready to run!
            if (readyCount > 0) {
                handleSelect(RelativeTime.ZERO);
            } else {
                handleSelect(nextTimeout);
            }

            runReady();
        }
    }


    /**
     * Execute tasks until there either
     * <ul>
     * <li>there are no ready tasks</li>
     * <li>there is a pending task (which may be of higher priority)</li>
     * </ul>
     */
    private static void runReady() {
        do {
            if (readyCount == 0) {
                return;
            }
            // start executing from the highest priority down to 0
            for (int p = Priority.size - 1; p >= 0; p--) {
                // execute all tasks with priority p
                LinkedList<TaskConfiguration> queue = ready.get(p);
                while (!queue.isEmpty()) {
                    TaskConfiguration tid = queue.removeFirst();
                    readyCount--;
                    tid.run();
                }
            }
        } while (pending.size() == 0);

    }

    /**
     * Request the disconnect of a scheduler. Marks all currently pending tasks as
     * ready because of disconnect. This will cause all tasks to run (as soon as
     * possible, respecting priorities and prerequisite tasks). Note that tasks
     * scheduled AFTER this call may still be delayed arbitrarily.
     */
    public static void shutdown() {
        // queueReady() while iterating would yield concurrent modification exn otherwise
        for (TaskConfiguration tid : new ArrayList<TaskConfiguration>(pending)) {
            tid.ctx.reasons.add(Reason.SHUTDOWN);
            queueReady(tid);
        }
        pending.clear();
    }


    /**
     * A handle to a file system object that can be selected on.
     */
    public static class FilePipe {
        private FilePipeThread filePipeThread;

        private FilePipe(FilePipeThread filePipeThread) {
            this.filePipeThread = filePipeThread;
        }

        public Pipe.SourceChannel getSource() {
            return filePipeThread.pipe.source();
        }

    }

    private static class FilePipeThread extends Thread {
        public File file;
        public Pipe pipe;

        FilePipeThread(File file) {
            this.file = file;
            try {
                pipe = SelectorProvider.provider().openPipe();
                pipe.source().configureBlocking(false);
                pipe.sink().configureBlocking(false);
            } catch (IOException e) {
                throw new RuntimeException("selector provider has no pipes");
            }
        }

        @Override
        public void run() {
            // has to be done in thread, blocks if file is a fifo
            FileChannel fileChannel;

            try {
                FileInputStream stream;
                stream = new FileInputStream(file);
                fileChannel = stream.getChannel();
            } catch (FileNotFoundException e) {
                throw new IOError(e);
            }

            ByteBuffer buffer = ByteBuffer.allocate(256);

            boolean quit = false;

            try {
                while (!quit) {
                    buffer.clear();

                    fileChannel.read(buffer);

                    buffer.flip();

                    pipe.sink().write(buffer);
                }
            } catch (IOException e) {
                quit = true;
            }

        }
    }

    public static FilePipe openFilePipe(File file) {
        FilePipeThread fpt = new FilePipeThread(file);
        fpt.setDaemon(true);
        fpt.start();
        return new FilePipe(fpt);
    }

    public static class AsyncProcess {
        // getIn, getOut, getErr

    }

    public static AsyncProcess openAsyncProcess(/*...*/) {
        throw new UnsupportedOperationException("not implemented yet");
    }
}