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