source/mircoders/global/JobQueue.java

   1 /*
   2  * Copyright (C) 2001, 2002 The Mir-coders group
   3  *
   4  * This file is part of Mir.
   5  *
   6  * Mir is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 2 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * Mir is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with Mir; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19  *
  20  * In addition, as a special exception, The Mir-coders gives permission to link
  21  * the code of this program with  any library licensed under the Apache Software License,
  22  * The Sun (tm) Java Advanced Imaging library (JAI), The Sun JIMI library
  23  * (or with modified versions of the above that use the same license as the above),
  24  * and distribute linked combinations including the two.  You must obey the
  25  * GNU General Public License in all respects for all of the code used other than
  26  * the above mentioned libraries.  If you modify this file, you may extend this
  27  * exception to your version of the file, but you are not obligated to do so.
  28  * If you do not wish to do so, delete this exception statement from your version.
  29  */
  30
  31 package mircoders.global;
  32
  33
  34 import java.util.ArrayList;
  35 import java.util.Calendar;
  36 import java.util.Date;
  37 import java.util.GregorianCalendar;
  38 import java.util.HashMap;
  39 import java.util.Iterator;
  40 import java.util.List;
  41 import java.util.Map;
  42
  43 import mir.log.LoggerWrapper;
  44
  45 // important: objects passed as data must not be altered once put into a job
  46
  47 public class JobQueue {
  48   private List jobHandlers;
  49   private Map identifierToJobHandler;
  50   private int nrJobs;
  51   private int jobCleanupTreshold;
  52   private JobQueueRunner queueRunner;
  53   private Thread thread;
  54   private LoggerWrapper logger;
  55   private long lastCleanup;
  56
  57   public static final int STATUS_CREATED = -1;
  58   public static final int STATUS_PENDING = 0;
  59   public static final int STATUS_PROCESSING = 1;
  60   public static final int STATUS_PROCESSED = 2;
  61   public static final int STATUS_CANCELLED = 3;
  62   public static final int STATUS_ABORTED = 4;
  63
  64   public static final int PRIORITY_NORMAL = 100;
  65   public static final int PRIORITY_LOW = 10;
  66   public static final int PRIORITY_HIGH = 1000;
  67
  68   public static final int FINISHEDJOBS_LOGSIZE = 10;
  69
  70   public JobQueue(LoggerWrapper aLogger) {
  71     logger = aLogger;
  72     jobHandlers = new ArrayList();
  73     identifierToJobHandler = new HashMap();
  74     nrJobs = 0;
  75     lastCleanup = 0;
  76     jobCleanupTreshold = 900; // seconds
  77     queueRunner = new JobQueueRunner(logger);
  78     thread = new Thread(queueRunner, "JobQueue");
  79     thread.setDaemon(true);
  80     thread.start();
  81   }
  82
  83   public String appendJob(Job aJob, String aDescription) {
  84     try {
  85       if (System.currentTimeMillis() - lastCleanup > 60000)
  86         cleanupJobList();
  87     }
  88     catch (Throwable t) {
  89       logger.error("error while cleaning up joblist: " + t.toString());
  90     }
  91
  92     synchronized (jobHandlers) {
  93       JobHandler jobHandler = new JobHandler(aJob, Integer.toString(nrJobs), aDescription);
  94       nrJobs++;
  95       jobHandlers.add(jobHandler);
  96       identifierToJobHandler.put(jobHandler.getIdentifier(), jobHandler);
  97       jobHandler.setPending();
  98
  99       jobHandlers.notify();
 100
 101       return jobHandler.getIdentifier();
 102     }
 103   }
 104
 105   public List getJobsInfo() {
 106     List result = new ArrayList();
 107
 108     synchronized (jobHandlers) {
 109       Iterator i = jobHandlers.iterator();
 110
 111       while (i.hasNext()) {
 112         result.add(0, ((JobHandler) i.next()).getJobInfo());
 113       }
 114     }
 115
 116     return result;
 117   }
 118
 119   private void cleanupJobList() {
 120     List toRemove = new ArrayList();
 121
 122     synchronized (jobHandlers) {
 123       Iterator i = jobHandlers.iterator();
 124
 125       Calendar tresholdCalendar = new GregorianCalendar();
 126       tresholdCalendar.add(Calendar.SECOND, -jobCleanupTreshold);
 127       Date treshold = tresholdCalendar.getTime();
 128
 129       while (i.hasNext()) {
 130         JobHandler jobHandler = (JobHandler) i.next();
 131
 132         synchronized (jobHandler) {
 133           if (jobHandler.isFinished() && jobHandler.getLastChange().before(treshold)) {
 134             toRemove.add(jobHandler);
 135           }
 136         }
 137       }
 138
 139       jobHandlers.removeAll(toRemove);
 140     }
 141
 142     lastCleanup = System.currentTimeMillis();
 143   }
 144
 145   /**
 146    * Returns when a new producer job has become available
 147    */
 148   private JobHandler acquirePendingJob() throws InterruptedException {
 149     int priorityFound= 0;
 150     JobHandler jobFound;
 151
 152     synchronized (jobHandlers) {
 153       do {
 154         jobFound = null;
 155         Iterator i = jobHandlers.iterator();
 156         while (i.hasNext()) {
 157           JobHandler job = (JobHandler) i.next();
 158
 159           if (job.isPending() && (jobFound==null || priorityFound<job.getPriority())) {
 160             jobFound = job;
 161             priorityFound = job.getPriority();
 162           }
 163         }
 164
 165         if (jobFound==null) {
 166           jobHandlers.wait();
 167         }
 168
 169       } while (jobFound==null);
 170
 171       return jobFound;
 172     }
 173   }
 174
 175   public void cancelJobs(List aJobs) {
 176     synchronized (jobHandlers) {
 177       Iterator i = aJobs.iterator();
 178
 179       while (i.hasNext()) {
 180         ((JobHandler) identifierToJobHandler.get(i.next())).cancelOrAbortJob();
 181       }
 182     }
 183   }
 184
 185   public void cancelAllJobs() {
 186     synchronized (jobHandlers) {
 187       Iterator i = jobHandlers.iterator();
 188
 189       while (i.hasNext()) {
 190         ((JobHandler) i.next()).cancelOrAbortJob();
 191       }
 192     }
 193   }
 194
 195   public interface Job {
 196     /**
 197      * This method should cause the run() method to terminate as soon as
 198      * possible.
 199      */
 200     void abort();
 201
 202     /**
 203      * This method should perform the actions associated with the job.
 204      * @return <code>true</code> if terminated normally, <code>false</code> if aborted
 205      */
 206     boolean run();
 207   }
 208
 209   public static class JobInfo {
 210     private String identifier;
 211     private Date lastChange;
 212     private int status;
 213     private long runningTime;
 214     private int priority;
 215     private String description;
 216
 217     private JobInfo(String aDescription, int aStatus, Date aLastChange, String anIdentifier, long aRunningTime, int aPriority) {
 218       description = aDescription;
 219       lastChange = aLastChange;
 220       status = aStatus;
 221       identifier = anIdentifier;
 222       priority = aPriority;
 223       runningTime = aRunningTime;
 224     }
 225
 226     public String getDescription() {
 227       return description;
 228     }
 229
 230     public int getStatus() {
 231       return status;
 232     }
 233
 234     public int getPriority() {
 235       return priority;
 236     }
 237
 238     public Date getLastChange() {
 239       return lastChange;
 240     }
 241
 242     public String getIdentifier() {
 243       return identifier;
 244     }
 245
 246     public long getRunningTime() {
 247       return runningTime;
 248     }
 249   }
 250
 251   public class JobHandler {
 252     private Job job;
 253     private String identifier;
 254     private String description;
 255
 256     private Date lastChange;
 257     private long starttime;
 258     private long endtime;
 259     private int status;
 260     private int priority;
 261     private boolean hasRun;
 262
 263     public JobHandler(Job aJob, String anIdentifier, String aDescription, int aPriority) {
 264       job = aJob;
 265       description = aDescription;
 266       identifier = anIdentifier;
 267       priority = aPriority;
 268       status = STATUS_CREATED;
 269     }
 270
 271     public JobHandler(Job aJob, String anIdentifier, String aDescription) {
 272       this(aJob, anIdentifier, aDescription, PRIORITY_NORMAL);
 273     }
 274
 275     public JobInfo getJobInfo() {
 276       return new JobInfo(getDescription(), getStatus(), getLastChange(), getIdentifier(), getRunningTime(), priority);
 277     }
 278
 279     private void runJob() {
 280       if (setProcessing()) {
 281         if (job.run())
 282           setProcessed();
 283         else
 284           setAborted();
 285       }
 286     }
 287
 288     private void cancelOrAbortJob() {
 289       synchronized (this) {
 290         if (isPending())
 291           setCancelled();
 292         if (isProcessing())
 293           job.abort();
 294       }
 295     }
 296
 297     public int getStatus() {
 298       synchronized(this) {
 299         return status;
 300       }
 301     }
 302
 303     public String getIdentifier() {
 304       return identifier;
 305     }
 306
 307     public String getDescription() {
 308       return description;
 309     }
 310
 311     public long getRunningTime() {
 312       synchronized(this) {
 313         long result = 0;
 314
 315         if (hasRun) {
 316           if (isFinished())
 317             result = endtime;
 318           else
 319             result = System.currentTimeMillis();
 320
 321           result = result - starttime;
 322         }
 323
 324         return result;
 325       }
 326     }
 327
 328     public int getPriority() {
 329       return priority;
 330     }
 331
 332     private boolean setProcessing() {
 333       return setStatus(STATUS_PENDING, STATUS_PROCESSING);
 334     }
 335
 336     private void setProcessed() {
 337       setStatus(STATUS_PROCESSING, STATUS_PROCESSED);
 338     }
 339
 340     private void setAborted() {
 341       setStatus(STATUS_PROCESSING, STATUS_ABORTED);
 342     }
 343
 344     private void setPending() {
 345       setStatus(STATUS_CREATED, STATUS_PENDING);
 346     }
 347
 348     private boolean setCancelled() {
 349       return setStatus(STATUS_PENDING, STATUS_CANCELLED);
 350     }
 351
 352     public boolean hasBeenProcessed() {
 353       return getStatus() == STATUS_PROCESSED;
 354     }
 355
 356     public boolean hasBeenAborted() {
 357       return getStatus() == STATUS_ABORTED;
 358     }
 359
 360     public boolean isCancelled() {
 361       return getStatus() == STATUS_CANCELLED;
 362     }
 363
 364     public boolean isFinished() {
 365       return hasBeenProcessed() || hasBeenAborted() || isCancelled();
 366     }
 367
 368     public boolean isProcessing() {
 369       return getStatus() == STATUS_PROCESSING;
 370     }
 371
 372     public boolean isPending() {
 373       return getStatus() == STATUS_PENDING;
 374     }
 375
 376     public Date getLastChange() {
 377       synchronized (this) {
 378         return lastChange;
 379       }
 380     }
 381
 382     private boolean setStatus(int anOldStatus, int aNewStatus) {
 383       synchronized(this) {
 384         if (status == anOldStatus) {
 385           status = aNewStatus;
 386           lastChange = (new GregorianCalendar()).getTime();
 387           if (isProcessing()) {
 388             starttime = System.currentTimeMillis();
 389             hasRun = true;
 390           }
 391
 392           if (isFinished()) {
 393             endtime = System.currentTimeMillis();
 394           }
 395           return true;
 396         }
 397                                 return false;
 398       }
 399     }
 400   }
 401
 402   private class JobQueueRunner implements Runnable {
 403
 404     public JobQueueRunner(LoggerWrapper aLogger) {
 405       logger = aLogger;
 406     }
 407
 408     public void run() {
 409       logger.debug("starting JobQueueRunner");
 410
 411       try {
 412         while (true) {
 413           try {
 414             JobHandler job = acquirePendingJob();
 415
 416             logger.debug("  starting job ("+job.getIdentifier()+"): " +job.getDescription());
 417             job.runJob();
 418             logger.debug("  finished job ("+job.getIdentifier()+"): " +job.getDescription());
 419
 420             job=null;
 421           }
 422           catch (InterruptedException e) {
 423           }
 424         }
 425       }
 426       finally {
 427         logger.warn("JobQueueRunner terminated");
 428       }
 429     }
 430   }
 431 }
 432