TorchAgentParallelRunner

class pams.runners.TorchAgentParallelRunner(settings, prng=None, logger=None, simulator_class=<class 'pams.simulator.Simulator'>)

Multi Process Agent Parallel runner class for agents using PyTorch. This is experimental.

This runner is pams.runners.MultiProcessAgentParallelRunner configured for agents whose pams.agents.Agent.submit_orders() uses PyTorch, e.g., to predict the order price by a neural network model. The simulation results are identical to those of pams.runners.SequentialRunner with the same settings and the same seed as long as the agents are deterministic (see below).

PyTorch is an optional dependency of pams. It is imported when this runner is created, and an ImportError is raised if it is not installed.

In addition to pams.runners.MultiProcessAgentParallelRunner, this runner

• starts the worker processes by spawn by default (default_start_method), which can be changed by simulation.startMethod. fork is unsafe for PyTorch: CUDA cannot be initialized in a forked process, and only the forking thread is copied to a forked process, so a worker forked after PyTorch has started its thread pools on the main process can deadlock. Because of spawn, the code running this runner must be under if __name__ == "__main__": on every platform, including Linux.

• sets the number of threads of PyTorch on each worker process by torch.set_num_threads() to simulation.torchNumThreads. By default, PyTorch uses about as many threads as the CPU cores in every process, so numParallel workers would oversubscribe the CPUs. The default is the number of threads of PyTorch on the main process, torch.get_num_threads(), divided by the number of workers that can run tasks at the same time, i.e., the smaller of numParallel and the largest maxNormalOrders of the sessions (at least one).

• sets the sharing strategy of torch.multiprocessing to torch_sharing_strategy, i.e., file_system, on the main process when it is set up and on each worker process (see below).

Importing PyTorch registers the reductions of tensors to multiprocessing.reduction.ForkingPickler, with which concurrent.futures.ProcessPoolExecutor pickles the tasks. Therefore, the tensors held by agents, e.g., the parameters of their models, are moved to shared memory when they are sent to a worker process for the first time, and each task sends only their handles instead of their data. This also happens with pams.runners.MultiProcessAgentParallelRunner, but on Linux, the default sharing strategy, file_descriptor, duplicates a file descriptor for each tensor in each task, because every task pickles the tensors of all the agents. It is several times slower, and it fails, e.g., with “Too many open files”, once the tensors of the tasks in flight exceed the limit of open files (1024 by default on many Linux systems, which 50 agents with 6 tensors each already exceed with numParallel 4). The file_system strategy, the only one on Windows and macOS, sends the name of the shared memory instead. Still, every tensor costs opening the shared memory in every task, and the shared memory, i.e., /dev/shm on Linux, must be large enough for the tensors of all the agents, e.g., by --shm-size of Docker, whose default is 64 MB.

The worker processes see the same memory as the main process, so in-place modifications of the tensors in pams.agents.Agent.submit_orders(), e.g., by training a model, change the tensors on the main process, unlike other modifications of the agents on the worker processes, which are discarded. Keep the tensors read-only in pams.agents.Agent.submit_orders().

Subclasses can override _get_worker_initializer and _get_worker_initargs, e.g., to load a model once per worker process. In that case, the initializer should call _initialize_torch_worker with the arguments given by this class, too.

Note

For the results identical to pams.runners.SequentialRunner, agents must derive their randomness from their own pseudo random number generator prng, e.g., by torch.Generator().manual_seed(self.prng.randrange(2**32)), because only its state is synchronized from the worker processes. The global random number generator of PyTorch on each worker process is independent of the main process and of the other workers. Also, the results of some operations of PyTorch, e.g., matrix products, can differ in rounding with the number of threads and with the device, and such a difference can change the course of the simulation. To compare with pams.runners.SequentialRunner, run it after torch.set_num_threads(n) and set simulation.torchNumThreads to n.

Note

This runner does not choose the device. The worker processes see the same devices as the main process, and agents choose the device in pams.agents.Agent.submit_orders(). Every worker process using CUDA creates its own CUDA context, which takes hundreds of megabytes of the GPU memory. CUDA tensors held by agents would be sent by CUDA IPC, which is not supported on Windows, so keep the tensors held by agents on the CPU. Moving them to the GPU in pams.agents.Agent.submit_orders() copies them in every task, because each task gets new copies of the agents. To copy a model to the GPU only once per worker process, load it by the worker initializer into a module-level variable and use it in pams.agents.Agent.submit_orders().

default_start_method: Optional[str] = 'spawn'

start method of the worker processes used when simulation.startMethod is not set in the config. spawn is used because fork is unsafe for PyTorch.

Type:

Optional[str]

torch_num_threads: Optional[int]

number of threads of PyTorch on each worker process given by simulation.torchNumThreads. None means the default, i.e., the number of threads of PyTorch on the main process divided by the smaller of num_parallel and the largest max_normal_orders of the sessions (at least one).

Type:

Optional[int]

torch_sharing_strategy: Optional[str] = 'file_system'

sharing strategy of torch.multiprocessing set on the main process and on the worker processes. It must be one of torch.multiprocessing.get_all_sharing_strategies(). None keeps the current strategy.

Type:

Optional[str]