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.Calendar;
  35 import java.util.Date;
  36 import java.util.GregorianCalendar;
  37 import java.util.HashMap;
  38 import java.util.Iterator;
  39 import java.util.List;
  40 import java.util.Map;
  41 import java.util.Vector;
  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 Vector 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 Vector();
  73     identifierToJobHandler = new HashMap();
  74     nrJobs = 0;
  75     lastCleanup = 0;
  76     jobCleanupTreshold = 900; // seconds
  77     queueRunner = new JobQueueRunner(logger);
  78     thread = new Thread(queueRunner);
  79     thread.start();
  80   }
  81
  82   public String appendJob(Job aJob, String aDescription) {
  83     try {
  84       if (System.currentTimeMillis() - lastCleanup > 60000)
  85         cleanupJobList();
  86     }
  87     catch (Throwable t) {
  88       logger.error("error while cleaning up joblist: " + t.toString());
  89     }
  90
  91     synchronized (jobHandlers) {
  92       JobHandler jobHandler = new JobHandler(aJob, Integer.toString(nrJobs), aDescription);
  93       nrJobs++;
  94       jobHandlers.add(jobHandler);
  95       identifierToJobHandler.put(jobHandler.getIdentifier(), jobHandler);
  96       jobHandler.setPending();
  97
  98       return jobHandler.getIdentifier();
  99     }
 100   }
 101
 102   public List getJobsInfo() {
 103     List result = new Vector();
 104
 105     synchronized (jobHandlers) {
 106       Iterator i = jobHandlers.iterator();
 107
 108       while (i.hasNext()) {
 109         result.add(0, ((JobHandler) i.next()).getJobInfo());
 110       }
 111     }
 112
 113     return result;
 114   }
 115
 116   private void cleanupJobList() {
 117     List toRemove = new Vector();
 118     synchronized (jobHandlers) {
 119       Iterator i = jobHandlers.iterator();
 120
 121       Calendar tresholdCalendar = new GregorianCalendar();
 122       tresholdCalendar.add(Calendar.SECOND, -jobCleanupTreshold);
 123       Date treshold = tresholdCalendar.getTime();
 124
 125       while (i.hasNext()) {
 126         JobHandler jobHandler = (JobHandler) i.next();
 127
 128         synchronized (jobHandler) {
 129           if (jobHandler.isFinished() && jobHandler.getLastChange().before(treshold)) {
 130             toRemove.add(jobHandler);
 131           }
 132         }
 133       }
 134
 135       jobHandlers.removeAll(toRemove);
 136     }
 137
 138     lastCleanup = System.currentTimeMillis();
 139   }
 140
 141   private JobHandler acquirePendingJob() {
 142     synchronized (jobHandlers) {
 143       int priorityFound= 0;
 144       JobHandler jobFound;
 145
 146       jobFound = null;
 147       Iterator i = jobHandlers.iterator();
 148       while (i.hasNext()) {
 149         JobHandler job = (JobHandler) i.next();
 150
 151         if (job.isPending() && (jobFound==null || priorityFound<job.getPriority())) {
 152           jobFound = job;
 153           priorityFound = job.getPriority();
 154         }
 155       }
 156
 157       return jobFound;
 158     }
 159   }
 160
 161   public void cancelJobs(List aJobs) {
 162     synchronized (jobHandlers) {
 163       Iterator i = aJobs.iterator();
 164
 165       while (i.hasNext()) {
 166         ((JobHandler) identifierToJobHandler.get(i.next())).cancelOrAbortJob();
 167       }
 168     }
 169   }
 170
 171   public interface Job {
 172     void abort();
 173
 174     /**
 175      *
 176      *
 177      * @return <code>true</code> if terminated normally, <code>false</code> if aborted
 178      */
 179     boolean run();
 180   }
 181
 182   public static class JobInfo {
 183     private String identifier;
 184     private Date lastChange;
 185     private int status;
 186     private long runningTime;
 187     private int priority;
 188     private String description;
 189
 190     private JobInfo(String aDescription, int aStatus, Date aLastChange, String anIdentifier, long aRunningTime, int aPriority) {
 191       description = aDescription;
 192       lastChange = aLastChange;
 193       status = aStatus;
 194       identifier = anIdentifier;
 195       priority = aPriority;
 196       runningTime = aRunningTime;
 197     }
 198
 199     public String getDescription() {
 200       return description;
 201     }
 202
 203     public int getStatus() {
 204       return status;
 205     }
 206
 207     public int getPriority() {
 208       return priority;
 209     }
 210
 211     public Date getLastChange() {
 212       return lastChange;
 213     }
 214
 215     public String getIdentifier() {
 216       return identifier;
 217     }
 218
 219     public long getRunningTime() {
 220       return runningTime;
 221     }
 222   }
 223
 224   public class JobHandler {
 225     private Job job;
 226     private String identifier;
 227     private String description;
 228
 229     private Date lastChange;
 230     private long starttime;
 231     private long endtime;
 232     private int status;
 233     private int priority;
 234     private boolean hasRun;
 235
 236     public JobHandler(Job aJob, String anIdentifier, String aDescription, int aPriority) {
 237       job = aJob;
 238       description = aDescription;
 239       identifier = anIdentifier;
 240       priority = aPriority;
 241       status = STATUS_CREATED;
 242     }
 243
 244     public JobHandler(Job aJob, String anIdentifier, String aDescription) {
 245       this(aJob, anIdentifier, aDescription, PRIORITY_NORMAL);
 246     }
 247
 248     public JobInfo getJobInfo() {
 249       return new JobInfo(getDescription(), getStatus(), getLastChange(), getIdentifier(), getRunningTime(), priority);
 250     }
 251
 252     private void runJob() {
 253       if (setProcessing()) {
 254         if (job.run())
 255           setProcessed();
 256         else
 257           setAborted();
 258       }
 259     };
 260
 261     private void cancelOrAbortJob() {
 262       synchronized (this) {
 263         if (isPending())
 264           setCancelled();
 265         if (isProcessing())
 266           job.abort();
 267       }
 268     };
 269
 270     public int getStatus() {
 271       synchronized(this) {
 272         return status;
 273       }
 274     }
 275
 276     public String getIdentifier() {
 277       return identifier;
 278     }
 279
 280     public String getDescription() {
 281       return description;
 282     }
 283
 284     public long getRunningTime() {
 285       synchronized(this) {
 286         long result = 0;
 287
 288         if (hasRun) {
 289           if (isFinished())
 290             result = endtime;
 291           else
 292             result = System.currentTimeMillis();
 293
 294           result = result - starttime;
 295         }
 296
 297         return result;
 298       }
 299     }
 300
 301     public int getPriority() {
 302       return priority;
 303     }
 304
 305     private boolean setProcessing() {
 306       return setStatus(STATUS_PENDING, STATUS_PROCESSING);
 307     }
 308
 309     private void setProcessed() {
 310       setStatus(STATUS_PROCESSING, STATUS_PROCESSED);
 311     }
 312
 313     private void setAborted() {
 314       setStatus(STATUS_PROCESSING, STATUS_ABORTED);
 315     }
 316
 317     private void setPending() {
 318       setStatus(STATUS_CREATED, STATUS_PENDING);
 319     }
 320
 321     private boolean setCancelled() {
 322       return setStatus(STATUS_PENDING, STATUS_CANCELLED);
 323     }
 324
 325     public boolean hasBeenProcessed() {
 326       return getStatus() == STATUS_PROCESSED;
 327     }
 328
 329     public boolean hasBeenAborted() {
 330       return getStatus() == STATUS_ABORTED;
 331     }
 332
 333     public boolean isCancelled() {
 334       return getStatus() == STATUS_CANCELLED;
 335     }
 336
 337     public boolean isFinished() {
 338       return hasBeenProcessed() || hasBeenAborted() || isCancelled();
 339     }
 340
 341     public boolean isProcessing() {
 342       return getStatus() == STATUS_PROCESSING;
 343     }
 344
 345     public boolean isPending() {
 346       return getStatus() == STATUS_PENDING;
 347     }
 348
 349     public Date getLastChange() {
 350       synchronized (this) {
 351         return lastChange;
 352       }
 353     }
 354
 355     private boolean setStatus(int anOldStatus, int aNewStatus) {
 356       synchronized(this) {
 357         if (status == anOldStatus) {
 358           status = aNewStatus;
 359           lastChange = (new GregorianCalendar()).getTime();
 360           if (isProcessing()) {
 361             starttime = System.currentTimeMillis();
 362             hasRun = true;
 363           }
 364
 365           if (isFinished()) {
 366             endtime = System.currentTimeMillis();
 367           }
 368           return true;
 369         }
 370         else {
 371           return false;
 372         }
 373       }
 374     }
 375   }
 376
 377   private class JobQueueRunner implements Runnable {
 378     private LoggerWrapper logger;
 379
 380     public JobQueueRunner(LoggerWrapper aLogger) {
 381       logger = aLogger;
 382     }
 383
 384     public void run() {
 385       logger.debug("starting JobQueueRunner");
 386
 387       try {
 388         while (true) {
 389           JobHandler job = acquirePendingJob();
 390           if (job != null) {
 391             logger.debug("  starting job ("+job.getIdentifier()+"): " +job.getDescription());
 392             job.runJob();
 393             logger.debug("  finished job ("+job.getIdentifier()+"): " +job.getDescription());
 394           }
 395           else {
 396             try {
 397               Thread.sleep(1500);
 398             }
 399             catch (InterruptedException e) {
 400             }
 401           }
 402         }
 403       }
 404       finally {
 405         logger.warn("JobQueueRunner terminated");
 406       }
 407     }
 408   }
 409 }
 410