def export_table(host, port, auth_key, db, table, directory, fields, format, error_queue, progress_info, stream_semaphore, exit_event):
    writer = None

    try:
        # This will open at least one connection for each rdb_call_wrapper, which is
        # a little wasteful, but shouldn't be a big performance hit
        conn_fn = lambda: r.connect(host, port, auth_key=auth_key)
        rdb_call_wrapper(conn_fn, "count", get_table_size, db, table, progress_info)
        table_info = rdb_call_wrapper(conn_fn, "info", write_table_metadata, db, table, directory)

        with stream_semaphore:
            task_queue = SimpleQueue()
            writer = launch_writer(format, directory, db, table, fields, task_queue, error_queue)
            writer.start()

            rdb_call_wrapper(conn_fn, "table scan", read_table_into_queue, db, table,
                             table_info["primary_key"], task_queue, progress_info, exit_event)
    except (r.RqlError, r.RqlDriverError) as ex:
        error_queue.put((RuntimeError, RuntimeError(ex.message), traceback.extract_tb(sys.exc_info()[2])))
    except:
        ex_type, ex_class, tb = sys.exc_info()
        error_queue.put((ex_type, ex_class, traceback.extract_tb(tb)))
    finally:
        if writer is not None and writer.is_alive():
            task_queue.put(("exit", "event")) # Exit is triggered by sending a message with two objects
            writer.join()
        else:
            error_queue.put((RuntimeError, RuntimeError("writer unexpectedly stopped"),
                             traceback.extract_tb(sys.exc_info()[2])))

class Logger(object):
    def __init__(self, logfilepath):
        try:
            os.remove(logfilepath)
        except OSError:
            pass
        
        self.logfilepath = logfilepath
        self.logq = SimpleQueue()
        
        self.tags = ''
        self.num_tags = 0
        
    def add_tag(self, tag):
        #self.log("adding tag {}".format(tag))
        self.num_tags += 1
        
        if self.tags != '':
            self.tags = self.tags + '.' + tag
        else:
            self.tags = tag
            
    def remove_tag(self):
        #self.log("removing tag")
        tags = self.tags.split('.')
        self.tags = ".".join(tags[:-1])
        self.num_tags -= 1
        
    def get_tag_part(self):
        if self.tags != '':
            return self.tags + ": "
        else:
            return ''
        
    def log(self, message, start_group=None, end_group=None):
        assert(type(message)==str)        
        self.logq.put(" "*self.num_tags*4 + self.get_tag_part() + message + '\n')
            
    def getlog(self):
        return self.logq.get()
            
    def getlogs(self, n=None):
        logs = []
        if n == None:
            while not self.logq.empty():
                logs.append(self.getlog())
        else:
            assert(type(n)==int)
            while not (self.logq.empty() or len(logs) == n):
                logs.append(self.getlog())
                
        return logs
        
    def write_to_file(self):        
        # mode 'a' for append
        with open(self.logfilepath, 'a') as f:
            f.writelines(self.getlogs())

    def test_can_pickle_via_queue(self):
        """
        https://github.com/andresriancho/w3af/issues/8748
        """
        sq = SimpleQueue()
        u1 = URL('http://www.w3af.com/')
        sq.put(u1)
        u2 = sq.get()

        self.assertEqual(u1, u2)

def QuiverPlotter(num):
	data_q = SimpleQueue()

	plot = Process(target=quiverPlotter,args=(data_q,num))
	plot.start()

	try:
		while True:
			data = (yield)
			if data_q.empty() == False:
				continue
			data_q.put(data)
	except GeneratorExit:
		plot.join()

def Plotter3D(plots,scale):
	data_q = SimpleQueue()

	plot = Process(target=plotter3D,args=(data_q,plots,scale))
	plot.start()

	data = {}
	try:
		while True:
			data.update((yield))
			if data_q.empty() == False:
				continue
			data_q.put(data)
	except GeneratorExit:
		pass

    def __init__(self, key, task_group, randomize):
        self.key = key
        self.gen_worker = task_group['gen_worker']
        self.task_ids = task_group['task_ids']
        self.is_parallel = task_group['is_parallel']
        if self.is_parallel:
            self.randomize = randomize
            if self.randomize:
                random.shuffle(self.task_ids)
        else:
            self.randomize = False
        self.result_queue = SimpleQueue()
        self.task_queue = SimpleQueue()

        # Don't expose queues file descriptors over Popen to, say, tarantool
        # running tests.
        set_fd_cloexec(self.result_queue._reader.fileno())
        set_fd_cloexec(self.result_queue._writer.fileno())
        set_fd_cloexec(self.task_queue._reader.fileno())
        set_fd_cloexec(self.task_queue._writer.fileno())

        for task_id in self.task_ids:
            self.task_queue.put(task_id)
        self.worker_ids = set()
        self.done = False
        self.done_task_ids = set()

class StatusTracker(object):
    def __init__(self):
        self.logq = SimpleQueue()
        self.history = []
        
    def put(self, msg):
        assert(type(msg)==str)
        self.logq.put(msg)
        
    def flushq(self):
        while not self.logq.empty():
            self.history.append(self.logq.get())
        self.prune_history()
            
    def prune_history(self):
        self.history = self.history[-100:]

    def __init__(self, data_structure, processes, scan_function, init_args, _mp_init_function):
        """ Init the scanner.

        data_structure is a world.DataSet
        processes is the number of child processes to use
        scan_function is the function to use for scanning
        init_args are the arguments passed to the init function
        _mp_init_function is the function used to init the child processes
        """
        assert (isinstance(data_structure, world.DataSet))
        self.data_structure = data_structure
        self.list_files_to_scan = data_structure._get_list()
        self.processes = processes
        self.scan_function = scan_function

        # Queue used by processes to pass results
        self.queue = SimpleQueue()
        init_args.update({'queue': self.queue})
        # NOTE TO SELF: initargs doesn't handle kwargs, only args!
        # Pass a dict with all the args
        self.pool = multiprocessing.Pool(processes=processes, initializer=_mp_init_function, initargs=(init_args,))

        # TODO: make this automatic amount
        # Recommended time to sleep between polls for results
        self.SCAN_START_SLEEP_TIME = 0.001
        self.SCAN_MIN_SLEEP_TIME = 1e-6
        self.SCAN_MAX_SLEEP_TIME = 0.1
        self.scan_sleep_time = self.SCAN_START_SLEEP_TIME
        self.queries_without_results = 0
        self.last_time = time()
        self.MIN_QUERY_NUM = 1
        self.MAX_QUERY_NUM = 5

        # Holds a friendly string with the name of the last file scanned
        self._str_last_scanned = None

def launch_graph_plot():
    q = SimpleQueue()
    Pyro4.config.HOST="10.1.1.2"
    daemon = Pyro4.Daemon()
    ns = Pyro4.locateNS()
    p = Process(target=_launch_daemon, args=(daemon, q,))
    p.start()
    graph_plot = GraphPlotPanel()
    while True:
        if not q.empty():
            item = q.get()
            if item[0] == 'time':
                print "got queue:", item
                graph_plot.set_time(item[1])
            elif item[0] == 'vertex_color':
                pass
        graph_plot.run()
        fpsClock.tick(60)

def DensityPlotter(num,size):
	# num = size/scale
	range = [[-size,size],[-size,size]]

	data_q = SimpleQueue()

	plot = Process(target=imagedraw,args=(data_q,num))
	plot.start()

	while True:
		x = (yield)

		if data_q.empty() == False:
			continue

		hist,_,_ = np.histogram2d(x[:,0],x[:,1],bins=num,range=range)
		avg = np.average(hist)
		hist = (hist - avg)/avg
		data_q.put(hist.astype(np.float32))

class RangerControlServer(HTTPServer):
    def __init__(self, fm):
        self.fm = fm
        self.queue = SimpleQueue()
        self.goDie = False
        HTTPServer.__init__(self, ("127.0.0.1", 5964), RangerControlHandler)

    def start(self):
        self.thread = threading.Thread(target=self.process)
        self.thread.start()

    def stop(self):
        self.shutdown()

    def process(self):
        self.serve_forever()

    def check_messages(self):
        if self.queue.empty():
            return None
        return self.queue.get()

    def act_on_messages(self):
        msg = self.check_messages()
        if msg == None:
            return False

        action, arg = msg
        match = re.match(r"/cdtab-(\S+)", action)
        if match != None:
            tab = match.group(1)
            if not (tab in self.fm.tabs):
                self.fm.tab_open(tab, arg)
            else:
                self.fm.tabs[tab].enter_dir(arg)
        elif action == "/cd":
            self.fm.enter_dir(arg)
        elif action == "/cdfirst":
            first_tab = self.fm._get_tab_list()[0]
            self.fm.tabs[first_tab].enter_dir(arg)
        else:
            self.fm.notify("Unknown server command", bad=True)
        return True

 def __init__(self, logfilepath):
     try:
         os.remove(logfilepath)
     except OSError:
         pass
     
     self.logfilepath = logfilepath
     self.logq = SimpleQueue()
     
     self.tags = ''
     self.num_tags = 0

class LinePlotter:
	def __init__(self,*args,**kwargs):
		self.data_q = SimpleQueue()
		self.data = {}

		self.plot = LinePlotterProcess(self.data_q)
		self.plot.add_plot(*args,**kwargs)

	def show(self):
		self.plot.start()

	def add_plot(self,*args,**kwargs):
		self.plot.add_plot(*args,**kwargs)

	def send(self,data):
		if data == GeneratorExit:
			self.plot.join()

		self.data.update(data)
		if self.data_q.empty() != False:
			self.data_q.put(data)

class ErrorMonitor:
    def __init__(self):
        self.pipe = SimpleQueue()
        self.message = None

    def main(self):
        while True:
            message = self.pipe.get()
            if message != 'Q':
                self.message = message[1:]
                LongJump.longjump()
                break
            else:
                self.pipe = None
                break
                    
    def haserror(self):
        """ master only """
        return self.message is not None
    def start(self):
        """ master only """
        self.thread = Thread(target=self.main)
        self.thread.daemon = True
        self.thread.start()
    def join(self):
        """ master only """
        try:
            self.pipe.put('Q')
            self.thread.join()
        except:
            pass
        finally:
            self.thread = None

    def slaveraise(self, type, error, traceback):
        """ slave only """
        message = 'E' * 1 + pickle.dumps((type,
            ''.join(tb.format_exception(type, error, traceback))))
        if self.pipe is not None:
            self.pipe.put(message)

    def setup(self):
        self.pid = os.getpid()
        self.worker_nums = self.config['workers']
        self.worker_class = SyncWorker
        self.queue = SimpleQueue()
        self.setup_logger()
        self.setup_signals()

        addresses = self.config['binds']
        self.sockets = create_sockets(addresses, self.logger)

        addresses_str = ', '.join(map(format_addr_str, addresses))
        self.logger.info('Arbiter booted')
        self.logger.info('Listening on: %s (%s)', addresses_str, self.pid)
        self.logger.info('Using worker: %s', self.worker_class)

def main():
    global TCP_SEND_PORT
    global TCP_SEND_IP
    global TCP_RECEIVE_IP
    global TCP_RECEIVE_PORT
    global key_store
    global eventual_requests
    global eventual_write_lock
    global eventual_read_lock
    key_store = {}
    eventual_requests = {}
    eventual_write_lock = threading.Lock()
    eventual_read_lock = threading.Lock()
    signal.signal(signal.SIGINT, signal_handler)
    TCP_RECEIVE_IP = TCP_SEND_IP = socket.gethostbyname(socket.gethostname())
    TCP_SEND_PORT = int(sys.argv[1])
    TCP_RECEIVE_PORT = int(sys.argv[2])
    BUFFER_SIZE = 1024
    listener = threading.Thread(target=listening_thread, args=[BUFFER_SIZE])
    listener.daemon = True
    listener.start()
    message_queue = SimpleQueue()
    worker = threading.Thread(target=worker_thread, args=[message_queue])
    worker.daemon = True
    worker.start()

    while 1:
        command = str(raw_input(bcolors.HEADER +  bcolors.UNDERLINE + "Enter Message:\n" + bcolors.ENDC))
        messages = []
        if command.endswith('.txt'):
            messages = readFile(command)
        else:
            messages.append(command)
        message_queue.put(messages)
        print bcolors.OKBLUE +  'System time is ' + \
                str(datetime.datetime.now().strftime("%H:%M:%S:%f")) + bcolors.ENDC

 def __init__(self, max_workers=None):
     _check_system_limits()
     if max_workers is None:
         self._max_workers = multiprocessing.cpu_count()
     else:
         self._max_workers = max_workers
     self._call_queue = multiprocessing.Queue(self._max_workers + EXTRA_QUEUED_CALLS)
     self._call_queue._ignore_epipe = True
     self._result_queue = SimpleQueue()
     self._work_ids = queue.Queue()
     self._queue_management_thread = None
     self._processes = {}
     self._shutdown_thread = False
     self._shutdown_lock = threading.Lock()
     self._broken = False
     self._queue_count = 0
     self._pending_work_items = {}

    def __init__(self, max_workers=None):
        """Initializes a new ProcessPoolExecutor instance.

        Args:
            max_workers: The maximum number of processes that can be used to
                execute the given calls. If None or not given then as many
                worker processes will be created as the machine has processors.
        """
        _check_system_limits()

        if max_workers is None:
            self._max_workers = multiprocessing.cpu_count() or 1
        else:
            if max_workers <= 0:
                raise ValueError("max_workers must be greater than 0")

            self._max_workers = max_workers

        # Make the call queue slightly larger than the number of processes to
        # prevent the worker processes from idling. But don't make it too big
        # because futures in the call queue cannot be cancelled.
        self._call_queue = multiprocessing.Queue(self._max_workers +
                                                 EXTRA_QUEUED_CALLS)
        # Killed worker processes can produce spurious "broken pipe"
        # tracebacks in the queue's own worker thread. But we detect killed
        # processes anyway, so silence the tracebacks.
        self._call_queue._ignore_epipe = True
        self._result_queue = SimpleQueue()
        self._work_ids = queue.Queue()
        self._queue_management_thread = None
        # Map of pids to processes
        self._processes = {}

        # Shutdown is a two-step process.
        self._shutdown_thread = False
        self._shutdown_lock = threading.Lock()
        self._broken = False
        self._queue_count = 0
        self._pending_work_items = {}

class BaseMultiprocessingRunner(BaseRunner):
    def __init__(self):
        super(BaseMultiprocessingRunner, self).__init__()
        self.numprocs = max(multiprocessing.cpu_count() - 1, 1)
        self.map_input_queue = SimpleQueue()
        self.map_output_queue = SimpleQueue()

    def run_map(self):
        for item in iter(self.map_input_queue.get, self.STOP_MSG):
            self.job.map(item, self.map_output_queue.put)
        self.map_output_queue.put(self.STOP_MSG)
        if self.debug:
            debug_print("Output : STOP sent")

    def run_enumerate(self):
        for inp in self.job.enumerate():
            self.map_input_queue.put(inp)
        for work in range(self.numprocs):
            self.map_input_queue.put(self.STOP_MSG)
        if self.debug:
            debug_print("Input: STOP sent")

    def run(self, job):
        self.job = job
        # Process that reads the input file
        self.enumeration_process = multiprocessing.Process(target=self.run_enumerate, args=())

        self.mappers = [multiprocessing.Process(target=self.run_map, args=()) for i in range(self.numprocs)]

        self.enumeration_process.start()
        for mapper in self.mappers:
            mapper.start()
        r = self.run_reduce()
        self.enumeration_process.join()
        for mapper in self.mappers:
            mapper.join()
        return r

class ProcessPoolExecutor(_base.Executor):
    def __init__(self, max_workers=None):
        """Initializes a new ProcessPoolExecutor instance.

        Args:
            max_workers: The maximum number of processes that can be used to
                execute the given calls. If None or not given then as many
                worker processes will be created as the machine has processors.
        """
        _check_system_limits()

        if max_workers is None:
            self._max_workers = multiprocessing.cpu_count()
        else:
            self._max_workers = max_workers

        # Make the call queue slightly larger than the number of processes to
        # prevent the worker processes from idling. But don't make it too big
        # because futures in the call queue cannot be cancelled.
        self._call_queue = multiprocessing.Queue(self._max_workers +
                                                 EXTRA_QUEUED_CALLS)
        # Killed worker processes can produce spurious "broken pipe"
        # tracebacks in the queue's own worker thread. But we detect killed
        # processes anyway, so silence the tracebacks.
        self._call_queue._ignore_epipe = True
        self._result_queue = SimpleQueue()
        self._work_ids = queue.Queue()
        self._queue_management_thread = None
        # Map of pids to processes
        self._processes = {}

        # Shutdown is a two-step process.
        self._shutdown_thread = False
        self._shutdown_lock = threading.Lock()
        self._broken = False
        self._queue_count = 0
        self._pending_work_items = {}

    def _start_queue_management_thread(self):
        # When the executor gets lost, the weakref callback will wake up
        # the queue management thread.
        def weakref_cb(_, q=self._result_queue):
            q.put(None)
        if self._queue_management_thread is None:
            # Start the processes so that their sentinels are known.
            self._adjust_process_count()
            self._queue_management_thread = threading.Thread(
                    target=_queue_management_worker,
                    args=(weakref.ref(self, weakref_cb),
                          self._processes,
                          self._pending_work_items,
                          self._work_ids,
                          self._call_queue,
                          self._result_queue))
            self._queue_management_thread.daemon = True
            self._queue_management_thread.start()
            _threads_queues[self._queue_management_thread] = self._result_queue

    def _adjust_process_count(self):
        for _ in range(len(self._processes), self._max_workers):
            p = multiprocessing.Process(
                    target=_process_worker,
                    args=(self._call_queue,
                          self._result_queue))
            p.start()
            self._processes[p.pid] = p

    def submit(self, fn, *args, **kwargs):
        with self._shutdown_lock:
            if self._broken:
                raise BrokenProcessPool('A child process terminated '
                    'abruptly, the process pool is not usable anymore')
            if self._shutdown_thread:
                raise RuntimeError('cannot schedule new futures after shutdown')

            f = _base.Future()
            w = _WorkItem(f, fn, args, kwargs)

            self._pending_work_items[self._queue_count] = w
            self._work_ids.put(self._queue_count)
            self._queue_count += 1
            # Wake up queue management thread
            self._result_queue.put(None)

            self._start_queue_management_thread()
            return f
    submit.__doc__ = _base.Executor.submit.__doc__

    def shutdown(self, wait=True):
        with self._shutdown_lock:
            self._shutdown_thread = True
        if self._queue_management_thread:
            # Wake up queue management thread
            self._result_queue.put(None)
            if wait:
                self._queue_management_thread.join()
        # To reduce the risk of opening too many files, remove references to
        # objects that use file descriptors.
        self._queue_management_thread = None
        self._call_queue = None
#.........这里部分代码省略.........