Quadcap Embeddable Server

com/quadcap/util/threads/PeriodicScheduler.java

Go to the documentation of this file.

package com.quadcap.util.threads;

/* Copyright 1997 - 2003 Quadcap Software.  All rights reserved.
 *
 * This software is distributed under the Quadcap Free Software License.
 * This software may be used or modified for any purpose, personal or
 * commercial.  Open Source redistributions are permitted.  Commercial
 * redistribution of larger works derived from, or works which bundle
 * this software requires a "Commercial Redistribution License"; see
 * http://www.quadcap.com/purchase.
 *
 * Redistributions qualify as "Open Source" under  one of the following terms:
 *   
 *    Redistributions are made at no charge beyond the reasonable cost of
 *    materials and delivery.
 *
 *    Redistributions are accompanied by a copy of the Source Code or by an
 *    irrevocable offer to provide a copy of the Source Code for up to three
 *    years at the cost of materials and delivery.  Such redistributions
 *    must allow further use, modification, and redistribution of the Source
 *    Code under substantially the same terms as this license.
 *
 * Redistributions of source code must retain the copyright notices as they
 * appear in each source code file, these license terms, and the
 * disclaimer/limitation of liability set forth as paragraph 6 below.
 *
 * Redistributions in binary form must reproduce this Copyright Notice,
 * these license terms, and the disclaimer/limitation of liability set
 * forth as paragraph 6 below, in the documentation and/or other materials
 * provided with the distribution.
 *
 * The Software is provided on an "AS IS" basis.  No warranty is
 * provided that the Software is free of defects, or fit for a
 * particular purpose.  
 *
 * Limitation of Liability. Quadcap Software shall not be liable
 * for any damages suffered by the Licensee or any third party resulting
 * from use of the Software.
 */

import java.io.ByteArrayOutputStream;
import java.io.PrintWriter;

import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import com.quadcap.util.DList;
import com.quadcap.util.DListItem;
import com.quadcap.util.Debug;
import com.quadcap.util.ListException;

/**
 * This class acts as a scheduler for periodic tasks, which are named and
 * execute repeatedly with a specified interval.  Each task is a
 * <a href="com.quadcap.util.threads.Command.html#_top_">Command</a>
 * object, as used by e.g.,
 * <a href="com.quadcap.util.threads.StreamWorker.html#_top_">
 * StreamWorker</a>.<p>
 *
 * Two flavors of this class are expressed here -- the first class
 * takes a <A href="com.quadcap.util.threads.Stream.html#_top_">Stream</a>
 * parameter.  When each task becomes ready, the Command object is
 * written to the Stream, and it is presumed that a StreamWorker task
 * will be sitting at the other end of the Stream, executing the
 * Commands that come down the pipe.<p>
 *
 * The other flavor of this class takes a context argument, which can
 * be any Java object (i.e., this class doesn't ever look at the context.)
 * When each task becomes ready, it is simply executed within the
 * PeriodicScheduler thread.
 *
 * @author Stan Bailes
 */
public class PeriodicScheduler extends Thread implements DebugMonitor {
    DList                               active          = new DList();
    Stream                              stream          = null;
    Object                              context         = null;
    boolean                             terminate       = false;
    Hashtable                           schedule        = new Hashtable();

    /**
     * Construct a new PeriodicScheduler which will write Commands to the
     * specified Stream.
     *
     * @param stream the Stream used to write Commands when they become
     *          ready.
     */
    public PeriodicScheduler(ThreadGroup group, String name, Object obj) {
        super(group, name);
        if (obj instanceof Stream) {
            this.stream = (Stream)obj;
        } else {
            this.context = obj;
        }
        setDaemon(true);
    }

    /**
     * Add a new task, with the specified name, action, and interval.
     * The task will <b>NOT</b> be executed 'now'; rather its first
     * execution will be scheduled at time 'now + interval'.
     *
     * @param name the name of this task.  Any existing task with the
     *          same name is replaced by this one.
     * @param c the Command action associated with this task.
     * @param interval the interval (in ms) between invocations of this
     *          task.
     */
    public void add(String name, Command c, long interval) {
        //Debug.println(3, "add " + name);
        synchronized (active) {
            ScheduleItem s = (ScheduleItem)schedule.get(name);
            if (s == null) {
                s = new ScheduleItem(name);
                schedule.put(name, s);
            } else {
                remove(s);
            }
            s.command = c;
            s.interval = interval;
            schedule(s);
        }
    }

    /**
     * Add a new task, with the specified name, action, and interval.
     * The task will <b>NOT</b> be executed 'now'; rather its first
     * execution will be scheduled at time 'now + interval'.
     *
     * @param name the name of this task.  Any existing task with the
     *          same name is replaced by this one.
     * @param r the Runnable action associated with this task.
     * @param interval the interval (in ms) between invocations of this
     *          task.
     */
    public void add(String name, Runnable r, long interval) {
        //Debug.println(3, "add " + name);
        synchronized (active) {
            ScheduleItem s = (ScheduleItem)schedule.get(name);
            if (s == null) {
                s = new ScheduleItem(name);
                schedule.put(name, s);
            } else {
                remove(s);
            }
            s.runnable = r;
            s.interval = interval;
            schedule(s);
        }
    }

    void remove(ScheduleItem item) {
        //Debug.println(3, "remove " + item.name);
        synchronized (active) {
            DListItem head = null;
            try {
                head = active.head();
            } catch (ListException e) {
            }
            DListItem d = head;
            if (d != null) do {
                ScheduleItem s = (ScheduleItem)d.obj;
                if (item == s) {
                    active.unlink(d);
                    break;
                }
                d = d.next;
            } while (d != head);
            active.notify();
        }
    }

    /**
     * Add the specified schedule item to the queue.
     *
     * @param item the item to schedule
     */
    void schedule(ScheduleItem item) {
        //Debug.println(6, "schedule " + item.name);
        item.checkDate(new Date());
        synchronized (active) {
            DListItem head = null;
            try {
                head = active.head();
            } catch (ListException e) {
            }
            DListItem d = head;
            if (d == null || item.date.before(((ScheduleItem)d.obj).date)) {
                active.addFront(item);
            } else {
                while (d.next != head) {
                    d = d.next;
                    ScheduleItem s = (ScheduleItem)d.obj;
                    if (item.date.before(s.date)) break;
                }
                active.addBefore(d, item);
            }
            active.notify();
        }
    }

    /**
     * Thread method, runs the main scheduler event loop:  Get the next
     * task to execute (the active list is sorted by execution time) and
     * wait that long.  If another thread puts a new task in the queue,
     * we'll get interrupted, but than's ok.  Then, look at the head of the
     * active list again and, if the task is ready, execute it.
     */
    public void run() {
        //Debug.println(2, "starting: " + getName());
        while (!terminate) {
            ScheduleItem s = null;
            synchronized (active) {
                DListItem head = null;
                try {
                    head = active.head();
                } catch (ListException e) {
                }

                try {
                    if (head == null) {
                        active.wait();
                    } else {
                        long ms = msUntil(((ScheduleItem)head.obj).date);
                        //Debug.println(6, "Next scheduled event in " + ms + " ms");
                        if (ms > 0) active.wait(ms);
                    }
                } catch (InterruptedException e) {
                }
                
                try {
                    head = active.head();
                    s = (ScheduleItem)head.obj;
                    if (s.ready()) {
                        active.popFront();
                    } else {
                        s = null;
                    }
                } catch (ListException e) {
                    s = null;
                }
            }               

            if (s != null) {
                try {
                    if (stream != null) {
                        //Debug.println(6, "Dispatching: " + s.name);
                        if (s.runnable != null) {
                            stream.write(s.runnable);
                        } else {
                            stream.write(s.command);
                        }
                    }
                    else if (context != null) {
                        //Debug.println(6, "Running: " + s.name);
                        if (s.runnable != null) {
                            if (!s.isRunning()) {
                                new Thread(s.runnable).start();
                            }
                        } else {
                            s.command.execute(context);
                        }
                        //Debug.println(6, "Done: " + s.name);
                    }
                } catch (Exception e) {
                    Debug.print(e);
                }
                schedule(s);
            }
        }
        //Debug.println(2, "terminating " + getName());
    }

    /**
     * Handle a debug monitor command directed at this instance.
     */
    public int handleInteractiveCommand(String cmd, Vector args) {
        if (cmd.equals("list")) {
            Enumeration enum = schedule.elements();
            while (enum.hasMoreElements()) {
                ScheduleItem s = (ScheduleItem)enum.nextElement();
                System.out.println("agent: " + s.name + ", interval = " +
                                   s.interval + " ms, next scheduled: " +
                                   s.date);
            }
        } else if (cmd.equals("queue")) {
            synchronized(active) {
                DListItem head = null;
                try {
                    head = active.head();
                } catch (ListException e) {
                }
                DListItem d = head;
                if (d != null) {
                    do {
                        ScheduleItem s = (ScheduleItem)d.obj;
                        System.out.println("agent: " + s.name +
                                           ", interval = " +
                                           s.interval +
                                           " ms, next scheduled: " +
                                           s.date);
                        d = d.next;
                    } while (d != head);
                }
            }
        } else {
            System.out.println("usage: ");
            System.out.println("   list");
            return -1;
        }
        return 0;
    }

    public String toString() {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        PrintWriter pos = new PrintWriter(bos);
        synchronized(active) {
            DListItem head = null;
            try {
                head = active.head();
            } catch (ListException e) {
            }
            DListItem d = head;
            if (d != null) {
                do {
                    ScheduleItem s = (ScheduleItem)d.obj;
                    pos.println("agent:  " + s.name);
                    pos.println("   interval: " + s.interval + " ms");
                    pos.println("   next: " + s.date);
                    pos.println("   running: " + s.isRunning());
                    d = d.next;
                } while (d != head);
            }
        }
        pos.flush();
        return bos.toString();
    }

    /**
     * How many ms between now and then (d)?
     *
     * @param d then
     * @return then - now
     */
    long msUntil(Date d) {
        long now = new Date().getTime();
        long then = d.getTime();
        return then - now;
    }

    /**
     * Terminate this scheduler.
     */
    public void terminate() {
        terminate = true;
        synchronized (active) {
            active.notify();
        }
    }
    
}

/**
 * This class is private to PeriodicScheduler: it represents a single
 * scheduled item, and gets placed in the queue and the hashtable.
 */
class ScheduleItem implements Runnable {
    Command                     command         = null;
    Runnable                    runnable        = null;
    String                      name            = null;
    Date                        date            = new Date();
    long                        interval        = 5000;
    boolean                     running         = false;

    ScheduleItem(String name) {
        this.name = name;
    }
    
    void checkDate(Date now) {
        if (interval < 0) throw new RuntimeException("negative interval");
        //date.setTime(now.getTime() + interval);
        date.setTime(date.getTime() + interval);
        if (date.before(now))
            date.setTime(now.getTime() + interval);
    }

    boolean ready() {
        return date.before(new Date());
    }

    boolean isRunning() {
        return running;
    }

    public void run() {
        running = true;
        try {
            runnable.run();
        } catch (Throwable t) {
            Debug.print(t);
        } finally {
            running = false;
        }
    }
}