def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool, worker_handler, task_handler, result_handler, cache):
        debug('finalizing pool')
        worker_handler._state = TERMINATE
        task_handler._state = TERMINATE
        debug('helping task handler/workers to finish')
        cls._help_stuff_finish(inqueue, task_handler, len(pool))
        if not (result_handler.is_alive() or len(cache) == 0):
            raise AssertionError
            result_handler._state = TERMINATE
            outqueue.put(None)
            debug('joining worker handler')
            if threading.current_thread() is not worker_handler:
                worker_handler.join(1e+100)
            if pool and hasattr(pool[0], 'terminate'):
                debug('terminating workers')
                for p in pool:
                    if p.exitcode is None:
                        p.terminate()

            debug('joining task handler')
            if threading.current_thread() is not task_handler:
                task_handler.join(1e+100)
            debug('joining result handler')
            if threading.current_thread() is not result_handler:
                result_handler.join(1e+100)
            pool and hasattr(pool[0], 'terminate') and debug('joining pool workers')
            for p in pool:
                if p.is_alive():
                    debug('cleaning up worker %d' % p.pid)
                    p.join()

        return

def eventCreator():
    aLotOfData = []
    es_conn = tools.get_es_connection()
    while True:
        d = q.get()
        m = json.loads(d)
        data = {
            '_type': 'netflow_lhcopn'
        }
        if not 'data'in m:
            print(threading.current_thread().name, 'no data in this message!')
            q.task_done()
            continue

        source = m['data']['src_site']
        destination = m['data']['dst_site']
        data['MA'] = 'capc.cern'
        data['srcInterface'] = source
        data['dstInterface'] = destination
        ts = m['data']['timestamp']
        th = m['data']['throughput']
        dati = datetime.utcfromtimestamp(float(ts))
        data['_index'] = "network_weather-" + \
            str(dati.year) + "." + str(dati.month) + "." + str(dati.day)
        data['timestamp'] = int(float(ts) * 1000)
        data['utilization'] = int(th)
        # print(data)
        aLotOfData.append(copy.copy(data))

        q.task_done()
        if len(aLotOfData) > 10:
            succ = tools.bulk_index(aLotOfData, es_conn=es_conn, thread_name=threading.current_thread().name)
            if succ is True:
                aLotOfData = []

 def free(self):
     i = threading.current_thread().ident
     if i in self.buff_dict:
         buff = self.buff_dict.pop(threading.current_thread().ident)
         buff.seek(0)
         buff.truncate()
         self.buff_queue.append(buff)

 def release(self, *args):
     print('*' * 120, file=sys.stderr)
     print('release called: thread id:', current_thread(), 'shared:', self._is_shared, file=sys.stderr)
     traceback.print_stack()
     RWLockWrapper.release(self)
     print('release done: thread id:', current_thread(), 'is_shared:', self._shlock.is_shared, 'is_exclusive:', self._shlock.is_exclusive, file=sys.stderr)
     print('_' * 120, file=sys.stderr)

def _manager_worker_process(output_queue, futures, is_shutdown):
    """ This worker process manages taking output responses and
    tying them back to the future keyed on the initial transaction id.
    Basically this can be thought of as the delivery worker.

    It should be noted that there are one of these threads and it must
    be an in process thread as the futures will not serialize across
    processes..

    :param output_queue: The queue holding output results to return
    :param futures: The mapping of tid -> future
    :param is_shutdown: Condition variable marking process shutdown
    """
    log.info("starting up manager worker: %s", threading.current_thread())
    while not is_shutdown.is_set():
        try:
            workitem = output_queue.get()
            future = futures.get(workitem.work_id, None)
            log.debug("dequeue manager response: %s", workitem)
            if not future: continue
            if workitem.is_exception:
                future.set_exception(workitem.response)
            else: future.set_result(workitem.response)
            log.debug("updated future result: %s", future)
            del futures[workitem.work_id]
        except Exception as ex:
            log.exception("error in manager")
    log.info("manager worker shutting down: %s", threading.current_thread())

def worker_func():
    print('worker thread started in %s' % (threading.current_thread()))
    # 改变随机数生成器的种子
    random.seed()
    # 让线程睡眠s随机一段时间
    time.sleep(random.random())
    print('worker thread finished in %s' % (threading.current_thread()))

 def _thread_worker(self):
     """
     A worker that does actual jobs
     """
     self.log.debug("Starting shooter thread %s", th.current_thread().name)
     while not self.quit.is_set():
         try:
             task = self.task_queue.get(timeout=1)
             if not task:
                 self.log.info(
                     "%s got killer task.", th.current_thread().name)
                 break
             timestamp, missile, marker = task
             planned_time = self.start_time + (timestamp / 1000.0)
             delay = planned_time - time.time()
             if delay > 0:
                 time.sleep(delay)
             self.gun.shoot(missile, marker, self.results)
         except (KeyboardInterrupt, SystemExit):
             break
         except Empty:
             if self.quit.is_set():
                 self.log.debug(
                     "Empty queue. Exiting thread %s",
                     th.current_thread().name)
                 return
         except Full:
             self.log.warning(
                 "Couldn't put to result queue because it's full")
     self.log.debug("Exiting shooter thread %s", th.current_thread().name)

    def _worker(self):
        '''
        This is the worker which will get the image from 'inbox',
        calculate the hash and puts the result in 'outbox'
        '''

        while not self.shutdown.isSet():
            
            try:
                image_path = self.inbox.get_nowait()
            except Empty:
                print 'no data found. isset: ' , self.done.isSet()
                if not self.done.isSet():
                    with self.empty:
                        self.empty.wait()
                        continue
                else:
                    break

            if not os.path.exists(image_path):
                self.error((image_path, 'Image Does not Exist'))
                
            try:
                print '[%s] Processing %s' % (current_thread().ident, image_path)
                image_hash = average_hash(image_path)
                self.outbox.put((image_hash, image_path))
            except IOError as err:
                print 'ERROR: Got %s for image : %s' % (image_path, err)
        print 'Worker %s has done processing.' % current_thread().ident

    def init(self, params):
        self.params = dict(params)
        # OpenERP session setup
        self.session_id = self.params.pop("session_id", None) or uuid.uuid4().hex
        self.session = self.httpsession.get(self.session_id)
        if not self.session:
            self.session = session.OpenERPSession()
            self.httpsession[self.session_id] = self.session

        # set db/uid trackers - they're cleaned up at the WSGI
        # dispatching phase in openerp.service.wsgi_server.application
        if self.session._db:
            threading.current_thread().dbname = self.session._db
        if self.session._uid:
            threading.current_thread().uid = self.session._uid

        self.context = self.params.pop('context', {})
        self.debug = self.params.pop('debug', False) is not False
        # Determine self.lang
        lang = self.params.get('lang', None)
        if lang is None:
            lang = self.context.get('lang')
        if lang is None:
            lang = self.httprequest.cookies.get('lang')
        if lang is None:
            lang = self.httprequest.accept_languages.best
        if not lang:
            lang = 'en_US'
        # tranform 2 letters lang like 'en' into 5 letters like 'en_US'
        lang = babel.core.LOCALE_ALIASES.get(lang, lang)
        # we use _ as seprator where RFC2616 uses '-'
        self.lang = lang.replace('-', '_')

def remove_heart_log(*args, **kwargs):
    if six.PY2:
        if threading.current_thread().name == 'MainThread':
            debug_log(*args, **kwargs)
    else:
        if threading.current_thread() == threading.main_thread():
            debug_log(*args, **kwargs)

def send_request():
    import requests
    print threading.current_thread().name
    url = 'http://localhost:9999/hello/' + threading.current_thread().name
    response=requests.get(url)

    print response.content

 def wrapper(*args, **kw):
     print("entering %s for thread %s:%s"
           %(fn.func_name, getpid(), current_thread()))
     ret = fn(*args, **kw)
     print("leaving %s for thread %s:%s"
           %(fn.func_name, getpid(), current_thread()))
     return ret

    def _threaded_resolve_AS():
        """Get an ASN from the queue, resolve it, return its routes to the
        *main* process and repeat until signaled to stop.
        This function is going to be spawned as a thread.
        """
        while True:
            current_AS = q.get()
            if current_AS == 'KILL':
                q.task_done()
                break

            try:
                resp = comm.get_routes_by_autnum(current_AS, ipv6_enabled=True)
                if resp is None:
                    raise LookupError
                routes = parsers.parse_AS_routes(resp)
            except LookupError:
                logging.warning("{}: {}: No Object found for {}"
                                .format(mp.current_process().name,
                                        threading.current_thread().name,
                                        current_AS))
                routes = None
            except Exception as e:
                logging.error("{}: {}: Failed to resolve DB object {}. {}"
                              .format(mp.current_process().name,
                                      threading.current_thread().name,
                                      current_AS, e))
                routes = None
            result_q.put((current_AS, routes))
            q.task_done()

def show_thread(q, extraByteCodes):
    for i in range(5):
        for j in range(extraByteCodes):
            pass
        # q.put(threading.current_thread().name)
        print threading.current_thread().name
    return

def loop():
    print 'thread %s is running...' % threading.current_thread().name
    n = 0
    while n < 5:
        n = n + 1
        print 'thread %s >> %s' % (threading.current_thread().name, n)
    print 'thread %s ended.' % threading.current_thread().name

	def submit(self, items, submitType):
		""" Submit the news items to the forum """

		# If there is nothing to submit, just return
		if not items: 
			return

		curDate  = datetime.date.today()
		newMonth = (curDate == datetime.date(curDate.year, curDate.month, 1))
		#subject  = u'Ежедневные новости науки - ' + months[curDate.month - 1] + u', ' + str(curDate.year).encode('utf-8')
		subject  = u'Ежедневные новости науки - {0}, {1}'.format(months[curDate.month - 1], str(curDate.year)) 

		if newMonth or subject.encode('utf-8') != self.getCurrentTopic('subject'):
			self.log.info("{0} It's a new month".format(current_thread().name))
			self.mech.open("http://www.ateism.ru/forum/posting.php?mode=post&f=35")
			self.mech.select_form(nr=1)
			self.mech.form['subject'] = subject.encode('utf-8') 
		else:
			self.mech.open(self.getCurrentTopic())
			self.mech.select_form(nr=1)

		self.mech.form['message'] = u''.join([unicode(i) for i in items]).encode('utf-8')

		self.log.info(u'{0} Submitting {1} items... Submit type: {2}'.format(current_thread().name, len(items), submitType))
		#self.mech.submit(name='post')
		if submitType == 'post' or submitType == 'preview':
			self.mech.submit(name=submitType)
		else:
			self.log.warning(u'{0} Unknown submit type: [{1}]'.format(current_thread().name, submitType))

		self.log.info(u'{0} Submission [OK]'.format(current_thread().name))
		
		if newMonth:
			self.updateTopicURL(subject)

def run_using_threadpool(fn_to_execute, inputs, pool_size):
  """For internal use only; no backwards-compatibility guarantees.

  Runs the given function on given inputs using a thread pool.

  Args:
    fn_to_execute: Function to execute
    inputs: Inputs on which given function will be executed in parallel.
    pool_size: Size of thread pool.
  Returns:
    Results retrieved after executing the given function on given inputs.
  """

  # ThreadPool crashes in old versions of Python (< 2.7.5) if created
  # from a child thread. (http://bugs.python.org/issue10015)
  if not hasattr(threading.current_thread(), '_children'):
    threading.current_thread()._children = weakref.WeakKeyDictionary()
  pool = ThreadPool(min(pool_size, len(inputs)))
  try:
    # We record and reset logging level here since 'apitools' library Beam
    # depends on updates the logging level when used with a threadpool -
    # https://github.com/google/apitools/issues/141
    # TODO: Remove this once above issue in 'apitools' is fixed.
    old_level = logging.getLogger().level
    return pool.map(fn_to_execute, inputs)
  finally:
    pool.terminate()
    logging.getLogger().setLevel(old_level)

def run_script_remote(node):
    print current_thread().name+": Execution of "+script+".sh on node "+node+" started."
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.WarningPolicy())
    client.connect(node, username='princeton_multisurf')
    
    channel = client.invoke_shell()
    channel.send('sudo chmod 744 '+script+'.sh\n')
    out = ''
    while not out.endswith('$ '):
        resp = channel.recv(1024)
        out += resp

    # Reading the output back seems to be the only way to 
    # make sure the update finishes
    channel.send('./'+script+'.sh\n')
    out = ''
    while not out.endswith('$ '):
        resp = channel.recv(1024)
        out += resp

    #add the newline to the node output
    out += '\n'

    # write the update's output to a log file, just for sanity
    f = open(node+'_'+script+'.log', 'wb')
    f.write(out)
    f.close()

    client.close()
    print current_thread().name+": Finished on node "+node+"\nCheck this node's script log file to make sure there were no errors."   

    def run(self):
        while True:
            try:
                url, deep = self.pool.urls_queue.get(block = True, timeout = 5)
                #url, deep = self.pool.urls_queue.get(block = False)
            except Queue.Empty:
                #time.sleep(5) # 非阻塞时使用
                logging.debug("%s, qsize: %s"%(threading.current_thread().getName(),
                                  self.pool.urls_queue.qsize()))
                if not self.pool.busy():
                    break
            else:
                self.running = True
                #print "%s, url=%s, deep=%s, qsize=%s"%(
                #      threading.current_thread().getName(), url, deep, self.pool.urls_queue.qsize())
                _spider = Spider(url, deep, self.key)
                links = _spider.spider()
                if links:
                    #print "++++++++++ In WorkThread: %s, links_lenth: %s ++++++++++"%(
                    #        threading.current_thread().getName(), len(links))
                    for i in links:
                        hash_code = md5.new(i.encode('utf-8')).hexdigest()

                        self.pool.saved_urls_lock.acquire()
                        if not hash_code in self.pool.saved_urls:
                            self.pool.saved_urls[hash_code] = i
                            self.pool.urls_queue.put((i, deep - 1))
                        #千万不能忘记解锁啊!!!
                        self.pool.saved_urls_lock.release()
                self.running = False
                self.pool.urls_queue.task_done()
        logging.info("%s, ended*******\n"%(threading.current_thread().getName()))

 def fetch_results():
   threading.current_thread().fetch_results_error = None
   try:
     new_client = self.create_impala_client()
     new_client.fetch(query, handle)
   except ImpalaBeeswaxException as e:
     threading.current_thread().fetch_results_error = e