JobManager


1   /*******************************************************************************
2    * Copyright (c) 2000, 2007 IBM Corporation and others.
3    * All rights reserved. This program and the accompanying materials
4    * are made available under the terms of the Eclipse Public License v1.0
5    * which accompanies this distribution, and is available at
6    * http://www.eclipse.org/legal/epl-v10.html
7    *
8    * Contributors:
9    *     IBM Corporation - initial API and implementation
10   *******************************************************************************/
11  package org.eclipse.jdt.internal.core.search.processing;
12  
13  import org.eclipse.core.runtime.*;
14  import org.eclipse.core.runtime.IProgressMonitor;
15  import org.eclipse.core.runtime.OperationCanceledException;
16  import org.eclipse.core.runtime.SubProgressMonitor;
17  import org.eclipse.core.runtime.jobs.Job;
18  import org.eclipse.jdt.internal.core.util.Messages;
19  import org.eclipse.jdt.internal.core.util.Util;
20  
21  public abstract class JobManager implements Runnable   {
22  
23      /* queue of jobs to execute */
24      protected IJob[] awaitingJobs = new IJob[10];
25      protected int jobStart = 0;
26      protected int jobEnd = -1;
27      protected boolean executing = false;
28  
29      /* background processing */
30      protected Thread   processingThread;
31      protected Job progressJob;
32  
33      /* counter indicating whether job execution is enabled or not, disabled if <= 0 
34          it cannot go beyond 1 */
35      private int enableCount = 1;
36  
37      public static boolean VERBOSE = false;
38      /* flag indicating that the activation has completed */
39      public boolean activated = false;
40      
41      private int awaitingClients = 0;
42  
43      /**
44       * Invoked exactly once, in background, before starting processing any job
45       */
46      public void activateProcessing() {
47          this.activated = true;
48      }
49      /**
50       * Answer the amount of awaiting jobs.
51       */
52      public synchronized int awaitingJobsCount() {
53          // pretend busy in case concurrent job attempts performing before activated
54          return this.activated ? this.jobEnd - this.jobStart + 1 : 1;
55      }
56      /**
57       * Answers the first job in the queue, or null if there is no job available
58       * Until the job has completed, the job manager will keep answering the same job.
59       */
60      public synchronized IJob currentJob() {
61          if (this.enableCount > 0 && this.jobStart <= this.jobEnd)
62              return this.awaitingJobs[this.jobStart];
63          return null;
64      }
65      public void disable() {
66          this.enableCount--;
67          if (VERBOSE)
68              Util.verbose("DISABLING background indexing"); //$NON-NLS-1$
69      }
70      /**
71       * Remove the index from cache for a given project.
72       * Passing null as a job family discards them all.
73       */
74      public void discardJobs(String   jobFamily) {
75  
76          if (VERBOSE)
77              Util.verbose("DISCARD   background job family - " + jobFamily); //$NON-NLS-1$
78  
79          try {
80              IJob currentJob;
81              // cancel current job if it belongs to the given family
82              synchronized(this){
83                  currentJob = this.currentJob();
84                  disable();
85              }
86              if (currentJob != null && (jobFamily == null || currentJob.belongsTo(jobFamily))) {
87                  currentJob.cancel();
88  
89                  // wait until current active job has finished
90                  while (this.processingThread != null && this.executing){
91                      try {
92                          if (VERBOSE)
93                              Util.verbose("-> waiting end of current background job - " + currentJob); //$NON-NLS-1$
94                          Thread.sleep(50);
95                      } catch(InterruptedException   e){
96                          // ignore
97                      }
98                  }
99              }
100 
101             // flush and compact awaiting jobs
102             int loc = -1;
103             synchronized(this) {
104                 for (int i = this.jobStart; i <= this.jobEnd; i++) {
105                     currentJob = this.awaitingJobs[i];
106                     if (currentJob != null) { // sanity check
107                         this.awaitingJobs[i] = null;
108                         if (!(jobFamily == null || currentJob.belongsTo(jobFamily))) { // copy down, compacting
109                             this.awaitingJobs[++loc] = currentJob;
110                         } else {
111                             if (VERBOSE)
112                                 Util.verbose("-> discarding background job  - " + currentJob); //$NON-NLS-1$
113                             currentJob.cancel();
114                         }
115                     }
116                 }
117                 this.jobStart = 0;
118                 this.jobEnd = loc;
119             }
120         } finally {
121             enable();
122         }
123         if (VERBOSE)
124             Util.verbose("DISCARD   DONE with background job family - " + jobFamily); //$NON-NLS-1$
125     }
126     public synchronized void enable() {
127         this.enableCount++;
128         if (VERBOSE)
129             Util.verbose("ENABLING  background indexing"); //$NON-NLS-1$
130         this.notifyAll(); // wake up the background thread if it is waiting (context must be synchronized)          
131     }
132     protected synchronized boolean isJobWaiting(IJob request) {
133         for (int i = this.jobEnd; i > this.jobStart; i--) // don't check job at jobStart, as it may have already started
134             if (request.equals(this.awaitingJobs[i])) return true;
135         return false;
136     }
137     /**
138      * Advance to the next available job, once the current one has been completed.
139      * Note: clients awaiting until the job count is zero are still waiting at this point.
140      */
141     protected synchronized void moveToNextJob() {
142         //if (!enabled) return;
143 
144         if (this.jobStart <= this.jobEnd) {
145             this.awaitingJobs[this.jobStart++] = null;
146             if (this.jobStart > this.jobEnd) {
147                 this.jobStart = 0;
148                 this.jobEnd = -1;
149             }
150         }
151     }
152     /**
153      * When idle, give chance to do something
154      */
155     protected void notifyIdle(long idlingTime) {
156         // do nothing
157     }
158     /**
159      * This API is allowing to run one job in concurrence with background processing.
160      * Indeed since other jobs are performed in background, resource sharing might be 
161      * an issue.Therefore, this functionality allows a given job to be run without
162      * colliding with background ones.
163      * Note: multiple thread might attempt to perform concurrent jobs at the same time,
164      *            and should synchronize (it is deliberately left to clients to decide whether
165      *            concurrent jobs might interfere or not. In general, multiple read jobs are ok).
166      *
167      * Waiting policy can be:
168      *      IJobConstants.ForceImmediateSearch
169      *      IJobConstants.CancelIfNotReadyToSearch
170      *      IJobConstants.WaitUntilReadyToSearch
171      *
172      */
173     public boolean performConcurrentJob(IJob searchJob, int waitingPolicy, IProgressMonitor progress) {
174         if (VERBOSE)
175             Util.verbose("STARTING  concurrent job - " + searchJob); //$NON-NLS-1$
176 
177         searchJob.ensureReadyToRun();
178 
179         boolean status = IJob.FAILED;
180         try {
181             int concurrentJobWork = 100;
182             if (progress != null)
183                 progress.beginTask("", concurrentJobWork); //$NON-NLS-1$
184             if (awaitingJobsCount() > 0) {
185                 switch (waitingPolicy) {
186     
187                     case IJob.ForceImmediate :
188                         if (VERBOSE)
189                             Util.verbose("-> NOT READY - forcing immediate - " + searchJob);//$NON-NLS-1$
190                         try {
191                             disable(); // pause indexing
192                             status = searchJob.execute(progress == null ? null : new SubProgressMonitor(progress, concurrentJobWork));
193                         } finally {
194                             enable();
195                         }
196                         if (VERBOSE)
197                             Util.verbose("FINISHED  concurrent job - " + searchJob); //$NON-NLS-1$
198                         return status;
199     
200                     case IJob.CancelIfNotReady :
201                         if (VERBOSE)
202                             Util.verbose("-> NOT READY - cancelling - " + searchJob); //$NON-NLS-1$
203                         if (VERBOSE)
204                             Util.verbose("CANCELED concurrent job - " + searchJob); //$NON-NLS-1$
205                         throw new OperationCanceledException();
206     
207                     case IJob.WaitUntilReady :
208                         IProgressMonitor subProgress = null;
209                         try {
210                             int totalWork = 1000;
211                             if (progress != null) {
212                                 subProgress = new SubProgressMonitor(progress, concurrentJobWork * 8 / 10);
213                                 subProgress.beginTask("", totalWork); //$NON-NLS-1$
214                                 concurrentJobWork = concurrentJobWork * 2 / 10;
215                             }
216                             // use local variable to avoid potential NPE (see bug 20435 NPE when searching java method
217                             // and bug 42760 NullPointerException in JobManager when searching)
218                             Thread   t = this.processingThread;
219                             int originalPriority = t == null ? -1 : t.getPriority();
220                             try {
221                                 if (t != null)
222                                     t.setPriority(Thread.currentThread().getPriority());
223                                 synchronized(this) {
224                                     this.awaitingClients++;
225                                 }
226                                 IJob previousJob = null;
227                                 int awaitingJobsCount;
228                                 int lastJobsCount = totalWork;
229                                 float lastWorked = 0;
230                                 float totalWorked = 0;
231                                 while ((awaitingJobsCount = awaitingJobsCount()) > 0) {
232                                     if (subProgress != null && subProgress.isCanceled())
233                                         throw new OperationCanceledException();
234                                     IJob currentJob = currentJob();
235                                     // currentJob can be null when jobs have been added to the queue but job manager is not enabled
236                                     if (currentJob != null && currentJob != previousJob) {
237                                         if (VERBOSE)
238                                             Util.verbose("-> NOT READY - waiting until ready - " + searchJob);//$NON-NLS-1$
239                                         if (subProgress != null) {
240                                             subProgress.subTask(
241                                                 Messages.bind(Messages.manager_filesToIndex, Integer.toString(awaitingJobsCount)));
242                                             // ratio of the amount of work relative to the total work
243                                             float ratio = awaitingJobsCount < totalWork ? 1 : ((float) totalWork) / awaitingJobsCount;
244                                             if (lastJobsCount > awaitingJobsCount) {
245                                                 totalWorked += (lastJobsCount - awaitingJobsCount) * ratio;
246                                             } else {
247                                                 // more jobs were added, just increment by the ratio
248                                                 totalWorked += ratio;
249                                             }
250                                             if (totalWorked - lastWorked >= 1) {
251                                                 subProgress.worked((int) (totalWorked - lastWorked));
252                                                 lastWorked = totalWorked;
253                                             }
254                                             lastJobsCount = awaitingJobsCount;
255                                         }
256                                         previousJob = currentJob;
257                                     }
258                                     try {
259                                         if (VERBOSE)
260                                             Util.verbose("-> GOING TO SLEEP - " + searchJob);//$NON-NLS-1$
261                                         Thread.sleep(50);
262                                     } catch (InterruptedException   e) {
263                                         // ignore
264                                     }
265                                 }
266                             } finally {
267                                 synchronized(this) {
268                                     this.awaitingClients--;
269                                 }
270                                 if (t != null && originalPriority > -1 && t.isAlive())
271                                     t.setPriority(originalPriority);
272                             }
273                         } finally {
274                             if (subProgress != null)
275                                 subProgress.done();
276                         }
277                 }
278             }
279             status = searchJob.execute(progress == null ? null : new SubProgressMonitor(progress, concurrentJobWork));
280         } finally {
281             if (progress != null)
282                 progress.done();
283             if (VERBOSE)
284                 Util.verbose("FINISHED  concurrent job - " + searchJob); //$NON-NLS-1$
285         }
286         return status;
287     }
288     public abstract String   processName();
289     
290     public synchronized void request(IJob job) {
291 
292         job.ensureReadyToRun();
293 
294         // append the job to the list of ones to process later on
295         int size = this.awaitingJobs.length;
296         if (++this.jobEnd == size) { // when growing, relocate jobs starting at position 0
297             this.jobEnd -= this.jobStart;
298             System.arraycopy(this.awaitingJobs, this.jobStart, this.awaitingJobs = new IJob[size * 2], 0, this.jobEnd);
299             this.jobStart = 0;
300         }
301         this.awaitingJobs[this.jobEnd] = job;
302         if (VERBOSE) {
303             Util.verbose("REQUEST   background job - " + job); //$NON-NLS-1$
304             Util.verbose("AWAITING JOBS count: " + awaitingJobsCount()); //$NON-NLS-1$
305         }
306         notifyAll(); // wake up the background thread if it is waiting
307     }
308     /**
309      * Flush current state
310      */
311     public synchronized void reset() {
312         if (VERBOSE)
313             Util.verbose("Reset"); //$NON-NLS-1$
314 
315         if (this.processingThread != null) {
316             discardJobs(null); // discard all jobs
317         } else {
318             /* initiate background processing */
319             this.processingThread = new Thread  (this, this.processName());
320             this.processingThread.setDaemon(true);
321             // less prioritary by default, priority is raised if clients are actively waiting on it
322             this.processingThread.setPriority(Thread.NORM_PRIORITY-1); 
323             this.processingThread.start();
324         }
325     }
326     /**
327      * Infinite loop performing resource indexing
328      */
329     public void run() {
330 
331         long idlingStart = -1;
332         activateProcessing();
333         try {
334             class ProgressJob extends Job {
335                 ProgressJob(String   name) {
336                     super(name);
337                 }
338                 protected IStatus run(IProgressMonitor monitor) {
339                     int awaitingJobsCount;
340                     while (!monitor.isCanceled() && (awaitingJobsCount = awaitingJobsCount()) > 0) {
341                         monitor.subTask(Messages.bind(Messages.manager_filesToIndex, Integer.toString(awaitingJobsCount))); 
342                         try {
343                             Thread.sleep(500);
344                         } catch (InterruptedException   e) {
345                             // ignore
346                         }
347                     }
348                     return Status.OK_STATUS;
349                 }
350             }
351             this.progressJob = null;
352             while (this.processingThread != null) {
353                 try {
354                     IJob job;
355                     synchronized (this) {
356                         // handle shutdown case when notifyAll came before the wait but after the while loop was entered
357                         if (this.processingThread == null) continue;
358 
359                         // must check for new job inside this sync block to avoid timing hole
360                         if ((job = currentJob()) == null) {
361                             if (this.progressJob != null) {
362                                 this.progressJob.cancel();
363                                 this.progressJob = null;
364                             }
365                             if (idlingStart < 0)
366                                 idlingStart = System.currentTimeMillis();
367                             else
368                                 notifyIdle(System.currentTimeMillis() - idlingStart);
369                             this.wait(); // wait until a new job is posted (or reenabled:38901)
370                         } else {
371                             idlingStart = -1;
372                         }
373                     }
374                     if (job == null) {
375                         notifyIdle(System.currentTimeMillis() - idlingStart);
376                         // just woke up, delay before processing any new jobs, allow some time for the active thread to finish
377                         Thread.sleep(500);
378                         continue;
379                     }
380                     if (VERBOSE) {
381                         Util.verbose(awaitingJobsCount() + " awaiting jobs"); //$NON-NLS-1$
382                         Util.verbose("STARTING background job - " + job); //$NON-NLS-1$
383                     }
384                     try {
385                         this.executing = true;
386                         if (this.progressJob == null) {
387                             this.progressJob = new ProgressJob(Messages.manager_indexingInProgress); 
388                             this.progressJob.setPriority(Job.LONG);
389                             this.progressJob.setSystem(true);
390                             this.progressJob.schedule();
391                         }
392                         /*boolean status = */job.execute(null);
393                         //if (status == FAILED) request(job);
394                     } finally {
395                         this.executing = false;
396                         if (VERBOSE)
397                             Util.verbose("FINISHED background job - " + job); //$NON-NLS-1$
398                         moveToNextJob();
399                         if (this.awaitingClients == 0)
400                             Thread.sleep(50);
401                     }
402                 } catch (InterruptedException   e) { // background indexing was interrupted
403                 }
404             }
405         } catch (RuntimeException   e) {
406             if (this.processingThread != null) { // if not shutting down
407                 // log exception
408                 Util.log(e, "Background Indexer Crash Recovery"); //$NON-NLS-1$
409                 
410                 // keep job manager alive
411                 this.discardJobs(null);
412                 this.processingThread = null;
413                 this.reset(); // this will fork a new thread with no waiting jobs, some indexes will be inconsistent
414             }
415             throw e;
416         } catch (Error   e) {
417             if (this.processingThread != null && !(e instanceof ThreadDeath  )) {
418                 // log exception
419                 Util.log(e, "Background Indexer Crash Recovery"); //$NON-NLS-1$
420                 
421                 // keep job manager alive
422                 this.discardJobs(null);
423                 this.processingThread = null;
424                 this.reset(); // this will fork a new thread with no waiting jobs, some indexes will be inconsistent
425             }
426             throw e;
427         }
428     }
429     /**
430      * Stop background processing, and wait until the current job is completed before returning
431      */
432     public void shutdown() {
433 
434         if (VERBOSE)
435             Util.verbose("Shutdown"); //$NON-NLS-1$
436 
437         disable();
438         discardJobs(null); // will wait until current executing job has completed
439         Thread   thread = this.processingThread;
440         try {
441             if (thread != null) { // see http://bugs.eclipse.org/bugs/show_bug.cgi?id=31858
442                 synchronized (this) {
443                     this.processingThread = null; // mark the job manager as shutting down so that the thread will stop by itself
444                     this.notifyAll(); // ensure its awake so it can be shutdown
445                 }
446                 // in case processing thread is handling a job
447                 thread.join();
448             }
449             Job job = this.progressJob;
450             if (job != null) {
451                 job.cancel();
452                 job.join();
453             }
454         } catch (InterruptedException   e) {
455             // ignore
456         }
457     }
458     public String   toString() {
459         StringBuffer   buffer = new StringBuffer  (10);
460         buffer.append("Enable count:").append(this.enableCount).append('\n'); //$NON-NLS-1$
461         int numJobs = this.jobEnd - this.jobStart + 1;
462         buffer.append("Jobs in queue:").append(numJobs).append('\n'); //$NON-NLS-1$
463         for (int i = 0; i < numJobs && i < 15; i++) {
464             buffer.append(i).append(" - job["+i+"]: ").append(this.awaitingJobs[this.jobStart+i]).append('\n'); //$NON-NLS-1$ //$NON-NLS-2$
465         }
466         return buffer.toString();
467     }   
468 }
469
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags