随着时间的推移和执行的继续，我们的 dask 调度程序进程似乎在内存中膨胀。目前我们看到它使用 5GB 的内存，这似乎很高，因为所有数据都应该存在于工作节点上：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
31172 atoz 20 0 5486944 5.071g 7100 S 23.8 65.0 92:38.64 dask-scheduler
启动调度程序时，我们的内存使用量将低于 1GB。重新启动网络执行 client.restart() 似乎没有帮助，只有杀死调度程序进程本身并重新启动才能释放内存。
每个执行的单个任务的预期内存使用量是多少？调度程序真的只维护指向哪个工作人员包含未来结果的指针吗？
- - 编辑 - -
我认为我主要关心的是为什么 client.restart() 似乎没有释放调度程序进程使用的内存。我显然不希望它释放所有内存，而是回到基本水平。我们正在使用 client.map 在不同输入列表中执行我们的函数。执行后，一遍又一遍地重新启动客户端并拍摄调度程序内存的快照，我们看到以下增长：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
27955 atoz 20 0 670556 507212 13536 R 43.7 6.2 1:23.61 dask-scheduler
27955 atoz 20 0 827308 663772 13536 S 1.7 8.1 16:25.85 dask-scheduler
27955 atoz 20 0 859652 696408 13536 S 4.0 8.5 19:18.04 dask-scheduler
27955 atoz 20 0 1087160 923912 13536 R 62.3 11.3 20:03.15 dask-scheduler
27955 atoz 20 0 1038904 875788 13536 S 3.7 10.7 23:57.07 dask-scheduler
27955 atoz 20 0 1441060 1.163g 12976 S 4.3 14.9 35:54.45 dask-scheduler
27955 atoz 20 0 1646204 1.358g 12976 S 4.3 17.4 37:05.86 dask-scheduler
27955 atoz 20 0 1597652 1.312g 12976 S 4.7 16.8 37:40.13 dask-scheduler
我想我只是惊讶于在执行 client.restart() 之后我们没有看到内存使用回到某个基线。
----进一步编辑---- 关于我们正在运行的更多信息，因为建议是如果我们传递大型数据结构，则将它们直接发送给工作人员。
我们发送一个字典作为每个任务的输入，当 json 转储字典时，大多数都在 1000 个字符以下。
---- 甚至进一步的编辑：转载问题---- 我们今天再次转载了这个问题。我关闭了调度程序并重新启动它，我们有大约 5.4 GB 的可用内存，然后我们运行我将在下面粘贴的函数，跨越 69614 个真正保存一些基于文件的信息的字典对象（我们所有的工作人员都映射到同一个NFS 数据存储，我们使用 Dask 作为分布式文件分析系统。
这是函数（注意：squarewheels4 是一个自制的惰性文件提取和分析包，它使用 Acora 和 libarchive 作为其基础，用于从压缩存档中获取文件并为文件编制索引。）
def get_mrc_failures(file_dict):
from squarewheels4.platforms.ucs.b_series import ChassisTechSupport
from squarewheels4.files.ucs.managed.chassis import CIMCTechSupportFile
import re
dimm_info_re = re.compile(r"(?P<slot>[^\|]+)\|(?P<size>\d+)\|.*\|(?P<pid>\S+)")
return_dict = file_dict
return_dict["return_code"] = "NOT_FILLED_OUT"
filename = "{file_path}{file_sha1}/{file_name}".format(**file_dict)
try:
sw = ChassisTechSupport(filename)
except Exception as e:
return_dict["return_code"] = "SW_LOAD_ERROR"
return_dict["error_msg"] = str(e)
return return_dict
server_dict = {}
cimcs = sw.getlist("CIMC*.tar.gz")
if not cimcs:
return_dict["return_code"] = "NO_CIMCS"
return_dict["keys_list"] = str(sw.getlist("*"))
return return_dict
for cimc in cimcs:
if not isinstance(cimc, CIMCTechSupportFile): continue
cimc_id = cimc.number
server_dict[cimc_id] = {}
# Get MRC file
try:
mrc = cimc["*MrcOut.txt"]
except KeyError:
server_dict[cimc_id]["response_code"] = "NO_MRC"
continue
# see if our end of file marker is there, should look like:
# --- END OF FILE (Done!
whole_mrc = mrc.read().splitlines()
last_10 = whole_mrc[-10:]
eof_line = [l for l in last_10 if b"END OF FILE" in l]
server_dict[cimc_id]["response_code"] = "EOF_FOUND" if eof_line else "EOF_MISSING"
if eof_line:
continue
# get DIMM types
hit_inventory_line = False
dimm_info = []
dimm_error_lines = []
equals_count = 0
for line in whole_mrc:
# regex each line... sigh
if b"DIMM Inventory" in line:
hit_inventory_line = True
if not hit_inventory_line:
continue
if hit_inventory_line and b"=========" in line:
equals_count += 1
if equals_count > 2:
break
continue
if equals_count < 2:
continue
# we're in the dimm section and not out of it yet
line = str(line)
reg = dimm_info_re.match(line)
if not reg:
#bad :/
dimm_error_lines.append(line)
continue
dimm_info.append(reg.groupdict())
server_dict[cimc_id]["dimm_info"] = dimm_info
server_dict[cimc_id]["dimm_error_lines"] = dimm_error_lines
return_dict["return_code"] = "COMPLETED"
return_dict["server_dict"] = server_dict
return return_dict
```
期货的生成方式如下：
futures = client.map(function_name, file_list)
在这种状态之后，我的目标是尝试恢复并让 dask 释放它分配的内存，这是我的努力：在取消期货之前：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
21914 atoz 20 0 6257840 4.883g 2324 S 0.0 62.6 121:21.93 dask-scheduler
atoz@atoz-sched:~$ free -h
total used free shared buff/cache available
Mem: 7.8G 7.1G 248M 9.9M 415M 383M
Swap: 8.0G 4.3G 3.7G
在取消期货时：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
21914 atoz 20 0 6258864 5.261g 5144 R 60.0 67.5 122:16.38 dask-scheduler
atoz@atoz-sched:~$ free -h
total used free shared buff/cache available
Mem: 7.8G 7.5G 176M 9.4M 126M 83M
Swap: 8.0G 4.1G 3.9G
取消期货后：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
21914 atoz 20 0 6243760 5.217g 4920 S 0.0 66.9 123:13.80 dask-scheduler
atoz@atoz-sched:~$ free -h
total used free shared buff/cache available
Mem: 7.8G 7.5G 186M 9.4M 132M 96M
Swap: 8.0G 4.1G 3.9G
做了一个 client.restart() 之后
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
21914 atoz 20 0 6177424 5.228g 4912 S 2.7 67.1 123:20.04 dask-scheduler
atoz@atoz-sched:~$ free -h
total used free shared buff/cache available
Mem: 7.8G 7.5G 196M 9.4M 136M 107M
Swap: 8.0G 4.0G 4.0G
不管我在分布式系统中运行什么，我的期望是在取消期货之后它至少会恢复到接近正常水平......并且在执行 client.restart() 之后，我们肯定会接近我们的正常基线。我在这里错了吗？
--- 第二次重现 ---- 使用以下步骤重现了行为（尽管不是完全内存耗尽）：
这是我的工人功能
def get_fault_list_v2(file_dict):
import libarchive
return_dict = file_dict
filename = "{file_path}{file_sha1}/{file_name}".format(**file_dict)
with libarchive.file_reader(filename) as arc:
for e in arc:
pn = e.pathname
return return_dict
我在 68617 次迭代/文件中运行它
在运行之前，我们看到使用了这么多内存：PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND 12256 atoz 20 0 1345848 1.107g 7972 S 1.7 14.2 47:15.24 dask-scheduler
atoz@atoz-sched:~$ free -h
total used free shared buff/cache available
Mem: 7.8G 3.1G 162M 22M 4.5G 4.3G
Swap: 8.0G 3.8G 4.2G
运行后我们看到了这么多：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
12256 atoz 20 0 2461004 2.133g 8024 S 1.3 27.4 66:41.46 dask-scheduler
执行完 client.restart 后，我们看到：
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
12256 atoz 20 0 2462756 2.134g 8144 S 6.6 27.4 66:42.61 dask-scheduler