Attachment #29054 for bug #53418

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

Lines 619-624 Link Here

(-)src/core/org/apache/jmeter/resources/messages_fr.properties (+1 lines)
619	number_of_threads=Nombre d'unit\u00E9s (utilisateurs) \:	619	number_of_threads=Nombre d'unit\u00E9s (utilisateurs) \:
620	obsolete_test_element=Cet \u00E9l\u00E9ment de test est obsol\u00E8te	620	obsolete_test_element=Cet \u00E9l\u00E9ment de test est obsol\u00E8te
621	once_only_controller_title=Contr\u00F4leur Ex\u00E9cution unique	621	once_only_controller_title=Contr\u00F4leur Ex\u00E9cution unique
		622	ondemand_threadgroup=Groupe d'unit\u00E9s progressives
622	opcode=Code d'op\u00E9ration	623	opcode=Code d'op\u00E9ration
623	open=Ouvrir...	624	open=Ouvrir...
624	option=Options	625	option=Options

Lines 625-630 Link Here

(-)src/core/org/apache/jmeter/resources/messages.properties (+1 lines)
625	number_of_threads=Number of Threads (users)\:	625	number_of_threads=Number of Threads (users)\:
626	obsolete_test_element=This test element is obsolete	626	obsolete_test_element=This test element is obsolete
627	once_only_controller_title=Once Only Controller	627	once_only_controller_title=Once Only Controller
		628	ondemand_threadgroup=Progressive Thread Group
628	opcode=opCode	629	opcode=opCode
629	open=Open...	630	open=Open...
630	option=Options	631	option=Options




import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.jmeter.JMeter;
import org.apache.jmeter.testbeans.TestBean;

import org.apache.jmeter.testelement.TestElement;
import org.apache.jmeter.testelement.TestListener;
import org.apache.jmeter.testelement.TestPlan;
import org.apache.jmeter.threads.AbstractThreadGroup;
import org.apache.jmeter.threads.JMeterContextService;
import org.apache.jmeter.threads.JMeterThread;
import org.apache.jmeter.threads.JMeterThreadMonitor;
import org.apache.jmeter.threads.ListenerNotifier;
import org.apache.jmeter.threads.TestCompiler;
import org.apache.jmeter.threads.AbstractThreadGroup;
import org.apache.jmeter.threads.SetupThreadGroup;
import org.apache.jmeter.threads.PostThreadGroup;
import org.apache.jmeter.threads.SetupThreadGroup;
import org.apache.jmeter.threads.TestCompiler;
import org.apache.jmeter.util.JMeterUtils;
import org.apache.jorphan.collections.HashTree;
import org.apache.jorphan.collections.ListedHashTree;

public class StandardJMeterEngine implements JMeterEngine, JMeterThreadMonitor, Runnable {
    private static final Logger log = LoggingManager.getLoggerForClass();

    private static final long WAIT_TO_DIE = JMeterUtils.getPropDefault("jmeterengine.threadstop.wait", 5 * 1000); // 5 seconds

    // Should we exit at end of the test? (only applies to server, because host is non-null)
    private static final boolean exitAfterTest =
        JMeterUtils.getPropDefault("server.exitaftertest", false);  // $NON-NLS-1$


    /** Whether to call System.exit(1) if threads won't stop */
    private static final boolean SYSTEM_EXIT_ON_STOP_FAIL = JMeterUtils.getPropDefault("jmeterengine.stopfail.system.exit", true);
    
    /** JMeterThread => its JVM thread */
    private final Map<JMeterThread, Thread> allThreads;

    /** Flag to show whether test is running. Set to false to stop creating more threads. */
    private volatile boolean running = false;



    private final String host;

    private List<AbstractThreadGroup> groups = Collections.synchronizedList(new ArrayList<AbstractThreadGroup>());

    public static void stopEngineNow() {
        if (engine != null) {// May be null if called from Unit test
            engine.stopTest(true);

        if (engine == null) {
            return false;// e.g. not yet started
        }
        boolean wasStopped = false;
        // ConcurrentHashMap does not need synch. here
        for (AbstractThreadGroup threadGroup : engine.groups) {
            wasStopped = wasStopped || threadGroup.stopThread(threadName, now);
            if (thrd.getThreadName().equals(threadName)){
                thrd.stop();
                thrd.interrupt();
                if (now) {
                    Thread t = entry.getValue();
                    if (t != null) {
                        t.interrupt();
                    }
                }
                return true;
            }
        }
        return wasStopped;
    }

    // End of code to allow engine to be controlled remotely


    public StandardJMeterEngine(String host) {
        this.host = host;
        this.allThreads = new ConcurrentHashMap<JMeterThread, Thread>();
        // Hack to allow external control
        engine = this;
    }

    // Called by JMeter thread when it finishes
    public synchronized void threadFinished(JMeterThread thread) {
        log.info("Ending thread " + thread.getThreadName());
        for (AbstractThreadGroup threadGroup : groups) {
            threadGroup.threadFinished(thread);
        }
    }

    public synchronized void stopTest() {

            engine = null;
            if (now) {
                tellThreadsToStop();
                pause(10 * countStillActiveThreads());
                boolean stopped = verifyThreadsStopped();
                if (!stopped) {  // we totally failed to stop the test
                    if (JMeter.isNonGUI()) {

                String groupName = startThreadGroup(group, groupCount, setupSearcher, testLevelElements, notifier);
                if (serialized && setupIter.hasNext()) {
                    log.info("Waiting for setup thread group: "+groupName+" to finish before starting next setup group");
                    pauseWhileRemainingThreads();
                        pause(1000);
                    }
                }
            }    
            log.info("Waiting for all setup thread groups To Exit");

            String groupName=startThreadGroup(group, groupCount, searcher, testLevelElements, notifier);
            if (serialized && iter.hasNext()) {
                log.info("Waiting for thread group: "+groupName+" to finish before starting next group");
                pauseWhileRemainingThreads();
                    pause(1000);
                }
            }
        } // end of thread groups
        if (groupCount == 0){ // No TGs found

                String groupName = startThreadGroup(group, groupCount, postSearcher, testLevelElements, notifier);
                if (serialized && postIter.hasNext()) {
                    log.info("Waiting for post thread group: "+groupName+" to finish before starting next post group");
                    pauseWhileRemainingThreads();
                        pause(1000);
                    }
                }
            }
            waitThreadsStopped(); // wait for Post threads to stop

        notifyTestListenersOfEnd(testListenersSave);
    }

    /**
     * Loop with 1s pause while thread groups have threads running 
     */
    private void pauseWhileRemainingThreads() {
        while (running && groupsHaveThreads()) {
            pause(1000);
        }
    }

    /**
     * @return true if remaining threads
     */
    private boolean groupsHaveThreads() {
        return countStillActiveThreads()>0;
    }

    /**
     * @return total of active threads in all Thread Groups
     */
    private int countStillActiveThreads() {
        int reminingThreads= 0;
        for (AbstractThreadGroup threadGroup : groups) {
            reminingThreads += threadGroup.numberOfActiveThreads();
        }
        return reminingThreads; 
    }
    
    private String startThreadGroup(AbstractThreadGroup group, int groupCount, SearchByClass<?> searcher, List<?> testLevelElements, ListenerNotifier notifier)
    {
            int numThreads = group.getNumThreads();

            }
            ListedHashTree threadGroupTree = (ListedHashTree) searcher.getSubTree(group);
            threadGroupTree.add(group, testLevelElements);
            
            JMeterThread[] jmThreads = 
                    new JMeterThread[numThreads];
            for (int i = 0; running && i < numThreads; i++) {
                final JMeterThread jmeterThread = new JMeterThread(cloneTree(threadGroupTree), this, notifier);
                jmeterThread.setThreadNum(i);


                group.scheduleThread(jmeterThread);

                jmThreads[i] = jmeterThread;
                newThread.setName(threadName);
                allThreads.put(jmeterThread, newThread);
                newThread.start();
            } // end of thread startup for this thread group
            group.setJMeterThreads(jmThreads);
            group.start();
            groups.add(group);
            return groupName;
    }

    /**
     * @return boolean true if all threads of all Threead Groups stopped
     */
    private boolean verifyThreadsStopped() {
        boolean stoppedAll = true;
        // ConcurrentHashMap does not need synch. here
        for (AbstractThreadGroup threadGroup : groups) {
            stoppedAll = stoppedAll && threadGroup.verifyThreadsStopped();
                if (t.isAlive()) {
                    try {
                        t.join(WAIT_TO_DIE);
                    } catch (InterruptedException e) {
                    }
                    if (t.isAlive()) {
                        stoppedAll = false;
                        log.warn("Thread won't exit: " + t.getName());
                    }
                }
            }
        }
        return stoppedAll;
    }

    /**
     * Wait for Group Threads to stop
     */
    private void waitThreadsStopped() {
        // ConcurrentHashMap does not need synch. here
        for (AbstractThreadGroup threadGroup : groups) {
            threadGroup.waitThreadsStopped();
                while (t.isAlive()) {
                    try {
                        t.join(WAIT_TO_DIE);
                    } catch (InterruptedException e) {
                    }
                }
            }
        }
    }


     */
    private void tellThreadsToStop() {
        // ConcurrentHashMap does not need protecting
        for (AbstractThreadGroup threadGroup : groups) {
            threadGroup.tellThreadsToStop();
            item.stop(); // set stop flag
            item.interrupt(); // interrupt sampler if possible
            Thread t = entry.getValue();
            if (t != null ) { // Bug 49734
                t.interrupt(); // also interrupt JVM thread
            }
        }
    }


     */
    private void stopAllThreads() {
        // ConcurrentHashMap does not need synch. here
        for (AbstractThreadGroup threadGroup : groups) {
            threadGroup.stop();
        }
    }


Lines 184-190 Link Here

(-)src/core/org/apache/jmeter/threads/ThreadGroup.java (-1 / +1 lines)
184	* @param thread	184	* @param thread
185	* @param group	185	* @param group
186	*/	186	*/
187	private void scheduleThread(JMeterThread thread, ThreadGroup group) {	187	protected void scheduleThread(JMeterThread thread, ThreadGroup group) {
188	// if true the Scheduler is enabled	188	// if true the Scheduler is enabled
189	if (group.getScheduler()) {	189	if (group.getScheduler()) {
190	long now = System.currentTimeMillis();	190	long now = System.currentTimeMillis();

Lines 19-24 Link Here

(-)src/core/org/apache/jmeter/threads/AbstractThreadGroup.java (+163 lines)
19	package org.apache.jmeter.threads;	19	package org.apache.jmeter.threads;
20		20
21	import java.io.Serializable;	21	import java.io.Serializable;
		22	import java.util.Map;
		23	import java.util.Map.Entry;
		24	import java.util.concurrent.ConcurrentHashMap;
22		25
23	import org.apache.jmeter.control.Controller;	26	import org.apache.jmeter.control.Controller;
24	import org.apache.jmeter.control.LoopController;	27	import org.apache.jmeter.control.LoopController;
Lines 29-34 Link Here
29	import org.apache.jmeter.testelement.property.IntegerProperty;	32	import org.apache.jmeter.testelement.property.IntegerProperty;
30	import org.apache.jmeter.testelement.property.JMeterProperty;	33	import org.apache.jmeter.testelement.property.JMeterProperty;
31	import org.apache.jmeter.testelement.property.TestElementProperty;	34	import org.apache.jmeter.testelement.property.TestElementProperty;
		35	import org.apache.jmeter.util.JMeterUtils;
		36	import org.apache.jorphan.logging.LoggingManager;
		37	import org.apache.log.Logger;
32		38
33	/**	39	/**
34	* ThreadGroup holds the settings for a JMeter thread group.	40	* ThreadGroup holds the settings for a JMeter thread group.
Lines 39-44 Link Here
39		45
40	private static final long serialVersionUID = 240L;	46	private static final long serialVersionUID = 240L;
41		47
		48	private static final long WAIT_TO_DIE = JMeterUtils.getPropDefault("jmeterengine.threadstop.wait", 5 * 1000); // 5 seconds
		49
		50	private static final Logger log = LoggingManager.getLoggerForClass();
		51
42	/** Action to be taken when a Sampler error occurs */	52	/** Action to be taken when a Sampler error occurs */
43	public final static String ON_SAMPLE_ERROR = "ThreadGroup.on_sample_error"; // int	53	public final static String ON_SAMPLE_ERROR = "ThreadGroup.on_sample_error"; // int
44		54
Lines 65-70 Link Here
65	// @GuardedBy("this")	75	// @GuardedBy("this")
66	private int numberOfThreads = 0; // Number of active threads in this group	76	private int numberOfThreads = 0; // Number of active threads in this group
67		77
		78	private JMeterThread[] jmThreads;
		79
		80	private Map<JMeterThread, Thread> allThreads = new ConcurrentHashMap<JMeterThread, Thread>();
		81
68	/** {@inheritDoc} */	82	/** {@inheritDoc} */
69	public boolean isDone() {	83	public boolean isDone() {
70	return getSamplerController().isDone();	84	return getSamplerController().isDone();
Lines 217-220 Link Here
217	}	231	}
218		232
219	public abstract void scheduleThread(JMeterThread thread);	233	public abstract void scheduleThread(JMeterThread thread);
		234
		235	/**
		236	* Default implementation starts threads immediately
		237	*/
		238	public void start() {
		239	for (int i = 0; i < jmThreads.length; i++) {
		240	Thread newThread = new Thread(jmThreads[i]);
		241	newThread.setName(jmThreads[i].getThreadName());
		242	registerStartedThread(jmThreads[i], newThread);
		243	newThread.start();
		244	}
		245	}
		246
		247	/**
		248	* Register Thread when it starts
		249	* @param jMeterThread {@link JMeterThread}
		250	* @param newThread Thread
		251	*/
		252	protected final void registerStartedThread(JMeterThread jMeterThread, Thread newThread) {
		253	allThreads.put(jMeterThread, newThread);
		254	}
		255
		256	/**
		257	*
		258	* @param jmThreads JMeterThread[]
		259	*/
		260	public final void setJMeterThreads(JMeterThread[] jmThreads) {
		261	this.jmThreads = jmThreads;
		262	}
		263
		264	/**
		265	* @return JMeterThread[]
		266	*/
		267	protected final JMeterThread[] getJMeterThreads() {
		268	return this.jmThreads;
		269	}
		270
		271	/**
		272	* Stop thread called threadName:
		273	* <ol>
		274	* <li>stop JMeter thread</li>
		275	* <li>interrupt JMeter thread</li>
		276	* <li>interrupt underlying thread</li>
		277	* <ol>
		278	* @param threadName String thread name
		279	* @param now boolean for stop
		280	* @return true if thread stopped
		281	*/
		282	public boolean stopThread(String threadName, boolean now) {
		283	for(Entry<JMeterThread, Thread> entry : allThreads.entrySet()){
		284	JMeterThread thrd = entry.getKey();
		285	if (thrd.getThreadName().equals(threadName)){
		286	thrd.stop();
		287	thrd.interrupt();
		288	if (now) {
		289	Thread t = entry.getValue();
		290	if (t != null) {
		291	t.interrupt();
		292	}
		293	}
		294	return true;
		295	}
		296	}
		297	return false;
298	}
299
300	/**
301	* Called by JMeter thread when it finishes
302	*/
303	public void threadFinished(JMeterThread thread) {
304	allThreads.remove(thread);
305	}
306
307	/**
308	* For each thread, invoke:
309	* <ul>
310	* <li>{@link JMeterThread#stop()} - set stop flag</li>
311	* <li>{@link JMeterThread#interrupt()} - interrupt sampler</li>
312	* <li>{@link Thread#interrupt()} - interrupt JVM thread</li>
313	* </ul>
314	*/
315	public void tellThreadsToStop() {
316	for (Entry<JMeterThread, Thread> entry : allThreads.entrySet()) {
317	JMeterThread item = entry.getKey();
318	item.stop(); // set stop flag
319	item.interrupt(); // interrupt sampler if possible
320	Thread t = entry.getValue();
321	if (t != null ) { // Bug 49734
322	t.interrupt(); // also interrupt JVM thread
323	}
324	}
325	}
326
327	/**
328	* For each thread, invoke:
329	* <ul>
330	* <li>{@link JMeterThread#stop()} - set stop flag</li>
331	* </ul>
332	*/
333	public void stop() {
334	for (JMeterThread item : allThreads.keySet()) {
335	item.stop();
336	}
337	}
338
339	/**
340	* @return number of active threads
341	*/
342	public int numberOfActiveThreads() {
343	return allThreads.size();
344	}
345
346	/**
347	* @return boolean true if all threads stopped
348	*/
349	public boolean verifyThreadsStopped() {
350	boolean stoppedAll = true;
351	for (Thread t : allThreads.values()) {
352	if (t != null) {
353	if (t.isAlive()) {
354	try {
355	t.join(WAIT_TO_DIE);
356	} catch (InterruptedException e) {
357	}
358	if (t.isAlive()) {
359	stoppedAll = false;
360	log.warn("Thread won't exit: " + t.getName());
361	}
362	}
363	}
364	}
365	return stoppedAll;
366	}
367
368	/**
369	* Wait for all Group Threads to stop
370	*/
371	public void waitThreadsStopped() {
372	for (Thread t : allThreads.values()) {
373	if (t != null) {
374	while (t.isAlive()) {
375	try {
376	t.join(WAIT_TO_DIE);
377	} catch (InterruptedException e) {
378	}
379	}
380	}
381	}
382	}
220	}	383	}




/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.jmeter.threads;

import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.jmeter.util.JMeterUtils;

/**
 * Thread Group implementation that creates Threads progressively
 */
public class OnDemandThreadGroup extends ThreadGroup {
    /** How often to check for shutdown during ramp-up, default 1000ms */
    private static final int RAMPUP_GRANULARITY =
            JMeterUtils.getPropDefault("jmeterthread.rampup.granularity", 1000); // $NON-NLS-1$

    /**
     * 
     */
    private static final long serialVersionUID = 1326448504092168570L;

    private Thread threadStarter;

    /**
     * Was test stopped
     */
    private AtomicBoolean stopped = new AtomicBoolean(false);

    /**
     * 
     */
    public OnDemandThreadGroup() {
        super();
    }

    /**
     * @see org.apache.jmeter.threads.AbstractThreadGroup#start()
     */
    @Override
    public void start() {
        stopped.set(false);
        this.threadStarter = new Thread(new ThreadStarter());
        threadStarter.start();  
        try {
            threadStarter.join();
        } catch (InterruptedException e) {
            return;
        }
    }
    
    /**
     * Starts Threads using ramp up
     */
    private class ThreadStarter implements Runnable {

        public ThreadStarter() {
            super();
        }
        
        public void run() {
            final JMeterThread[] jMeterThreads = getJMeterThreads();
            
            int rampUp = getRampUp();
            float perThreadDelay = ((float) (rampUp * 1000) / (float) getNumThreads());
            if (getScheduler()) {
                long now = System.currentTimeMillis();
                // set the start time for the Thread
                if (getDelay() > 0) {// Duration is in seconds
                    delayBy(getDelay() * 1000, "start");
                } else {
                    long start = getStartTime();
                    if (start < now) {
                        start = now; // Force a sensible start time
                        // No delay
                    } else {
                        delayBy(start-now, "start");
                    }
                }
            }
            for (int i = 0; i < jMeterThreads.length; i++) {
                try {
                    if(!stopped.get()) {
                        Thread.sleep(Math.round(perThreadDelay));
                        Thread newThread = new Thread(jMeterThreads[i]);
                        newThread.setName(jMeterThreads[i].getThreadName());
                        registerStartedThread(jMeterThreads[i], newThread);
                        newThread.start();
                    }
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

    /**
     * Wait for delay with RAMPUP_GRANULARITY
     * @param delay delay in ms
     * @param type Delay type
     */
    protected final void delayBy(long delay, String type) {
        if (delay > 0) {
            long start = System.currentTimeMillis();
            long end = start + delay;
            long now=0;
            long pause = RAMPUP_GRANULARITY;
            while(!stopped.get() && (now = System.currentTimeMillis()) < end) {
                long togo = end - now;
                if (togo < pause) {
                    pause = togo;
                }
                try {
                    Thread.sleep(pause); // delay between checks
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }
    /**
     * @see org.apache.jmeter.threads.AbstractThreadGroup#stop()
     */
    @Override
    public void stop() {
        stopped.set(true);
        super.stop();
    }
    
    @Override
    public void scheduleThread(JMeterThread thread)
    {
        thread.setInitialDelay(0);
        scheduleThread(thread, this);
    }
    
    /**
     * This will schedule the time for the JMeterThread.
     *
     * @param thread
     * @param group
     */
    protected void scheduleThread(JMeterThread thread, ThreadGroup group) {
        // if true the Scheduler is enabled
        if (group.getScheduler()) {
            long now = System.currentTimeMillis();
            // set the start time for the Thread
            if (group.getDelay() > 0) {// Duration is in seconds
                thread.setStartTime(group.getDelay() * 1000 + now);
            } else {
                long start = group.getStartTime();
                if (start < now) {
                    start = now; // Force a sensible start time
                }
                thread.setStartTime(start);
            }

            // set the endtime for the Thread
            if (group.getDuration() > 0) {// Duration is in seconds
                thread.setEndTime(group.getDuration() * 1000 + (thread.getStartTime()));
            } else {
                thread.setEndTime(group.getEndTime());
            }

            // Enables the scheduler
            thread.setScheduled(true);
        }
    }
}

Lines 196-202 Link Here

(-)src/core/org/apache/jmeter/threads/JMeterThread.java (-1 / +3 lines)
196	*	196	*
197	*/	197	*/
198	private void stopScheduler() {	198	private void stopScheduler() {
199	long delay = System.currentTimeMillis() - endTime;	199	long now = System.currentTimeMillis();
		200	long delay = now - endTime;
200	if ((delay >= 0)) {	201	if ((delay >= 0)) {
201	running = false;	202	running = false;
202	}	203	}
Lines 788-793 Link Here
788	*/	789	*/
789	protected final void delayBy(long delay, String type) {	790	protected final void delayBy(long delay, String type) {
790	if (delay > 0) {	791	if (delay > 0) {
		792	System.out.println(type+"=>"+threadName+" delaying by:"+delay);
791	long start = System.currentTimeMillis();	793	long start = System.currentTimeMillis();
792	long end = start + delay;	794	long end = start + delay;
793	long now=0;	795	long now=0;




/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.jmeter.threads.gui;

import org.apache.jmeter.testelement.TestElement;
import org.apache.jmeter.threads.OnDemandThreadGroup;
import org.apache.jmeter.threads.ThreadGroup;

/**
 * GUI for {@link OnDemandThreadGroup}
 */
public class OnDemandThreadGroupGui extends ThreadGroupGui {

    /**
     * 
     */
    private static final long serialVersionUID = 7903310220568526814L;

    /**
     * 
     */
    public OnDemandThreadGroupGui() {
        super();
    }

    /**
     * @see org.apache.jmeter.threads.gui.ThreadGroupGui#createTestElement()
     */
    @Override
    public TestElement createTestElement() {
        ThreadGroup tg = new OnDemandThreadGroup();
        modifyTestElement(tg);
        return tg;
    }
    
    /**
     * @return String label key
     */
    public String getLabelResource() {
        return "ondemand_threadgroup"; // $NON-NLS-1$
    }
}

Return to bug 53418

Lines 21-34 Link Here

(-)src/core/org/apache/jmeter/engine/StandardJMeterEngine.java (-74 / +66 lines)
21	import java.io.PrintWriter;	21	import java.io.PrintWriter;
22	import java.io.StringWriter;	22	import java.io.StringWriter;
23	import java.util.ArrayList;	23	import java.util.ArrayList;
		24	import java.util.Collections;
24	import java.util.Date;	25	import java.util.Date;
25	import java.util.Iterator;	26	import java.util.Iterator;
26	import java.util.LinkedList;	27	import java.util.LinkedList;
27	import java.util.List;	28	import java.util.List;
28	import java.util.Map;
29	import java.util.Map.Entry;
30	import java.util.Properties;	29	import java.util.Properties;
31	import java.util.concurrent.ConcurrentHashMap;
32		30
33	import org.apache.jmeter.JMeter;	31	import org.apache.jmeter.JMeter;
34	import org.apache.jmeter.testbeans.TestBean;	32	import org.apache.jmeter.testbeans.TestBean;
Lines 36-49 Link Here
36	import org.apache.jmeter.testelement.TestElement;	34	import org.apache.jmeter.testelement.TestElement;
37	import org.apache.jmeter.testelement.TestListener;	35	import org.apache.jmeter.testelement.TestListener;
38	import org.apache.jmeter.testelement.TestPlan;	36	import org.apache.jmeter.testelement.TestPlan;
		37	import org.apache.jmeter.threads.AbstractThreadGroup;
39	import org.apache.jmeter.threads.JMeterContextService;	38	import org.apache.jmeter.threads.JMeterContextService;
40	import org.apache.jmeter.threads.JMeterThread;	39	import org.apache.jmeter.threads.JMeterThread;
41	import org.apache.jmeter.threads.JMeterThreadMonitor;	40	import org.apache.jmeter.threads.JMeterThreadMonitor;
42	import org.apache.jmeter.threads.ListenerNotifier;	41	import org.apache.jmeter.threads.ListenerNotifier;
43	import org.apache.jmeter.threads.TestCompiler;
44	import org.apache.jmeter.threads.AbstractThreadGroup;
45	import org.apache.jmeter.threads.SetupThreadGroup;
46	import org.apache.jmeter.threads.PostThreadGroup;	42	import org.apache.jmeter.threads.PostThreadGroup;
		43	import org.apache.jmeter.threads.SetupThreadGroup;
		44	import org.apache.jmeter.threads.TestCompiler;
47	import org.apache.jmeter.util.JMeterUtils;	45	import org.apache.jmeter.util.JMeterUtils;
48	import org.apache.jorphan.collections.HashTree;	46	import org.apache.jorphan.collections.HashTree;
49	import org.apache.jorphan.collections.ListedHashTree;	47	import org.apache.jorphan.collections.ListedHashTree;
Lines 58-65 Link Here
58	public class StandardJMeterEngine implements JMeterEngine, JMeterThreadMonitor, Runnable {	56	public class StandardJMeterEngine implements JMeterEngine, JMeterThreadMonitor, Runnable {
59	private static final Logger log = LoggingManager.getLoggerForClass();	57	private static final Logger log = LoggingManager.getLoggerForClass();
60		58
61	private static final long WAIT_TO_DIE = JMeterUtils.getPropDefault("jmeterengine.threadstop.wait", 5 * 1000); // 5 seconds
62
63	// Should we exit at end of the test? (only applies to server, because host is non-null)	59	// Should we exit at end of the test? (only applies to server, because host is non-null)
64	private static final boolean exitAfterTest =	60	private static final boolean exitAfterTest =
65	JMeterUtils.getPropDefault("server.exitaftertest", false); // $NON-NLS-1$	61	JMeterUtils.getPropDefault("server.exitaftertest", false); // $NON-NLS-1$
Lines 92-101 Link Here
92		88
93	/** Whether to call System.exit(1) if threads won't stop */	89	/** Whether to call System.exit(1) if threads won't stop */
94	private static final boolean SYSTEM_EXIT_ON_STOP_FAIL = JMeterUtils.getPropDefault("jmeterengine.stopfail.system.exit", true);	90	private static final boolean SYSTEM_EXIT_ON_STOP_FAIL = JMeterUtils.getPropDefault("jmeterengine.stopfail.system.exit", true);
95		91
96	/** JMeterThread => its JVM thread */
97	private final Map<JMeterThread, Thread> allThreads;
98
99	/** Flag to show whether test is running. Set to false to stop creating more threads. */	92	/** Flag to show whether test is running. Set to false to stop creating more threads. */
100	private volatile boolean running = false;	93	private volatile boolean running = false;
101		94
Lines 111-116 Link Here
111		104
112	private final String host;	105	private final String host;
113		106
		107	private List<AbstractThreadGroup> groups = Collections.synchronizedList(new ArrayList<AbstractThreadGroup>());
		108
114	public static void stopEngineNow() {	109	public static void stopEngineNow() {
115	if (engine != null) {// May be null if called from Unit test	110	if (engine != null) {// May be null if called from Unit test
116	engine.stopTest(true);	111	engine.stopTest(true);
Lines 139-160 Link Here
139	if (engine == null) {	134	if (engine == null) {
140	return false;// e.g. not yet started	135	return false;// e.g. not yet started
141	}	136	}
		137	boolean wasStopped = false;
142	// ConcurrentHashMap does not need synch. here	138	// ConcurrentHashMap does not need synch. here
143	for(Entry<JMeterThread, Thread> entry : engine.allThreads.entrySet()){	139	for (AbstractThreadGroup threadGroup : engine.groups) {
144	JMeterThread thrd = entry.getKey();	140	wasStopped = wasStopped \|\| threadGroup.stopThread(threadName, now);
145	if (thrd.getThreadName().equals(threadName)){
146	thrd.stop();
147	thrd.interrupt();
148	if (now) {
149	Thread t = entry.getValue();
150	if (t != null) {
151	t.interrupt();
152	}
153	}
154	return true;
155	}
156	}	141	}
157	return false;	142	return wasStopped;
158	}	143	}
159		144
160	// End of code to allow engine to be controlled remotely	145	// End of code to allow engine to be controlled remotely
Lines 165-171 Link Here
165		150
166	public StandardJMeterEngine(String host) {	151	public StandardJMeterEngine(String host) {
167	this.host = host;	152	this.host = host;
168	this.allThreads = new ConcurrentHashMap<JMeterThread, Thread>();
169	// Hack to allow external control	153	// Hack to allow external control
170	engine = this;	154	engine = this;
171	}	155	}
Lines 268-274 Link Here
268	// Called by JMeter thread when it finishes	252	// Called by JMeter thread when it finishes
269	public synchronized void threadFinished(JMeterThread thread) {	253	public synchronized void threadFinished(JMeterThread thread) {
270	log.info("Ending thread " + thread.getThreadName());	254	log.info("Ending thread " + thread.getThreadName());
271	allThreads.remove(thread);	255	for (AbstractThreadGroup threadGroup : groups) {
		256	threadGroup.threadFinished(thread);
		257	}
272	}	258	}
273		259
274	public synchronized void stopTest() {	260	public synchronized void stopTest() {
Lines 292-298 Link Here
292	engine = null;	278	engine = null;
293	if (now) {	279	if (now) {
294	tellThreadsToStop();	280	tellThreadsToStop();
295	pause(10 * allThreads.size());	281	pause(10 * countStillActiveThreads());
296	boolean stopped = verifyThreadsStopped();	282	boolean stopped = verifyThreadsStopped();
297	if (!stopped) { // we totally failed to stop the test	283	if (!stopped) { // we totally failed to stop the test
298	if (JMeter.isNonGUI()) {	284	if (JMeter.isNonGUI()) {
Lines 381-389 Link Here
381	String groupName = startThreadGroup(group, groupCount, setupSearcher, testLevelElements, notifier);	367	String groupName = startThreadGroup(group, groupCount, setupSearcher, testLevelElements, notifier);
382	if (serialized && setupIter.hasNext()) {	368	if (serialized && setupIter.hasNext()) {
383	log.info("Waiting for setup thread group: "+groupName+" to finish before starting next setup group");	369	log.info("Waiting for setup thread group: "+groupName+" to finish before starting next setup group");
384	while (running && allThreads.size() > 0) {	370	pauseWhileRemainingThreads();
385	pause(1000);
386	}
387	}	371	}
388	}	372	}
389	log.info("Waiting for all setup thread groups To Exit");	373	log.info("Waiting for all setup thread groups To Exit");
Lines 414-422 Link Here
414	String groupName=startThreadGroup(group, groupCount, searcher, testLevelElements, notifier);	398	String groupName=startThreadGroup(group, groupCount, searcher, testLevelElements, notifier);
415	if (serialized && iter.hasNext()) {	399	if (serialized && iter.hasNext()) {
416	log.info("Waiting for thread group: "+groupName+" to finish before starting next group");	400	log.info("Waiting for thread group: "+groupName+" to finish before starting next group");
417	while (running && allThreads.size() > 0) {	401	pauseWhileRemainingThreads();
418	pause(1000);
419	}
420	}	402	}
421	} // end of thread groups	403	} // end of thread groups
422	if (groupCount == 0){ // No TGs found	404	if (groupCount == 0){ // No TGs found
Lines 443-451 Link Here
443	String groupName = startThreadGroup(group, groupCount, postSearcher, testLevelElements, notifier);	425	String groupName = startThreadGroup(group, groupCount, postSearcher, testLevelElements, notifier);
444	if (serialized && postIter.hasNext()) {	426	if (serialized && postIter.hasNext()) {
445	log.info("Waiting for post thread group: "+groupName+" to finish before starting next post group");	427	log.info("Waiting for post thread group: "+groupName+" to finish before starting next post group");
446	while (running && allThreads.size() > 0) {	428	pauseWhileRemainingThreads();
447	pause(1000);
448	}
449	}	429	}
450	}	430	}
451	waitThreadsStopped(); // wait for Post threads to stop	431	waitThreadsStopped(); // wait for Post threads to stop
Lines 454-459 Link Here
454	notifyTestListenersOfEnd(testListenersSave);	434	notifyTestListenersOfEnd(testListenersSave);
455	}	435	}
456		436
		437	/**
		438	* Loop with 1s pause while thread groups have threads running
		439	*/
		440	private void pauseWhileRemainingThreads() {
		441	while (running && groupsHaveThreads()) {
		442	pause(1000);
		443	}
		444	}
		445
		446	/**
		447	* @return true if remaining threads
		448	*/
		449	private boolean groupsHaveThreads() {
		450	return countStillActiveThreads()>0;
		451	}
		452
		453	/**
		454	* @return total of active threads in all Thread Groups
		455	*/
		456	private int countStillActiveThreads() {
		457	int reminingThreads= 0;
		458	for (AbstractThreadGroup threadGroup : groups) {
		459	reminingThreads += threadGroup.numberOfActiveThreads();
		460	}
		461	return reminingThreads;
		462	}
		463
457	private String startThreadGroup(AbstractThreadGroup group, int groupCount, SearchByClass<?> searcher, List<?> testLevelElements, ListenerNotifier notifier)	464	private String startThreadGroup(AbstractThreadGroup group, int groupCount, SearchByClass<?> searcher, List<?> testLevelElements, ListenerNotifier notifier)
458	{	465	{
459	int numThreads = group.getNumThreads();	466	int numThreads = group.getNumThreads();
Lines 478-483 Link Here
478	}	485	}
479	ListedHashTree threadGroupTree = (ListedHashTree) searcher.getSubTree(group);	486	ListedHashTree threadGroupTree = (ListedHashTree) searcher.getSubTree(group);
480	threadGroupTree.add(group, testLevelElements);	487	threadGroupTree.add(group, testLevelElements);
		488
		489	JMeterThread[] jmThreads =
		490	new JMeterThread[numThreads];
481	for (int i = 0; running && i < numThreads; i++) {	491	for (int i = 0; running && i < numThreads; i++) {
482	final JMeterThread jmeterThread = new JMeterThread(cloneTree(threadGroupTree), this, notifier);	492	final JMeterThread jmeterThread = new JMeterThread(cloneTree(threadGroupTree), this, notifier);
483	jmeterThread.setThreadNum(i);	493	jmeterThread.setThreadNum(i);
Lines 493-537 Link Here
493		503
494	group.scheduleThread(jmeterThread);	504	group.scheduleThread(jmeterThread);
495		505
496	Thread newThread = new Thread(jmeterThread);	506	jmThreads[i] = jmeterThread;
497	newThread.setName(threadName);
498	allThreads.put(jmeterThread, newThread);
499	newThread.start();
500	} // end of thread startup for this thread group	507	} // end of thread startup for this thread group
		508	group.setJMeterThreads(jmThreads);
		509	group.start();
		510	groups.add(group);
501	return groupName;	511	return groupName;
502	}	512	}
503		513
		514	/**
		515	* @return boolean true if all threads of all Threead Groups stopped
		516	*/
504	private boolean verifyThreadsStopped() {	517	private boolean verifyThreadsStopped() {
505	boolean stoppedAll = true;	518	boolean stoppedAll = true;
506	// ConcurrentHashMap does not need synch. here	519	// ConcurrentHashMap does not need synch. here
507	for (Thread t : allThreads.values()) {	520	for (AbstractThreadGroup threadGroup : groups) {
508	if (t != null) {	521	stoppedAll = stoppedAll && threadGroup.verifyThreadsStopped();
509	if (t.isAlive()) {
510	try {
511	t.join(WAIT_TO_DIE);
512	} catch (InterruptedException e) {
513	}
514	if (t.isAlive()) {
515	stoppedAll = false;
516	log.warn("Thread won't exit: " + t.getName());
517	}
518	}
519	}
520	}	522	}
521	return stoppedAll;	523	return stoppedAll;
522	}	524	}
523		525
		526	/**
		527	* Wait for Group Threads to stop
		528	*/
524	private void waitThreadsStopped() {	529	private void waitThreadsStopped() {
525	// ConcurrentHashMap does not need synch. here	530	// ConcurrentHashMap does not need synch. here
526	for (Thread t : allThreads.values()) {	531	for (AbstractThreadGroup threadGroup : groups) {
527	if (t != null) {	532	threadGroup.waitThreadsStopped();
528	while (t.isAlive()) {
529	try {
530	t.join(WAIT_TO_DIE);
531	} catch (InterruptedException e) {
532	}
533	}
534	}
535	}	533	}
536	}	534	}
537		535
Lines 545-558 Link Here
545	*/	543	*/
546	private void tellThreadsToStop() {	544	private void tellThreadsToStop() {
547	// ConcurrentHashMap does not need protecting	545	// ConcurrentHashMap does not need protecting
548	for (Entry<JMeterThread, Thread> entry : allThreads.entrySet()) {	546	for (AbstractThreadGroup threadGroup : groups) {
549	JMeterThread item = entry.getKey();	547	threadGroup.tellThreadsToStop();
550	item.stop(); // set stop flag
551	item.interrupt(); // interrupt sampler if possible
552	Thread t = entry.getValue();
553	if (t != null ) { // Bug 49734
554	t.interrupt(); // also interrupt JVM thread
555	}
556	}	548	}
557	}	549	}
558		550
Lines 570-577 Link Here
570	*/	562	*/
571	private void stopAllThreads() {	563	private void stopAllThreads() {
572	// ConcurrentHashMap does not need synch. here	564	// ConcurrentHashMap does not need synch. here
573	for (JMeterThread item : allThreads.keySet()) {	565	for (AbstractThreadGroup threadGroup : groups) {
574	item.stop();	566	threadGroup.stop();
575	}	567	}
576	}	568	}
577		569

Line 0 Link Here

(-)src/core/org/apache/jmeter/threads/OnDemandThreadGroup.java (+184 lines)
		1	/*
		2	* Licensed to the Apache Software Foundation (ASF) under one or more
		3	* contributor license agreements. See the NOTICE file distributed with
		4	* this work for additional information regarding copyright ownership.
		5	* The ASF licenses this file to You under the Apache License, Version 2.0
		6	* (the "License"); you may not use this file except in compliance with
		7	* the License. You may obtain a copy of the License at
		8	*
		9	* http://www.apache.org/licenses/LICENSE-2.0
		10	*
		11	* Unless required by applicable law or agreed to in writing, software
		12	* distributed under the License is distributed on an "AS IS" BASIS,
		13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
		14	* See the License for the specific language governing permissions and
		15	* limitations under the License.
		16	*
		17	*/
		18
		19	package org.apache.jmeter.threads;
		20
		21	import java.util.concurrent.atomic.AtomicBoolean;
		22
		23	import org.apache.jmeter.util.JMeterUtils;
		24
		25	/**
		26	* Thread Group implementation that creates Threads progressively
		27	*/
		28	public class OnDemandThreadGroup extends ThreadGroup {
		29	/** How often to check for shutdown during ramp-up, default 1000ms */
		30	private static final int RAMPUP_GRANULARITY =
		31	JMeterUtils.getPropDefault("jmeterthread.rampup.granularity", 1000); // $NON-NLS-1$
		32
		33	/**
		34	*
		35	*/
		36	private static final long serialVersionUID = 1326448504092168570L;
		37
		38	private Thread threadStarter;
		39
		40	/**
		41	* Was test stopped
		42	*/
		43	private AtomicBoolean stopped = new AtomicBoolean(false);
		44
		45	/**
		46	*
		47	*/
		48	public OnDemandThreadGroup() {
		49	super();
		50	}
		51
		52	/**
		53	* @see org.apache.jmeter.threads.AbstractThreadGroup#start()
		54	*/
		55	@Override
		56	public void start() {
		57	stopped.set(false);
		58	this.threadStarter = new Thread(new ThreadStarter());
		59	threadStarter.start();
		60	try {
		61	threadStarter.join();
		62	} catch (InterruptedException e) {
		63	return;
		64	}
65	}
66
67	/**
68	* Starts Threads using ramp up
69	*/
70	private class ThreadStarter implements Runnable {
71
72	public ThreadStarter() {
73	super();
74	}
75
76	public void run() {
77	final JMeterThread[] jMeterThreads = getJMeterThreads();
78
79	int rampUp = getRampUp();
80	float perThreadDelay = ((float) (rampUp * 1000) / (float) getNumThreads());
81	if (getScheduler()) {
82	long now = System.currentTimeMillis();
83	// set the start time for the Thread
84	if (getDelay() > 0) {// Duration is in seconds
85	delayBy(getDelay() * 1000, "start");
86	} else {
87	long start = getStartTime();
88	if (start < now) {
89	start = now; // Force a sensible start time
90	// No delay
91	} else {
92	delayBy(start-now, "start");
93	}
94	}
95	}
96	for (int i = 0; i < jMeterThreads.length; i++) {
97	try {
98	if(!stopped.get()) {
99	Thread.sleep(Math.round(perThreadDelay));
100	Thread newThread = new Thread(jMeterThreads[i]);
101	newThread.setName(jMeterThreads[i].getThreadName());
102	registerStartedThread(jMeterThreads[i], newThread);
103	newThread.start();
104	}
105	} catch (InterruptedException e) {
106	break;
107	}
108	}
109	}
110	}
111
112	/**
113	* Wait for delay with RAMPUP_GRANULARITY
114	* @param delay delay in ms
115	* @param type Delay type
116	*/
117	protected final void delayBy(long delay, String type) {
118	if (delay > 0) {
119	long start = System.currentTimeMillis();
120	long end = start + delay;
121	long now=0;
122	long pause = RAMPUP_GRANULARITY;
123	while(!stopped.get() && (now = System.currentTimeMillis()) < end) {
124	long togo = end - now;
125	if (togo < pause) {
126	pause = togo;
127	}
128	try {
129	Thread.sleep(pause); // delay between checks
130	} catch (InterruptedException e) {
131	break;
132	}
133	}
134	}
135	}
136	/**
137	* @see org.apache.jmeter.threads.AbstractThreadGroup#stop()
138	*/
139	@Override
140	public void stop() {
141	stopped.set(true);
142	super.stop();
143	}
144
145	@Override
146	public void scheduleThread(JMeterThread thread)
147	{
148	thread.setInitialDelay(0);
149	scheduleThread(thread, this);
150	}
151
152	/**
153	* This will schedule the time for the JMeterThread.
154	*
155	* @param thread
156	* @param group
157	*/
158	protected void scheduleThread(JMeterThread thread, ThreadGroup group) {
159	// if true the Scheduler is enabled
160	if (group.getScheduler()) {
161	long now = System.currentTimeMillis();
162	// set the start time for the Thread
163	if (group.getDelay() > 0) {// Duration is in seconds
164	thread.setStartTime(group.getDelay() * 1000 + now);
165	} else {
166	long start = group.getStartTime();
167	if (start < now) {
168	start = now; // Force a sensible start time
169	}
170	thread.setStartTime(start);
171	}
172
173	// set the endtime for the Thread
174	if (group.getDuration() > 0) {// Duration is in seconds
175	thread.setEndTime(group.getDuration() * 1000 + (thread.getStartTime()));
176	} else {
177	thread.setEndTime(group.getEndTime());
178	}
179
180	// Enables the scheduler
181	thread.setScheduled(true);
182	}
183	}
184	}

Line 0 Link Here

(-)src/core/org/apache/jmeter/threads/gui/OnDemandThreadGroupGui.java (+58 lines)
	1	/*
	2	* Licensed to the Apache Software Foundation (ASF) under one or more
	3	* contributor license agreements. See the NOTICE file distributed with
	4	* this work for additional information regarding copyright ownership.
	5	* The ASF licenses this file to You under the Apache License, Version 2.0
	6	* (the "License"); you may not use this file except in compliance with
	7	* the License. You may obtain a copy of the License at
	8	*
	9	* http://www.apache.org/licenses/LICENSE-2.0
	10	*
	11	* Unless required by applicable law or agreed to in writing, software
	12	* distributed under the License is distributed on an "AS IS" BASIS,
	13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	14	* See the License for the specific language governing permissions and
	15	* limitations under the License.
	16	*
	17	*/
	18
	19	package org.apache.jmeter.threads.gui;
	20
	21	import org.apache.jmeter.testelement.TestElement;
	22	import org.apache.jmeter.threads.OnDemandThreadGroup;
	23	import org.apache.jmeter.threads.ThreadGroup;
	24
	25	/**
	26	* GUI for {@link OnDemandThreadGroup}
	27	*/
	28	public class OnDemandThreadGroupGui extends ThreadGroupGui {
	29
	30	/**
	31	*
	32	*/
	33	private static final long serialVersionUID = 7903310220568526814L;
	34
	35	/**
	36	*
	37	*/
	38	public OnDemandThreadGroupGui() {
	39	super();
	40	}
	41
	42	/**
	43	* @see org.apache.jmeter.threads.gui.ThreadGroupGui#createTestElement()
	44	*/
	45	@Override
	46	public TestElement createTestElement() {
	47	ThreadGroup tg = new OnDemandThreadGroup();
	48	modifyTestElement(tg);
	49	return tg;
	50	}
	51
	52	/**
	53	* @return String label key
	54	*/
	55	public String getLabelResource() {
	56	return "ondemand_threadgroup"; // $NON-NLS-1$
	57	}
	58	}