Python multiprocessing - Debugging OSError: [Errno 12] Cannot allocate memory
Python multiprocessing - Debugging OSError: [Errno 12] Cannot allocate memory
我面临以下问题。我正在尝试并行化更新文件的函数，但由于 OSError: [Errno 12] Cannot allocate memory，我无法启动 Pool()。我已经开始在服务器上四处查看，这不像我使用的是旧的、弱的 one/out 实际内存。
参见 htop：
此外，free -m 显示除了 ~7GB 的交换内存外，我还有足够的可用内存：
我尝试处理的文件也不是那么大。我将在下面粘贴我的代码（和堆栈跟踪），大小如下：
所用的predictionmatrix数据帧占用了大约。 80MB 根据 pandasdataframe.memory_usage()
文件 geo.geojson 是 2MB
我该如何调试呢？我可以检查什么以及如何检查？感谢任何 tips/tricks!
代码：
def parallelUpdateJSON(paramMatch, predictionmatrix, data):
for feature in data['features']:
currentfeature = predictionmatrix[(predictionmatrix['SId']==feature['properties']['cellId']) & paramMatch]
if (len(currentfeature) > 0):
feature['properties'].update({"style": {"opacity": currentfeature.AllActivity.item()}})
else:
feature['properties'].update({"style": {"opacity": 0}})
def writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix):
with open('geo.geojson') as f:
data = json.load(f)
paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
pool = Pool()
func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
pool.map(func, data)
pool.close()
pool.join()
with open('output.geojson', 'w') as outfile:
json.dump(data, outfile)
堆栈跟踪：
---------------------------------------------------------------------------
OSError Traceback (most recent call last)
<ipython-input-428-d6121ed2750b> in <module>()
----> 1 writeGeoJSON(6, 15, baseline)
<ipython-input-427-973b7a5a8acc> in writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix)
14 print("Start loop")
15 paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
---> 16 pool = Pool(2)
17 func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
18 print(predictionmatrix.memory_usage())
/usr/lib/python3.5/multiprocessing/context.py in Pool(self, processes, initializer, initargs, maxtasksperchild)
116 from .pool import Pool
117 return Pool(processes, initializer, initargs, maxtasksperchild,
--> 118 context=self.get_context())
119
120 def RawValue(self, typecode_or_type, *args):
/usr/lib/python3.5/multiprocessing/pool.py in __init__(self, processes, initializer, initargs, maxtasksperchild, context)
166 self._processes = processes
167 self._pool = []
--> 168 self._repopulate_pool()
169
170 self._worker_handler = threading.Thread(
/usr/lib/python3.5/multiprocessing/pool.py in _repopulate_pool(self)
231 w.name = w.name.replace('Process', 'PoolWorker')
232 w.daemon = True
--> 233 w.start()
234 util.debug('added worker')
235
/usr/lib/python3.5/multiprocessing/process.py in start(self)
103 'daemonic processes are not allowed to have children'
104 _cleanup()
--> 105 self._popen = self._Popen(self)
106 self._sentinel = self._popen.sentinel
107 _children.add(self)
/usr/lib/python3.5/multiprocessing/context.py in _Popen(process_obj)
265 def _Popen(process_obj):
266 from .popen_fork import Popen
--> 267 return Popen(process_obj)
268
269 class SpawnProcess(process.BaseProcess):
/usr/lib/python3.5/multiprocessing/popen_fork.py in __init__(self, process_obj)
18 sys.stderr.flush()
19 self.returncode = None
---> 20 self._launch(process_obj)
21
22 def duplicate_for_child(self, fd):
/usr/lib/python3.5/multiprocessing/popen_fork.py in _launch(self, process_obj)
65 code = 1
66 parent_r, child_w = os.pipe()
---> 67 self.pid = os.fork()
68 if self.pid == 0:
69 try:
OSError: [Errno 12] Cannot allocate memory
更新
根据@robyschek 的解决方案，我已将代码更新为：
global g_predictionmatrix
def worker_init(predictionmatrix):
global g_predictionmatrix
g_predictionmatrix = predictionmatrix
def parallelUpdateJSON(paramMatch, data_item):
for feature in data_item['features']:
currentfeature = predictionmatrix[(predictionmatrix['SId']==feature['properties']['cellId']) & paramMatch]
if (len(currentfeature) > 0):
feature['properties'].update({"style": {"opacity": currentfeature.AllActivity.item()}})
else:
feature['properties'].update({"style": {"opacity": 0}})
def use_the_pool(data, paramMatch, predictionmatrix):
pool = Pool(initializer=worker_init, initargs=(predictionmatrix,))
func = partial(parallelUpdateJSON, paramMatch)
pool.map(func, data)
pool.close()
pool.join()
def writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix):
with open('geo.geojson') as f:
data = json.load(f)
paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
use_the_pool(data, paramMatch, predictionmatrix)
with open('trentino-grid.geojson', 'w') as outfile:
json.dump(data, outfile)
我仍然遇到同样的错误。另外，根据 documentation，map() 应该将我的 data 分成块，所以我认为它不应该复制我的 80MBs rownum 次。我可能是错的... :)
另外我注意到，如果我使用较小的输入（~11MB 而不是 80MB），我不会收到错误。所以我想我正在尝试使用太多内存，但我无法想象它是如何从 80MB 变成 16GB RAM 无法处理的。
使用 multiprocessing.Pool 时，启动进程的默认方式是 fork。 fork 的问题是整个过程是重复的。 (see details here)。因此，如果您的主进程已经在使用大量内存，则该内存将被复制，达到 MemoryError。例如，如果您的主进程使用 2GB 内存而您使用 8 个子进程，则您将需要 18GB 内存。
您应该尝试使用不同的启动方法，例如 'forkserver' 或 'spawn':
from multiprocessing import set_start_method, Pool
set_start_method('forkserver')
# You can then start your Pool without each process
# cloning your entire memory
pool = Pool()
func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
pool.map(func, data)
这些方法避免了复制您的 Process 的工作区，但由于您需要重新加载正在使用的模块，因此启动速度可能会慢一些。
我们有过几次这样的经历。根据我的系统管理员的说法，unix 中有 "a bug"，如果你的内存不足，如果你的进程达到最大文件描述符限制，它会引发同样的错误。
我们有文件描述符泄漏，引发的错误是 [Errno 12] 无法分配内存#012OSError。
所以你应该看看你的脚本并仔细检查问题是否不是创建了太多的 FD 而不是
我面临以下问题。我正在尝试并行化更新文件的函数，但由于 OSError: [Errno 12] Cannot allocate memory，我无法启动 Pool()。我已经开始在服务器上四处查看，这不像我使用的是旧的、弱的 one/out 实际内存。
参见 htop：
free -m 显示除了 ~7GB 的交换内存外，我还有足够的可用内存：
所用的predictionmatrix数据帧占用了大约。 80MB 根据 pandasdataframe.memory_usage()
文件 geo.geojson 是 2MB
我该如何调试呢？我可以检查什么以及如何检查？感谢任何 tips/tricks!
代码：
def parallelUpdateJSON(paramMatch, predictionmatrix, data):
for feature in data['features']:
currentfeature = predictionmatrix[(predictionmatrix['SId']==feature['properties']['cellId']) & paramMatch]
if (len(currentfeature) > 0):
feature['properties'].update({"style": {"opacity": currentfeature.AllActivity.item()}})
else:
feature['properties'].update({"style": {"opacity": 0}})
def writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix):
with open('geo.geojson') as f:
data = json.load(f)
paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
pool = Pool()
func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
pool.map(func, data)
pool.close()
pool.join()
with open('output.geojson', 'w') as outfile:
json.dump(data, outfile)
堆栈跟踪：
---------------------------------------------------------------------------
OSError Traceback (most recent call last)
<ipython-input-428-d6121ed2750b> in <module>()
----> 1 writeGeoJSON(6, 15, baseline)
<ipython-input-427-973b7a5a8acc> in writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix)
14 print("Start loop")
15 paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
---> 16 pool = Pool(2)
17 func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
18 print(predictionmatrix.memory_usage())
/usr/lib/python3.5/multiprocessing/context.py in Pool(self, processes, initializer, initargs, maxtasksperchild)
116 from .pool import Pool
117 return Pool(processes, initializer, initargs, maxtasksperchild,
--> 118 context=self.get_context())
119
120 def RawValue(self, typecode_or_type, *args):
/usr/lib/python3.5/multiprocessing/pool.py in __init__(self, processes, initializer, initargs, maxtasksperchild, context)
166 self._processes = processes
167 self._pool = []
--> 168 self._repopulate_pool()
169
170 self._worker_handler = threading.Thread(
/usr/lib/python3.5/multiprocessing/pool.py in _repopulate_pool(self)
231 w.name = w.name.replace('Process', 'PoolWorker')
232 w.daemon = True
--> 233 w.start()
234 util.debug('added worker')
235
/usr/lib/python3.5/multiprocessing/process.py in start(self)
103 'daemonic processes are not allowed to have children'
104 _cleanup()
--> 105 self._popen = self._Popen(self)
106 self._sentinel = self._popen.sentinel
107 _children.add(self)
/usr/lib/python3.5/multiprocessing/context.py in _Popen(process_obj)
265 def _Popen(process_obj):
266 from .popen_fork import Popen
--> 267 return Popen(process_obj)
268
269 class SpawnProcess(process.BaseProcess):
/usr/lib/python3.5/multiprocessing/popen_fork.py in __init__(self, process_obj)
18 sys.stderr.flush()
19 self.returncode = None
---> 20 self._launch(process_obj)
21
22 def duplicate_for_child(self, fd):
/usr/lib/python3.5/multiprocessing/popen_fork.py in _launch(self, process_obj)
65 code = 1
66 parent_r, child_w = os.pipe()
---> 67 self.pid = os.fork()
68 if self.pid == 0:
69 try:
OSError: [Errno 12] Cannot allocate memory
更新
根据@robyschek 的解决方案，我已将代码更新为：
global g_predictionmatrix
def worker_init(predictionmatrix):
global g_predictionmatrix
g_predictionmatrix = predictionmatrix
def parallelUpdateJSON(paramMatch, data_item):
for feature in data_item['features']:
currentfeature = predictionmatrix[(predictionmatrix['SId']==feature['properties']['cellId']) & paramMatch]
if (len(currentfeature) > 0):
feature['properties'].update({"style": {"opacity": currentfeature.AllActivity.item()}})
else:
feature['properties'].update({"style": {"opacity": 0}})
def use_the_pool(data, paramMatch, predictionmatrix):
pool = Pool(initializer=worker_init, initargs=(predictionmatrix,))
func = partial(parallelUpdateJSON, paramMatch)
pool.map(func, data)
pool.close()
pool.join()
def writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix):
with open('geo.geojson') as f:
data = json.load(f)
paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
use_the_pool(data, paramMatch, predictionmatrix)
with open('trentino-grid.geojson', 'w') as outfile:
json.dump(data, outfile)
我仍然遇到同样的错误。另外，根据 documentation，map() 应该将我的 data 分成块，所以我认为它不应该复制我的 80MBs rownum 次。我可能是错的... :)
另外我注意到，如果我使用较小的输入（~11MB 而不是 80MB），我不会收到错误。所以我想我正在尝试使用太多内存，但我无法想象它是如何从 80MB 变成 16GB RAM 无法处理的。
使用 multiprocessing.Pool 时，启动进程的默认方式是 fork。 fork 的问题是整个过程是重复的。 (see details here)。因此，如果您的主进程已经在使用大量内存，则该内存将被复制，达到 MemoryError。例如，如果您的主进程使用 2GB 内存而您使用 8 个子进程，则您将需要 18GB 内存。
您应该尝试使用不同的启动方法，例如 'forkserver' 或 'spawn':
from multiprocessing import set_start_method, Pool
set_start_method('forkserver')
# You can then start your Pool without each process
# cloning your entire memory
pool = Pool()
func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
pool.map(func, data)
这些方法避免了复制您的 Process 的工作区，但由于您需要重新加载正在使用的模块，因此启动速度可能会慢一些。
我们有过几次这样的经历。根据我的系统管理员的说法，unix 中有 "a bug"，如果你的内存不足，如果你的进程达到最大文件描述符限制，它会引发同样的错误。
我们有文件描述符泄漏，引发的错误是 [Errno 12] 无法分配内存#012OSError。
所以你应该看看你的脚本并仔细检查问题是否不是创建了太多的 FD 而不是