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()));
 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 previous one is still running");
 173       }
 174       Console.debug( format("Worker %s queued", getWorker()));
 175       final var runnable = new Runnable()
 176       {
 177         private boolean started = false;
 178
 179         private boolean completed = false;
 180
 181         Future<?> future = null;
 182
 183         @Override
 184         public void run()
 185         {
 186           try
 187           {
 188             if (!started)
 189             {
 190               Console.debug(format("Worker %s started", getWorker()));
 191               getWorker().startUp();
 192               started = true;
 193             }
 194             else if (!completed)
 195             {
 196               Console.debug(format("Polling worker %s", getWorker()));
 197               if (getWorker().poll())
 198               {
 199                 Console.debug(format("Worker %s finished", getWorker()));
 200                 completed = true;
 201               }
 202             }
 203           } catch (Throwable th)
 204           {
 205             Console.debug(format("Worker %s failed", getWorker()), th);
 206             completed = true;
 207           }
 208           if (completed)
 209           {
 210             final var worker = getWorker();
 211             if (!isRegistered())
 212               PollableWorkerManager.super.worker = null;
 213             Console.debug(format("Finalizing completed worker %s", worker));
 214             worker.done();
 215             // almost impossible, but the future may be null at this point
 216             // let it throw NPE to cancel forcefully
 217             future.cancel(false);
 218           }
 219         }
 220       };
 221       runnable.future = task = executor.scheduleWithFixedDelay(runnable, 10,
 222               1000, TimeUnit.MILLISECONDS);
 223     }
 224
 225     synchronized protected void cancel()
 226     {
 227       if (!isWorking())
 228       {
 229         return;
 230       }
 231       Console.debug(format("Cancelling worker %s", getWorker()));
 232       task.cancel(false);
 233       executor.submit(() -> {
 234         final var worker = getWorker();
 235         if (!isRegistered())
 236           PollableWorkerManager.super.worker = null;
 237         if (worker != null)
 238         {
 239           worker.cancel();
 240           Console.debug(format("Finalizing cancelled worker %s", worker));
 241           worker.done();
 242         }
 243       });
 244     }
 245   }
 246
 247   private final ScheduledExecutorService executor = Executors
 248           .newSingleThreadScheduledExecutor();
 249
 250   private final Map<AlignCalcWorkerI, WorkerManager> registered = synchronizedMap(
 251           new HashMap<>());
 252
 253   private final Map<AlignCalcWorkerI, WorkerManager> oneshot = synchronizedMap(
 254           new WeakHashMap<>());
 255
 256   private final List<AlignCalcListener> listeners = new CopyOnWriteArrayList<>();
 257
 258   private WorkerManager createManager(AlignCalcWorkerI worker)
 259   {
 260     if (worker instanceof PollableAlignCalcWorkerI)
 261     {
 262       return new PollableWorkerManager((PollableAlignCalcWorkerI) worker);
 263     }
 264     else
 265     {
 266       return new SimpleWorkerManager(worker);
 267     }
 268   }
 269
 270   @Override
 271   public void registerWorker(AlignCalcWorkerI worker)
 272   {
 273     Objects.requireNonNull(worker);
 274     WorkerManager manager = createManager(worker);
 275     registered.putIfAbsent(worker, manager);
 276     startWorker(worker);
 277   }
 278
 279   @Override
 280   public List<AlignCalcWorkerI> getWorkers()
 281   {
 282     List<AlignCalcWorkerI> result = new ArrayList<>(registered.size());
 283     result.addAll(registered.keySet());
 284     return result;
 285   }
 286
 287   @Override
 288   public List<AlignCalcWorkerI> getWorkersOfClass(
 289           Class<? extends AlignCalcWorkerI> cls)
 290   {
 291     List<AlignCalcWorkerI> collected = new ArrayList<>();
 292     for (var worker : getWorkers())
 293     {
 294       if (worker.getClass().equals(cls))
 295       {
 296         collected.add(worker);
 297       }
 298     }
 299     return unmodifiableList(collected);
 300   }
 301
 302   @Override
 303   public void removeWorker(AlignCalcWorkerI worker)
 304   {
 305     registered.remove(worker);
 306   }
 307
 308   @Override
 309   public void removeWorkerForAnnotation(AlignmentAnnotation annot)
 310   {
 311     synchronized (registered)
 312     {
 313       for (var worker : getWorkers())
 314       {
 315         if (worker.involves(annot))
 316         {
 317           removeWorker(worker);
 318         }
 319       }
 320     }
 321   }
 322
 323   @Override
 324   public void removeWorkersOfClass(Class<? extends AlignCalcWorkerI> cls)
 325   {
 326     synchronized (registered)
 327     {
 328       for (var worker : getWorkers())
 329       {
 330         if (worker.getClass().equals(cls))
 331         {
 332           removeWorker(worker);
 333         }
 334       }
 335     }
 336   }
 337
 338   @Override
 339   public void disableWorker(AlignCalcWorkerI worker)
 340   {
 341     // Null pointer check might be needed
 342     registered.get(worker).setEnabled(false);
 343   }
 344
 345   @Override
 346   public void enableWorker(AlignCalcWorkerI worker)
 347   {
 348     // Null pointer check might be needed
 349     registered.get(worker).setEnabled(true);
 350   }
 351
 352   @Override
 353   public boolean isDisabled(AlignCalcWorkerI worker)
 354   {
 355     if (registered.containsKey(worker))
 356     {
 357       return !registered.get(worker).isEnabled();
 358     }
 359     else
 360     {
 361       return false;
 362     }
 363   }
 364
 365   @Override
 366   public boolean isWorking(AlignCalcWorkerI worker)
 367   {
 368     var manager = registered.get(worker);
 369     if (manager == null)
 370       manager = oneshot.get(worker);
 371     if (manager == null)
 372       return false;
 373     else
 374       return manager.isWorking();
 375   }
 376
 377   @Override
 378   public boolean isWorkingWithAnnotation(AlignmentAnnotation annot)
 379   {
 380     synchronized (registered)
 381     {
 382       for (var entry : registered.entrySet())
 383         if (entry.getKey().involves(annot) && entry.getValue().isWorking())
 384           return true;
 385     }
 386     synchronized (oneshot)
 387     {
 388       for (var entry : registered.entrySet())
 389         if (entry.getKey().involves(annot) && entry.getValue().isWorking())
 390           return true;
 391     }
 392     return false;
 393   }
 394
 395   @Override
 396   public boolean isWorking()
 397   {
 398     synchronized (registered)
 399     {
 400       for (var manager : registered.values())
 401         if (manager.isWorking())
 402           return true;
 403     }
 404     synchronized (oneshot)
 405     {
 406       for (var manager : oneshot.values())
 407         if (manager.isWorking())
 408           return true;
 409     }
 410     return false;
 411   }
 412
 413   @Override
 414   public void startWorker(AlignCalcWorkerI worker)
 415   {
 416     Objects.requireNonNull(worker);
 417     var manager = registered.get(worker);
 418     if (manager == null)
 419     {
 420       Console.warn("Starting unregistered worker " + worker);
 421       manager = createManager(worker);
 422       oneshot.put(worker, manager);
 423     }
 424     manager.restart();
 425   }
 426
 427   @Override
 428   public void restartWorkers()
 429   {
 430     synchronized (registered)
 431     {
 432       for (var manager : registered.values())
 433       {
 434         manager.restart();
 435       }
 436     }
 437   }
 438
 439   @Override
 440   public void cancelWorker(AlignCalcWorkerI worker)
 441   {
 442     Objects.requireNonNull(worker);
 443     var manager = registered.get(worker);
 444     if (manager == null)
 445       manager = oneshot.get(worker);
 446     if (manager == null)
 447     {
 448       throw new NoSuchElementException();
 449     }
 450     manager.cancel();
 451   }
 452
 453   private void notifyQueued(AlignCalcWorkerI worker)
 454   {
 455     for (AlignCalcListener listener : listeners)
 456     {
 457       listener.workerQueued(worker);
 458     }
 459   }
 460
 461   private void notifyStarted(AlignCalcWorkerI worker)
 462   {
 463     for (AlignCalcListener listener : listeners)
 464     {
 465       listener.workerStarted(worker);
 466     }
 467   }
 468
 469   private void notifyCompleted(AlignCalcWorkerI worker)
 470   {
 471     for (AlignCalcListener listener : listeners)
 472     {
 473       try
 474       {
 475         listener.workerCompleted(worker);
 476       } catch (RuntimeException e)
 477       {
 478         e.printStackTrace();
 479       }
 480     }
 481   }
 482
 483   private void notifyCancelled(AlignCalcWorkerI worker)
 484   {
 485     for (AlignCalcListener listener : listeners)
 486     {
 487       try
 488       {
 489         listener.workerCancelled(worker);
 490       } catch (RuntimeException e)
 491       {
 492         e.printStackTrace();
 493       }
 494     }
 495   }
 496
 497   private void notifyExceptional(AlignCalcWorkerI worker,
 498           Throwable throwable)
 499   {
 500     for (AlignCalcListener listener : listeners)
 501     {
 502       try
 503       {
 504         listener.workerExceptional(worker, throwable);
 505       } catch (RuntimeException e)
 506       {
 507         e.printStackTrace();
 508       }
 509     }
 510   }
 511
 512   @Override
 513   public void addAlignCalcListener(AlignCalcListener listener)
 514   {
 515     listeners.add(listener);
 516   }
 517
 518   @Override
 519   public void removeAlignCalcListener(AlignCalcListener listener)
 520   {
 521     listeners.remove(listener);
 522   }
 523
 524   @Override
 525   public void shutdown()
 526   {
 527     executor.shutdownNow();
 528     listeners.clear();
 529     registered.clear();
 530   }
 531
 532 }