blocks|key|281095|text|Bash脚本而不是Python，但我经常使用它来进行简单的并行处理：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|281096|#!/usr/bin/env+bash
waitForNProcs()
{
+nprocs=$(pgrep+-f+$procName+%7C+wc+-l)
+while+[+$nprocs+-gt+$MAXPROCS+];+do
++sleep+$SLEEPTIME
++nprocs=$(pgrep+-f+$procName+%7C+wc+-l)
+done
}
SLEEPTIME=3
MAXPROCS=10
procName=myPython.py
for+file+in+./data/*.txt;+do
+waitForNProcs
+./$procName+$file+&
done|code-block|syntax|javascript|281097|或者对于非常简单的情况，另一个选项是xargs，其中P设置procs的数量|281098|find+./data/+%7C+grep+txt+%7C+xargs+-P10+-I+SUB+./myPython.py+SUB+|281099|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|N|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|O|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

Bash script rather than Python, but I use it often for simple parallel processing:

<pre><code>#!/usr/bin/env bash
waitForNProcs()
{
 nprocs=$(pgrep -f $procName | wc -l)
 while [ $nprocs -gt $MAXPROCS ]; do
 sleep $SLEEPTIME
 nprocs=$(pgrep -f $procName | wc -l)
 done
}
SLEEPTIME=3
MAXPROCS=10
procName=myPython.py
for file in ./data/*.txt; do
 waitForNProcs
 ./$procName $file &amp;
done
</code></pre>

Or for very simple cases, another option is xargs where P sets the number of procs

<pre><code>find ./data/ | grep txt | xargs -P10 -I SUB ./myPython.py SUB 
</code></pre>

blocks|key|497939|text|您应该使用流程主管。一种方法是使用Circus提供的应用程序接口来“编程”，文档站点现在离线了，但我认为这只是一个暂时的问题，无论如何，你可以使用马戏团来处理这个问题。另一种方法是使用supervisord并将进程的参数numprocs设置为您拥有的内核数量。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|497940|使用Circus的示例：|497941|from+circus+import+get_arbiter

arbiter+=+get_arbiter("myprogram",+numprocesses=3)
try:
++++arbiter.start()
finally:
++++arbiter.stop()|code-block|syntax|javascript|497942|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/mozilla-services/circus/|1|http://supervisord.org/^0|32|8|H|6|0|2K|B|1|0|0|0^^$0|@$1|2|3|4|5|6|7|W|8|@$9|X|A|Y|B|C]]|D|@$9|Z|A|10|1|11]|$9|12|A|13|1|14]]|E|$]]|$1|F|3|G|5|6|7|15|8|@]|D|@]|E|$]]|$1|H|3|I|5|J|7|16|8|@]|D|@]|E|$K|L]]|$1|M|3|-4|5|6|7|17|8|@]|D|@]|E|$]]]|N|$O|$5|P|Q|R|E|$S|T]]|U|$5|P|Q|R|E|$S|V]]]]

You should use a process supervisor. One approach would be using the API provided by <a href="https://github.com/mozilla-services/circus/" rel="nofollow">Circus</a> to do that "programatically", the documentation site is now offline but I think its just a temporary problem, anyway, you can use the Circus to handle this. Another approach would be using the <a href="http://supervisord.org" rel="nofollow">supervisord</a> and setting the parameter <code>numprocs</code> of the process to the number of cores you have.

An example using Circus:

<pre><code>from circus import get_arbiter

arbiter = get_arbiter("myprogram", numprocesses=3)
try:
 arbiter.start()
finally:
 arbiter.stop()
</code></pre>

blocks|key|281215|text|虽然有很多关于使用multiprocessing.pool的答案，但关于如何使用multiprocessing.Process的代码片段并不多，当内存使用很重要时，这确实更有好处。启动1000个进程将使CPU过载并杀死内存。如果每个进程及其数据管道都是内存密集型的，则操作系统或Python本身将限制并行进程的数量。我开发了下面的代码来限制同时批量提交给CPU的作业数量。批处理大小可以与CPU核心的数量成比例地缩放。在我的windows电脑上，每批作业的效率最高可达CPU的4倍。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|281216|import+multiprocessing
def+func_to_be_multiprocessed(q,data):
++++q.put(('s'))
q+=+multiprocessing.Queue()
worker+=+[]
for+p+in+range(number_of_jobs):
++++worker[p].append(multiprocessing.Process(target=func_to_be_multiprocessed,+\
++++++++args=(q,data)...))
num_cores+=+multiprocessing.cpu_count()
Scaling_factor_batch_jobs+=+3.0
num_jobs_per_batch+=+num_cores+*+Scaling_factor_batch_jobs
num_of_batches+=+number_of_jobs+//+num_jobs_per_batch
for+i_batch+in+range(num_of_batches):
++++floor_job+=+i_batch+*+num_jobs_per_batch
++++ceil_job++=+floor_job+%2B+num_jobs_per_batch
++++for+p+in+worker[floor_job+:+ceil_job]:
+++++++++++++++++++++++++++++++++++++++++worker.start()
++++for+p+in+worker[floor_job+:+ceil_job]:
+++++++++++++++++++++++++++++++++++++++++worker.join()
for+p+in+worker[ceil_job+:]:
+++++++++++++++++++++++++++worker.start()
for+p+in+worker[ceil_job+:]:
+++++++++++++++++++++++++++worker.join()
for+p+in+multiprocessing.active_children():
+++++++++++++++++++++++++++p.terminate()
result+=+[]
for+p+in+worker:
+++result.append(q.get())|code-block|syntax|javascript|281217|唯一的问题是，如果任何批处理中的任何作业无法完成并导致挂起情况，则批处理的其余作业将不会启动。因此，要处理的函数必须具有适当的错误处理例程。|281218|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

While there are many answers about using multiprocessing.pool, there are not many code snippets on how to use multiprocessing.Process, which is indeed more beneficial when memory usage matters. starting 1000 processes will overload the CPU and kill the memory. If each process and its data pipelines are memory intensive, OS or Python itself will limit the number of parallel processes. I developed the below code to limit the simultaneous number of jobs submitted to the CPU in batches. The batch size can be scaled proportional to the number of CPU cores. In my windows PC, the number of jobs per batch can be efficient upto 4 times the CPU coures available.

<pre><code>import multiprocessing
def func_to_be_multiprocessed(q,data):
 q.put(('s'))
q = multiprocessing.Queue()
worker = []
for p in range(number_of_jobs):
 worker[p].append(multiprocessing.Process(target=func_to_be_multiprocessed, \
 args=(q,data)...))
num_cores = multiprocessing.cpu_count()
Scaling_factor_batch_jobs = 3.0
num_jobs_per_batch = num_cores * Scaling_factor_batch_jobs
num_of_batches = number_of_jobs // num_jobs_per_batch
for i_batch in range(num_of_batches):
 floor_job = i_batch * num_jobs_per_batch
 ceil_job = floor_job + num_jobs_per_batch
 for p in worker[floor_job : ceil_job]:
 worker.start()
 for p in worker[floor_job : ceil_job]:
 worker.join()
for p in worker[ceil_job :]:
 worker.start()
for p in worker[ceil_job :]:
 worker.join()
for p in multiprocessing.active_children():
 p.terminate()
result = []
for p in worker:
 result.append(q.get())
</code></pre>

The only problem is, if any of the job in any batch could not complete and leads to a hanging situation, rest of the batches of jobs will not be initiated. So, the function to be processed must have proper error handling routines.

I'd like to run multiple instances of program.py simultaneously, while limiting the number of instances running at the same time (e.g. to the number of CPU cores available on my system). For example, if I have 10 cores and have to do 1000 runs of program.py in total, only 10 instances will be created and running at any given time. 

I've tried using the multiprocessing module, multithreading, and using queues, but there's nothing that seemed to me to lend itself to an easy implementation. The biggest problem I have is finding a way to limit the number of processes running simultaneously. This is important because if I create 1000 processes at once, it becomes equivalent to a fork bomb. I don't need the results returned from the processes programmatically (they output to disk), and the processes all run independently of each other. 

Can anyone please give me suggestions or an example of how I could implement this in python, or even bash? I'd post the code I've written so far using queues, but it doesn't work as intended and might already be down the wrong path. 

Many thanks.

Multiprocessing in Python while limiting the number of running processes

数据管道

Python

我希望同时运行多个program.py实例，同时限制同时运行的实例数量(例如，限制系统上可用的CPU核心数量)。例如，如果我有10个内核，总共需要运行1000次program.py，那么在任何给定时间都只会创建并运行10个实例。我尝试过使用多处理模块、多线程和队列，但对我来说，似乎没有任何一种方法适合于简单的实现。我遇...

问Python中的多进程，同时限制正在运行的进程的数量
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Python中的多进程，同时限制正在运行的进程的数量EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Python中的多进程，同时限制正在运行的进程的数量
EN