def run(self):
		try:
			global threads
			global max_threads
			global threads_spawned_count
			while running == True:
				if len(threads) < max_threads:
					try:
						debugMsg(str(self)+" INITIATOR Spawning new thread since len(threads) is less than " + str(max_threads))
						current = someThreadFunction(threads_spawned_count)
						threads.append([threads_spawned_count,current])
						current.start()
						del current
						threads_spawned_count = threads_spawned_count + 1
						debugMsg(str(self)+" INITIATOR Running " + str(len(threads)) + " threads so far")
						print "INITIATOR Active Threads " + str(threading.activeCount()) + " including thread timeout timers"
					except:
						debugMsg("Unable to spawn thread, probably os limit")
						time.sleep(1)
				else:
					debugMsg(str(self)+" INITIATOR Waiting for a thread to timeout / die which will reduce threads_count")
					time.sleep(randrange(2,5,1))
					debugMsg(str(self)+" INITIATOR Running Threads " + str(threads))
					print "INITIATIOR " + str(len(threads)) + " threads in stack"
					print "INITIATIOR Active Threads:" + str(threading.activeCount()) + " including thread timeout timers"
					time.sleep(1)
		except Exception, e:
			debugMsg("WARNING: a initiator has died")
			debugMsg(str(e))
			global initiator_count
			initiator_count = initiator_count - 1

def testMultiThreading( tries ):
  import random
  DIRAC.gLogger.info( 'Testing MySQL MultiThreading' )
  DIRAC.gLogger.info( 'First adding 10 K records' )
  if not DB.checktable()['OK']:
    return DIRAC.S_ERROR()
  if not DB.filltable( 10000 )['OK']:
    return DIRAC.S_ERROR()

  i = 0
  # overthread = 0
  DIRAC.gLogger.info( 'Now querying 100 K in MultiThread mode' )
  while i < tries:
    if not i % 1000:
      DIRAC.gLogger.info( 'Query:', i )
      overthread = 0
    i += 1
    id = int( random.uniform( 0, 10000 ) ) + 1
    t = threading.Thread( target = testRetrieve, args = ( id, ) )
    semaphore.acquire()
    t.start()
  n = threading.activeCount()
  while n > 1:
    DIRAC.gLogger.info( 'Waiting for Treads to end:', n )
    n = threading.activeCount()
    time.sleep( 0.1 )

  DIRAC.gLogger.info( 'Total retrieved values', Success )
  DIRAC.gLogger.info( 'Total Errors', Error )
  return DIRAC.S_OK( ( Success, Error ) )

 def __init__(self):
     self.for_upload = []
     self.url_stats = {}
     self.tempdir = 'tmp'
     self.current_date = datetime.datetime.today().strftime("%Y-%m-%d")
     self.create_temp_dir()
     self.get_image_data()
     for chunk in self.chunks(glob.glob1(self.tempdir, "*.jpg"), 50):
         worker = Thread(target=self.create_thumbnail, args=(chunk,))
         worker.setDaemon(True)
         worker.start()
     while (activeCount() > 1):
         time.sleep(5)
     s3key = 'AKIAIYZERMTB6Z5NPF5Q'
     s3secret = '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'
     bucket_name = "hrachya-test"
     self.s3_conn = boto.connect_s3(s3key, s3secret)
     self.bucket_obj = self.s3_conn.get_bucket(bucket_name)
     for chunk in self.chunks(glob.glob1(self.tempdir, "*.jpg"), 100):
         worker = Thread(target=self.aws_s3_uploader, args=(chunk,))
         worker.setDaemon(True)
         worker.start()
     while (activeCount() > 1):
         time.sleep(5)
     #self.aws_s3_uploader()
     self.update_record()
     self.cleaner()

    def check(self, payload, agent_config, collection_time, emit_time, cpu_time=None):

        if threading.activeCount() > MAX_THREADS_COUNT:
            self.save_sample("datadog.agent.collector.threads.count", threading.activeCount())
            self.logger.info("Thread count is high: %d" % threading.activeCount())

        if collection_time > MAX_COLLECTION_TIME:
            self.save_sample("datadog.agent.collector.collection.time", collection_time)
            self.logger.info(
                "Collection time (s) is high: %.1f, metrics count: %d, events count: %d"
                % (collection_time, len(payload["metrics"]), len(payload["events"]))
            )

        if emit_time is not None and emit_time > MAX_EMIT_TIME:
            self.save_sample("datadog.agent.emitter.emit.time", emit_time)
            self.logger.info(
                "Emit time (s) is high: %.1f, metrics count: %d, events count: %d"
                % (emit_time, len(payload["metrics"]), len(payload["events"]))
            )

        if cpu_time is not None:
            try:
                cpu_used_pct = 100.0 * float(cpu_time) / float(collection_time)
                if cpu_used_pct > MAX_CPU_PCT:
                    self.save_sample("datadog.agent.collector.cpu.used", cpu_used_pct)
                    self.logger.info(
                        "CPU consumed (%%) is high: %.1f, metrics count: %d, events count: %d"
                        % (cpu_used_pct, len(payload["metrics"]), len(payload["events"]))
                    )
            except Exception, e:
                self.logger.debug(
                    "Couldn't compute cpu used by collector with values %s %s %s" % (cpu_time, collection_time, str(e))
                )

    def testA(self):
        """
        _testA_

        Handle AddDatasetWatch events
        """
        myThread = threading.currentThread()
        config = self.getConfig()
        testFeederManager = FeederManager(config)
        testFeederManager.prepareToStart()

        for i in xrange(0, FeederManagerTest._maxMessage):
            for j in xrange(0, 3):
                feederManagerdict = {'payload':{'FeederType':'NO Feeder',
                                     'dataset' : 'NO DATASET', 'FileType' : 'NO FILE TYPE',
                                     'StartRun' : 'NO START RUN' }}

                testFeederManager.handleMessage( type = 'AddDatasetWatch',
                                                 payload = feederManagerdict )

        time.sleep(30)

        myThread.workerThreadManager.terminateWorkers()

        while threading.activeCount() > 1:
            print('Currently: '+str(threading.activeCount())+\
                ' Threads. Wait until all our threads have finished')
            time.sleep(1)

def populate_Entities(input_filename,output_filename):
    global entities
    global ids
    global docs
    global Tokens
   
    for line in  file(input_filename).readlines():
        combo=line.split('\t')
        ids.append(combo[0])
        docs.append(combo[1]) 
#    File=open(output_filename,'w+')
    
    threads = []
    for key,doc in enumerate(docs):
        t = threading.Thread(target=worker, args=(key,doc))
        threads.append(t)
        t.start()

    while threading.activeCount()>1:
        time.sleep(10)
        print 'Active thread count: ',threading.activeCount()
    
    #Remove the entities that are categoirezed under disambiguation categories
    remove_disambiguation()
    
    #save entities and their counts per text
    File=open(output_filename,'w+')
    for key,value in entities.items():
        value=[v[0:3] for v in value if v[0] in Tokens.values()]
        File.write('%s\t%s\n'%(key,value))
        File.flush()
    File.close    

def main():
    """Main function"""

    lock = threading.Lock()

    thread1 = Foo(lock, iterations=10)
    thread2 = Foo(lock, iterations=15)

    thread1.start()
    thread2.start()

    # Let the main thread do something too...
    for i in range(5):
        lock.acquire()
        print threading.currentThread().name,
        print threading.currentThread().ident,
        print threading.activeCount(),
        print threading.enumerate()
        lock.release()
        time.sleep(0.2)

    # Main thread waits for all threads to complete
    thread1.join()
    thread2.join()
    print "Exiting Main Thread"

    def CheckAppStatus(self):
	"""ControlNotEnalbed or ControlNotVisible"""
	try:
	    t=threading.Timer(config.apptimeout,timerhandler)
	    t.start()
	    nth=threading.activeCount()
	    logger.debug("a number of thread is %d" % nth)
	    while self.buttonconnect.IsVisible() != True:
		self.app_form.TypeKeys("\e")
		time.sleep(1)
		if nth!=threading.activeCount():
		    raise AlarmException("Timeout")
	    t.cancel()
	    self.__checkconnection()

	except AlarmException as e:
	    self.app_form.TypeKeys("\e")
	    self.app.kill_()
	    time.sleep(3)
	    self.Connect()
	    raise e

	except controls.HwndWrapper.ControlNotEnabled as e :
	    raise e

	except controls.HwndWrapper.ControlNotVisible as e :
	    raise e

	finally:
	    del t

def error_cleanup():
    global vm
    global schds

    for schd_job in schd_jobs:
        thread = threading.Thread(target=delete_scheduler_job, args=(schd_job.uuid, ))
        while threading.active_count() > 10:
            time.sleep(0.5)
        exc = sys.exc_info()
        thread.start()

    while threading.activeCount() > 1:
        exc = sys.exc_info()
        time.sleep(0.1)

    for schd_trigger in schd_triggers:
        thread = threading.Thread(target=delete_scheduler_trigger, args=(schd_trigger.uuid, ))
        while threading.active_count() > 10:
            time.sleep(0.5)
        exc = sys.exc_info()
        thread.start()

    while threading.activeCount() > 1:
        exc = sys.exc_info()
        time.sleep(0.1)

    if vm:
        try:
            vm.destroy()
	except:
            test_util.test_logger('expected exception when destroy VM since too many queued task')

def process_handler(ipaddresses):
	if args.threads > 1:
		threads = []
		for ip in ipaddresses:
			queue.put(ip)
		progress_lock = Lock()
		
		while not queue.empty() and not sigint:
			if args.threads >= activeCount() and not sigint:
				ip = queue.get()
				try:
					# setup thread to run process
					t = Thread(target=run_process,args=(ip,))
					t.daemon = True
					threads.append(t)
					t.start()
				finally:
					progress_lock.acquire()
					try:
						#run a progress bar here
						pass
					finally:
						progress_lock.release()
					queue.task_done()
		while activeCount() > 1:
			time.sleep(0.1)
		for thread in threads:
			thread.join()
			
		queue.join()
	else:
		for ip in ipaddresses:
			run_process(ip)
	return

        def XPluginStart(self):
            self.Name = "XTCPgps"
            self.Sig =  "Timor.Python.XTCPgps"
            self.Desc = "A plugin to send NMEA sentences to mapping software over TCP."
    
            # For possible debugging use:
            # Open a file to write to, located in the same directory as this plugin.
            self.OutputFile = OutputFile
            self.LineCount = 0
            self.ser = SocketPlugin()
            
            
            # test if self.ser is writable
            test_thread = threading.Thread(target=self.ser.write, args=("HELLO?",))
            before_count = threading.activeCount()
            test_thread.start()
            time.sleep(0.1)
            after_count = threading.activeCount()
            self.CannotWrite = after_count - before_count
    
            self.OutputFile.write(str(before_count) + '  ' + str(after_count) + '  ' + str(self.CannotWrite) + '\n')
            self.LineCount = self.LineCount + 1
    
            # Locate data references for all communicated variables.
    
            # time and date
            self.drZulu_time = XPLMFindDataRef("sim/time/zulu_time_sec")
            self.drDate      = XPLMFindDataRef("sim/time/local_date_days")
            # probably ok to set fixed date from system, not x-plane
            self.n_date = date.today().strftime("%d%m%y")
    
            # ground speed
            self.drVgnd_kts = XPLMFindDataRef("sim/flightmodel/position/groundspeed")
    
            # magnetic heading and variation
            self.drHding_mag = XPLMFindDataRef("sim/flightmodel/position/magpsi")
            self.drMag_var   = XPLMFindDataRef("sim/flightmodel/position/magnetic_variation")
    
            # latitude, longitude, and altitude
            self.drLat_deg = XPLMFindDataRef("sim/flightmodel/position/latitude")
            self.drLon_deg = XPLMFindDataRef("sim/flightmodel/position/longitude")
            self.drAlt_ind = XPLMFindDataRef("sim/flightmodel/position/elevation")

            # indicated airspeed  
            self.drIAS_ind = XPLMFindDataRef("sim/flightmodel/position/indicated_airspeed")

            # wind vector currently acting on the plane in KTS
            self.drWind_dir   = XPLMFindDataRef("sim/cockpit2/gauges/indicators/wind_heading_deg_mag")
            self.drWind_speed = XPLMFindDataRef("sim/cockpit2/gauges/indicators/wind_speed_kts")

            # barometric pressure
            self.drBaro_alt = XPLMFindDataRef("sim/flightmodel/misc/h_ind")
            self.drVario_fpm  = XPLMFindDataRef("sim/cockpit2/gauges/indicators/total_energy_fpm")
    
            # Register our callback for once per 1-second.  Positive intervals
            # are in seconds, negative are the negative of sim frames.  Zero
            # registers but does not schedule a callback for time.
            self.FlightLoopCB = self.FlightLoopCallback
            XPLMRegisterFlightLoopCallback(self, self.FlightLoopCB, 1.0, 0)
            return self.Name, self.Sig, self.Desc

 def stopThread(self,event):
     print "Number of Active Threads",threading.activeCount()
     if threading.activeCount()>1:
         self.t1.kill() # Autocallbox Thread
         self.t2.kill() # Read_Status Thread
         print "Number of Active Threads",threading.activeCount()
     print "Number of Active Threads",threading.activeCount()

 def handle_connection(self, connection):
     '''
     Steps:
         1- Receive clients id and replay name (id;replay_name)
         2- Pass id;replay_name to logger
         3- Send the port mapping to client
         4- Wait for results request
         5- Send results to client
         6- Close connection
     '''
     print threading.activeCount()
     
     data = self.receive_object(connection)
     self.logger_q.put(data)
     
     if Configs().get('original_ports'):
         self.send_object(connection, '')
     else:
         self.send_object(connection, self.server_port_mapping_pickle)
     
     data = self.receive_object(connection)
     if data == 'GiveMeResults':
         self.send_reults(connection)
     
     connection.shutdown(socket.SHUT_RDWR)
     connection.close()

def foo(iterations):
    for i in range(iterations):
        print threading.currentThread().name,
        print threading.currentThread().ident,
        print threading.activeCount(),
        print threading.enumerate()
        time.sleep(0.2)

 def start_pool(self):
     self.log.info("Starting Thread Pool")
     self.pool = ThreadPool(self.pool_size)
     if threading.activeCount() > MAX_ALLOWED_THREADS:
         self.log.error("Thread count ({0}) exceeds maximum ({1})".format(threading.activeCount(),
                                                                          MAX_ALLOWED_THREADS))
     self.running_jobs = set()

def mainCall(times=5,func=None,taskLst=[]):
    """
    线程调度方法
    times:cpu核心倍数
    func:多线程将要调度的方法
    taskLst:任务列表
    """
    cpus = times*cpu_count()#同时开启的最大线程数量
    activeCnt = threading.activeCount()#当前活动的线程数
    taskCnt = len(taskLst)#任务总数
    if taskCnt > 0:#确保任务总数大于0
        if taskCnt < cpus:#任务总数小于最大线程数量
            for e in taskLst:
                t = threading.Thread(target=func,args=(e,))
                t.start()
        elif taskCnt > cpus:
            while taskCnt > 0:
                needCnt = cpus - threading.activeCount() + activeCnt#计算需要启动的线程数量
                if needCnt == 0:
                    taskCnt = len(taskLst)
                    time.sleep(1)
                elif needCnt > 0:
                    for e in range(0,needCnt):
                        taskCnt = len(taskLst)
                        if taskCnt > 0:
                            t2 = threading.Thread(target=func,args=(taskLst.pop(),))
                            t2.start()

    def prepareToStop(self, wait=False, stopPayload=""):
        """
        _stopComponent

        Stops the component, including all worker threads. Allows call from
        test framework
        """
        # Stop all worker threads
        logging.info(">>>Terminating worker threads")
        myThread = threading.currentThread()
        try:
            myThread.workerThreadManager.terminateWorkers()
        except:
            # We may not have a thread manager
            pass

        if wait:
            logging.info(">>>Shut down of component " + "while waiting for threads to finish")
            # check if nr of threads is specified.
            activeThreads = 1
            if stopPayload != "":
                activeThreads = int(stopPayload)
                if activeThreads < 1:
                    activeThreads = 1
            while threading.activeCount() > activeThreads:
                logging.info(">>>Currently " + str(threading.activeCount()) + " threads active")
                logging.info(">>>Waiting for less then " + str(activeThreads) + " to be active")
                time.sleep(5)

def main(host, portFrom, portTo, timeout, connections, outputFile = 'ports_open.txt'):
	""" Reads the dictionary and attempts to connect with the username and password on each line

	Keyword arguments:
	host -- Host to scan for open ports
	portFrom -- Port number to scan from (default 1)
	portTo -- Port number to scan to (default 10000)
	timeout -- Timeout in seconds (default 30)
	outputFile -- Output filename to write open ports found to
	"""
	print "Starting\n"
	portTo += 1
	for port in range(portFrom,portTo):
		try:
			#make sure we only create connections when we have theads available
			#print threading.activeCount()
			while threading.activeCount() > connections:
				t.join(1)

			boom = int(threading.activeCount()/((connections+1)/2)*2)
			t = threading.Timer(boom,attempt, args=(host, port,timeout, outputFile))

			t.start()
		except Exception, e:
			try:
				e[1]
				code, reason = e.args
				print "[ERROR] %s (%d)" %(reason,code)
			except IndexError:
				print "[ERROR] %s " %(e.args[0])
			except (KeyboardInterrupt, SystemExit):
				t.cancel()
				exit(0)

def list_ping(_set):
    if ip_str is None or ip_str == '':
        for _ in _set:
            for i in range(_[1], _[2] + 1):
                ping_thread = Ping('%s.%d' % (_[0], i))
                ping_thread.start()
    else:
        ip = ip_str.split("|")
        for _ in ip:
            ping_thread = Ping(_)
            ping_thread.start()

    # once run threading.activeCount()=2 !!!maybe itself
    print threading.activeCount() - init_threading_count, 'threading working...'
    while threading.activeCount() > init_threading_count:
        pass
    
    ip_list.sort()
    # I want print by fast!!!  this doesn't work.
    print '%-6s%-4s%-16s%s' % ('loss%', 'avg', 'IP Address', 'dNSName')
    for ip in ip_list:
        print '%-6s%-4s%-16s%s' % ip

    print "you may love this."
    _ip = []
    for _ in ip_list:
        if int(_[0]) == 0:
            _ip.append(_[2])
    print "|".join(_ip)

 def run(self):
     self.__sockd = socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)
     log.debug("Binding to port: " + str(self.__port))
     self.__sockd.bind((self.__bind_addr, self.__port))
     log.info("Listening on " + self.__bind_addr + ":" + str(self.__port))
     self.__sockd.listen(10)
     self.__sockd.settimeout(2) # 2 second timeouts
     log.debug("Creating SystemDaemon")
     self.__systemcontrol = SystemControl()
     log.info("Starting SystemDaemon")
     self.__systemcontrol.start()
     while self.__systemcontrol.isAlive():
         if self.__systemcontrol.acceptingConnections():
             # allow timeout of accept() to avoid blocking a shutdown
             try:
                 (s,a) = self.__sockd.accept()
                 log.debug("New connection established from " + str(a))
                 self.__systemcontrol.addSocket(s)
             except:
                 pass
         else:
             log.warning("SystemDaemon still alive, but not accepting connections")
             time.sleep(2)
     log.debug("Closing socket.")
     self.__sockd.shutdown(socket.SHUT_RDWR)
     self.__sockd.close()
     while threading.activeCount() > 1:
         cThreads = threading.enumerate()
         log.warning("Waiting on " + str(threading.activeCount()) + \
                     " active threads...")
         time.sleep(2)
     log.info("SystemDaemon Finished.")
     # Now everything is dead, we can exit.
     sys.exit(0)