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