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