我正在执行下面的代码，它工作正常，但它不会生成到不同的进程，而是有时所有代码都在同一个进程中运行，有时是两个进程合二为一。我使用的是 4 个 cpu 的机器。这段代码有什么问题吗？
def f(values):
print(multiprocessing.current_process())
return values
def main():
p = Pool(4) #number of processes = number of CPUs
keys, values= zip(*data.items()) #ordered keys and values
processed_values= p.map( f, values )
result= dict( zip(keys, processed_values ) )
p.close() # no more tasks
p.join() # wrap up current tasks
结果是
<SpawnProcess(SpawnPoolWorker-1, started daemon)>
<SpawnProcess(SpawnPoolWorker-1, started daemon)>
<SpawnProcess(SpawnPoolWorker-1, started daemon)>
<SpawnProcess(SpawnPoolWorker-1, started daemon)>
而有时也像这样，
<SpawnProcess(SpawnPoolWorker-3, started daemon)>
<SpawnProcess(SpawnPoolWorker-2, started daemon)>
<SpawnProcess(SpawnPoolWorker-1, started daemon)>
<SpawnProcess(SpawnPoolWorker-3, started daemon)>
有时，
<SpawnProcess(SpawnPoolWorker-1, started …我正在尝试使用Python的多处理模块来生成一个服务器来接收UDP消息，稍微修改它们，然后将它们传递给以subprocess模块启动的grep进程。由于 Popen 子进程的标准输入接受文件描述符，这就是我想要传递的内容。
我遇到的问题是获取一个与服务器进程通信的文件描述符，我可以将其传递给 grep 子进程。我过去曾使用普通的 os.fork() 和 os.pipe() 完成此操作，但现在我想使用带有 spawn start 方法的多重处理。我尝试从 os.pipe 获取写入描述符，使其可继承，并通过 multiprocess.Process 将其作为参数传递给新进程。当我尝试使用 os.fdopen(fd, 'wb') 在另一个进程中打开它时，我收到错误文件描述符的 OSError 。这是我测试过的代码片段。
def _listen_syslog(ip_address, 端口, write_pipe):
f = os.fdopen(write_pipe, 'wb')
#做一些事情，比如写入文件
def Listen_syslog(ip_address, 端口):
r, w = os.pipe()
os.set_inheritable(w, True)
proc = mp.Process(target=_listen_syslog, args=(ip_address, port, w))
proc.start()
#这个进程不需要写，所以关闭它
os.close(w)
#这是我想传递给 grep 子进程 stdin 的描述符
#在使用 os.fork() 之前，类似的情况已经发生过
返回r
最后，如果无法使用通过 os.pipe() 创建的管道来执行此操作，我可以使用 multiprocessing.Pipe() 并使用连接对象 fileno() 函数中的文件描述符直接使用吗？更重要的是，只要我不将连接对象用于其他任何用途，这样做就安全吗？
Can anyone tell me why the following code gives the error "cannot start a process twice"? By my reckoning, p1 and p2 should already have been force-closed by the p.terminate() command
EDIT: Have added in some more code to give context - wanted to come up with a simple example but left out the while loops
import time
import os
from multiprocessing import Process
import datetime
def a():
print ("a starting")
time.sleep(30)
print ("a ending")
def b():
print ("b starting") …当初始化程序抛出如下错误时，脚本将不会停止。
我想在开始主进程之前中止（不要运行“do_something”）。
from multiprocessing import Pool
import contextlib
def initializer():
raise Exception("init failed")
def do_something(args):
# main process
pass
pool = Pool(1, initializer=initializer)
with contextlib.closing(pool):
try:
pool.map_async(do_something, [1]).get(100)
except:
pool.terminate()
控制台上永不停止的堆栈跟踪如下
...
Exception: init failed
Process ForkPoolWorker-18:
Traceback (most recent call last):
File "/home/hoge/anaconda3/lib/python3.6/multiprocessing/process.py", line 249, in _bootstrap
self.run()
File "/home/hoge/anaconda3/lib/python3.6/multiprocessing/process.py", line 93, in run
self._target(*self._args, **self._kwargs)
File "/home/hoge/anaconda3/lib/python3.6/multiprocessing/pool.py", line 103, in worker
initializer(*initargs)
File "hoge.py", line 5, in initializer
raise Exception("init failed")
Exception: init failed
...
我的解决方法是通过使用如下所示的全局标志来抑制初始化程序错误并在主进程开始时返回。
但我想学更好的。
def …我正在努力使用 Cython 正确并行化函数。基本上，问题是对一些数据进行分类。实际代码有点长，但最终它做了这样的事情：
def bin_var(double[:] dist,
double[:] values,
double[:] bin_def,
double[:] varg, long[:] count):
dbin = (bin_def[1] - bin_def[0]) / bin_def[2]
for n1 in range(values.size):
if (dist[n1] < bin_def[0]) or (dist[n1] >= bin_def[1]):
continue
else:
ni = int((dist - bin_def[0]) / dbin)
count[ni] += 1
varg[ni] += calc_something(values[ni])
# Compute the mean
for n1 in range(int(bin_def[2])):
varg[ni] /= count[ni]
该代码适合一些简单的并行化（values并且dist非常大）：需要将第一个for循环拆分为单独的进程，每个进程都处理自己版本的count和varg数组。完成后，必须通过对第二个for循环（更短）之前的count和的不同版本求和来将所有内容组合在一起。varg
也就是说，我花了两天时间试图了解如何在 Cython 中有效地实现这一点，并且我开始怀疑当前版本的语言是不可能的。请注意，仅在第一个循环中使用prangefrom …
multithreading multiprocessing cython python-multiprocessing
I'm trying to parallelize a script, but for an unknown reason the kernel just freeze without any errors thrown.
minimal working example:
from multiprocessing import Pool
def f(x):
return x*x
p = Pool(6)
print(p.map(f, range(10)))
Interestingly, all works fine if I define my function in another file then import it. How can I make it work without the need of another file?
I work with spyder (anaconda) and I have the same result if I run my code from the …
python windows parallel-processing multiprocessing python-multiprocessing
我写了一个python代码来使用多处理计算数字的平方。它显示了下面的错误。如何修复它。
import os
from multiprocessing import Process, current_process
def sqaure(number):
result=number*number
process_id=os.getpid()
print(f"Process ID : {process_id}")
print(f"the number {number} sqaures to {result}")
if __name__=='__main__':
numbers=[1,2,3,4]
processes=[]
for number in numbers:
process=Process(target=sqaure, args=(number,))
processes.append(process)
process.start()
错误：
Traceback (most recent call last):
File "C:/Users/Rohan/PycharmProjects/untitled1/hyu.py", line 2, in <module>
from multiprocessing import Process, current_process
ImportError: cannot import name 'Process' from 'multiprocessing' (C:\Users\Rohan\PycharmProjects\untitled1\multiprocessing.py)
我正在尝试使用multiprocessing.Pool.下载和提取 zip 文件。但是每次执行脚本时，只会下载3 个zip ，并且目录中看不到剩余的文件（CPU % 也达到 100%）。有人可以帮助我如何解决这个问题/建议更好的方法并遵循我尝试过的代码段。我对多处理完全陌生。我的目标是在不达到最大 CPU 的情况下并行下载多个文件。
import StringIO
import os
import sys
import zipfile
from multiprocessing import Pool, cpu_count
import requests
filePath = os.path.dirname(os.path.abspath(__file__))
print("filePath is %s " % filePath)
sys.path.append(filePath)
url = ["http://mlg.ucd.ie/files/datasets/multiview_data_20130124.zip",
"http://mlg.ucd.ie/files/datasets/movielists_20130821.zip",
"http://mlg.ucd.ie/files/datasets/bbcsport.zip",
"http://mlg.ucd.ie/files/datasets/movielists_20130821.zip",
"http://mlg.ucd.ie/files/datasets/3sources.zip"]
def download_zips(url):
file_name = url.split("/")[-1]
response = requests.get(url)
sourceZip = zipfile.ZipFile(StringIO.StringIO(response.content))
print("\n Downloaded {} ".format(file_name))
sourceZip.extractall(filePath)
print("extracted {} \n".format(file_name))
sourceZip.close()
if __name__ == "__main__":
print("There are {} CPUs on this machine ".format(cpu_count()))
pool = …嗨，我正在尝试在 python 中调试多进程。下面是我使用 Pool 运行多进程的一部分
pool = Pool(num_half_logical_cpus)
pool_result_dict = pool.starmap(process_batches, lstListSets)
但是，我无法达到任何断点。谁能指导我点击我设置的断点？
谢谢！
我有一个功能：
import time
def all_40k():
for _ in range(400000):
print('validate')
print('parsing')
print('inserting')
if __name__ == '__main__':
start_time = time.time()
all_40k()
print(f'used time:{time.time()-start_time}')
输出是：
used time:9.545064210891724
因为这个相同的函数重复了 40k 次，所以我希望有 4 个并行函数同时运行，每个函数运行 10k，理想情况下这会快 4 倍。
所以我首先尝试了多线程：
import threading
import time
def first_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def second_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def third_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def forth_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
thread1 = threading.Thread(target=first_10k)
thread2 = threading.Thread(target=second_10k) …python multithreading python-multithreading python-asyncio python-multiprocessing