1 package jalview.workers;
3 import java.util.HashMap;
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;
15 import static java.lang.String.format;
16 import static java.util.Collections.synchronizedMap;
17 import static java.util.Collections.unmodifiableList;
19 import java.util.ArrayList;
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;
29 public class AlignCalcManager2 implements AlignCalcManagerI2
31 private abstract class WorkerManager
33 protected volatile boolean enabled = true;
35 protected AlignCalcWorkerI worker;
37 WorkerManager(AlignCalcWorkerI worker)
42 protected AlignCalcWorkerI getWorker()
52 void setEnabled(boolean enabled)
54 this.enabled = enabled;
57 synchronized void restart()
74 protected boolean isRegistered()
76 return registered.containsKey(getWorker());
79 abstract boolean isWorking();
81 protected abstract void submit();
83 abstract void cancel();
86 private class SimpleWorkerManager extends WorkerManager
88 private Future<?> task = null;
90 SimpleWorkerManager(AlignCalcWorkerI worker)
98 return task != null && !task.isDone();
102 protected void submit()
104 if (task != null && !(task.isDone() || task.isCancelled()))
106 throw new IllegalStateException(
107 "Cannot submit new task if the prevoius one is still running");
110 format("Worker %s queued", getWorker()));
111 task = executor.submit(() -> {
114 Console.debug(format("Worker %s started", getWorker()));
116 Console.debug(format("Worker %s finished", getWorker()));
117 } catch (InterruptedException e)
119 Console.debug(format("Worker %s interrupted", getWorker()));
120 } catch (Throwable th)
122 Console.debug(format("Worker %s failed", getWorker()), th);
127 // delete worker reference so garbage collector can remove it
135 synchronized void cancel()
141 Console.debug(format("Cancelling worker %s", getWorker()));
146 private class PollableWorkerManager extends WorkerManager
148 private Future<?> task = null;
150 PollableWorkerManager(PollableAlignCalcWorkerI worker)
156 protected PollableAlignCalcWorkerI getWorker()
158 return (PollableAlignCalcWorkerI) super.getWorker();
164 return task != null && !task.isDone();
167 protected void submit()
169 if (task != null && !(task.isDone() || task.isCancelled()))
171 throw new IllegalStateException(
172 "Cannot submit new task if the previous one is still running");
174 Console.debug( format("Worker %s queued", getWorker()));
175 final var runnable = new Runnable()
177 private boolean started = false;
179 private boolean completed = false;
181 Future<?> future = null;
190 Console.debug(format("Worker %s started", getWorker()));
191 getWorker().startUp();
196 Console.debug(format("Polling worker %s", getWorker()));
197 if (getWorker().poll())
199 Console.debug(format("Worker %s finished", getWorker()));
203 } catch (Throwable th)
205 Console.debug(format("Worker %s failed", getWorker()), th);
210 final var worker = getWorker();
212 PollableWorkerManager.super.worker = null;
213 Console.debug(format("Finalizing completed worker %s", worker));
215 // almost impossible, but the future may be null at this point
216 // let it throw NPE to cancel forcefully
217 future.cancel(false);
221 runnable.future = task = executor.scheduleWithFixedDelay(runnable, 10,
222 1000, TimeUnit.MILLISECONDS);
225 synchronized protected void cancel()
231 Console.debug(format("Cancelling worker %s", getWorker()));
233 executor.submit(() -> {
234 final var worker = getWorker();
236 PollableWorkerManager.super.worker = null;
240 Console.debug(format("Finalizing cancelled worker %s", worker));
247 private final ScheduledExecutorService executor = Executors
248 .newSingleThreadScheduledExecutor();
250 private final Map<AlignCalcWorkerI, WorkerManager> registered = synchronizedMap(
253 private final Map<AlignCalcWorkerI, WorkerManager> oneshot = synchronizedMap(
254 new WeakHashMap<>());
256 private final List<AlignCalcListener> listeners = new CopyOnWriteArrayList<>();
258 private WorkerManager createManager(AlignCalcWorkerI worker)
260 if (worker instanceof PollableAlignCalcWorkerI)
262 return new PollableWorkerManager((PollableAlignCalcWorkerI) worker);
266 return new SimpleWorkerManager(worker);
271 public void registerWorker(AlignCalcWorkerI worker)
273 Objects.requireNonNull(worker);
274 WorkerManager manager = createManager(worker);
275 registered.putIfAbsent(worker, manager);
280 public List<AlignCalcWorkerI> getWorkers()
282 List<AlignCalcWorkerI> result = new ArrayList<>(registered.size());
283 result.addAll(registered.keySet());
288 public List<AlignCalcWorkerI> getWorkersOfClass(
289 Class<? extends AlignCalcWorkerI> cls)
291 List<AlignCalcWorkerI> collected = new ArrayList<>();
292 for (var worker : getWorkers())
294 if (worker.getClass().equals(cls))
296 collected.add(worker);
299 return unmodifiableList(collected);
303 public void removeWorker(AlignCalcWorkerI worker)
305 if (worker.isDeletable())
307 registered.remove(worker);
312 public void removeWorkerForAnnotation(AlignmentAnnotation annot)
314 synchronized (registered)
316 for (var worker : getWorkers())
318 if (worker.involves(annot))
320 removeWorker(worker);
327 public void removeWorkersOfClass(Class<? extends AlignCalcWorkerI> cls)
329 synchronized (registered)
331 for (var worker : getWorkers())
333 if (worker.getClass().equals(cls))
335 removeWorker(worker);
342 public void disableWorker(AlignCalcWorkerI worker)
344 // Null pointer check might be needed
345 registered.get(worker).setEnabled(false);
349 public void enableWorker(AlignCalcWorkerI worker)
351 // Null pointer check might be needed
352 registered.get(worker).setEnabled(true);
356 public boolean isDisabled(AlignCalcWorkerI worker)
358 if (registered.containsKey(worker))
360 return !registered.get(worker).isEnabled();
369 public boolean isWorking(AlignCalcWorkerI worker)
371 var manager = registered.get(worker);
373 manager = oneshot.get(worker);
377 return manager.isWorking();
381 public boolean isWorkingWithAnnotation(AlignmentAnnotation annot)
383 synchronized (registered)
385 for (var entry : registered.entrySet())
386 if (entry.getKey().involves(annot) && entry.getValue().isWorking())
389 synchronized (oneshot)
391 for (var entry : registered.entrySet())
392 if (entry.getKey().involves(annot) && entry.getValue().isWorking())
399 public boolean isWorking()
401 synchronized (registered)
403 for (var manager : registered.values())
404 if (manager.isWorking())
407 synchronized (oneshot)
409 for (var manager : oneshot.values())
410 if (manager.isWorking())
417 public void startWorker(AlignCalcWorkerI worker)
419 Objects.requireNonNull(worker);
420 var manager = registered.get(worker);
423 Console.warn("Starting unregistered worker " + worker);
424 manager = createManager(worker);
425 oneshot.put(worker, manager);
431 public void restartWorkers()
433 synchronized (registered)
435 for (var manager : registered.values())
443 public void cancelWorker(AlignCalcWorkerI worker)
445 Objects.requireNonNull(worker);
446 var manager = registered.get(worker);
448 manager = oneshot.get(worker);
451 throw new NoSuchElementException();
456 private void notifyQueued(AlignCalcWorkerI worker)
458 for (AlignCalcListener listener : listeners)
460 listener.workerQueued(worker);
464 private void notifyStarted(AlignCalcWorkerI worker)
466 for (AlignCalcListener listener : listeners)
468 listener.workerStarted(worker);
472 private void notifyCompleted(AlignCalcWorkerI worker)
474 for (AlignCalcListener listener : listeners)
478 listener.workerCompleted(worker);
479 } catch (RuntimeException e)
486 private void notifyCancelled(AlignCalcWorkerI worker)
488 for (AlignCalcListener listener : listeners)
492 listener.workerCancelled(worker);
493 } catch (RuntimeException e)
500 private void notifyExceptional(AlignCalcWorkerI worker,
503 for (AlignCalcListener listener : listeners)
507 listener.workerExceptional(worker, throwable);
508 } catch (RuntimeException e)
516 public void addAlignCalcListener(AlignCalcListener listener)
518 listeners.add(listener);
522 public void removeAlignCalcListener(AlignCalcListener listener)
524 listeners.remove(listener);
528 public void shutdown()
530 executor.shutdownNow();