platform/platform-impl/src/com/intellij/concurrency/JobImpl.java

   1 /*
   2  * Copyright 2000-2009 JetBrains s.r.o.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 /*
  18  * @author max
  19  */
  20 package com.intellij.concurrency;
  21
  22 import com.intellij.openapi.application.Application;
  23 import com.intellij.openapi.application.ApplicationManager;
  24 import com.intellij.util.containers.ContainerUtil;
  25
  26 import java.util.ArrayList;
  27 import java.util.List;
  28 import java.util.concurrent.*;
  29
  30 public class JobImpl<T> implements Job<T> {
  31   private final String myTitle;
  32   private final List<Callable<T>> myTasks = ContainerUtil.createEmptyCOWList();
  33   private final long myJobIndex = JobSchedulerImpl.currentJobIndex();
  34   private final int myPriority;
  35   private final List<PrioritizedFutureTask<T>> myFutures = ContainerUtil.createEmptyCOWList();
  36   private volatile boolean myCanceled = false;
  37
  38   public JobImpl(String title, int priority) {
  39     myTitle = title;
  40     myPriority = priority;
  41   }
  42
  43   public String getTitle() {
  44     return myTitle;
  45   }
  46
  47   public void addTask(Callable<T> task) {
  48     checkNotStarted();
  49
  50     myTasks.add(task);
  51   }
  52
  53   public void addTask(Runnable task, T result) {
  54     addTask(Executors.callable(task, result));
  55   }
  56
  57   public void addTask(Runnable task) {
  58     addTask(Executors.callable(task, (T)null));
  59   }
  60
  61   public List<T> scheduleAndWaitForResults() throws Throwable {
  62     checkCanSchedule();
  63     final Application application = ApplicationManager.getApplication();
  64     boolean callerHasReadAccess = application != null && application.isReadAccessAllowed();
  65
  66     createFutures(callerHasReadAccess, false);
  67
  68     // Don't bother scheduling if we only have one processor or only one task
  69     if (JobSchedulerImpl.CORES_COUNT >= 2 && myFutures.size() >= 2) {
  70       for (PrioritizedFutureTask<T> future : myFutures) {
  71         JobSchedulerImpl.execute(future);
  72       }
  73     }
  74
  75     // http://gafter.blogspot.com/2006/11/thread-pool-puzzler.html
  76     for (PrioritizedFutureTask<T> future : myFutures) {
  77       future.run();
  78     }
  79
  80     return waitForTermination();
  81   }
  82
  83   private void createFutures(boolean callerHasReadAccess, final boolean reportExceptions) {
  84     int startTaskIndex = JobSchedulerImpl.currentTaskIndex();
  85     for (final Callable<T> task : myTasks) {
  86       final PrioritizedFutureTask<T> future = new PrioritizedFutureTask<T>(task, myJobIndex, startTaskIndex++, myPriority, callerHasReadAccess,
  87                                                                            reportExceptions);
  88       myFutures.add(future);
  89     }
  90   }
  91
  92   public List<T> waitForTermination() throws Throwable {
  93     List<T> results = new ArrayList<T>(myFutures.size());
  94
  95     Throwable ex = null;
  96     for (Future<T> f : myFutures) {
  97       try {
  98         results.add(f.get());
  99       }
 100       catch (CancellationException ignore) {
 101       }
 102       catch (ExecutionException e) {
 103         cancel();
 104
 105         Throwable cause = e.getCause();
 106         if (cause != null) {
 107           ex = cause;
 108         }
 109       }
 110     }
 111
 112     // Future.get() exits when currently running is canceled, thus awaiter may get control before spawned tasks actually terminated,
 113     // that's why additional join logic.
 114     for (PrioritizedFutureTask<T> future : myFutures) {
 115       future.awaitTermination();
 116     }
 117
 118     if (ex != null) throw ex;
 119
 120     return results;
 121   }
 122
 123   public void cancel() {
 124     checkScheduled();
 125     if (myCanceled) return;
 126     myCanceled = true;
 127
 128     for (Future<T> future : myFutures) {
 129       future.cancel(false);
 130     }
 131   }
 132
 133   public boolean isCanceled() {
 134     checkScheduled();
 135     return myCanceled;
 136   }
 137
 138   public void schedule() {
 139     checkCanSchedule();
 140
 141     createFutures(false, true);
 142
 143     for (PrioritizedFutureTask<T> future : myFutures) {
 144       JobSchedulerImpl.execute(future);
 145     }
 146   }
 147
 148   public boolean isDone() {
 149     checkScheduled();
 150
 151     for (Future<T> future : myFutures) {
 152       if (!future.isDone()) return false;
 153     }
 154
 155     return true;
 156   }
 157
 158   private void checkCanSchedule() {
 159     checkNotStarted();
 160     if (myTasks.isEmpty()) {
 161       throw new IllegalStateException("No tasks to run. You can't schedule a job which has no tasks");
 162     }
 163   }
 164
 165   private void checkNotStarted() {
 166     if (!myFutures.isEmpty()) {
 167       throw new IllegalStateException("Already running. You can't call this method for a job which is already scheduled");
 168     }
 169   }
 170
 171   private void checkScheduled() {
 172     if (myFutures.isEmpty()) {
 173       throw new IllegalStateException("Cannot call this method for not yet started job");
 174     }
 175   }
 176 }