我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用multiprocessing.Pipe()。
def __init__(self, maxsize=0): if maxsize <= 0: maxsize = _multiprocessing.SemLock.SEM_VALUE_MAX self._maxsize = maxsize self._reader, self._writer = Pipe(duplex=False) self._rlock = Lock() self._opid = os.getpid() if sys.platform == 'win32': self._wlock = None else: self._wlock = Lock() self._sem = BoundedSemaphore(maxsize) self._after_fork() if sys.platform != 'win32': register_after_fork(self, Queue._after_fork)
def _construct_pipes(core_classes_map): """Creates all the pipes needed to connect the cores""" # Create the first pipe receiver, sender = Pipe(duplex=False) # The input pipe of the pipeline is the sender end (introduced the packages to the first core) input_pipe = sender for core_class in core_classes_map: # If no kwargs passed, initialize as empty object if Pipeline.KEY_KWARGS not in core_class: core_class[Pipeline.KEY_KWARGS] = {} # The input pipe of a core is the end that receives packages core_class[Pipeline.KEY_KWARGS]['pipe_in'] = receiver # Create the inter-core pipe receiver, sender = Pipe(duplex=False) # The output pipe of a core is the end that sends the result core_class[Pipeline.KEY_KWARGS]['pipe_out'] = sender # The output pipe of the pipeline is the receiver end of the last core (in order to receive its result) output_pipe = receiver return input_pipe, output_pipe
def __init__(self, env_fns): """ envs: list of gym environments to run in subprocesses """ self.closed = False nenvs = len(env_fns) self.remotes, self.work_remotes = zip(*[Pipe() for _ in range(nenvs)]) self.ps = [Process(target=worker, args=(work_remote, remote, CloudpickleWrapper(env_fn))) for (work_remote, remote, env_fn) in zip(self.work_remotes, self.remotes, env_fns)] for p in self.ps: p.daemon = True # if the main process crashes, we should not cause things to hang p.start() for remote in self.work_remotes: remote.close() self.remotes[0].send(('get_spaces', None)) self.action_space, self.observation_space = self.remotes[0].recv()
def main(): jobs = [] pipe_list = [] for i in range(5): # ????????(???????) recv_end , send_end = multiprocessing.Pipe(False) p = multiprocessing.Process(target=worker, args=(i, send_end)) jobs.append(p) pipe_list.append(recv_end) p.start() for proc in jobs: proc.join() result_list = [x.recv() for x in pipe_list] print(result_list) for x in pipe_list: x.close()
def __init__(self, env_m, worker_idx): # These are instantiated in the *parent* process # currently. Probably will want to change this. The parent # does need to obtain the relevant Spaces at some stage, but # that's doable. self.worker_idx = worker_idx self.env_m = env_m self.m = len(env_m) self.parent_conn, self.child_conn = multiprocessing.Pipe() self.joiner = multiprocessing.Process(target=self.run) self._clear_state() self.start() # Parent only! self.child_conn.close()
def test_job_wrapper_fills_pipe_with_exception_info(self): module_name = self.test_jobs_module.constants.THROW_AN_EXCEPTION_JOB_NAME config = {} to_job, to_self = Pipe() job_wrapper(to_self, module_name, config, MockStatusUpdater(), entry_point_group_name="hoplite.test_jobs") exec_info = to_job.recv() # Get to the bottom level of the exception information while ('type' not in exec_info) and (exec_info is not None): exec_info = exec_info.get('previous_exception', None) try: self.assertEqual(exec_info['type'], str(TypeError)) self.assertEqual(exec_info['message'], "THE SKY IS FALLING!!") except Exception, e: raise e finally: to_job.close() to_self.close()
def test_job_wrapper_fills_pipe_with_exception_info_bubble_up(self): module_name = self.test_jobs_module.constants.JOB_FAILED_EXCEPTION_JOB_NAME config = {} to_job, to_self = Pipe() job_wrapper(to_self, module_name, config, MockStatusUpdater(), entry_point_group_name="hoplite.test_jobs") exec_info = to_job.recv() exec_info = exec_info.get('previous_exception', None) try: self.assertEqual(exec_info['address'], "10.2.1.1") self.assertEqual(exec_info['uuid'], "5") self.assertIsInstance(exec_info['traceback'], types.TracebackType) # Get to the very bottom level of the exception information exec_info = exec_info.get('previous_exception', None) self.assertEqual(exec_info['message'], "Test Message") self.assertEqual(exec_info['type'], "Test Type String") self.assertEqual(exec_info['exception_object'], "pickled_string") except Exception, e: raise e finally: to_job.close() to_self.close()
def __init__(self, host='192.168.1.1'): self.host = host self.sequence = 1 self.timer = 0.2 self.com_watchdog_timer = threading.Timer(self.timer, self.commwdg) self.lock = threading.Lock() self.speed = 0.2 self.at(ardrone.at.config, 'general:navdata_demo', 'TRUE') self.at(ardrone.at.config, 'control:altitude_max', '20000') self.video_pipe, video_pipe_other = multiprocessing.Pipe() self.nav_pipe, nav_pipe_other = multiprocessing.Pipe() self.com_pipe, com_pipe_other = multiprocessing.Pipe() self.network_process = ardrone.network.ARDroneNetworkProcess(self.host, nav_pipe_other, video_pipe_other, com_pipe_other) self.network_process.start() self.ipc_thread = ardrone.network.IPCThread(self) self.ipc_thread.start() self.image = PIL.Image.new('RGB', (640, 360)) self.navdata = dict() self.time = 0
def main(): parent_conn, child_conn = Pipe() child = Process(target=work, args=(child_conn,)) for item in ( 42, 'some string', {'one': 1}, CustomClass(), None, ): print( "PRNT: send: {}".format(item) ) parent_conn.send(item) child.start() child.join()
def setup_workers(self): # work only once if self._initialized: return self._initialized = True self.model.cleargrads() for i in six.moves.range(1, len(self.gpus)): pipe, worker_end = multiprocessing.Pipe() worker = _Worker(i, worker_end, self.model, self.gpus, self.da, int(float(self.batch) / len(self.gpus) / self.train_batch_divide), self) worker.start() self._workers.append(worker) self._pipes.append(pipe) with cuda.Device(self.gpus[0]): self.model.to_gpu(self.gpus[0]) if len(self.gpus) > 1: communication_id = nccl.get_unique_id() self._send_message(("set comm_id", communication_id)) self.communication = nccl.NcclCommunicator(len(self.gpus), communication_id, 0)
def setup_workers(self): # work only once if self._initialized: return self._initialized = True self.model.zerograds() for i in six.moves.range(1, len(self.gpus)): pipe, worker_end = multiprocessing.Pipe() worker = _Worker(i, worker_end, self.model, self.gpus, self.da, int(self.batch / len(self.gpus) / self.train_batch_divide), self) worker.start() self._workers.append(worker) self._pipes.append(pipe) with cuda.Device(self.gpus[0]): self.model.to_gpu(self.gpus[0]) if len(self.gpus) > 1: communication_id = nccl.get_unique_id() self._send_message(("set comm_id", communication_id)) self.communication = nccl.NcclCommunicator(len(self.gpus), communication_id, 0)
def __init__(self): testIfName() self.stop = mp.Event() self.plot = False self.plotFunct = None self.plotHistory = 100000 self.samplerate = 0 self.nChannelsInData = 1 self.saveData = False self.saveDataFormat = "csv" self.saveDataFilename = "data" self.configDone = False self.inputToPlot_write_end, self.inputToPlot_read_end = mp.Pipe() self.inputToFile_write_end, self.inputToFile_read_end = mp.Pipe() self.output_write_end, self.output_read_end = mp.Pipe() self.processes = {} self.rdy = {} self.inputChannels = [] self.activeChannels = {}
def __call__(self, *args, **kargs): def function_process(pipe, function, args, kargs): pipe.send(function(*args, **kargs)) # A: result got from function process p_pipe, c_pipe = multiprocessing.Pipe() p = Process(target=function_process, args=(c_pipe, self.function, args, kargs)) p.start() p.join(self._timeout_threshold) # Wait if p.exception: # if there is other Error raise RuntimeError elif p.is_alive(): # if passes the timeout threshold, terminate function process p.terminate() raise TimeoutError('Timeout') else: return p_pipe.recv() # return result from A
def test_recursion(self): rconn, wconn = self.Pipe(duplex=False) self._test_recursion(wconn, []) time.sleep(DELTA) result = [] while rconn.poll(): result.append(rconn.recv()) expected = [ [], [0], [0, 0], [0, 1], [1], [1, 0], [1, 1] ] self.assertEqual(result, expected)
def test_spawn_close(self): # We test that a pipe connection can be closed by parent # process immediately after child is spawned. On Windows this # would have sometimes failed on old versions because # child_conn would be closed before the child got a chance to # duplicate it. conn, child_conn = self.Pipe() p = self.Process(target=self._echo, args=(child_conn,)) p.daemon = True p.start() child_conn.close() # this might complete before child initializes msg = latin('hello') conn.send_bytes(msg) self.assertEqual(conn.recv_bytes(), msg) conn.send_bytes(SENTINEL) conn.close() p.join()
def test_large_fd_transfer(self): # With fd > 256 (issue #11657) if self.TYPE != 'processes': self.skipTest("only makes sense with processes") conn, child_conn = self.Pipe(duplex=True) p = self.Process(target=self._writefd, args=(child_conn, b"bar", True)) p.daemon = True p.start() with open(test_support.TESTFN, "wb") as f: fd = f.fileno() for newfd in range(256, MAXFD): if not self._is_fd_assigned(newfd): break else: self.fail("could not find an unassigned large file descriptor") os.dup2(fd, newfd) try: reduction.send_handle(conn, newfd, p.pid) finally: os.close(newfd) p.join() with open(test_support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"bar")
def test_timeout(self): old_timeout = socket.getdefaulttimeout() try: socket.setdefaulttimeout(0.1) parent, child = multiprocessing.Pipe(duplex=True) l = multiprocessing.connection.Listener(family='AF_INET') p = multiprocessing.Process(target=self._test_timeout, args=(child, l.address)) p.start() child.close() self.assertEqual(parent.recv(), 123) parent.close() conn = l.accept() self.assertEqual(conn.recv(), 456) conn.close() l.close() p.join(10) finally: socket.setdefaulttimeout(old_timeout) # # Test what happens with no "if __name__ == '__main__'" #
def test_ignore(self): conn, child_conn = multiprocessing.Pipe() try: p = multiprocessing.Process(target=self._test_ignore, args=(child_conn,)) p.daemon = True p.start() child_conn.close() self.assertEqual(conn.recv(), 'ready') time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) time.sleep(0.1) conn.send(1234) self.assertEqual(conn.recv(), 1234) time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) self.assertEqual(conn.recv_bytes(), b'x'*(1024*1024)) time.sleep(0.1) p.join() finally: conn.close()
def test_ignore_listener(self): conn, child_conn = multiprocessing.Pipe() try: p = multiprocessing.Process(target=self._test_ignore_listener, args=(child_conn,)) p.daemon = True p.start() child_conn.close() address = conn.recv() time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) time.sleep(0.1) client = multiprocessing.connection.Client(address) self.assertEqual(client.recv(), 'welcome') p.join() finally: conn.close() # # #
def start_pollers(configs): """start_pollers starts a set of pollers for specified configurations. Args: configs: Configurations for the pollers. Returns: Array of poller info (process, comm pipe). """ pollers = [] for config in configs: parent_pipe, child_pipe = Pipe() p = Process(target=poll_scm, args=(config, child_pipe,)) pollers.append(PollerInfo( process=p, pipe=parent_pipe)) l.info('Starting poller for {}.'.format(config.name)) p.start() return pollers
def __init__(self, nenvs, map_name): """ envs: list of gym environments to run in subprocesses """ self.remotes, self.work_remotes = zip(*[Pipe() for _ in range(nenvs)]) self.ps = [] i = 0 for (work_remote, ) in zip(self.work_remotes, ): self.ps.append( Process(target=worker, args=(work_remote, map_name, i))) i += 1 # # self.ps = [Process(target=worker, args=(work_remote, (map_name))) # for (work_remote,) in zip(self.work_remotes,)] for p in self.ps: p.start() self.remotes[0].send(('get_spaces', 1)) self.action_space, self.observation_space = self.remotes[0].recv() #print("action_space: ", self.action_space, " observation_space: ", self.observation_space)
def _accept_new_connection(self, s): # accepting the connection clt_sock, clt_info = s.accept() # Getting the service ability new_abl = self.callback() # Giving to the service ability the informations about the client new_abl.set_opt(self.client_info_name, '{}:{}'.format(clt_info[0], clt_info[1])) # Creating the pipes in_pipe_in, in_pipe_out = multiprocessing.Pipe() out_pipe_in, out_pipe_out = multiprocessing.Pipe() new_abl.add_in_pipe(in_pipe_out) new_abl.add_out_pipe(out_pipe_in) # Starting the service ability new_abl.start() return clt_sock, in_pipe_in, out_pipe_out, new_abl
def _accept_new_connection(self, s): # accepting the connection clt_sock, clt_info = s.accept() # Getting the service ability new_abl = self.callback() # Giving to the service ability the information about the client if not isinstance(self.client_info_name, type(None)): new_abl.set_opt(self.client_info_name, '{}:{}'.format(clt_info[0], clt_info[1])) # Creating the pipes in_pipe_in, in_pipe_out = multiprocessing.Pipe() out_pipe_in, out_pipe_out = multiprocessing.Pipe() new_abl.add_in_pipe(in_pipe_out) new_abl.add_out_pipe(out_pipe_in) # Starting the service ability new_abl.start() return clt_sock, in_pipe_in, out_pipe_out, new_abl
def test_basic_usage(self): """""" pipp, pipc = Pipe() pips, pipr = Pipe() self.print_bar() print('Test Task Interface') ret_process = ProcessTask(id='test-1', target=testfun, args=(5,), status_monitor_pipe=pipc, result_pipe=pips, result_hook_function=result_callback) ret_process.start() print('Test get threads status') time.sleep(1) #print(ret_process.subthreads_count) threads_status = pipp.recv() self.assertIsInstance(threads_status, dict) #print pipr.recv() #print pipr.recv() #print pipr.recv() #print pipr.recv() self.print_end_bar()
def runpool(): parsecli() try: httpd = ThreadedTCPServer((LISTEN, PORT), Proxy) except OSError as e: print(e) return mainsock = httpd.socket if hasattr(socket, "fromshare"): workers = MAX_WORKERS for i in range(workers-1): (pipeout, pipein) = multiprocessing.Pipe() p = multiprocessing.Process(target=start_worker, args=(pipeout,)) p.daemon = True p.start() while p.pid == None: time.sleep(1) pipein.send(mainsock.share(p.pid)) serve_forever(httpd)
def test_thousand(self): if self.TYPE == 'manager': self.skipTest('test not appropriate for {}'.format(self.TYPE)) passes = 1000 lock = self.Lock() conn, child_conn = self.Pipe(False) for j in range(self.N): p = self.Process(target=self._test_thousand_f, args=(self.barrier, passes, child_conn, lock)) p.start() for i in range(passes): for j in range(self.N): self.assertEqual(conn.recv(), i) # # #
def test_fd_transfer(self): if self.TYPE != 'processes': self.skipTest("only makes sense with processes") conn, child_conn = self.Pipe(duplex=True) p = self.Process(target=self._writefd, args=(child_conn, b"foo")) p.daemon = True p.start() self.addCleanup(test.support.unlink, test.support.TESTFN) with open(test.support.TESTFN, "wb") as f: fd = f.fileno() if msvcrt: fd = msvcrt.get_osfhandle(fd) reduction.send_handle(conn, fd, p.pid) p.join() with open(test.support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"foo")
def test_dont_merge(self): a, b = self.Pipe() self.assertEqual(a.poll(0.0), False) self.assertEqual(a.poll(0.1), False) p = self.Process(target=self._child_dont_merge, args=(b,)) p.start() self.assertEqual(a.recv_bytes(), b'a') self.assertEqual(a.poll(1.0), True) self.assertEqual(a.poll(1.0), True) self.assertEqual(a.recv_bytes(), b'b') self.assertEqual(a.poll(1.0), True) self.assertEqual(a.poll(1.0), True) self.assertEqual(a.poll(0.0), True) self.assertEqual(a.recv_bytes(), b'cd') p.join() # # Test of sending connection and socket objects between processes #
def test_wait_timeout(self): from multiprocessing.connection import wait expected = 5 a, b = multiprocessing.Pipe() start = time.time() res = wait([a, b], expected) delta = time.time() - start self.assertEqual(res, []) self.assertLess(delta, expected * 2) self.assertGreater(delta, expected * 0.5) b.send(None) start = time.time() res = wait([a, b], 20) delta = time.time() - start self.assertEqual(res, [a]) self.assertLess(delta, 0.4)
def test_ignore_listener(self): conn, child_conn = multiprocessing.Pipe() try: p = multiprocessing.Process(target=self._test_ignore_listener, args=(child_conn,)) p.daemon = True p.start() child_conn.close() address = conn.recv() time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) time.sleep(0.1) client = multiprocessing.connection.Client(address) self.assertEqual(client.recv(), 'welcome') p.join() finally: conn.close()
def listen_and_send(self, send_data): '''Listen on socket and send data in response. :param bytes send_data: data to send ''' self.socket_pipe, child_pipe = multiprocessing.Pipe() self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.socket.bind(os.path.join(self.socket_dir, 'sock')) self.socket.listen(1) def worker(sock, pipe, send_data_): conn, addr = sock.accept() pipe.send(conn.makefile('rb').read()) conn.sendall(send_data_) conn.close() self.proc = multiprocessing.Process(target=worker, args=(self.socket, child_pipe, send_data)) self.proc.start() self.socket.close()
def newproc(self): global plock self.timer_update() self.pq, self.cq = Queue(1), Queue(1) # two queue needed # self.pc, self.cc = Pipe() self.p = Process( target = standalone_headless_isolated, args=(self.pq, self.cq, plock) ) self.p.daemon = True self.p.start() self.reset_count = 0 # how many times has this instance been reset() ed self.step_count = 0 self.timer_update() return # send x to the process
def blocking_run(self): parent_conn, child_conn = Pipe() q = Queue() q.put(self.parameters) self.p = Process(target=job_process, args=(self.job_id, self.job_class, q, child_conn, self.server_url, self.log_filename, )) self.p.start() while self.p.is_alive(): while parent_conn.poll(): self.output_recieved_from_job(parent_conn.recv()) time.sleep(1) self.p.join() while parent_conn.poll(): self.output_recieved_from_job(parent_conn.recv()) if self.terminated: self.result = {'job_id':self.job_id, 'success':False, 'retcode':1, 'exception':'Terminated by server', 'progress':'terminated'} else: self.result = q.get() self.result['progress'] = self.status parent_conn.close() return self.result
def __init__(self, conn, keepAlive=60): self.conn = conn self.connLock = threading.Lock() self.requests = multiprocessing.Pipe(False) self.keepAlive = keepAlive self.resTable = {} self.resTableLock = threading.Lock() self.lastTaskId = 0 self.workers = [] self.lastRead = time.time() self._stopping = False self._stoppingLock = threading.Lock() self.startWorker(self.sendWorker) self.startWorker(self.recvWorker) self.startWorker(self.mainWorker) atexit.register(self.cancel)
def _pipe(self): """On Windows we use a pipe to emulate a Linux style character buffer.""" if NIX: return None if not self.__pipe: target_function = self._get_target_function() if not target_function: return None self.__pipe, child_conn = Pipe(duplex=False) self._listener = Process(target=target_function, args=(child_conn,)) self._listener.start() return self.__pipe
