src/jalview/workers/AlignCalcManager2.java

   1 package jalview.workers;
   2
   3 import java.util.HashMap;
   4 import java.util.List;
   5 import java.util.Map;
   6 import java.util.NoSuchElementException;
   7 import java.util.Objects;
   8 import java.util.concurrent.CopyOnWriteArrayList;
   9 import java.util.concurrent.Executors;
  10 import java.util.concurrent.Future;
  11 import java.util.concurrent.ScheduledExecutorService;
  12 import java.util.concurrent.TimeUnit;
  13
  14 import static java.lang.String.format;
  15 import static java.util.Collections.synchronizedMap;
  16 import static java.util.Collections.unmodifiableList;
  17
  18 import java.util.ArrayList;
  19
  20 import jalview.api.AlignCalcListener;
  21 import jalview.api.AlignCalcManagerI2;
  22 import jalview.api.AlignCalcWorkerI;
  23 import jalview.api.PollableAlignCalcWorkerI;
  24 import jalview.bin.Cache;
  25 import jalview.datamodel.AlignmentAnnotation;
  26
  27 public class AlignCalcManager2 implements AlignCalcManagerI2
  28 {
  29   private abstract class WorkerManager
  30   {
  31     static final int IDLE = 0;
  32     static final int QUEUED = 1;
  33     static final int RUNNING = 2;
  34     static final int CANCELLING = 3;
  35
  36     protected volatile int state = IDLE;
  37     protected volatile boolean enabled = true;
  38
  39     protected final AlignCalcWorkerI worker;
  40
  41     WorkerManager(AlignCalcWorkerI worker)
  42     {
  43       this.worker = worker;
  44     }
  45
  46     AlignCalcWorkerI getWorker()
  47     {
  48       return worker;
  49     }
  50
  51     boolean isEnabled()
  52     {
  53       return enabled;
  54     }
  55
  56     void setEnabled(boolean enabled)
  57     {
  58       this.enabled = enabled;
  59     }
  60
  61     synchronized protected void setState(int state)
  62     {
  63       this.state = state;
  64     }
  65
  66     int getState()
  67     {
  68       return state;
  69     }
  70
  71     void restart()
  72     {
  73       if (!isEnabled())
  74       {
  75         return;
  76       }
  77       if (state == IDLE)
  78       {
  79         submit();
  80       }
  81       else if (state == QUEUED)
  82       {
  83         // job already queued, do nothing
  84       }
  85       else if (state == RUNNING)
  86       {
  87         cancel();
  88         submit();
  89       }
  90       else if (state == CANCELLING)
  91       {
  92         submit();
  93       }
  94     }
  95
  96     protected abstract void submit();
  97
  98     abstract void cancel();
  99   }
 100
 101
 102   private class SimpleWorkerManager extends WorkerManager
 103   {
 104     private Future<?> task = null;
 105
 106     SimpleWorkerManager(AlignCalcWorkerI worker)
 107     {
 108       super(worker);
 109     }
 110
 111     @Override
 112     protected void submit()
 113     {
 114       if (task != null && !(task.isDone() || task.isCancelled()))
 115       {
 116         throw new IllegalStateException(
 117             "Cannot submit new task if the prevoius one is still running");
 118       }
 119       Cache.log.debug(format("Worker %s queued",
 120               worker.getClass().getName()));
 121       setState(QUEUED);
 122       task = executor.submit(() -> {
 123         setState(RUNNING);
 124         try
 125         {
 126           Cache.log.debug(format("Worker %s started",
 127                   worker.getClass().getName()));
 128           worker.run();
 129           Cache.log.debug(format("Worker %s finished",
 130                   worker.getClass().getName()));
 131         }
 132         catch (InterruptedException e)
 133         {
 134           Cache.log.debug(format("Worker %s interrupted",
 135                   worker.getClass().getName()));
 136         }
 137         catch (Throwable th)
 138         {
 139           Cache.log.debug(format("Worker %s failed",
 140                   worker.getClass().getName()), th);
 141         }
 142         finally
 143         {
 144           // fixme: should not set to idle if another task is already queued for execution
 145           setState(IDLE);
 146         }
 147       });
 148     }
 149
 150     @Override
 151     synchronized void cancel()
 152     {
 153       if (task == null || state == IDLE || state == CANCELLING)
 154       {
 155         return;
 156       }
 157       Cache.log.debug(format("Cancelling worker %s",
 158               worker.getClass().getName()));
 159       setState(CANCELLING);
 160       task.cancel(true);
 161       if (task.isCancelled())
 162       {
 163         setState(IDLE);
 164       }
 165     }
 166   }
 167
 168
 169   private class PollableWorkerManager extends WorkerManager
 170   {
 171     private final PollableAlignCalcWorkerI worker;
 172     private Future<?> task = null;
 173
 174     PollableWorkerManager(PollableAlignCalcWorkerI worker)
 175     {
 176       super(worker);
 177       this.worker = worker;
 178     }
 179
 180     protected void submit()
 181     {
 182       if (task != null && !(task.isDone() || task.isCancelled()))
 183       {
 184         throw new IllegalStateException(
 185             "Cannot submit new task if the prevoius one is still running");
 186       }
 187       Cache.log.debug(format("Worker %s queued",
 188               worker.getClass().getName()));
 189       setState(QUEUED);
 190       final var runnable = new Runnable()
 191       {
 192         private boolean started = false;
 193         private boolean completed = false;
 194         Future<?> future = null;
 195
 196         @Override
 197         public void run()
 198         {
 199           try
 200           {
 201             if (!started)
 202             {
 203               Cache.log.debug(format("Worker %s started",
 204                       worker.getClass().getName()));
 205               setState(RUNNING);
 206               worker.startUp();
 207               started = true;
 208             }
 209             else if (!completed)
 210             {
 211               if (worker.poll())
 212               {
 213                 Cache.log.debug(format("Worker %s finished",
 214                         worker.getClass().getName()));
 215                 completed = true;
 216                 setState(IDLE);
 217               }
 218             }
 219           } catch (Throwable th)
 220           {
 221             Cache.log.debug(format("Worker %s failed",
 222                     worker.getClass().getName()), th);
 223             completed = true;
 224             setState(IDLE);
 225           }
 226           if (completed)
 227           {
 228             try
 229             {
 230               future.cancel(false);
 231             }
 232             catch (NullPointerException ignored)
 233             {
 234               // extremely unlikely to happen
 235             }
 236           }
 237         }
 238       };
 239       runnable.future = task = executor.scheduleWithFixedDelay(
 240               runnable, 10, 1000, TimeUnit.MILLISECONDS);
 241     }
 242
 243     synchronized protected void cancel()
 244     {
 245       if (task == null || state == IDLE || state == CANCELLING)
 246       {
 247         return;
 248       }
 249       Cache.log.debug(format("Cancelling worker %s",
 250               worker.getClass().getName()));
 251       setState(CANCELLING);
 252       task.cancel(false);
 253       if (task.isCancelled())
 254       {
 255         setState(IDLE);
 256       }
 257       executor.submit(() -> {
 258         worker.cancel();
 259       });
 260     }
 261   }
 262
 263
 264   private final ScheduledExecutorService executor =
 265           Executors.newSingleThreadScheduledExecutor();
 266   private final Map<AlignCalcWorkerI, WorkerManager> registered =
 267           synchronizedMap(new HashMap<>());
 268
 269   private final List<AlignCalcListener> listeners =
 270           new CopyOnWriteArrayList<>();
 271
 272
 273   @Override
 274   public void registerWorker(AlignCalcWorkerI worker)
 275   {
 276     Objects.requireNonNull(worker);
 277     WorkerManager manager = (worker instanceof PollableAlignCalcWorkerI) ?
 278             new PollableWorkerManager((PollableAlignCalcWorkerI) worker) :
 279               new SimpleWorkerManager(worker);
 280     registered.putIfAbsent(worker, manager);
 281     startWorker(worker);
 282   }
 283
 284   @Override
 285   public List<AlignCalcWorkerI> getWorkers()
 286   {
 287     return List.copyOf(registered.keySet());
 288   }
 289
 290   @Override
 291   public List<AlignCalcWorkerI> getWorkersOfClass(
 292           Class<? extends AlignCalcWorkerI> cls)
 293   {
 294     List<AlignCalcWorkerI> collected = new ArrayList<>();
 295     for (var worker : getWorkers())
 296     {
 297       if (worker.getClass().equals(cls))
 298       {
 299         collected.add(worker);
 300       }
 301     }
 302     return unmodifiableList(collected);
 303   }
 304
 305   @Override
 306   public void removeWorker(AlignCalcWorkerI worker)
 307   {
 308     registered.remove(worker);
 309   }
 310
 311   @Override
 312   public void removeWorkerForAnnotation(AlignmentAnnotation annot)
 313   {
 314     synchronized (registered)
 315     {
 316       for (var worker : getWorkers())
 317       {
 318         if (worker.involves(annot) && worker.isDeletable())
 319         {
 320           removeWorker(worker);
 321         }
 322       }
 323     }
 324   }
 325
 326   @Override
 327   public void removeWorkersOfClass(Class<? extends AlignCalcWorkerI> cls)
 328   {
 329     synchronized (registered)
 330     {
 331       for (var worker : getWorkers())
 332       {
 333         if (worker.getClass().equals(cls))
 334         {
 335           removeWorker(worker);
 336         }
 337       }
 338     }
 339   }
 340
 341   @Override
 342   public void disableWorker(AlignCalcWorkerI worker)
 343   {
 344     // Null pointer check might be needed
 345     registered.get(worker).setEnabled(false);
 346   }
 347
 348   @Override
 349   public void enableWorker(AlignCalcWorkerI worker)
 350   {
 351     // Null pointer check might be needed
 352     registered.get(worker).setEnabled(true);
 353   }
 354
 355   @Override
 356   public boolean isDisabled(AlignCalcWorkerI worker)
 357   {
 358     if (registered.containsKey(worker))
 359     {
 360       return !registered.get(worker).isEnabled();
 361     }
 362     else
 363     {
 364       return false;
 365     }
 366   }
 367
 368   @Override
 369   public boolean isWorking(AlignCalcWorkerI worker)
 370   {
 371     if (!registered.containsKey(worker))
 372     {
 373       return false;
 374     }
 375     else
 376     {
 377       return registered.get(worker).getState() == WorkerManager.RUNNING;
 378     }
 379   }
 380
 381   @Override
 382   public boolean isWorkingWithAnnotation(AlignmentAnnotation annot)
 383   {
 384     synchronized (registered)
 385     {
 386       for (var entry : registered.entrySet())
 387       {
 388         if (entry.getKey().involves(annot) &&
 389                 entry.getValue().getState() == WorkerManager.RUNNING)
 390         {
 391           return true;
 392         }
 393       }
 394     }
 395     return false;
 396   }
 397
 398   @Override
 399   public boolean isWorking()
 400   {
 401     synchronized (registered)
 402     {
 403       for (var manager : registered.values())
 404       {
 405         if (manager.getState() == WorkerManager.RUNNING)
 406         {
 407           return true;
 408         }
 409       }
 410     }
 411     return false;
 412   }
 413
 414   @Override
 415   public void startWorker(AlignCalcWorkerI worker)
 416   {
 417     Objects.requireNonNull(worker);
 418     var manager = registered.get(worker);
 419     if (manager == null)
 420     {
 421       throw new NoSuchElementException();
 422     }
 423     manager.restart();
 424   }
 425
 426   @Override
 427   public void restartWorkers()
 428   {
 429     synchronized (registered)
 430     {
 431       for (var manager : registered.values())
 432       {
 433         manager.restart();
 434       }
 435     }
 436   }
 437
 438   @Override
 439   public void cancelWorker(AlignCalcWorkerI worker)
 440   {
 441     Objects.requireNonNull(worker);
 442     var manager = registered.get(worker);
 443     if (manager == null)
 444     {
 445       throw new NoSuchElementException();
 446     }
 447     manager.cancel();
 448   }
 449
 450   private void notifyQueued(AlignCalcWorkerI worker)
 451   {
 452     for (AlignCalcListener listener : listeners)
 453     {
 454       listener.workerQueued(worker);
 455     }
 456   }
 457
 458   private void notifyStarted(AlignCalcWorkerI worker)
 459   {
 460     for (AlignCalcListener listener : listeners)
 461     {
 462       listener.workerStarted(worker);
 463     }
 464   }
 465
 466   private void notifyCompleted(AlignCalcWorkerI worker)
 467   {
 468     for (AlignCalcListener listener : listeners)
 469     {
 470       try
 471       {
 472         listener.workerCompleted(worker);
 473       } catch (RuntimeException e)
 474       {
 475         e.printStackTrace();
 476       }
 477     }
 478   }
 479
 480   private void notifyCancelled(AlignCalcWorkerI worker)
 481   {
 482     for (AlignCalcListener listener : listeners)
 483     {
 484       try
 485       {
 486         listener.workerCancelled(worker);
 487       } catch (RuntimeException e)
 488       {
 489         e.printStackTrace();
 490       }
 491     }
 492   }
 493
 494   private void notifyExceptional(AlignCalcWorkerI worker,
 495           Throwable throwable)
 496   {
 497     for (AlignCalcListener listener : listeners)
 498     {
 499       try
 500       {
 501         listener.workerExceptional(worker, throwable);
 502       } catch (RuntimeException e)
 503       {
 504         e.printStackTrace();
 505       }
 506     }
 507   }
 508
 509   @Override
 510   public void addAlignCalcListener(AlignCalcListener listener)
 511   {
 512     listeners.add(listener);
 513   }
 514
 515   @Override
 516   public void removeAlignCalcListener(AlignCalcListener listener)
 517   {
 518     listeners.remove(listener);
 519   }
 520
 521   @Override
 522   public void shutdown()
 523   {
 524     executor.shutdownNow();
 525     listeners.clear();
 526     registered.clear();
 527   }
 528
 529 }