platform/vcs-impl/src/com/intellij/openapi/vcs/changes/UpdateRequestsQueue.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 package com.intellij.openapi.vcs.changes;
  17
  18 import com.intellij.ide.startup.impl.StartupManagerImpl;
  19 import com.intellij.openapi.application.ApplicationManager;
  20 import com.intellij.openapi.application.ModalityState;
  21 import com.intellij.openapi.diagnostic.Logger;
  22 import com.intellij.openapi.progress.SomeQueue;
  23 import com.intellij.openapi.project.Project;
  24 import com.intellij.openapi.startup.StartupManager;
  25 import com.intellij.openapi.util.Getter;
  26 import com.intellij.openapi.vcs.ProjectLevelVcsManager;
  27 import com.intellij.util.Consumer;
  28 import com.intellij.util.concurrency.Semaphore;
  29 import com.intellij.util.io.storage.HeavyProcessLatch;
  30 import org.jetbrains.annotations.Nullable;
  31 import org.jetbrains.annotations.TestOnly;
  32
  33 import javax.swing.*;
  34 import java.util.ArrayList;
  35 import java.util.List;
  36 import java.util.concurrent.ScheduledExecutorService;
  37 import java.util.concurrent.TimeUnit;
  38 import java.util.concurrent.atomic.AtomicReference;
  39
  40 /**
  41  * ChangeListManager updates scheduler.
  42  * Tries to zip several update requests into one (if starts and see several requests in the queue)
  43  * own inner synchronization
  44  */
  45 @SomeQueue
  46 public class UpdateRequestsQueue {
  47   private final Logger LOG = Logger.getInstance("#com.intellij.openapi.vcs.changes.UpdateRequestsQueue");
  48   private static final String ourHeavyLatchOptimization = "vcs.local.changes.track.heavy.latch";
  49   private final Project myProject;
  50   private final AtomicReference<ScheduledExecutorService> myExecutor;
  51   private final Runnable myDelegate;
  52   private final Object myLock;
  53   private volatile boolean myStarted;
  54   private volatile boolean myStopped;
  55   private volatile boolean myIgnoreBackgroundOperation;
  56
  57   private boolean myRequestSubmitted;
  58   private boolean myRequestRunning;
  59   private final List<Runnable> myWaitingUpdateCompletionQueue;
  60   private final List<Semaphore> myWaitingUpdateCompletionSemaphores = new ArrayList<Semaphore>();
  61   private final ProjectLevelVcsManager myPlVcsManager;
  62   //private final ScheduledSlowlyClosingAlarm mySharedExecutor;
  63   private final StartupManager myStartupManager;
  64   private final boolean myTrackHeavyLatch;
  65   private final Getter<Boolean> myIsStoppedGetter;
  66
  67   public UpdateRequestsQueue(final Project project, final AtomicReference<ScheduledExecutorService> executor, final Runnable delegate) {
  68     myProject = project;
  69     myExecutor = executor;
  70     myTrackHeavyLatch = Boolean.parseBoolean(System.getProperty(ourHeavyLatchOptimization));
  71
  72     myDelegate = delegate;
  73     myPlVcsManager = ProjectLevelVcsManager.getInstance(myProject);
  74     myStartupManager = StartupManager.getInstance(myProject);
  75     myLock = new Object();
  76     myWaitingUpdateCompletionQueue = new ArrayList<Runnable>();
  77     // not initialized
  78     myStarted = false;
  79     myStopped = false;
  80     myIsStoppedGetter = new Getter<Boolean>() {
  81       @Override
  82       public Boolean get() {
  83         return isStopped();
  84       }
  85     };
  86   }
  87
  88   public void initialized() {
  89     LOG.debug("Initialized for project: " + myProject.getName());
  90     myStarted = true;
  91   }
  92
  93   public Getter<Boolean> getIsStoppedGetter() {
  94     return myIsStoppedGetter;
  95   }
  96
  97   public boolean isStopped() {
  98     return myStopped;
  99   }
 100
 101   public void schedule() {
 102     synchronized (myLock) {
 103       if (! myStarted && ApplicationManager.getApplication().isUnitTestMode()) return;
 104
 105       if (! myStopped) {
 106         if (! myRequestSubmitted) {
 107           final MyRunnable runnable = new MyRunnable();
 108           myRequestSubmitted = true;
 109           myExecutor.get().schedule(runnable, 300, TimeUnit.MILLISECONDS);
 110           LOG.debug("Scheduled for project: " + myProject.getName() + ", runnable: " + runnable.hashCode());
 111         }
 112       }
 113     }
 114   }
 115
 116   public void pause() {
 117     synchronized (myLock) {
 118       myStopped = true;
 119     }
 120   }
 121
 122   public void forceGo() {
 123     synchronized (myLock) {
 124       myStopped = false;
 125       myRequestSubmitted = false;
 126       myRequestRunning = false;
 127     }
 128     schedule();
 129   }
 130
 131   public void go() {
 132     synchronized (myLock) {
 133       myStopped = false;
 134     }
 135     schedule();
 136   }
 137
 138   public void stop() {
 139     LOG.debug("Calling stop for project: " + myProject.getName());
 140     final List<Runnable> waiters = new ArrayList<Runnable>(myWaitingUpdateCompletionQueue.size());
 141     synchronized (myLock) {
 142       myStopped = true;
 143       waiters.addAll(myWaitingUpdateCompletionQueue);
 144       myWaitingUpdateCompletionQueue.clear();
 145     }
 146     LOG.debug("Calling runnables in stop for project: " + myProject.getName());
 147     // do not run under lock
 148     for (Runnable runnable : waiters) {
 149       runnable.run();
 150     }
 151     LOG.debug("Stop finished for project: " + myProject.getName());
 152   }
 153
 154   @TestOnly
 155   public void waitUntilRefreshed() {
 156     while (true) {
 157       final Semaphore semaphore = new Semaphore();
 158       synchronized (myLock) {
 159         if (!myRequestSubmitted && !myRequestRunning) {
 160           return;
 161         }
 162
 163         if (!myRequestRunning) {
 164           myExecutor.get().schedule(new MyRunnable(), 0, TimeUnit.MILLISECONDS);
 165         }
 166
 167         semaphore.down();
 168         myWaitingUpdateCompletionSemaphores.add(semaphore);
 169       }
 170       if (!semaphore.waitFor(100*1000)) {
 171         LOG.error("Too long VCS update");
 172         return;
 173       }
 174     }
 175   }
 176
 177   private void freeSemaphores() {
 178     synchronized (myLock) {
 179       for (Semaphore semaphore : myWaitingUpdateCompletionSemaphores) {
 180         semaphore.up();
 181       }
 182       myWaitingUpdateCompletionSemaphores.clear();
 183     }
 184   }
 185
 186   public void invokeAfterUpdate(final Runnable afterUpdate, final InvokeAfterUpdateMode mode, final String title,
 187                                 @Nullable final Consumer<VcsDirtyScopeManager> dirtyScopeManagerFiller, final ModalityState state) {
 188     LOG.debug("invokeAfterUpdate for project: " + myProject.getName());
 189     final CallbackData data = CallbackData.create(afterUpdate, title, state, mode, myProject);
 190
 191     VcsDirtyScopeManagerProxy managerProxy = null;
 192     if (dirtyScopeManagerFiller != null) {
 193       managerProxy  = new VcsDirtyScopeManagerProxy();
 194       dirtyScopeManagerFiller.consume(managerProxy);
 195     }
 196
 197     // can ask stopped without a lock
 198     if (! myStopped) {
 199       if (managerProxy != null) {
 200         managerProxy.callRealManager(VcsDirtyScopeManager.getInstance(myProject));
 201       }
 202     }
 203
 204     synchronized (myLock) {
 205       if (! myStopped) {
 206         myWaitingUpdateCompletionQueue.add(data.getCallback());
 207         schedule();
 208       }
 209     }
 210     // do not run under lock; stopped cannot be switched into not stopped - can check without lock
 211     if (myStopped) {
 212       LOG.debug("invokeAfterUpdate: stopped, invoke right now for project: " + myProject.getName());
 213       SwingUtilities.invokeLater(new Runnable() {
 214         public void run() {
 215           if (!myProject.isDisposed()) {
 216             afterUpdate.run();
 217           }
 218         }
 219       });
 220       return;
 221     }
 222     // invoke progress if needed
 223     if (data.getWrapperStarter() != null) {
 224       data.getWrapperStarter().run();
 225     }
 226     LOG.debug("invokeAfterUpdate: exit for project: " + myProject.getName());
 227   }
 228
 229   // true = do not execute
 230   private boolean checkHeavyOperations() {
 231     if (myIgnoreBackgroundOperation) return false;
 232     return myPlVcsManager.isBackgroundVcsOperationRunning() || myTrackHeavyLatch && HeavyProcessLatch.INSTANCE.isRunning();
 233   }
 234
 235   // true = do not execute
 236   private boolean checkLifeCycle() {
 237     return !myStarted || !((StartupManagerImpl)myStartupManager).startupActivityPassed();
 238   }
 239
 240   private class MyRunnable implements Runnable {
 241     public void run() {
 242       final List<Runnable> copy = new ArrayList<Runnable>(myWaitingUpdateCompletionQueue.size());
 243       try {
 244         synchronized (myLock) {
 245           if (!myRequestSubmitted) return;
 246
 247           LOG.assertTrue(!myRequestRunning);
 248           myRequestRunning = true;
 249           if (myStopped) {
 250             myRequestSubmitted = false;
 251             LOG.debug("MyRunnable: STOPPED, project: " + myProject.getName() + ", runnable: " + hashCode());
 252             return;
 253           }
 254
 255           if (checkLifeCycle() || checkHeavyOperations()) {
 256             LOG.debug("MyRunnable: reschedule, project: " + myProject.getName() + ", runnable: " + hashCode());
 257             myRequestSubmitted = false;
 258             // try again after time
 259             schedule();
 260             return;
 261           }
 262
 263           copy.addAll(myWaitingUpdateCompletionQueue);
 264           myRequestSubmitted = false;
 265         }
 266
 267         LOG.debug("MyRunnable: INVOKE, project: " + myProject.getName() + ", runnable: " + hashCode());
 268         myDelegate.run();
 269         LOG.debug("MyRunnable: invokeD, project: " + myProject.getName() + ", runnable: " + hashCode());
 270       } finally {
 271         synchronized (myLock) {
 272           myRequestRunning = false;
 273           LOG.debug("MyRunnable: delete executed, project: " + myProject.getName() + ", runnable: " + hashCode());
 274           if (! copy.isEmpty()) {
 275             myWaitingUpdateCompletionQueue.removeAll(copy);
 276           }
 277
 278           if (! myWaitingUpdateCompletionQueue.isEmpty() && ! myRequestSubmitted && ! myStopped) {
 279             LOG.error("No update task to handle request(s)");
 280           }
 281         }
 282         // do not run under lock
 283         for (Runnable runnable : copy) {
 284           runnable.run();
 285         }
 286         freeSemaphores();
 287         LOG.debug("MyRunnable: Runnables executed, project: " + myProject.getName() + ", runnable: " + hashCode());
 288       }
 289     }
 290   }
 291
 292   public void setIgnoreBackgroundOperation(boolean ignoreBackgroundOperation) {
 293     myIgnoreBackgroundOperation = ignoreBackgroundOperation;
 294   }
 295 }