Python multiprocessing 模块，set_start_method() 实例源码

def __init__(
        self,
        worker_class,
        concurrent_workers,
    ):
        self.logger = logger.logger.Logger(
            logger_name='Supervisor',
        )

        self.worker_class = worker_class
        self.concurrent_workers = concurrent_workers

        self.task = self.worker_class()

        self.workers_processes = []

        self.should_work_event = threading.Event()
        self.should_work_event.set()

        multiprocessing.set_start_method(
            method='spawn',
            force=True,
        )

def test_set_get(self):
        multiprocessing.set_forkserver_preload(PRELOAD)
        count = 0
        old_method = multiprocessing.get_start_method()
        try:
            for method in ('fork', 'spawn', 'forkserver'):
                try:
                    multiprocessing.set_start_method(method, force=True)
                except ValueError:
                    continue
                self.assertEqual(multiprocessing.get_start_method(), method)
                ctx = multiprocessing.get_context()
                self.assertEqual(ctx.get_start_method(), method)
                self.assertTrue(type(ctx).__name__.lower().startswith(method))
                self.assertTrue(
                    ctx.Process.__name__.lower().startswith(method))
                self.check_context(multiprocessing)
                count += 1
        finally:
            multiprocessing.set_start_method(old_method, force=True)
        self.assertGreaterEqual(count, 1)

def match_game(self):
        multiprocessing.set_start_method('spawn')
        # TODO restart runners for running games?
        logger.info('Matchmaker started')
        while True:
            wait = gen.sleep(5)
            starttime = ioloop.IOLoop.current().time()
            players = yield gamequeue.find().sort([('_id', 1)]).limit(10).to_list(length=10)
            while len(players) >= 2:
                random.shuffle(players)
                p0, p1, players = players[0], players[1], players[2:]
                p0['token'], p1['token'] = create_token(), create_token()
                queue_ids = [p0.pop('_id'), p1.pop('_id')]
                game = {'player0': p0, 'player1': p1, 'turn': 0, 'status': 'new'}
                insert_result = yield games.insert_one(game)
                game_idstr = str(insert_result.inserted_id)
                runner_name = 'runner-%s' % game_idstr
                logger.info('Launching Process "%s"', runner_name)
                p = multiprocessing.Process(target=Runner.start_game, args=(game_idstr,), name=runner_name, daemon=True)
                p.start()
                # TODO keep track of spawned runner processes
                yield gamequeue.delete_many({'_id': {'$in': queue_ids}})
            endtime = ioloop.IOLoop.current().time()
            logger.debug('MatchMaker ran for %.3fms', 1000 * (endtime - starttime))
            yield wait

def start(self, slave_addr, task):
        self._task = task

        def _start(id, slave_addr, task):
            from multiprocessing import Process
            import multiprocessing
            #multiprocessing.set_start_method('spawn')
            Process(target=_worker_main, args=(id, slave_addr, task)).start()

        from concurrent.futures import ProcessPoolExecutor
        print("[Worker {0}] Create".format(self.id))
        _start(self.id, slave_addr, task)
        #executor = ProcessPoolExecutor()
        #loop = asyncio.get_event_loop()
        #asyncio.ensure_future(loop.run_in_executor(ProcessPoolExecutor(), _worker_main, self.id, slave_addr, task))
        #asyncio.ensure_future(_start(self.id, slave_addr, task))
        #yield from asyncio.sleep(10)
        print("***")

def test_set_get(self):
        multiprocessing.set_forkserver_preload(PRELOAD)
        count = 0
        old_method = multiprocessing.get_start_method()
        try:
            for method in ('fork', 'spawn', 'forkserver'):
                try:
                    multiprocessing.set_start_method(method, force=True)
                except ValueError:
                    continue
                self.assertEqual(multiprocessing.get_start_method(), method)
                ctx = multiprocessing.get_context()
                self.assertEqual(ctx.get_start_method(), method)
                self.assertTrue(type(ctx).__name__.lower().startswith(method))
                self.assertTrue(
                    ctx.Process.__name__.lower().startswith(method))
                self.check_context(multiprocessing)
                count += 1
        finally:
            multiprocessing.set_start_method(old_method, force=True)
        self.assertGreaterEqual(count, 1)

def __init__(self):
            # multiprocessing.set_start_method('spawn')
            self.layers = {}
            self.stop_event = multiprocessing.get_context('spawn').Event()
            self.input_prompts = multiprocessing.get_context('spawn').Queue()
            self.show_monitor = False

def run():

    # ask user for difficulty
    q_app = QtWidgets.QApplication([])
    q_widget = QtWidgets.QWidget()
    dialog = QtWidgets.QMessageBox(q_widget)
    dialog.addButton('Easy', QtWidgets.QMessageBox.ActionRole)
    dialog.addButton('Medium', QtWidgets.QMessageBox.ActionRole)
    dialog.addButton('Hard', QtWidgets.QMessageBox.ActionRole)
    dialog.addButton('Impossible', QtWidgets.QMessageBox.ActionRole)
    dialog.setText('Choose difficulty:')
    ret = dialog.exec_()

    easy, medium, hard, impossible = range(4)
    sim_time = None
    if ret == easy:
        sim_time = 1
    elif ret == medium:
        sim_time = 3
    elif ret == hard:
        sim_time = 5
    elif ret == impossible:
        sim_time = 8

    mp.set_start_method('spawn')
    gui_process = mp.Process(target=start_client.main)
    gui_process.start()

    run_game.main(BlackAgent='human', WhiteAgent='monte_carlo',
                  sim_time=sim_time, gui=True)

def main(args):
    import multiprocessing

    try:
        multiprocessing.set_start_method("fork")
    except RuntimeError as ex:
        log.warn("multiprocessing.set_start_method: " + str(ex))

    if not args.command:
        # No command was given.
        args.app.arg_parser.print_help()
        return 1

    # In the case fileConfig undid the command line, which has precedence.
    args.applyLoggingOpts(args.log_levels, args.log_files)

    if args.db_url:
        args.config.set(MAIN_SECT, SA_KEY, args.db_url)
        # Don't want commands and such to use this, so reset.
        args.db_url = None
    elif "MISHMASH_DBURL" in os.environ:
        log.verbose("Using environment MISHMASH_DBURL over configuration: {}"
                    .format(os.environ["MISHMASH_DBURL"]))
        args.config.set(MAIN_SECT, SA_KEY, os.environ["MISHMASH_DBURL"])

    try:
        # Run command
        retval = args.command_func(args, args.config) or 0
    except (KeyboardInterrupt, PromptExit) as ex:
        # PromptExit raised when CTRL+D during prompt, or prompts disabled
        retval = 0
    except (sql_exceptions.ArgumentError,
            sql_exceptions.OperationalError) as ex:
        _pErr("Database error")
        retval = 1
    except Exception as ex:
        log.exception(ex)
        _pErr("General error")
        retval = 2

    return retval

def test_context(self):
        for method in ('fork', 'spawn', 'forkserver'):
            try:
                ctx = multiprocessing.get_context(method)
            except ValueError:
                continue
            self.assertEqual(ctx.get_start_method(), method)
            self.assertIs(ctx.get_context(), ctx)
            self.assertRaises(ValueError, ctx.set_start_method, 'spawn')
            self.assertRaises(ValueError, ctx.set_start_method, None)
            self.check_context(ctx)

def test_semaphore_tracker(self):
        import subprocess
        cmd = '''if 1:
            import multiprocessing as mp, time, os
            mp.set_start_method("spawn")
            lock1 = mp.Lock()
            lock2 = mp.Lock()
            os.write(%d, lock1._semlock.name.encode("ascii") + b"\\n")
            os.write(%d, lock2._semlock.name.encode("ascii") + b"\\n")
            time.sleep(10)
        '''
        r, w = os.pipe()
        p = subprocess.Popen([sys.executable,
                             '-c', cmd % (w, w)],
                             pass_fds=[w],
                             stderr=subprocess.PIPE)
        os.close(w)
        with open(r, 'rb', closefd=True) as f:
            name1 = f.readline().rstrip().decode('ascii')
            name2 = f.readline().rstrip().decode('ascii')
        _multiprocessing.sem_unlink(name1)
        p.terminate()
        p.wait()
        time.sleep(2.0)
        with self.assertRaises(OSError) as ctx:
            _multiprocessing.sem_unlink(name2)
        # docs say it should be ENOENT, but OSX seems to give EINVAL
        self.assertIn(ctx.exception.errno, (errno.ENOENT, errno.EINVAL))
        err = p.stderr.read().decode('utf-8')
        p.stderr.close()
        expected = 'semaphore_tracker: There appear to be 2 leaked semaphores'
        self.assertRegex(err, expected)
        self.assertRegex(err, 'semaphore_tracker: %r: \[Errno' % name1)

#
# Mixins
#

def run(args):
    # create dummy environment to be able to create model
    env = gym.make(args.environment)
    assert isinstance(env.observation_space, Box)
    assert isinstance(env.action_space, Discrete)
    print("Observation space:", env.observation_space)
    print("Action space:", env.action_space)

    # create main model
    model = create_model(env, args)
    model.summary()
    env.close()

    # force runner processes to use cpu
    os.environ["CUDA_VISIBLE_DEVICES"] = ""
    # for better compatibility with Theano and Tensorflow
    multiprocessing.set_start_method('spawn')

    # create shared buffer for sharing weights
    blob = pickle.dumps(model.get_weights(), pickle.HIGHEST_PROTOCOL)
    shared_buffer = Array('c', len(blob))
    shared_buffer.raw = blob

    # create fifos and threads for all runners
    fifos = []
    for i in range(args.num_runners):
        fifo = Queue(args.queue_length)
        fifos.append(fifo)
        process = Process(target=runner, args=(shared_buffer, fifo, args))
        process.start()

    # start trainer in main thread
    trainer(model, fifos, shared_buffer, args)

def test_context(self):
        for method in ('fork', 'spawn', 'forkserver'):
            try:
                ctx = multiprocessing.get_context(method)
            except ValueError:
                continue
            self.assertEqual(ctx.get_start_method(), method)
            self.assertIs(ctx.get_context(), ctx)
            self.assertRaises(ValueError, ctx.set_start_method, 'spawn')
            self.assertRaises(ValueError, ctx.set_start_method, None)
            self.check_context(ctx)

def test_semaphore_tracker(self):
        import subprocess
        cmd = '''if 1:
            import multiprocessing as mp, time, os
            mp.set_start_method("spawn")
            lock1 = mp.Lock()
            lock2 = mp.Lock()
            os.write(%d, lock1._semlock.name.encode("ascii") + b"\\n")
            os.write(%d, lock2._semlock.name.encode("ascii") + b"\\n")
            time.sleep(10)
        '''
        r, w = os.pipe()
        p = subprocess.Popen([sys.executable,
                             '-c', cmd % (w, w)],
                             pass_fds=[w],
                             stderr=subprocess.PIPE)
        os.close(w)
        with open(r, 'rb', closefd=True) as f:
            name1 = f.readline().rstrip().decode('ascii')
            name2 = f.readline().rstrip().decode('ascii')
        _multiprocessing.sem_unlink(name1)
        p.terminate()
        p.wait()
        time.sleep(2.0)
        with self.assertRaises(OSError) as ctx:
            _multiprocessing.sem_unlink(name2)
        # docs say it should be ENOENT, but OSX seems to give EINVAL
        self.assertIn(ctx.exception.errno, (errno.ENOENT, errno.EINVAL))
        err = p.stderr.read().decode('utf-8')
        p.stderr.close()
        expected = 'semaphore_tracker: There appear to be 2 leaked semaphores'
        self.assertRegex(err, expected)
        self.assertRegex(err, 'semaphore_tracker: %r: \[Errno' % name1)

#
# Mixins
#

def main():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    asmbot.log("ASM version {} booting".format(asmbot.__version__), tag="ASM LDR")

    asmbot.log("Loading config", tag="PAM LDR")
    with open("config.json", "r") as f:
        cts = f.read()
        tkd = json.loads(cts)

    asmbot.log("Launching ASM", tag="ASM LDR")

    mp.set_start_method("spawn")
    executor = ThreadPoolExecutor(max_workers=int(tkd["shard_count"]))
    processes = []

    for i in range(0, int(tkd["shard_count"])):
        args = {
            "token": tkd["token"],
            "shard_id": i,
            "shard_count": int(tkd["shard_count"]),
            "script": tkd["script"],
            "guild_blacklist": tkd["guild_blacklist"],
            "guild_exempt_list": tkd["guild_exempt_list"]
        }

        loop.create_task(launch_process(executor, asmbotlauncher.initialize_asmbot, **args))

        if i != int(tkd["shard_count"]) - 1:
            loop.run_until_complete(asyncio.sleep(10))

    asmbot.log("Running", tag="ASM LDR")

    try:
        loop.run_until_complete(asyncio.gather(*asyncio.Task.all_tasks(loop)))

    except KeyboardInterrupt:
        pass

    finally:
        asmbot.log("Shutting down", tag="ASM LDR")
        for process in processes:
            process.join()

    asmbot.log("Shutdown finalized", tag="ASM LDR")