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