Java > Open Source Codes > hudson > model > Queue


1 package hudson.model;
2
3 import hudson.Util;
4 import hudson.model.Node.Mode;
5 import hudson.util.OneShotEvent;
6
7 import java.io.BufferedReader ;
8 import java.io.File ;
9 import java.io.FileInputStream ;
10 import java.io.FileOutputStream ;
11 import java.io.IOException ;
12 import java.io.InputStreamReader ;
13 import java.io.PrintWriter ;
14 import java.util.Calendar ;
15 import java.util.GregorianCalendar ;
16 import java.util.HashMap ;
17 import java.util.HashSet ;
18 import java.util.Iterator ;
19 import java.util.LinkedList ;
20 import java.util.List ;
21 import java.util.Map ;
22 import java.util.Map.Entry;
23 import java.util.Set ;
24 import java.util.TreeSet ;
25 import java.util.logging.Level ;
26 import java.util.logging.Logger ;
27
28 /**
29  * Build queue.
30  *
31  * <p>
32  * This class implements the core scheduling logic.
33  *
34  * @author Kohsuke Kawaguchi
35  */
36 public class Queue {
37     /**
38      * Items in the queue ordered by {@link Item#timestamp}.
39      *
40      * <p>
41      * This consists of {@link Item}s that cannot be run yet
42      * because its time has not yet come.
43      */
44     private final Set <Item> queue = new TreeSet <Item>();
45
46     /**
47      * {@link Project}s that can be built immediately
48      * but blocked because another build is in progress.
49      */
50     private final Set <AbstractProject> blockedProjects = new HashSet <AbstractProject>();
51
52     /**
53      * {@link Project}s that can be built immediately
54      * that are waiting for available {@link Executor}.
55      */
56     private final List <AbstractProject> buildables = new LinkedList <AbstractProject>();
57
58     /**
59      * Data structure created for each idle {@link Executor}.
60      * This is an offer from the queue to an executor.
61      *
62      * <p>
63      * It eventually receives a {@link #project} to build.
64      */
65     private static class JobOffer {
66         final Executor executor;
67
68         /**
69          * Used to wake up an executor, when it has an offered
70          * {@link Project} to build.
71          */
72         final OneShotEvent event = new OneShotEvent();
73         /**
74          * The project that this {@link Executor} is going to build.
75          * (Or null, in which case event is used to trigger a queue maintenance.)
76          */
77         AbstractProject project;
78
79         public JobOffer(Executor executor) {
80             this.executor = executor;
81         }
82
83         public void set(AbstractProject p) {
84             assert this.project==null;
85             this.project = p;
86             event.signal();
87         }
88
89         public boolean isAvailable() {
90             return project==null && !executor.getOwner().isOffline();
91         }
92
93         public Node getNode() {
94             return executor.getOwner().getNode();
95         }
96
97         public boolean isNotExclusive() {
98             return getNode().getMode()== Mode.NORMAL;
99         }
100     }
101
102     private final Map <Executor,JobOffer> parked = new HashMap <Executor,JobOffer>();
103
104     /**
105      * Loads the queue contents that was {@link #save() saved}.
106      */
107     public synchronized void load() {
108         // write out the contents of the queue
109 try {
110             File queueFile = getQueueFile();
111             if(!queueFile.exists())
112                 return;
113
114             BufferedReader in = new BufferedReader (new InputStreamReader (new FileInputStream (queueFile)));
115             String line;
116             while((line=in.readLine())!=null) {
117                 AbstractProject j = Hudson.getInstance().getItemByFullName(line,AbstractProject.class);
118                 if(j!=null)
119                     j.scheduleBuild();
120             }
121             in.close();
122             // discard the queue file now that we are done
123 queueFile.delete();
124         } catch(IOException e) {
125             LOGGER.log(Level.WARNING, "Failed to load the queue file "+getQueueFile(),e);
126         }
127     }
128
129     /**
130      * Persists the queue contents to the disk.
131      */
132     public synchronized void save() {
133         // write out the contents of the queue
134 try {
135             PrintWriter w = new PrintWriter (new FileOutputStream (
136                 getQueueFile()));
137             for (Item i : getItems())
138                 w.println(i.project.getName());
139             w.close();
140         } catch(IOException e) {
141             LOGGER.log(Level.WARNING, "Failed to write out the queue file "+getQueueFile(),e);
142         }
143     }
144
145     private File getQueueFile() {
146         return new File (Hudson.getInstance().getRootDir(),"queue.txt");
147     }
148
149     /**
150      * Schedule a new build for this project.
151      *
152      * @return
153      * true if the project is actually added to the queue.
154      * false if the queue contained it and therefore the add()
155      * was noop
156      */
157     public synchronized boolean add( AbstractProject p ) {
158         if(contains(p))
159             return false; // no double queueing
160
161         LOGGER.fine(p.getName()+" added to queue");
162
163         // put the item in the queue
164 Calendar due = new GregorianCalendar ();
165         due.add(Calendar.SECOND, p.getQuietPeriod());
166         queue.add(new Item(due,p));
167
168         scheduleMaintenance(); // let an executor know that a new item is in the queue.
169 return true;
170     }
171
172     public synchronized void cancel( AbstractProject<?,?> p ) {
173         LOGGER.fine("Cancelling "+p.getName());
174         for (Iterator itr = queue.iterator(); itr.hasNext();) {
175             Item item = (Item) itr.next();
176             if(item.project==p) {
177                 itr.remove();
178                 return;
179             }
180         }
181         blockedProjects.remove(p);
182         buildables.remove(p);
183     }
184
185     public synchronized boolean isEmpty() {
186         return queue.isEmpty() && blockedProjects.isEmpty() && buildables.isEmpty();
187     }
188
189     private synchronized Item peek() {
190         return queue.iterator().next();
191     }
192
193     /**
194      * Gets a snapshot of items in the queue.
195      */
196     public synchronized Item[] getItems() {
197         Item[] r = new Item[queue.size()+blockedProjects.size()+buildables.size()];
198         queue.toArray(r);
199         int idx=queue.size();
200         Calendar now = new GregorianCalendar ();
201         for (AbstractProject p : blockedProjects) {
202             r[idx++] = new Item(now, p, true, false);
203         }
204         for (AbstractProject p : buildables) {
205             r[idx++] = new Item(now, p, false, true);
206         }
207         return r;
208     }
209
210     /**
211      * Returns true if this queue contains the said project.
212      */
213     public synchronized boolean contains(AbstractProject p) {
214         if(blockedProjects.contains(p) || buildables.contains(p))
215             return true;
216         for (Item item : queue) {
217             if (item.project == p)
218                 return true;
219         }
220         return false;
221     }
222
223     /**
224      * Called by the executor to fetch something to build next.
225      *
226      * This method blocks until a next project becomes buildable.
227      */
228     public AbstractProject pop() throws InterruptedException {
229         final Executor exec = Executor.currentExecutor();
230
231         // used in the finally block to check if we are returning from this method normally
232 // or abnormally via an exception
233 boolean successfulReturn = false;
234
235         try {
236             while(true) {
237                 final JobOffer offer = new JobOffer(exec);
238                 long sleep = -1;
239
240                 synchronized(this) {
241                     // consider myself parked
242 assert !parked.containsKey(exec);
243                     parked.put(exec,offer);
244
245                     // reuse executor thread to do a queue maintenance.
246 // at the end of this we get all the buildable jobs
247 // in the buildables field.
248 maintain();
249
250                     // allocate buildable jobs to executors
251 Iterator <AbstractProject> itr = buildables.iterator();
252                     while(itr.hasNext()) {
253                         AbstractProject p = itr.next();
254
255                         // one last check to make sure this build is not blocked.
256 if(p.isBuildBlocked()) {
257                             itr.remove();
258                             blockedProjects.add(p);
259                             continue;
260                         }
261                         
262                         JobOffer runner = choose(p);
263                         if(runner==null)
264                             // if we couldn't find the executor that fits,
265 // just leave it in the buildables list and
266 // check if we can execute other projects
267 continue;
268
269                         // found a matching executor. use it.
270 runner.set(p);
271                         itr.remove();
272                     }
273
274                     // we went over all the buildable projects and awaken
275 // all the executors that got work to do. now, go to sleep
276 // until this thread is awakened. If this executor assigned a job to
277 // itself above, the block method will return immediately.
278
279                     if(!queue.isEmpty()) {
280                         // wait until the first item in the queue is due
281 sleep = peek().timestamp.getTimeInMillis()-new GregorianCalendar ().getTimeInMillis();
282                         if(sleep <100) sleep =100; // avoid wait(0)
283 }
284                 }
285
286                 // this needs to be done outside synchronized block,
287 // so that executors can maintain a queue while others are sleeping
288 if(sleep ==-1)
289                     offer.event.block();
290                 else
291                     offer.event.block(sleep);
292
293                 synchronized(this) {
294                     // retract the offer object
295 assert parked.get(exec)==offer;
296                     parked.remove(exec);
297
298                     // am I woken up because I have a project to build?
299 if(offer.project!=null) {
300                         LOGGER.fine("Pop returning "+offer.project+" for "+exec.getName());
301                         // if so, just build it
302 return offer.project;
303                     }
304                     // otherwise run a queue maintenance
305 }
306             }
307         } finally {
308             synchronized(this) {
309                 // remove myself from the parked list
310 JobOffer offer = parked.remove(exec);
311                 if(offer!=null && offer.project!=null) {
312                     // we are already assigned a project,
313 // ask for someone else to build it.
314 // note that while this thread is waiting for CPU
315 // someone else can schedule this build again,
316 // so check the contains method first.
317 if(!contains(offer.project))
318                         buildables.add(offer.project);
319                 }
320
321                 // since this executor might have been chosen for
322 // maintenance, schedule another one. Worst case
323 // we'll just run a pointless maintenance, and that's
324 // fine.
325 scheduleMaintenance();
326             }
327         }
328     }
329
330     /**
331      * Chooses the executor to carry out the build for the given project.
332      *
333      * @return
334      * null if no {@link Executor} can run it.
335      */
336     private JobOffer choose(AbstractProject<?,?> p) {
337         if(Hudson.getInstance().isQuietingDown()) {
338             // if we are quieting down, don't run anything so that
339 // all executors will be free.
340 return null;
341         }
342
343         Node n = p.getAssignedNode();
344         if(n!=null) {
345             // if a project has assigned node, it can be only built on it
346 for (JobOffer offer : parked.values()) {
347                 if(offer.isAvailable() && offer.getNode()==n)
348                     return offer;
349             }
350             return null;
351         }
352
353         // otherwise let's see if the last node where this project was built is available
354 // it has up-to-date workspace, so that's usually preferable.
355 // (but we can't use an exclusive node)
356 n = p.getLastBuiltOn();
357         if(n!=null && n.getMode()==Mode.NORMAL) {
358             for (JobOffer offer : parked.values()) {
359                 if(offer.isAvailable() && offer.getNode()==n)
360                     return offer;
361             }
362         }
363
364         // duration of a build on a slave tends not to have an impact on
365 // the master/slave communication, so that means we should favor
366 // running long jobs on slaves.
367 AbstractBuild succ = p.getLastSuccessfulBuild();
368         if(succ!=null && succ.getDuration()>15*60*1000) {
369             // consider a long job to be > 15 mins
370 for (JobOffer offer : parked.values()) {
371                 if(offer.isAvailable() && offer.getNode() instanceof Slave && offer.isNotExclusive())
372                     return offer;
373             }
374         }
375
376         // lastly, just look for any idle executor
377 for (JobOffer offer : parked.values()) {
378             if(offer.isAvailable() && offer.isNotExclusive())
379                 return offer;
380         }
381
382         // nothing available
383 return null;
384     }
385
386     /**
387      * Checks the queue and runs anything that can be run.
388      *
389      * <p>
390      * When conditions are changed, this method should be invoked.
391      *
392      * This wakes up one {@link Executor} so that it will maintain a queue.
393      */
394     public synchronized void scheduleMaintenance() {
395         // this code assumes that after this method is called
396 // no more executors will be offered job except by
397 // the pop() code.
398 for (Entry<Executor, JobOffer> av : parked.entrySet()) {
399             if(av.getValue().project==null) {
400                 av.getValue().event.signal();
401                 return;
402             }
403         }
404     }
405
406
407     /**
408      * Queue maintenance.
409      *
410      * Move projects between {@link #queue}, {@link #blockedProjects}, and {@link #buildables}
411      * appropriately.
412      */
413     private synchronized void maintain() {
414         if(LOGGER.isLoggable(Level.FINE))
415             LOGGER.fine("Queue maintenance started "+this);
416
417         Iterator <AbstractProject> itr = blockedProjects.iterator();
418         while(itr.hasNext()) {
419             AbstractProject p = itr.next();
420             if(!p.isBuildBlocked()) {
421                 // ready to be executed
422 LOGGER.fine(p.getName()+" no longer blocked");
423                 itr.remove();
424                 buildables.add(p);
425             }
426         }
427
428         while(!queue.isEmpty()) {
429             Item top = peek();
430
431             if(!top.timestamp.before(new GregorianCalendar ()))
432                 return; // finished moving all ready items from queue
433
434             AbstractProject p = top.project;
435             if(!p.isBuildBlocked()) {
436                 // ready to be executed immediately
437 queue.remove(top);
438                 LOGGER.fine(p.getName()+" ready to build");
439                 buildables.add(p);
440             } else {
441                 // this can't be built now because another build is in progress
442 // set this project aside.
443 queue.remove(top);
444                 LOGGER.fine(p.getName()+" is blocked");
445                 blockedProjects.add(p);
446             }
447         }
448     }
449
450     /**
451      * Item in a queue.
452      */
453     public class Item implements Comparable <Item> {
454         /**
455          * This item can be run after this time.
456          */
457         public final Calendar timestamp;
458
459         /**
460          * Project to be built.
461          */
462         public final AbstractProject<?,?> project;
463
464         /**
465          * Unique number of this {@link Item}.
466          * Used to differentiate {@link Item}s with the same due date.
467          */
468         public final int id;
469
470         /**
471          * Build is blocked because another build is in progress.
472          * This flag is only used in {@link Queue#getItems()} for
473          * 'pseudo' items that are actually not really in the queue.
474          */
475         public final boolean isBlocked;
476
477         /**
478          * Build is waiting the executor to become available.
479          * This flag is only used in {@link Queue#getItems()} for
480          * 'pseudo' items that are actually not really in the queue.
481          */
482         public final boolean isBuildable;
483
484         public Item(Calendar timestamp, AbstractProject project) {
485             this(timestamp,project,false,false);
486         }
487
488         public Item(Calendar timestamp, AbstractProject project, boolean isBlocked, boolean isBuildable) {
489             this.timestamp = timestamp;
490             this.project = project;
491             this.isBlocked = isBlocked;
492             this.isBuildable = isBuildable;
493             synchronized(Queue.this) {
494                 this.id = iota++;
495             }
496         }
497
498         /**
499          * Gets a human-readable status message describing why it's in the queu.
500          */
501         public String getWhy() {
502             if(isBuildable) {
503                 Node node = project.getAssignedNode();
504                 Hudson hudson = Hudson.getInstance();
505                 if(node==hudson && hudson.getSlaves().isEmpty())
506                     node = null; // no master/slave. pointless to talk about nodes
507
508                 String name = null;
509                 if(node!=null) {
510                     if(node==hudson)
511                         name = "master";
512                     else
513                         name = node.getNodeName();
514                 }
515
516                 return "Waiting for next available executor"+(name==null?"":" on "+name);
517             }
518
519             if(isBlocked) {
520                 AbstractBuild<?, ?> build = project.getLastBuild();
521                 Executor e = build.getExecutor();
522                 String eta="";
523                 if(e!=null)
524                     eta = " (ETA:"+e.getEstimatedRemainingTime()+")";
525                 int lbn = build.getNumber();
526                 return "Build #"+lbn+" is already in progress"+eta;
527             }
528
529             long diff = timestamp.getTimeInMillis() - System.currentTimeMillis();
530             if(diff>0) {
531                 return "In the quiet period. Expires in "+ Util.getTimeSpanString(diff);
532             }
533
534             return "???";
535         }
536
537         public int compareTo(Item that) {
538             int r = this.timestamp.getTime().compareTo(that.timestamp.getTime());
539             if(r!=0) return r;
540
541             return this.id-that.id;
542         }
543
544     }
545
546     /**
547      * Unique number generator
548      */
549     private int iota=0;
550
551     private static final Logger LOGGER = Logger.getLogger(Queue.class.getName());
552 }
553

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags