def close(self, error=None):
     """ close down this channel with an optional error message.
         Note that closing of a channel tied to remote_exec happens
         automatically at the end of execution and cannot be done explicitely.
     """
     if self._executing:
         raise IOError("cannot explicitly close channel within remote_exec")
     if self._closed:
         self.gateway._trace(self, "ignoring redundant call to close()")
     if not self._closed:
         # state transition "opened/sendonly" --> "closed"
         # threads warning: the channel might be closed under our feet,
         # but it's never damaging to send too many CHANNEL_CLOSE messages
         # however, if the other side triggered a close already, we
         # do not send back a closed message.
         if not self._receiveclosed.isSet():
             put = self.gateway._send
             if error is not None:
                 put(Message.CHANNEL_CLOSE_ERROR, self.id, dumps_internal(error))
             else:
                 put(Message.CHANNEL_CLOSE, self.id)
             self._trace("sent channel close message")
         if isinstance(error, RemoteError):
             self._remoteerrors.append(error)
         self._closed = True         # --> "closed"
         self._receiveclosed.set()
         queue = self._items
         if queue is not None:
             queue.put(ENDMARKER)
         self.gateway._channelfactory._no_longer_opened(self.id)

 def on_message(self, message):
     #logging.info("got message %r", message)
     parsed = tornado.escape.json_decode(message)
     if 'command' in parsed:
         command = parsed['command'].lower()
         data = parsed['data'] if 'data' in parsed else None
         req_id = parsed['req_id'] if 'req_id' in parsed else None
         if hasattr(self, 'on_' + command):
             getattr(self, 'on_' + command)(data, req_id)
         elif hasattr(self, command + '_queue'):
             queue = getattr(self, command + '_queue')
             url = data.get('url')
             logging.debug("putting response in queue %s", url)
             if queue is not None:
                 if isinstance(queue, dict):
                     if url in queue:
                         queue[url].put(data)
                 else:
                     queue.put(data)
         elif data and req_id:
             if isinstance(data, dict):
                 args = data['args'] if 'args' in data else [data]
             else:
                 args = [data]
             logging.info("got callback[%s] (%r)" % (req_id, args,))
             ServerSocketHandler.send('callback', args, req_id)

def data_link( self, threadName, delay, in_file, queue):
    "Fetch data from a particular data link"

    # Open and read file
    ifile = open(in_file,'rb')

    # Read data (get packets)
    data = ifile.read(packet_size)
    
    while data:

        # Push bytes into Queue
        queue.put(data)

        # Sleep the thread
        time.sleep(delay)

        # Read data (get packets)
        data = ifile.read(packet_size)

        print("Fetching from %s" % threadName)

    # Close file
    ifile.close()
    return

def ucs(source, target, graph):
    """ Uniform-cost graph search """
    queue = queue.PriorityQueue() # fringe
    queue.put((0, source))

    parent = {source:None}
    visited = {}

    while not queue.empty():
        (d, v_in) = queue.get()

        if v_in not in visited or d < visited[v_in]:

            if v_in == target:
                return (d, build_path(parent, target))

            for v_out in graph.adj(v_in):
                cost = graph.distance(v_in, v_out) + d
                if v_out not in visited:
                    queue.put((cost, v_out))
                    parent[v_out] = v_in

            visited[v_in] = cost

    return None

def command(queue, stop):
    commands = ['error',
                'move',
                'set safety',
                'fire',
                'move reticle',
                'store location',
                'get reticle',
                'get altitude',
                'get azimuth',
                'get safety',
                'get locations',
                '0x0',
                '0x10',
                '0x11',
                '0x12',
                '0x13',
                '0x14']
    count = 0
    while True:
        if stop.is_set():
            break
        if count % 10 == 11:
            command = random.choice(commands)
            queue.put([command, 0])
        count += 1
        time.sleep(1)

def absoluteFade(indexes, rgb, fadeTime):
    '''Is given a color to fade to, and executes fade'''
    if not fadeTime:
        fadeTime = 1 / frameRate
    for c in rgb:
        c = makeEightBit(c)
    #Calculates how many individual fade frames are needed
    alterations = int(fadeTime * frameRate)
    queueList = []
    queueLock.acquire()
    while not queue.empty():
        queueList.append(queue.get())
        queue.task_done()
    #Amount of frames that need to be added to queue
    appends = alterations - len(queueList)
    #fill out the queue with blank dictionaries to populate
    if appends > 0:
        for i in range(abs(appends)):
            queueList.append({})
    #Iterate down indexes, figure out what items in queue need to be altered
    for i in indexes:
        #INVESTIGATE: THIS MIGHT BE THE SOURCE OF FLASHING ISSUES AT THE START OF A COMMAND
        start = pixels[i]
        bridgeGenerator = bridgeValues(alterations, start, rgb)
        for m in range(alterations):
            queueList[m][i] = next(bridgeGenerator)
    #If this command overrides a previous command to the pixel, it should wipe any commands remaining
        if appends < 0:
            for r in range(abs(appends)):
                if i in queueList[alterations + r]:
                    del queueList[alterations + r][i]
    while queueList:
        queue.put(queueList.pop(0))
    queueLock.release()

def multiCommand(commands):
    maxAlterations = int(max([i[2] for i in commands]) * frameRate)
    queueList = []
    queueLock.acquire()
    while not queue.empty():
        queueList.append(queue.get())
        queue.task_done()
    appends = maxAlterations - len(queueList)
    if appends > 0:
        for i in range(abs(appends)):
            queueList.append({})
    for c in commands:
        commandAlterations = int(c[2] * frameRate)
        for i in range(c[0][0], c[0][1]):
            start = pixels[i]
            bridgeGenerator = bridgeValues(commandAlterations, start, c[1])
            for m in range(commandAlterations):
                queueList[m][i] = next(bridgeGenerator)
        if appends < 0:
            for r in range(abs(appends)):
                if i in queueList[commandAlterations + r]:
                    del queueList[commandAlterations + r][i]
    while queueList:
        queue.put(queueList.pop(0))
    queueLock.release()

def new_request(http_request):
    form = forms.RequestForm()
    items_formset = forms.ItemRequestFormSet()

    if http_request.method == "POST":
        form = forms.RequestForm(http_request.POST)
        items_formset = forms.ItemRequestFormSet(http_request.POST)
        if form.is_valid() and items_formset.is_valid():
            logger.debug("creating the new request.")
            request = form.save()
            for item_data in items_formset.cleaned_data:
                if item_data:
                    logger.debug("adding a new item.")
                    item_request = models.ItemRequest()
                    item_request.request = request
                    item_request.item = item_data['item']
                    item_request.quantity = item_data['quantity']
                    item_request.save()
            logger.debug("new request created.")

            logger.debug("Sending new request to the queue.")
            queue.put(request)

            form = forms.RequestForm()
            items_formset = forms.ItemRequestFormSet()

    return render(http_request, "new_request.html",
                  {'form': form, 'items_formset': items_formset})

  def use_cached_files(self, cache_key, results_dir=None):
    if self._localcache.has(cache_key):
      return self._localcache.use_cached_files(cache_key, results_dir)

    queue = multiprocessing.Queue()
    try:
      response = self._request('GET', cache_key)
      if response is not None:
        threading.Thread(
          target=_log_if_no_response,
          args=(
            60,
            "\nStill downloading artifacts (either they're very large or the connection to the cache is slow)",
            queue.get,
          )
        ).start()
        # Delegate storage and extraction to local cache
        byte_iter = response.iter_content(self.READ_SIZE_BYTES)
        res = self._localcache.store_and_use_artifact(cache_key, byte_iter, results_dir)
        queue.put(None)
        return res
    except Exception as e:
      logger.warn('\nError while reading from remote artifact cache: {0}\n'.format(e))
      queue.put(None)
      # TODO(peiyu): clean up partially downloaded local file if any
      return UnreadableArtifact(cache_key, e)

    return False

def reader(pipe,queue):
	"""Target for threading for scrolling BASH function below."""
	try:
		with pipe:
			for line in iter(pipe.readline,b''):
				queue.put((pipe, line))
	finally: queue.put(None)

def producer(queue):
    while True:
        time.sleep(1)
        print("Queue size: %d" % queue.qsize())
        print("Put Widget")
        widget = random.choice(('drum', 'stick', 'trombone', 'swordfish'))
        queue.put(widget, block=False)

def init(name, number=10):
	global cnt
	global visited
	# blog_name = input('输入博客名称:')
	# thread_num = input('输入启动线程数:')
	blog_name = name.lower()
	th_num = int(number)
	url = 'http://blog.csdn.net/' + blog_name + '/'
	opener = urllib.request.build_opener(urllib.request.HTTPHandler)
	headers = [
		('User-Agent', 'Mozilla/5.0 (Windows NT 6.3; WOW64; Trident/7.0; rv:11.0) like Gecko')
	]
	urllib.request.install_opener(opener)
	opener.addheaders = headers

	queue.put(url)
	visited |= {url}
	cnt = 0

	for i in range(th_num):
		t = CsdnBlogSpider(queue,opener,blog_name)
		t.setDaemon(True)
		t.start()
	queue.join()
	print('--------end!!!-----')
	print('共抓取:' + str(cnt))

 def support_test_send_reply_to(self, address, queue):
     message = uuid.uuid4().hex
     with self.context.socket(roles['speaker']) as socket:
         socket.connect("tcp://%s" % address)
         socket.send(nw0.sockets._serialise(message))
         reply = nw0.sockets._unserialise(socket.recv())
     queue.put(reply)

def run_server(tmpdir, handler_class, stop_event, queue):  # pragma: no cover
    """
    Runs an HTTP server serving files from given tmpdir in a separate
    process.  When it's ready, it sends a URL to the server over a
    queue so the main process (the HTTP client) can start making
    requests of it.
    """
    class HTTPRequestHandler(handler_class):
        def translate_path(self, path):
            path = handler_class.translate_path(self, path)
            path = os.path.join(
                tmpdir,
                os.path.relpath(path, os.getcwd()))
            return path

    server = socketserver.TCPServer(("127.0.0.1", 0), HTTPRequestHandler)
    domain, port = server.server_address
    url = "http://{0}:{1}/".format(domain, port)

    # Set a reasonable timeout so that invalid requests (which may occur during
    # testing) do not cause the entire test suite to hang indefinitely
    server.timeout = 0.1

    queue.put(url)

    # Using server.serve_forever does not work here since it ignores the
    # timeout value set above. Having an explicit loop also allows us to kill
    # the server from the parent thread.
    while not stop_event.isSet():
        server.handle_request()

    server.server_close()

 def parse_potential_recent_deaths(self, refresh):
     from queue import Queue, Empty
     queue = Queue()
     for name, info in self.chars.items():
         if (info.is_online() or time.time() - info.last_online() < 1200) and info.vocation != 'N':
             queue.put(name)
     task_count = queue.qsize()
     def get_info():
         while True:
             try:
                 name = queue.get(block=False)
                 tasks_left = queue.qsize()
             except Empty:
                 return
             info = tibiacom.char_info(name)
             self.chars[name].deaths = info["deaths"]
             refresh()
             queue.task_done()
             print("pzlock update: %d/%d" % ((task_count - tasks_left), task_count))
     threads = []
     for i in range(10):
         thrd = threading.Thread(target=get_info)
         thrd.start()
         threads.append(thrd)
     queue.join()
     for t in threads:
         t.join()

def running(queue):
    
    for x in range(5):
        text = 'message ' + str(x)
        print('PUT:', text)
        queue.put(text)
        time.sleep(4)
    queue.put('last')

def worker(url, queue): # get queue as argument
    r = requests.get(url)

    soup = BeautifulSoup(r.text, "html.parser")
    data = soup.find("span", {"class": "text"}).get_text()

    # send result to main thread using queue
    queue.put(data)

 def prepare_fitness(self):
     """
     Here we request fitness calculation.  Prepare place for result
     and put our string into a queue.  A running worker-thread will
     pick it up from the queue and calculate.
     """
     self.result_dict = {}
     queue.put((str(self), self.result_dict))

def download_t(uid, cat, start, queue, t_name):
        global EXIT
        url = "http://api.douban.com/people/{0}/collection?cat={1}&tag=&status=&start-index={2}&max-results=50&alt=atom&apikey={3}".format(uid, cat, start,APIKEY)
        filename = '{0}_{1}_{2}'.format(uid, cat, start)
        while True:
                try:
                        fh = open("api_limit", "a+")
                        fcntl.flock(fh.fileno(), LOCK_EX|LOCK_NB)
                        break
                except IOError:
                        fh.close()
        if(fh.tell() != 0):
                os.lseek(fh.fileno(), -20, os.SEEK_END)
                line = fh.read(19)
                seconds = float(line.split()[1])
        # just be careful! 
        while float(time.time()) - seconds <= 1.8:
                pass
        fh.write("[LOG] {0:<24} {1} {2}\n".format(time.ctime(), url[:-40], time.time().__repr__()[0:-4]))
        fcntl.flock(fh, LOCK_UN)
        fh.close()
        #print('start dl:{0} {1}'.format(t_name, time.ctime()))
        try:
                req = urllib.request.Request(url, headers= {'Accept-encoding':'gzip'})
                rec = urllib.request.urlopen(req)
                compressed_data = rec.read()
                compressed_fh = io.BytesIO(compressed_data)
                gzipper = gzip.GzipFile(fileobj = compressed_fh);
                data = gzipper.read().decode('utf8')
                #data = urllib.request.urlopen(url).read().decode('utf8')
        except (urllib.error.URLError, ValueError) as e:
                if hasattr(e, 'reason'):
                        print("<h4>Cannot connected to the server</h4>")
                        print("<h4>url:</h4>",url)
                        EXIT = True
                        return
                if hasattr(e, 'code'):
                        print("<h4>Return code:",e.code,"error</h4>")
                        print("<h4>",e.msg,"</h4>")
                        print("<h4>url:</h4>",url)
                        EXIT = True
                        return
        fw = open(ROOTDIR + r'htdocs/cache_datas/' + filename, "w+", encoding = 'utf8')
        fw.write(data)
        fw.close()
        #print('creat file:' ,filename, time.ctime())
        if start == 1:
                fr = open(ROOTDIR + r'htdocs/cache_datas/' + filename, 'r', encoding = 'utf8')
                for line in fr.readlines():
                        if 'totalResults' in line:
                                max_item = int(line.split('>')[1].split('<')[0])
                                dl_nr = max_item//50
                                break
                if max_item > 50:
                        for i in range(dl_nr):
                                start += 50
                                queue.put([uid, cat, start, queue])

def test():
    for i in range(500):
        queue.put('init pro ' + str(i))
    for i in range(2):
        p = Producer()
        p.start()
    for i in range(5):
        c = Consumer()
        c.start()