standardjmeterengine.java

来自「测试工具」· Java 代码 · 共 550 行 · 第 1/2 页

            log.info("Ending thread " + thread.getThreadName());
            if (!serialized && !schcdule_run && !startingGroups && allThreads.size() == 0 ) {
            	log.info("Stopping test");
            	stopTest();
            }
        } catch (Throwable e) {
            log.fatalError("Call to threadFinished should never throw an exception - this can deadlock JMeter",e);
        }
	}

	public synchronized void stopTest() {
		Thread stopThread = new Thread(new StopTest());
		stopThread.start();
	}

	public synchronized void stopTest(boolean b) {
		Thread stopThread = new Thread(new StopTest(b));
		stopThread.start();
	}

	private class StopTest implements Runnable {
		boolean now;

		private StopTest() {
			now = true;
		}

		private StopTest(boolean b) {
			now = b;
		}

		public void run() {
			if (running) {
				running = false;
				if (now) {
					tellThreadsToStop();
				} else {
					stopAllThreads();
				}
				try {
					Thread.sleep(10 * allThreads.size());
				} catch (InterruptedException e) {
				}
				boolean stopped = verifyThreadsStopped();
				if (stopped || now) {
					notifyTestListenersOfEnd();
				}
			}
		}
	}

	public void run() {
		log.info("Running the test!");
		running = true;

		SearchByClass testPlan = new SearchByClass(TestPlan.class);
		getTestTree().traverse(testPlan);
		Object[] plan = testPlan.getSearchResults().toArray();
		if (plan.length == 0) {
			System.err.println("Could not find the TestPlan!");
			log.error("Could not find the TestPlan!");
			System.exit(1);
		}
		if (((TestPlan) plan[0]).isSerialized()) {
			serialized = true;
		}
        JMeterContextService.startTest();
        try {
        	compileTree();
	    } catch (RuntimeException e) {
	    	log.error("Error occurred compiling the tree:",e);
	    	JMeterUtils.reportErrorToUser("Error occurred compiling the tree: - see log file");
	    	return; // no point continuing
        }
		/**
		 * Notification of test listeners needs to happen after function
		 * replacement, but before setting RunningVersion to true.
		 */
		testListeners = new SearchByClass(TestListener.class);
		getTestTree().traverse(testListeners);
		
		//	Merge in any additional test listeners
		// currently only used by the function parser
		testListeners.getSearchResults().addAll(testList);
		testList.clear(); // no longer needed
		
		if (!startListenersLater )notifyTestListenersOfStart();
		getTestTree().traverse(new TurnElementsOn());
        if (startListenersLater)notifyTestListenersOfStart();

		List testLevelElements = new LinkedList(getTestTree().list(getTestTree().getArray()[0]));
		removeThreadGroups(testLevelElements);
		SearchByClass searcher = new SearchByClass(ThreadGroup.class);
		getTestTree().traverse(searcher);
		TestCompiler.initialize();
		// for each thread group, generate threads
		// hand each thread the sampler controller
		// and the listeners, and the timer
		Iterator iter = searcher.getSearchResults().iterator();

		/*
		 * Here's where the test really starts. Run a Full GC now: it's no harm
		 * at all (just delays test start by a tiny amount) and hitting one too
		 * early in the test can impair results for short tests.
		 */
		System.gc();

		notifier = new ListenerNotifier();

		schcdule_run = true;
		JMeterContextService.getContext().setSamplingStarted(true);
		int groupCount = 0;
        JMeterContextService.clearTotalThreads();
        startingGroups = true;
		while (iter.hasNext()) {
			groupCount++;
			ThreadGroup group = (ThreadGroup) iter.next();
			int numThreads = group.getNumThreads();
            JMeterContextService.addTotalThreads(numThreads);
			boolean onErrorStopTest = group.getOnErrorStopTest();
			boolean onErrorStopThread = group.getOnErrorStopThread();
			String groupName = group.getName();
			int rampUp = group.getRampUp();
			float perThreadDelay = ((float) (rampUp * 1000) / (float) numThreads);
			log.info("Starting " + numThreads + " threads for group " + groupName + ". Ramp up = " + rampUp + ".");

			if (onErrorStopTest) {
				log.info("Test will stop on error");
			} else if (onErrorStopThread) {
				log.info("Thread will stop on error");
			} else {
				log.info("Continue on error");
			}

            ListedHashTree threadGroupTree = (ListedHashTree) searcher.getSubTree(group);
            threadGroupTree.add(group, testLevelElements);
			for (int i = 0; running && i < numThreads; i++) {
                final JMeterThread jmeterThread = new JMeterThread(cloneTree(threadGroupTree), this, notifier);
                jmeterThread.setThreadNum(i);
				jmeterThread.setThreadGroup(group);
				jmeterThread.setInitialContext(JMeterContextService.getContext());
				jmeterThread.setInitialDelay((int) (perThreadDelay * i));
				jmeterThread.setThreadName(groupName + " " + (groupCount) + "-" + (i + 1));

				scheduleThread(jmeterThread, group);

				// Set up variables for stop handling
				jmeterThread.setEngine(this);
				jmeterThread.setOnErrorStopTest(onErrorStopTest);
				jmeterThread.setOnErrorStopThread(onErrorStopThread);

				Thread newThread = new Thread(jmeterThread);
				newThread.setName(jmeterThread.getThreadName());
				allThreads.put(jmeterThread, newThread);
				if (serialized && !iter.hasNext() && i == numThreads - 1) // last thread
				{
					serialized = false;
				}
				newThread.start();
			}
			schcdule_run = false;
			if (serialized) {
				while (running && allThreads.size() > 0) {
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
					}
				}
			}
		}
        startingGroups = false;
	}

	/**
	 * This will schedule the time for the JMeterThread.
	 * 
	 * @param thread
	 * @param group
	 */
	private 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);
		}
	}

	private boolean verifyThreadsStopped() {
		boolean stoppedAll = true;
		Iterator iter = new HashSet(allThreads.keySet()).iterator();
		while (iter.hasNext()) {
			Thread t = (Thread) allThreads.get(iter.next());
			if (t != null && t.isAlive()) {
				try {
					t.join(WAIT_TO_DIE);
				} catch (InterruptedException e) {
				}
				if (t.isAlive()) {
					stoppedAll = false;
					log.info("Thread won't die: " + t.getName());
				}
			}
		}
		return stoppedAll;
	}

	private void tellThreadsToStop() {
		Iterator iter = new HashSet(allThreads.keySet()).iterator();
		while (iter.hasNext()) {
			JMeterThread item = (JMeterThread) iter.next();
			item.stop();
			Thread t = (Thread) allThreads.get(item);
			if (t != null) {
				t.interrupt();
			} else {
				log.warn("Lost thread: " + item.getThreadName());
				allThreads.remove(item);
			}
		}
	}

	public void askThreadsToStop() {
		engine.stopTest(false);
	}

	private void stopAllThreads() {
		Iterator iter = new HashSet(allThreads.keySet()).iterator();
		while (iter.hasNext()) {
			JMeterThread item = (JMeterThread) iter.next();
			item.stop();
		}
	}

	// Remote exit
	public void exit() {
		// Needs to be run in a separate thread to allow RMI call to return OK
		Thread t = new Thread() {
			public void run() {
				// log.info("Pausing");
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
				log.info("Bye");
				System.exit(0);
			}
		};
		log.info("Starting Closedown");
		t.start();
	}

	public void setProperties(Properties p) {
		log.info("Applying properties "+p);
		JMeterUtils.getJMeterProperties().putAll(p);
	}
}