Attachment #10245 for bug #33467

This Bugzilla instance is a read-only archive of historic NetBeans bug reports. To report a bug in NetBeans please follow the project's instructions for reporting issues.

View | Details | Raw Unified | Return to bug 33467
Collapse All | Expand All




 * <CODE>RequestProcessor</CODE> instance with limited throughput (probably
 * set to 1), the IDE would try to run all your requests in parallel otherwise.
 *
 * <P>
 * Since version 4.6 there is a support for interruption of long running tasks.
 * There always was a way how to cancel not yet running task using {@link RequestProcessor.Task#cancel }
 * but if the task was already running, one was out of luck. Since version 4.6
 * the thread running the task is interrupted and the Runnable can check for that
 * and terminate its execution sooner. In the runnable one shall check for 
 * thread interruption (done from {@link RequestProcessor.Task#cancel }) and 
 * if true, return immediatelly as in this example:
 * <PRE>
 * public void run () {
 *     while (veryLongTimeLook) {
 *       doAPieceOfIt ();
 *
 *       if (Thread.interrupted ()) return;
 *     }
 * }
 * </PRE>
 * 
 * @author Petr Nejedly, Jaroslav Tulach
 */
public final class RequestProcessor {

            final Item localItem;
            synchronized (processorLock) {
                notifyRunning();
                if (item != null) item.clear(null);
                item = new Item(this, RequestProcessor.this);
                localItem = item;
            }

        */
        public boolean cancel () {
            synchronized (processorLock) {
                boolean success;
                if (item == null) {
                    success = false;
                } else {
                    Processor p = item.getProcessor();
                    success = item.clear(null);
                    if (p != null) {
                        p.interruptTask (this);
                    }
                }
                if (success) notifyFinished(); // mark it as finished
                return success;
            }

                synchronized (processorLock) {
// correct line:    toRun = (item == null) ? !isFinished (): (item.clear() && !isFinished ());
// the same:        toRun = !isFinished () && (item == null ? true : item.clear ());
                    toRun = !isFinished () && (item == null || item.clear (null));
                }

                if (toRun) { //System.err.println("    ## running it synchronously");

        }
    }

    /** Called under the processorLock */
    Task askForWork (Processor worker) {
        if (stopped || queue.isEmpty()) { // no more work in this burst, return him
            processors.remove(worker);
            Processor.put(worker);
            running--;
            return null;
        } else { // we have some work for the worker, pass it
            Item i = (Item) queue.remove(0);
            Task t = i.getTask ();
            i.clear (worker);
            return t;
                return t;
            }
        }
    }


    /* One item representing the task pending in the pending queue */ 
    private static class Item extends Exception {
        private final RequestProcessor owner;
        private Object action;
        private boolean enqueued;
        
        Item (Task task, RequestProcessor rp) {

        }

        Task getTask() {
            Object a = action;
            return a instanceof Task ? (Task)a : null;
        }
            
        /** Annulate this request iff still possible.
         * @returns true if it was possible to skip this item, false
         * if the item was/is already processed */ 
        boolean clear(Processor processor) {
            synchronized (owner.processorLock) {
                action = processor;
                return enqueued ? owner.queue.remove(this) : true;
            }
        }
        
        Processor getProcessor () {
            Object a = action;
            return a instanceof Processor ? (Processor)a : null;
        }

        int getPriority() {
            return getTask ().getPriority();
        }
        
        public Throwable fillInStackTrace() {


        private RequestProcessor source;
        
        /** task we are working on */
        private RequestProcessor.Task todo;
        
        /* One minute of inactivity and the Thread will die if not assigned */
        private static final int INACTIVE_TIMEOUT = 60000;
	

                    }
		}
                
                Task todo;

                ErrorManager em = logger();
                boolean loggable = em.isLoggable(ErrorManager.INFORMATIONAL);
                if (loggable) em.log (ErrorManager.INFORMATIONAL, "Begining work " + getName ()); // NOI18N
                // while we have something to do
                for (;;) {
                    // need the same sync as interruptTask
                    synchronized (current.processorLock) {
                        todo = current.askForWork(this);
                        if (todo == null) break;
                    }
                    
                    if(todo != null) {
                        setPrio(todo.getPriority());
                        try {

                            doNotify(todo, e);
                        }
                        // do not catch e.g. OutOfMemoryError, etc.

                        // need the same sync as interruptTask
                        synchronized (current.processorLock) {
                            // to improve GC
                            todo = null;
                            // and to clear any possible interrupted state
                            // set by calling Task.cancel ()
                            Thread.interrupted();
                        }
                    }
                }
                
                if (loggable) logger ().log (ErrorManager.INFORMATIONAL, "Work finished " + getName ()); // NOI18N
	    }
	}

        /** Called under the processorLock */
        public void interruptTask (Task t) {
            if (t != todo) {
                // not running this task so
                return;
            }

            // otherwise interrupt this thread
            interrupt ();
        }
        
        /** @see "#20467" */
        private static void doNotify(RequestProcessor.Task todo, Throwable ex) {




        assertTrue("Task was performed even if it is canceled", !x.performed);
    }
    
    public void testCancelInterruptsTheRunningThread () throws Exception {
        RequestProcessor rp = new RequestProcessor ("Cancellable");
        
        class R implements Runnable {
            public boolean checkBefore;
            public boolean checkAfter;
            public boolean interrupted;
            
            public synchronized void run () {
                checkBefore = Thread.interrupted();
                
                notifyAll ();
                
                try {
                    wait ();
                    interrupted = false;
                } catch (InterruptedException ex) {
                    interrupted = true;
                }
                
                notifyAll ();
                
                try {
                    wait ();
                } catch (InterruptedException ex) {
                }
                
                checkAfter = Thread.interrupted();
                
                notifyAll ();
            }
        }
        
        R r = new R ();
        synchronized (r) {
            RequestProcessor.Task t = rp.post (r);
            r.wait ();
            assertTrue ("The task is already running", !t.cancel ());
            r.wait ();
            r.notifyAll ();
            r.wait ();
            assertTrue ("The task has been interrupted", r.interrupted);
            assertTrue ("Not before", !r.checkBefore);
            assertTrue ("Not after - as the notification was thru InterruptedException", !r.checkAfter);
        }
        
        // interrupt after the task has finished
        r = new R ();
        synchronized (r) {
            RequestProcessor.Task t = rp.post (r);
            r.wait ();
            r.notifyAll ();
            r.wait ();
            assertTrue ("The task is already running", !t.cancel ());
            r.notifyAll ();
            r.wait ();
            assertTrue ("The task has not been interrupted by exception", !r.interrupted);
            assertTrue ("Not interupted before", !r.checkBefore);
            assertTrue ("But interupted after", r.checkAfter);
        }
    }
    
    public void testInterruptedStatusIsClearedBetweenTwoTaskExecution () throws Exception {
        RequestProcessor rp = new RequestProcessor ("testInterruptedStatusIsClearedBetweenTwoTaskExecution");
        
        final RequestProcessor.Task[] task = new RequestProcessor.Task[1];
        // test interrupted status is cleared after task ends
        class Fail implements Runnable {
            public boolean checkBefore;
            public Thread runIn;
            public boolean goodThread;
            
            public synchronized void run () {
                if (runIn == null) {
                    runIn = Thread.currentThread();
                    task[0].schedule (0);
                } else {
                    goodThread = Thread.currentThread () == runIn;
                }
                    
                checkBefore = runIn.isInterrupted();
                // set the flag for next execution
                runIn.interrupt();
                
                notifyAll ();
            }
        }
        
        Fail f = new Fail ();
        synchronized (f) {
            task[0] = rp.post (f);
            
            // wait for the first execution
            f.wait ();
        }
        // wait for the second
        task[0].waitFinished ();
        
        assertTrue ("This shall be always true, but if not, than it does not mean too much"
            + " just that the tasks were not executed in the same thread. In such case it "
            + " this test does not do anything useful as it needs to execute the task twice "
            + " in the same thread", f.goodThread);
        
        assertTrue ("Interrupted state has been cleared between two executions of the task", !f.checkBefore);
            
    }
    
    private static void doGc (int count, Reference toClear) {
        java.util.ArrayList l = new java.util.ArrayList (count);
        while (count-- > 0) {

Return to bug 33467

Lines 63-68 Link Here

(-)src/org/openide/util/RequestProcessor.java (-26 / +82 lines)
63	* <CODE>RequestProcessor</CODE> instance with limited throughput (probably	63	* <CODE>RequestProcessor</CODE> instance with limited throughput (probably
64	* set to 1), the IDE would try to run all your requests in parallel otherwise.	64	* set to 1), the IDE would try to run all your requests in parallel otherwise.
65	*	65	*
		66	* <P>
		67	* Since version 4.6 there is a support for interruption of long running tasks.
		68	* There always was a way how to cancel not yet running task using {@link RequestProcessor.Task#cancel }
		69	* but if the task was already running, one was out of luck. Since version 4.6
		70	* the thread running the task is interrupted and the Runnable can check for that
		71	* and terminate its execution sooner. In the runnable one shall check for
		72	* thread interruption (done from {@link RequestProcessor.Task#cancel }) and
		73	* if true, return immediatelly as in this example:
		74	* <PRE>
		75	* public void run () {
		76	* while (veryLongTimeLook) {
		77	* doAPieceOfIt ();
		78	*
		79	* if (Thread.interrupted ()) return;
		80	* }
		81	* }
		82	* </PRE>
		83	*
66	* @author Petr Nejedly, Jaroslav Tulach	84	* @author Petr Nejedly, Jaroslav Tulach
67	*/	85	*/
68	public final class RequestProcessor {	86	public final class RequestProcessor {
Lines 356-362 Link Here
356	final Item localItem;	374	final Item localItem;
357	synchronized (processorLock) {	375	synchronized (processorLock) {
358	notifyRunning();	376	notifyRunning();
359	if (item != null) item.clear();	377	if (item != null) item.clear(null);
360	item = new Item(this, RequestProcessor.this);	378	item = new Item(this, RequestProcessor.this);
361	localItem = item;	379	localItem = item;
362	}	380	}
Lines 381-387 Link Here
381	*/	399	*/
382	public boolean cancel () {	400	public boolean cancel () {
383	synchronized (processorLock) {	401	synchronized (processorLock) {
384	boolean success = (item == null) ? false : item.clear();	402	boolean success;
		403	if (item == null) {
		404	success = false;
		405	} else {
		406	Processor p = item.getProcessor();
		407	success = item.clear(null);
		408	if (p != null) {
		409	p.interruptTask (this);
		410	}
		411	}
385	if (success) notifyFinished(); // mark it as finished	412	if (success) notifyFinished(); // mark it as finished
386	return success;	413	return success;
387	}	414	}
Lines 422-428 Link Here
422	synchronized (processorLock) {	449	synchronized (processorLock) {
423	// correct line: toRun = (item == null) ? !isFinished (): (item.clear() && !isFinished ());	450	// correct line: toRun = (item == null) ? !isFinished (): (item.clear() && !isFinished ());
424	// the same: toRun = !isFinished () && (item == null ? true : item.clear ());	451	// the same: toRun = !isFinished () && (item == null ? true : item.clear ());
425	toRun = !isFinished () && (item == null \|\| item.clear ());	452	toRun = !isFinished () && (item == null \|\| item.clear (null));
426	}	453	}
427		454
428	if (toRun) { //System.err.println(" ## running it synchronously");	455	if (toRun) { //System.err.println(" ## running it synchronously");
Lines 494-512 Link Here
494	}	521	}
495	}	522	}
496		523
		524	/** Called under the processorLock */
497	Task askForWork (Processor worker) {	525	Task askForWork (Processor worker) {
498	synchronized (processorLock) {	526	if (stopped \|\| queue.isEmpty()) { // no more work in this burst, return him
499	if (stopped \|\| queue.isEmpty()) { // no more work in this burst, return him	527	processors.remove(worker);
500	processors.remove(worker);	528	Processor.put(worker);
501	Processor.put(worker);	529	running--;
502	running--;	530	return null;
503	return null;	531	} else { // we have some work for the worker, pass it
504	} else { // we have some work for the worker, pass it	532	Item i = (Item) queue.remove(0);
505	Item i = (Item) queue.remove(0);	533	Task t = i.getTask ();
506	Task t = i.getTask ();	534	i.clear (worker);
507	i.clear ();	535	return t;
508	return t;
509	}
510	}	536	}
511	}	537	}
512		538
Lines 515-521 Link Here
515	/* One item representing the task pending in the pending queue */	541	/* One item representing the task pending in the pending queue */
516	private static class Item extends Exception {	542	private static class Item extends Exception {
517	private final RequestProcessor owner;	543	private final RequestProcessor owner;
518	private Task action;	544	private Object action;
519	private boolean enqueued;	545	private boolean enqueued;
520		546
521	Item (Task task, RequestProcessor rp) {	547	Item (Task task, RequestProcessor rp) {
Lines 525-545 Link Here
525	}	551	}
526		552
527	Task getTask() {	553	Task getTask() {
528	return action;	554	Object a = action;
		555	return a instanceof Task ? (Task)a : null;
529	}	556	}
530		557
531	/** Annulate this request iff still possible.	558	/** Annulate this request iff still possible.
532	* @returns true if it was possible to skip this item, false	559	* @returns true if it was possible to skip this item, false
533	* if the item was/is already processed */	560	* if the item was/is already processed */
534	boolean clear() {	561	boolean clear(Processor processor) {
535	synchronized (owner.processorLock) {	562	synchronized (owner.processorLock) {
536	action = null;	563	action = processor;
537	return enqueued ? owner.queue.remove(this) : true;	564	return enqueued ? owner.queue.remove(this) : true;
538	}	565	}
539	}	566	}
		567
		568	Processor getProcessor () {
		569	Object a = action;
		570	return a instanceof Processor ? (Processor)a : null;
		571	}
540		572
541	int getPriority() {	573	int getPriority() {
542	return action.getPriority();	574	return getTask ().getPriority();
543	}	575	}
544		576
545	public Throwable fillInStackTrace() {	577	public Throwable fillInStackTrace() {
Lines 603-608 Link Here
603		635
604	private RequestProcessor source;	636	private RequestProcessor source;
605		637
		638	/** task we are working on */
		639	private RequestProcessor.Task todo;
		640
606	/* One minute of inactivity and the Thread will die if not assigned */	641	/* One minute of inactivity and the Thread will die if not assigned */
607	private static final int INACTIVE_TIMEOUT = 60000;	642	private static final int INACTIVE_TIMEOUT = 60000;
608		643
Lines 663-675 Link Here
663	}	698	}
664	}	699	}
665		700
666	Task todo;
667
668	ErrorManager em = logger();	701	ErrorManager em = logger();
669	boolean loggable = em.isLoggable(ErrorManager.INFORMATIONAL);	702	boolean loggable = em.isLoggable(ErrorManager.INFORMATIONAL);
670	if (loggable) em.log (ErrorManager.INFORMATIONAL, "Begining work " + getName ()); // NOI18N	703	if (loggable) em.log (ErrorManager.INFORMATIONAL, "Begining work " + getName ()); // NOI18N
671	// while we have something to do	704	// while we have something to do
672	while((todo = current.askForWork(this)) != null) {	705	for (;;) {
		706	// need the same sync as interruptTask
		707	synchronized (current.processorLock) {
		708	todo = current.askForWork(this);
		709	if (todo == null) break;
		710	}
		711
673	if(todo != null) {	712	if(todo != null) {
674	setPrio(todo.getPriority());	713	setPrio(todo.getPriority());
675	try {	714	try {
Lines 685-699 Link Here
685	doNotify(todo, e);	724	doNotify(todo, e);
686	}	725	}
687	// do not catch e.g. OutOfMemoryError, etc.	726	// do not catch e.g. OutOfMemoryError, etc.
688		727
689	// to improve GC	728	// need the same sync as interruptTask
690	todo = null;	729	synchronized (current.processorLock) {
		730	// to improve GC
		731	todo = null;
		732	// and to clear any possible interrupted state
		733	// set by calling Task.cancel ()
		734	Thread.interrupted();
		735	}
691	}	736	}
692	}	737	}
693		738
694	if (loggable) logger ().log (ErrorManager.INFORMATIONAL, "Work finished " + getName ()); // NOI18N	739	if (loggable) logger ().log (ErrorManager.INFORMATIONAL, "Work finished " + getName ()); // NOI18N
695	}	740	}
696	}	741	}
		742
		743	/** Called under the processorLock */
		744	public void interruptTask (Task t) {
		745	if (t != todo) {
		746	// not running this task so
		747	return;
		748	}
		749
		750	// otherwise interrupt this thread
		751	interrupt ();
		752	}
697		753
698	/** @see "#20467" */	754	/** @see "#20467" */
699	private static void doNotify(RequestProcessor.Task todo, Throwable ex) {	755	private static void doNotify(RequestProcessor.Task todo, Throwable ex) {

Lines 564-569 Link Here

(-)test/unit/src/org/openide/util/RequestProcessorTest.java (+107 lines)
564	assertTrue("Task was performed even if it is canceled", !x.performed);	564	assertTrue("Task was performed even if it is canceled", !x.performed);
565	}	565	}
566		566
		567	public void testCancelInterruptsTheRunningThread () throws Exception {
		568	RequestProcessor rp = new RequestProcessor ("Cancellable");
		569
		570	class R implements Runnable {
		571	public boolean checkBefore;
		572	public boolean checkAfter;
		573	public boolean interrupted;
		574
		575	public synchronized void run () {
		576	checkBefore = Thread.interrupted();
		577
		578	notifyAll ();
		579
		580	try {
		581	wait ();
		582	interrupted = false;
		583	} catch (InterruptedException ex) {
		584	interrupted = true;
		585	}
		586
		587	notifyAll ();
		588
		589	try {
		590	wait ();
		591	} catch (InterruptedException ex) {
		592	}
		593
		594	checkAfter = Thread.interrupted();
		595
		596	notifyAll ();
		597	}
		598	}
		599
		600	R r = new R ();
		601	synchronized (r) {
		602	RequestProcessor.Task t = rp.post (r);
		603	r.wait ();
		604	assertTrue ("The task is already running", !t.cancel ());
		605	r.wait ();
		606	r.notifyAll ();
		607	r.wait ();
		608	assertTrue ("The task has been interrupted", r.interrupted);
		609	assertTrue ("Not before", !r.checkBefore);
		610	assertTrue ("Not after - as the notification was thru InterruptedException", !r.checkAfter);
		611	}
		612
		613	// interrupt after the task has finished
		614	r = new R ();
		615	synchronized (r) {
		616	RequestProcessor.Task t = rp.post (r);
		617	r.wait ();
		618	r.notifyAll ();
		619	r.wait ();
		620	assertTrue ("The task is already running", !t.cancel ());
		621	r.notifyAll ();
		622	r.wait ();
		623	assertTrue ("The task has not been interrupted by exception", !r.interrupted);
		624	assertTrue ("Not interupted before", !r.checkBefore);
		625	assertTrue ("But interupted after", r.checkAfter);
		626	}
		627	}
		628
		629	public void testInterruptedStatusIsClearedBetweenTwoTaskExecution () throws Exception {
		630	RequestProcessor rp = new RequestProcessor ("testInterruptedStatusIsClearedBetweenTwoTaskExecution");
631
632	final RequestProcessor.Task[] task = new RequestProcessor.Task[1];
633	// test interrupted status is cleared after task ends
634	class Fail implements Runnable {
635	public boolean checkBefore;
636	public Thread runIn;
637	public boolean goodThread;
638
639	public synchronized void run () {
640	if (runIn == null) {
641	runIn = Thread.currentThread();
642	task[0].schedule (0);
643	} else {
644	goodThread = Thread.currentThread () == runIn;
645	}
646
647	checkBefore = runIn.isInterrupted();
648	// set the flag for next execution
649	runIn.interrupt();
650
651	notifyAll ();
652	}
653	}
654
655	Fail f = new Fail ();
656	synchronized (f) {
657	task[0] = rp.post (f);
658
659	// wait for the first execution
660	f.wait ();
661	}
662	// wait for the second
663	task[0].waitFinished ();
664
665	assertTrue ("This shall be always true, but if not, than it does not mean too much"
666	+ " just that the tasks were not executed in the same thread. In such case it "
667	+ " this test does not do anything useful as it needs to execute the task twice "
668	+ " in the same thread", f.goodThread);
669
670	assertTrue ("Interrupted state has been cleared between two executions of the task", !f.checkBefore);
671
672	}
673
567	private static void doGc (int count, Reference toClear) {	674	private static void doGc (int count, Reference toClear) {
568	java.util.ArrayList l = new java.util.ArrayList (count);	675	java.util.ArrayList l = new java.util.ArrayList (count);
569	while (count-- > 0) {	676	while (count-- > 0) {