platform/platform-impl/src/com/intellij/concurrency/JobSchedulerImpl.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.Disposable;
  23 import com.intellij.openapi.application.impl.ApplicationImpl;
  24 import org.jetbrains.annotations.NonNls;
  25
  26 import java.util.concurrent.PriorityBlockingQueue;
  27 import java.util.concurrent.ThreadFactory;
  28 import java.util.concurrent.ThreadPoolExecutor;
  29 import java.util.concurrent.TimeUnit;
  30 import java.util.concurrent.locks.Lock;
  31 import java.util.concurrent.locks.ReentrantLock;
  32
  33 @NonNls
  34 public class JobSchedulerImpl extends JobScheduler implements Disposable {
  35   public static final int CORES_COUNT = /*1;//*/Runtime.getRuntime().availableProcessors();
  36
  37   private static final ThreadFactory WORKERS_FACTORY = new ThreadFactory() {
  38     int i;
  39     public Thread newThread(final Runnable r) {
  40       final Thread thread = new Thread(r, "JobScheduler pool " + i++ + "/" + CORES_COUNT);
  41       thread.setPriority(Thread.NORM_PRIORITY);
  42       return thread;
  43     }
  44   };
  45
  46   private static final Lock ourSuspensionLock = new ReentrantLock();
  47
  48   private static final PriorityBlockingQueue<Runnable> ourQueue = new PriorityBlockingQueue<Runnable>() {
  49     public Runnable poll() {
  50       final Runnable result = super.poll();
  51
  52       ourSuspensionLock.lock();
  53       try {
  54         return result;
  55       }
  56       finally {
  57         ourSuspensionLock.unlock();
  58       }
  59     }
  60
  61     public Runnable poll(final long timeout, final TimeUnit unit) throws InterruptedException {
  62       final Runnable result = super.poll(timeout, unit);
  63
  64       ourSuspensionLock.lock();
  65       try {
  66         return result;
  67       }
  68       finally {
  69         ourSuspensionLock.unlock();
  70       }
  71     }
  72   };
  73   private static final ThreadPoolExecutor ourExecutor = new ThreadPoolExecutor(CORES_COUNT, Integer.MAX_VALUE, 60 * 10, TimeUnit.SECONDS,
  74                                                                                ourQueue, WORKERS_FACTORY) {
  75     protected void beforeExecute(final Thread t, final Runnable r) {
  76       PrioritizedFutureTask task = (PrioritizedFutureTask)r;
  77       if (task.isParentThreadHasReadAccess()) {
  78         ApplicationImpl.setExceptionalThreadWithReadAccessFlag(true);
  79       }
  80       task.signalStarted();
  81
  82       // TODO: hook up JobMonitor into thread locals
  83       super.beforeExecute(t, r);
  84     }
  85
  86     protected void afterExecute(final Runnable r, final Throwable t) {
  87       super.afterExecute(r, t);
  88       ApplicationImpl.setExceptionalThreadWithReadAccessFlag(false);
  89       PrioritizedFutureTask task = (PrioritizedFutureTask)r;
  90       task.signalDone();
  91       // TODO: cleanup JobMonitor
  92     }
  93   };
  94
  95   private static volatile long ourJobsCounter = 0;
  96
  97   public static void execute(Runnable task) {
  98     ourExecutor.execute(task);
  99   }
 100
 101   public static int currentTaskIndex() {
 102     final PrioritizedFutureTask topTask = (PrioritizedFutureTask)ourQueue.peek();
 103     return topTask == null ? 0 : topTask.getTaskIndex();
 104   }
 105
 106   public static long currentJobIndex() {
 107     return ourJobsCounter++;
 108   }
 109
 110   public static void suspend() {
 111     ourSuspensionLock.lock();
 112   }
 113
 114   public static void resume() {
 115     ourSuspensionLock.unlock();
 116   }
 117
 118   public <T> Job<T> createJob(String title, int priority) {
 119     return new JobImpl<T>(title, priority);
 120   }
 121
 122   public void dispose() {
 123     ((ThreadPoolExecutor)getScheduler()).getQueue().clear();
 124   }
 125 }