gov/nasa/worldwind/BasicRetrievalService.java

   1 /*
   2 Copyright (C) 2001, 2006 United States Government
   3 as represented by the Administrator of the
   4 National Aeronautics and Space Administration.
   5 All Rights Reserved.
   6 */
   7 package gov.nasa.worldwind;
   8
   9 /**
  10  * Performs threaded retrieval of data.
  11  *
  12  * @author Tom Gaskins
  13  * @version $Id: BasicRetrievalService.java 1985 2007-06-09 00:33:37Z tgaskins $
  14  */
  15 public final class BasicRetrievalService extends WWObjectImpl
  16     implements RetrievalService, Thread.UncaughtExceptionHandler
  17 {
  18     // These constants are last-ditch values in case Configuration lacks defaults
  19     private static final int DEFAULT_QUEUE_SIZE = 100;
  20     private static final int DEFAULT_POOL_SIZE = 5;
  21     private static final long DEFAULT_STALE_REQUEST_LIMIT = 30000; // milliseconds
  22     private static final int DEFAULT_TIME_PRIORITY_GRANULARITY = 500; // milliseconds
  23
  24     private static final String RUNNING_THREAD_NAME_PREFIX = WorldWind.retrieveMessage(
  25         "BasicRetrievalService.RUNNING_THREAD_NAME_PREFIX", "ThreadStrings");
  26     private static final String IDLE_THREAD_NAME_PREFIX = WorldWind.retrieveMessage(
  27         "BasicRetrievalService.IDLE_THREAD_NAME_PREFIX", "ThreadStrings");
  28
  29     private RetrievalExecutor executor; // thread pool for running retrievers
  30     private java.util.concurrent.ConcurrentLinkedQueue<RetrievalTask> activeTasks; // tasks currently allocated a thread
  31     private int queueSize; // maximum queue size
  32
  33     /**
  34      * Encapsulates a single threaded retrieval as a {@link java.util.concurrent.FutureTask}.
  35      */
  36     private static class RetrievalTask extends java.util.concurrent.FutureTask<Retriever>
  37         implements RetrievalFuture, Comparable<RetrievalTask>
  38     {
  39         private Retriever retriever;
  40         private double priority; // retrieval secondary priority (primary priority is submit time)
  41
  42         private RetrievalTask(Retriever retriever, double priority)
  43         {
  44             super(retriever);
  45             this.retriever = retriever;
  46             this.priority = priority;
  47         }
  48
  49         public double getPriority()
  50         {
  51             return priority;
  52         }
  53
  54         public Retriever getRetriever()
  55         {
  56             return this.retriever;
  57         }
  58
  59         public void setException(Throwable throwable)
  60         {
  61             super.setException(throwable);
  62         }
  63
  64         @Override
  65         public void run()
  66         {
  67             if (this.isDone() || this.isCancelled())
  68                 return;
  69
  70             super.run();
  71         }
  72
  73         /**
  74          * @param that the task to compare with this one
  75          * @return 0 if task priorities are equal, -1 if priority of this is less than that, 1 otherwise
  76          * @throws IllegalArgumentException if <code>that</code> is null
  77          */
  78         public int compareTo(RetrievalTask that)
  79         {
  80             if (that == null)
  81             {
  82                 String msg = WorldWind.retrieveErrMsg("nullValue.RetrieverIsNull");
  83                 WorldWind.logger().log(java.util.logging.Level.FINE, msg);
  84                 throw new IllegalArgumentException(msg);
  85             }
  86
  87             if (this.priority > 0 && that.priority > 0) // only secondary priority used if either is negative
  88             {
  89                 // Requests submitted within different time-granularity periods are ordered exclusive of their
  90                 // client-specified priority.
  91                 long now = System.currentTimeMillis();
  92                 long thisElapsedTime = now - this.retriever.getSubmitTime();
  93                 long thatElapsedTime = now - that.retriever.getSubmitTime();
  94                 if (((thisElapsedTime - thatElapsedTime) / DEFAULT_TIME_PRIORITY_GRANULARITY) != 0)
  95                     return thisElapsedTime < thatElapsedTime ? -1 : 1;
  96             }
  97
  98             // The client-pecified priority is compared for requests submitted within the same granularity period.
  99             return this.priority == that.priority ? 0 : this.priority < that.priority ? -1 : 1;
 100         }
 101
 102         public boolean equals(Object o)
 103         {
 104             if (this == o)
 105                 return true;
 106             if (o == null || getClass() != o.getClass())
 107                 return false;
 108
 109             final RetrievalTask that = (RetrievalTask) o;
 110
 111             // Tasks are equal if their retrievers are equivalent
 112             return this.retriever.equals(that.retriever);
 113             // Priority and submint time are not factors in equality
 114         }
 115
 116         public int hashCode()
 117         {
 118             return this.retriever.getName().hashCode();
 119         }
 120     }
 121
 122     public void uncaughtException(Thread thread, Throwable throwable)
 123     {
 124         String message = WorldWind.retrieveErrMsg("BasicRetrievalService.UncaughtExceptionDuringRetrieval")
 125             + thread.getName();
 126         WorldWind.logger().log(java.util.logging.Level.FINE, message);
 127     }
 128
 129     private class RetrievalExecutor extends java.util.concurrent.ThreadPoolExecutor
 130     {
 131         private static final long THREAD_TIMEOUT = 2; // keep idle threads alive this many seconds
 132         private long staleRequestLimit; // reject requests older than this
 133
 134         private RetrievalExecutor(int poolSize, int queueSize)
 135         {
 136             super(poolSize, poolSize, THREAD_TIMEOUT, java.util.concurrent.TimeUnit.SECONDS,
 137                 new java.util.concurrent.PriorityBlockingQueue<Runnable>(queueSize),
 138                 new java.util.concurrent.ThreadFactory()
 139                 {
 140                     public Thread newThread(Runnable runnable)
 141                     {
 142                         Thread thread = new Thread(runnable);
 143                         thread.setDaemon(true);
 144                         thread.setPriority(Thread.MIN_PRIORITY);
 145                         thread.setUncaughtExceptionHandler(BasicRetrievalService.this);
 146                         return thread;
 147                     }
 148                 }, new java.util.concurrent.ThreadPoolExecutor.DiscardPolicy() // abandon task when queue is full
 149             {
 150                 // This listener is invoked only when the executor queue is a bounded queue and runs out of room.
 151                 // If the queue is a java.util.concurrent.PriorityBlockingQueue, this listener is never invoked.
 152                 public void rejectedExecution(Runnable runnable,
 153                     java.util.concurrent.ThreadPoolExecutor threadPoolExecutor)
 154                 {
 155                     // Interposes logging for rejected execution
 156                     RetrievalTask task = (RetrievalTask) runnable;
 157                     String name = task.getRetriever().getName();
 158                     String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ResourceRejected") + name;
 159                     WorldWind.logger().log(java.util.logging.Level.FINER, message);
 160
 161                     super.rejectedExecution(runnable, threadPoolExecutor);
 162                 }
 163             });
 164
 165             this.staleRequestLimit = Configuration.getLongValue(AVKey.RETRIEVAL_QUEUE_STALE_REQUEST_LIMIT,
 166                 DEFAULT_STALE_REQUEST_LIMIT);
 167         }
 168
 169         /**
 170          * @param thread   the thread the task is running on
 171          * @param runnable the <code>Retriever</code> running on the thread
 172          * @throws IllegalArgumentException if either <code>thread</code> or <code>runnable</code> is null
 173          */
 174         protected void beforeExecute(Thread thread, Runnable runnable)
 175         {
 176             WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
 177                 "BasicRetrievalService.EnteringBeforeExecute"));
 178
 179             if (thread == null)
 180             {
 181                 String msg = WorldWind.retrieveErrMsg("nullValue.ThreadIsNull");
 182                 WorldWind.logger().log(java.util.logging.Level.FINE, msg);
 183                 throw new IllegalArgumentException(msg);
 184             }
 185             if (runnable == null)
 186             {
 187                 String msg = WorldWind.retrieveErrMsg("nullValue.RunnableIsNull");
 188                 WorldWind.logger().log(java.util.logging.Level.FINE, msg);
 189                 throw new IllegalArgumentException(msg);
 190             }
 191
 192             RetrievalTask task = (RetrievalTask) runnable;
 193
 194             task.retriever.setBeginTime(System.currentTimeMillis());
 195             long limit = task.retriever.getStaleRequestLimit() >= 0
 196                 ? task.retriever.getStaleRequestLimit() : this.staleRequestLimit;
 197             if (task.retriever.getBeginTime() - task.retriever.getSubmitTime() > limit)
 198             {
 199                 // Task has been sitting on the queue too long
 200                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.CancellingTooOldRetrieval")
 201                     + task.getRetriever().getName();
 202                 WorldWind.logger().log(java.util.logging.Level.FINER, message);
 203                 task.cancel(true);
 204             }
 205
 206             if (BasicRetrievalService.this.activeTasks.contains(task))
 207             {
 208                 // Task is a duplicate
 209                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.CancellingDuplicateRetrieval") + task
 210                     .getRetriever().getName();
 211                 WorldWind.logger().log(java.util.logging.Level.FINER, message);
 212                 task.cancel(true);
 213             }
 214
 215             BasicRetrievalService.this.activeTasks.add(task);
 216
 217             thread.setName(RUNNING_THREAD_NAME_PREFIX + task.getRetriever().getName());
 218             thread.setPriority(Thread.MIN_PRIORITY); // Subordinate thread priority to rendering
 219             thread.setUncaughtExceptionHandler(BasicRetrievalService.this);
 220
 221             super.beforeExecute(thread, runnable);
 222
 223             WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
 224                 "BasicRetrievalService.LeavingBeforeExecute"));
 225         }
 226
 227         /**
 228          * @param runnable  the <code>Retriever</code> running on the thread
 229          * @param throwable an exception thrown during retrieval, will be null if no exception occurred
 230          * @throws IllegalArgumentException if <code>runnable</code> is null
 231          */
 232         protected void afterExecute(Runnable runnable, Throwable throwable)
 233         {
 234             WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
 235                 "BasicRetrievalService.EnteringAfterExecute"));
 236
 237             if (runnable == null)
 238             {
 239                 String msg = WorldWind.retrieveErrMsg("nullValue.RunnableIsNull");
 240                 WorldWind.logger().log(java.util.logging.Level.FINE, msg);
 241                 throw new IllegalArgumentException(msg);
 242             }
 243
 244             super.afterExecute(runnable, throwable);
 245
 246             RetrievalTask task = (RetrievalTask) runnable;
 247             BasicRetrievalService.this.activeTasks.remove(task);
 248             task.retriever.setEndTime(System.currentTimeMillis());
 249
 250             try
 251             {
 252                 if (throwable != null)
 253                 {
 254                     String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ExceptionDuringRetrieval") + task
 255                         .getRetriever().getName();
 256                     WorldWind.logger().log(java.util.logging.Level.FINE, message, throwable);
 257                 }
 258
 259                 task.get(); // Wait for task to finish, cancel or break
 260             }
 261             catch (java.util.concurrent.ExecutionException e)
 262             {
 263                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ExecutionExceptionDuringRetrieval")
 264                     + task.getRetriever().getName();
 265                 if (e.getCause() instanceof java.net.SocketTimeoutException)
 266                 {
 267                     WorldWind.logger()
 268                         .log(java.util.logging.Level.FINE, message + " " + e.getCause().getLocalizedMessage());
 269                 }
 270                 else
 271                 {
 272                     WorldWind.logger().log(java.util.logging.Level.FINE, message, e);
 273                 }
 274             }
 275             catch (InterruptedException e)
 276             {
 277                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.RetrievalOf_1")
 278                     + task.getRetriever().getName() + WorldWind
 279                     .retrieveErrMsg("BasicRetrievalService.WasInterrupted_2");
 280
 281                 WorldWind.logger().log(java.util.logging.Level.FINE, message, e);
 282             }
 283             catch (java.util.concurrent.CancellationException e)
 284             {
 285                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.RetrievalOf_1")
 286                     + task.getRetriever().getName() + WorldWind.retrieveErrMsg("BasicRetrievalService.WasCancelled_2");
 287
 288                 WorldWind.logger().log(java.util.logging.Level.FINE, message);
 289             }
 290             finally
 291             {
 292
 293                 Thread.currentThread().setName(IDLE_THREAD_NAME_PREFIX);
 294
 295                 WorldWind.logger().log(java.util.logging.Level.FINEST, WorldWind.retrieveErrMsg(
 296                     "BasicRetrievalService.LeavingAfterExecute"));
 297             }
 298         }
 299     }
 300
 301     public BasicRetrievalService()
 302     {
 303         Integer poolSize = Configuration.getIntegerValue(AVKey.RETRIEVAL_POOL_SIZE, DEFAULT_POOL_SIZE);
 304         this.queueSize = Configuration.getIntegerValue(AVKey.RETRIEVAL_QUEUE_SIZE, DEFAULT_QUEUE_SIZE);
 305
 306         // this.executor runs the retrievers, each in their own thread
 307         this.executor = new RetrievalExecutor(poolSize, this.queueSize);
 308
 309         // this.activeTasks holds the list of currently executing tasks (*not* those pending on the queue)
 310         this.activeTasks = new java.util.concurrent.ConcurrentLinkedQueue<RetrievalTask>();
 311     }
 312
 313     /**
 314      * @param retriever the retriever to run
 315      * @return a future object that can be used to query the request status of cancel the request.
 316      * @throws IllegalArgumentException if <code>retrieer</code> is null or has no name
 317      */
 318     public RetrievalFuture runRetriever(Retriever retriever)
 319     {
 320         if (retriever == null)
 321         {
 322             String msg = WorldWind.retrieveErrMsg("nullValue.RetrieverIsNull");
 323             WorldWind.logger().log(java.util.logging.Level.FINE, msg);
 324             throw new IllegalArgumentException(msg);
 325         }
 326         if (retriever.getName() == null)
 327         {
 328             String message = WorldWind.retrieveErrMsg("nullValue.RetrieverNameIsNull");
 329             WorldWind.logger().log(java.util.logging.Level.FINE, message);
 330             throw new IllegalArgumentException(message);
 331         }
 332
 333         // Add with secondary priority that removes most recently added requests first.
 334         return this.runRetriever(retriever, (double) (Long.MAX_VALUE - System.currentTimeMillis()));
 335     }
 336
 337     /**
 338      * @param retriever the retriever to run
 339      * @param priority  the secondary priority of the retriever, or negative if it is to be the primary priority
 340      * @return a future object that can be used to query the request status of cancel the request.
 341      * @throws IllegalArgumentException if <code>retriever</code> is null or has no name
 342      */
 343     public synchronized RetrievalFuture runRetriever(Retriever retriever, double priority)
 344     {
 345         if (retriever == null)
 346         {
 347             String message = WorldWind.retrieveErrMsg("nullValue.RetrieverIsNull");
 348             WorldWind.logger().log(java.util.logging.Level.FINE, message);
 349             throw new IllegalArgumentException(message);
 350         }
 351
 352         if (retriever.getName() == null)
 353         {
 354             String message = WorldWind.retrieveErrMsg("nullValue.RetrieverNameIsNull");
 355             WorldWind.logger().log(java.util.logging.Level.FINE, message);
 356             throw new IllegalArgumentException(message);
 357         }
 358
 359         if (this.isFull())
 360         {
 361             String name = retriever.getName();
 362             String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ResourceRejected") + name;
 363             WorldWind.logger().log(java.util.logging.Level.FINER, message);
 364         }
 365
 366         RetrievalTask task = new RetrievalTask(retriever, priority);
 367         retriever.setSubmitTime(System.currentTimeMillis());
 368
 369         // Do not queue duplicates.
 370         if (this.activeTasks.contains(task) || this.executor.getQueue().contains(task))
 371             return null;
 372
 373         this.executor.execute(task);
 374
 375         return task;
 376     }
 377
 378     /**
 379      * @param poolSize the number of threads in the thread pool
 380      * @throws IllegalArgumentException if <code>poolSize</code> is non-positive
 381      */
 382     public void setRetrieverPoolSize(int poolSize)
 383     {
 384         if (poolSize < 1)
 385         {
 386             String message = WorldWind.retrieveErrMsg("BasicRetrievalService.RetrieverPoolSizeIsLessThanOne");
 387             WorldWind.logger().log(java.util.logging.Level.FINE, message);
 388             throw new IllegalArgumentException(message);
 389         }
 390
 391         this.executor.setCorePoolSize(poolSize);
 392         this.executor.setMaximumPoolSize(poolSize);
 393     }
 394
 395     public int getRetrieverPoolSize()
 396     {
 397         return this.executor.getCorePoolSize();
 398     }
 399
 400     private boolean hasRetrievers()
 401     {
 402         Thread[] threads = new Thread[Thread.activeCount()];
 403         int numThreads = Thread.enumerate(threads);
 404         for (int i = 0; i < numThreads; i++)
 405         {
 406             if (threads[i].getName().startsWith(RUNNING_THREAD_NAME_PREFIX))
 407                 return true;
 408         }
 409         return false;
 410     }
 411
 412     public boolean hasActiveTasks()
 413     {
 414         return this.hasRetrievers();
 415     }
 416
 417     public boolean isFull()
 418     {
 419         return this.executor.getQueue().size() >= this.queueSize;
 420     }
 421
 422     public int getNumRetrieversPending()
 423     {
 424         // Could use same method to determine active tasks as hasRetrievers() above, but this method only advisory.
 425         return this.activeTasks.size() + this.executor.getQueue().size();
 426     }
 427
 428     /**
 429      * @param retriever the retriever to check
 430      * @return <code>true</code> if the retriever is being run or pending execution
 431      * @throws IllegalArgumentException if <code>retriever</code> is null
 432      */
 433     public boolean contains(Retriever retriever)
 434     {
 435         if (retriever == null)
 436         {
 437             String msg = WorldWind.retrieveErrMsg("nullValue.RetrieverIsNull");
 438             WorldWind.logger().log(java.util.logging.Level.FINE, msg);
 439             throw new IllegalArgumentException(msg);
 440         }
 441         RetrievalTask task = new RetrievalTask(retriever, 0d);
 442         return (this.activeTasks.contains(task) || this.executor.getQueue().contains(task));
 443     }
 444
 445     public double getProgress()
 446     {
 447         int totalContentLength = 0;
 448         int totalBytesRead = 0;
 449
 450         for (RetrievalTask task : this.activeTasks)
 451         {
 452             if (task.isDone())
 453                 continue;
 454
 455             Retriever retriever = task.getRetriever();
 456             try
 457             {
 458                 double tcl = retriever.getContentLength();
 459                 if (tcl > 0)
 460                 {
 461                     totalContentLength += tcl;
 462                     totalBytesRead += retriever.getContentLengthRead();
 463                 }
 464             }
 465             catch (Exception e)
 466             {
 467                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ExcptnRetrievingContentSizes") + (
 468                     retriever.getName() != null ? retriever.getName() : "");
 469                 WorldWind.logger().log(java.util.logging.Level.FINE, message, e);
 470             }
 471         }
 472
 473         for (Runnable runnable : this.executor.getQueue())
 474         {
 475             gov.nasa.worldwind.BasicRetrievalService.RetrievalTask task =
 476                 (gov.nasa.worldwind.BasicRetrievalService.RetrievalTask) runnable;
 477
 478             Retriever retriever = task.getRetriever();
 479             try
 480             {
 481                 double tcl = retriever.getContentLength();
 482                 if (tcl > 0)
 483                 {
 484                     totalContentLength += tcl;
 485                     totalBytesRead += retriever.getContentLengthRead();
 486                 }
 487             }
 488             catch (Exception e)
 489             {
 490                 String message = WorldWind.retrieveErrMsg("BasicRetrievalService.ExcptnRetrievingContentSizes") + (
 491                     retriever.getName() != null ? retriever.getName() : "");
 492                 WorldWind.logger().log(java.util.logging.Level.FINE, message, e);
 493             }
 494         }
 495
 496         // Compute an aggregated progress notification.
 497
 498         double progress;
 499
 500         if (totalContentLength < 1)
 501             progress = 0;
 502         else
 503             progress = Math.min(100.0, 100.0 * (double) totalBytesRead / (double) totalContentLength);
 504
 505         return progress;
 506     }
 507 }