/* * Copyright (C) The Apache Software Foundation. All rights reserved. * * This software is published under the terms of the Apache Software License * version 1.1, a copy of which has been included with this distribution in * the LICENSE file. */ package org.apache.james.core; import java.io.PrintStream; import java.util.*; import org.apache.avalon.cornerstone.services.scheduler.PeriodicTimeTrigger; import org.apache.avalon.cornerstone.services.scheduler.Target; import org.apache.avalon.cornerstone.services.scheduler.TimeScheduler; import org.apache.avalon.cornerstone.services.scheduler.TimeTrigger; import org.apache.avalon.framework.activity.Disposable; import org.apache.avalon.framework.activity.Initializable; import org.apache.avalon.framework.configuration.Configurable; import org.apache.avalon.framework.configuration.Configuration; import org.apache.avalon.framework.configuration.ConfigurationException; /** * This service provides a way to regularly schedule jobs. * currently(Oct '02) it is only handles one time events. This is all * James needs for now. If the service cannot cleanly generalized with * same performance, interface should be narrowed. */ public class TimeSchedulerImpl implements TimeScheduler, Initializable, Disposable, Configurable { // name to TimerTask map. private TimerEventCollection[] cache; private Thread[] triggerThread; /** number of processor threads */ private int threadCount; /** time between attempts to process schduled events, if there are * no schedulble events */ private long idleTime; /** @see Configurable#configure */ public void configure( final Configuration configuration ) throws ConfigurationException { this.threadCount = configuration.getChild("threadCount").getValueAsInteger(1); this.idleTime = configuration.getChild("idleTime").getValueAsLong(1000); } /** @see Initializable#initialize */ public void initialize() { cache = new TimerEventCollection[threadCount]; triggerThread = new Thread[threadCount]; // HBNOTE: replace this with tread pool. for ( int i = 0 ; i < threadCount ; i++ ) { cache[i] = new TimerEventCollection(); TimerEventProcessor eventProcessor = new TimerEventProcessor(cache[i],idleTime); triggerThread[i] = new Thread(eventProcessor,"timer-trigger-thread-"+i); } for ( int i = 0 ; i < threadCount ; i++ ) triggerThread[i].start(); } /** @see Disposable#dispose */ public void dispose() { for ( int i = 0 ; i < triggerThread.length ; i++ ) triggerThread[i].interrupt(); cache = null; } /** this method converts Avalon TimeTrigger mechanism to JDK Timer based * mechanism. * The Trigger is expected to be . * Warning: This method may have cause a very small and in most cases * inconsequential skew in the alarm mechanism. */ public void addTrigger(String name, TimeTrigger trigger, Target target ) { //log(); if ( ( trigger instanceof PeriodicTimeTrigger ) == false ) { throw new RuntimeException ("Currently only the periodic timer is supported"); } PeriodicTimeTrigger ptm = (PeriodicTimeTrigger)trigger; //System.out.println(name.hashCode()+", "+threadCount); //System.out.println(name.hashCode()+", "+threadCount+", "+(name.hashCode()%threadCount)); cache[name.hashCode()%threadCount].put(new TimerEvent(ptm.getOffset(),name,target)); } public void removeTrigger( String name ) throws NoSuchElementException { //log(); cache[name.hashCode()%threadCount].remove(name); } public void resetTrigger( String name ) throws NoSuchElementException { //log(); cache[name.hashCode()%threadCount].reset(name); } /* private static int counter = 0; private void log() { counter++; if ( counter % 500 == 0 ) for ( int i = 0 ; i < threadCount ; i++ ) System.out.println(i+":"+cache[i].toString()); } */ } /** Scheduled timer event */ class TimerEvent { /** event id */ final String id; /** target to trigger when event is scheduled for execution */ final Target trg; /** time offset from time of event creation */ final long offset; /** time in milliseconds when event will be exectued */ private final Long scheduledTime; /** * @param offset time offset from current time * @param id Event unique identifier * @param target sceduled callback target */ TimerEvent(long offset,String id,Target trg) { this.id = id; this.trg = trg; this.offset = offset; this.scheduledTime = new Long(System.currentTimeMillis()+offset); } /** * reset event. Recalcuate execution time * @return event with reset scheduled time. */ TimerEvent reset() { return new TimerEvent(offset,id,trg); } /** @return time of event execution in milliseconds */ Long getScheduledTime() { return scheduledTime; } } // class TimerEvent /** * collection of scheduled events. * Event lookup is done either on the basis of ID or * in time of execution sequence. */ class TimerEventCollection { /** map of scheduled event time -> scheduled event. */ private final SortedMap timerMap = new TreeMap(); /** map of event identifier -> scheduled event */ private final Map idMap = new HashMap(); /** add event to cache */ synchronized void put(TimerEvent event) { putInternal(event); } /** * Reset the event. This involves removing reinserting event * based on scheduled event time * @param id Event ID */ synchronized void reset(String id) { TimerEvent event = (TimerEvent)idMap.remove(id); if ( event != null ) { // event may have been removed for trigger. timerMap.remove(event.getScheduledTime()); putInternal(event.reset()); } } /** * remove event. * @param id Event ID */ synchronized void remove(String id) { TimerEvent event = (TimerEvent)idMap.remove(id); if ( event != null ) timerMap.remove(event.getScheduledTime()); } /** * return the first scheduled event. Typically called by * background thread(s) to obtain and fire schdeuled events * * @return event if there is a scheduled event. null indicates * that there were no scheduled event. Calling thread should try * later. */ synchronized TimerEvent getScheduledEvent() throws InterruptedException { if ( timerMap.isEmpty() ) return null; Long first = (Long)timerMap.firstKey(); long future = first.longValue()-System.currentTimeMillis(); if ( future > 0 ) return null; // first event is up for execution. TimerEvent event = (TimerEvent)timerMap.remove(first); idMap.remove(event.id); return event; } /** * internal method to add events to cache. Note: calling method * is responsible for synchornization */ private void putInternal(TimerEvent event) { // move by one millisec, if there is a stored event. // // Note: this way of ensuring uniqueness may be somewhat heavy // but not likely to get executed. // // collision can only happen if 2 events happen at the same // millisecond. There is a very low chance of collisions at // this granularity. while ( timerMap.containsKey(event.getScheduledTime()) ) { event = new TimerEvent(event.offset+1,event.id,event.trg); } timerMap.put(event.getScheduledTime(),event); idMap.put(event.id,event); } /* @return Cache size information. A statistics collection * process may want to log this information periodically to * avoid memory leaks. */ public synchronized String toString() { return timerMap.size()+", "+idMap.size(); } } // TimerEventCollection /** * Interacts with background threads via runnable interface to execute * schduled events */ class TimerEventProcessor implements Runnable { private final TimerEventCollection cache; private final long idleTime; /** * @param cache Collection of events. * @param idleTime If there is no suitable event in cache, wait * for idle time and try again. */ TimerEventProcessor(TimerEventCollection cache,long idleTime) { this.cache = cache; this.idleTime = idleTime; } /** @see Runnable#run */ public void run() { while ( !Thread.currentThread().isInterrupted() ) { try { TimerEvent event = cache.getScheduledEvent(); if ( event == null ) Thread.currentThread().sleep(idleTime); else event.trg.targetTriggered(event.id); } catch(Throwable t) { if ( t instanceof InterruptedException ) break; else t.printStackTrace(); } } } } // TimerEventProcessor