目标
- 无需在有无乱序数据时分开配置
- 避免设置活跃的 partition 个数
- 考虑 PrimitiveArrayPool 内存占用
- 尽量有效利用内存,使 Chunk 尽量大
- 尽量保证创建元数据成功,为保证内存不爆,可以拒绝写入
- 尽量在有无乱序情况下,iotdb参数不需要变化都能很好适应
- 尽量不阻塞写入
- 尽量不加入固定参数
写入流程各部分内存统计
RPC模块
- 一次请求的大小受限制 thrift_max_frame_size=67108864.(防止许用户一条SQL写入1亿个点等场景;或者写了一个大于2GB的bytes[]).
- 并发数受限制 rpc_max_concurrent_client_num=65535。
核心思想:
- Schema和历史resource单独分配大小;下文仅考虑其余写数据部分大小。
- 每个SG统计自身的chunk_metadata和unseal_resource大小;
- 全局ArrayPool统计buffered和out of buffer的array大小
- 系统统计总的大小
数据写入流程
写入线程
- 如果是非空的写入线程
- 在 StorageEngine 中检查SystemInfo是否为reject状态;如果是,则该写入线程循环(等待flush线程释放内存,system置回正常状态)再进行写入;如果等待max_waiting_time_when_insert_blocked后仍为reject状态,抛出写入异常;
- 进入对应的StorageGroupProcessor,获取 writeLock
- 进入对应分区的 TsFileProcessor:(1)获取已有的可写入的顺序或乱序 TsFileProcessor(2)如果没有可写入的TsFileProcessor,创建新的 TsFileProcessor
- 统计当前写入计划新增的内存占用,增加至TspInfo和SgInfo中:(1)新测点增加 chunk_metadata(2)TEXT 类型数据(3)TVList 中增加的 PrimitiveArray(4)flush内存
- 如果 SGInfo 增量超过阈值(storage_group_size_report_threshold=16M)
- 向SystemInfo进行上报(将当前 TsFileProcessor 传入);
synchronized(SystemInfo) {- 更新 SystemInfo 内存占用。
- 如果 SystemInfo 内存占用 < ,返回 true。
- 如果 总写入内存 * flush_proportion ≤ SystemInfo 内存占用 < 总写入内存 * reject_proportion, 执行 选择Memtable提交flush流程,返回 true。
- 如果 总写入内存 * reject_proportion ≤ SystemInfo 内存占用, SystemInfo 置为 reject 状态, 执行 选择Memtable提交flush流程,
- 如果 flag = true
- 如果 SystemInfo 内存占用 < 总写入内存,则返回 true
- 如果 SystemInfo 内存占用 ≥ 总写入内存,直接抛 写入Reject 异常
- 如果 flag = false,则返回 false
}
- 判断 向SystemInfo上报 的返回结果
- 如果返回 false,则 ,检查 SystemInfo 的 reject 状态如果不 reject或者该memtable被标记为shouldFlush,执行正常写入。如果等待 max_waiting_time_when_insert_blocked 后仍为reject状态,抛出写入异常
- 如果返回 true,则执行正常写入
- 如果捕获到 写入Reject 异常,reset SystemInfo,并继续向上抛
- 检查 workingMemtable 的 shouldFlush,如果为true,提交 Flush 任务,并根据文件大小判断是否需要 close。
- StorageGroupProsessor. 释放writeLock
- 如果是空的写入线程
- 进入对应的 StorageGroupProcessor,获取 writeLock
- 获取对应分区的 TsFileProcessor:如果(其 workingMemtable 不为空且 shouldFlush 为 true),则提交 flush 任务;否则直接返回。
- StorageGroupProsessor. 释放writeLock
选择Memtable提交flush流程:
- boolean flag = false
- 当前活跃内存 = SystemInfo 总内存 - SystemInfo flush内存
- 对 PriorityQueue 的每个 workingMemtable 逐个标记 shouldFlush,(直到标记的这些 TsFileProcessor 刷盘后 当前活跃内存 能降到 flush 阈值之下)
- 提交一个异步的空的写入线程(写入被标记的 Memtable 中)
- 判断此 workingMemtable 是否属于当前 TsFileProcessor,如果属于,flag = true
- 返回 flag
Flush 线程:
- 先更新 SystemInfo flush 内存
- 将 workingMemtable 移到 flushingMemtables 中
TsFile文件关闭逻辑:
- 一个insertPlan写入完成后,检查该TSP的 workingMemtable 的 shouldFlush 字段,如果为 true,再检查是否TsFile大小超过阈值,如果超过,flush memtable后将文件封口。
- TsFile关闭完成后,清空该TSPInfo,重置对应的 SGInfo 状态,并向SystemInfo报告重置后SGInfo
- 如果此时SystemInfo 为reject状态 且 `SystemInfo中统计的总内存 < 总写入内存 * reject_proportion`,将SystemInfo 置于正常状态
MTree内存控制:
注册时间序列时,如果总时间序列个数*estimate_series_size > 总内存*write_read_schema_free_memory_proportion:schema,此时拒绝注册,抛出异常。
相关参数整理
- 是否开启内存控制
enable_mem_control=true
以下参数只在开启内存控制时生效: - flush阈值(0.0--1.0)(关闭内存控制后无效)
当所有memtable实际占用大于总写入内存 * flush_proportion,触发flush
flush_proportion=0.4 - reject阈值(0.0--1.0)(关闭内存控制后无效)
当所有memtable实际占用大于总写入内存 * reject_proportion,阻塞写入,等待flush释放内存
reject_proportion=0.8 - array pool内存占总写入内存的大小比例(0.0--1.0)
buffered_arrays_memory_proportion=0.6 - sg上报阈值(bytes)(关闭内存控制后无效)
当一个sg内所有memtable的内存相比上次上报的增量大于这个值,向SystemInfo更新目前Sg的总大小
注意:当sg较多,例如1000时,需要考虑调小这个值。因为此时memtable需要写 16M * 1000 = 16G 才会向SystemInfo 汇报,比较危险。
storage_group_report_threshold=16777216 - 阻塞写入后的检查周期(ms)(关闭内存控制后无效)
当写入被SystemInfo拒绝后,客户端线程会以这个时间周期去检查SystemInfo的状态,直到flush线程释放掉一些内存,SystemInfo置回正常状态。
check_period_when_insert_blocked=50 - 阻塞写入后的最大等待时间(ms)(关闭内存控制后无效)
当写入阻塞时间超过这个值后,向客户端返回写入异常
max_waiting_time_when_insert_blocked=10000 - 预估一条序列在mtree中的大小(关闭内存控制后无效)
这个值用来限制可注册的序列个数
estimated_series_size=300
- memtable大小阈值(bytes)(开启内存控制后无效)
memtable_size_threshold=1073741824 - 写入、查询、schema、剩余内存占比
其中可注册的序列的个数由schema的内存(byte)除以estimated_series_size来确定
write_read_schema_free_memory_proportion=4:3:1:2 - array pool中的array长度
平均每个chunk的点数小于这个值时会造成内存的浪费,可以考虑调小这个值
primitive_array_size=128 - TsFile大小阈值
0代表只刷一个memtable就关文件;1代表刷两个memtable才关文件;大于1时如1G时,1个文件中会存在更多的memtable
目前设为1是考虑merge会把小文件合并为大文件;而且如果这个阈值过大且开启内存控制后,会导致内存中metadata积累较多,memtable越来越小
tsfile_size_threshold=1 - 平均chunk点数阈值(个数)
当memtable内平均每个序列的点数超过这个阈值时,触发flush
avg_series_point_number_threshold=10000
优点:
- 所有SG共享内存,不再对每个SG单独设置一个内存上限,因此创建序列(或今后改为序列活跃情况变化)时也不需要再更新SG;好处是内存利用率可以很高;
缺点:
- 部分步骤需要全局锁;目前看,假设array为k,SG info 写x延迟上报,则个memTable写入16MB后,会拿一次全局锁更新全局内存情况。
细节:关于Array Pool中分类型的数组如何管理?
解法: :
a. :
各类型数据在pool中的比例由schema中序列的数据类型比例决定。当pool中buffer已满,但是各类型的array的比例还未调整至schema中的比例时,申请(那些比例应该很高、但是buffer中还很少的)数据类型的size时,先当做OOP,待归还时逐渐将比例调好。另,为了加速比例调整,可以在此处触发flush;
按使用频度置换:
LRU缺点:有类型写入很快、有类型很慢,则抖动太厉害(慢速的加入池中,会踢出快类型的,然后还没被复用,就又被踢走了);LFU:负载变化时适应太慢;2Q。
细节:为什么当SG的info变化超过一定阈值才向SystemInfo汇报?
解法:为了减少全局锁。假设Sys预留1GB出来,有n个SG,那么每个SG info每增长1/n GB数据时,才需要跟全局同步一次,减少contention。
细节:String/byte[] 不需要buffer池的原因。
解:当接收到客户端一个string/byte[]时,接收线程已经占用了这么多内存了,(指针移动)。
细节:Array Pool中的字符型/byte[]如何管理?因为每个String的长度可能不同。
解法:Array Pool中有List<Binary[]> 用于做array的缓冲池,但是归还时,内部每个Binary均为null。此时有两种方法可选:
- 将String类型的数组申请每次都当做array Pool无法响应。缺点是每次都要触发2.2.2,带来全局锁。
在ArrayPool中虚拟地向Sys info汇报自身拥有不同大小的byte[] 若干个(即byte[][]),当plan中有string时,向arrayPool中申请一个可以容纳的byte[], array Pool中有,则arrayPool做计数,认为该byte[]被借走。flush时在归还。优点是不会增加全局锁,缺点是byte[]长度变化严重时,这里内存利用率会降低。【该方法仍然需要每次都要全局锁。】- 将该部分内存放入SG info中统计。
细节:TsFile什么时候关闭?
解法:指定TsFile的大小;或者由于SGInfo中Chunk_metadata过大导致刷磁盘时进行关闭。
在该方案中,只有图中红色部分是全局锁(ArrayPool, SystemInfo,reject信号量)
详细计算公式:
- WAL buffer: 一个存储组固定大小的buffer:b(永久)
- PrimitiveArrayPool 原始类型数组缓存(永久)
- memtable 写入数据使用
- memtable 排序使用
- String数组每次会清空(现状)。
- TVListAllocator 中 TVList 对象头缓存(永久)
- MemtablePool 中 Memtable 对象头缓存(永久)
- flush 之后在内存中积累的 ChunkMetadata 缓存大小为 K(临时)
- flush编码后的字节数组:一个Chunk编码后的大小(临时)
每个存储组维护 SGinfo: G
- G=C + U + B + b (ChunkMetadata + UnsealedResources + Bytes[] + WAL)
- 当前写入plan在该SG下的数据总内存占用为 T
当前注册的时间序列数量 N
ArrayPool维护 ArrayPoolInfo:A
- Buffered Array 内存 B,已用B_u
- OOP内存 O
- A=B_u+O
系统级别维护SystemInfo:S
- 若写入正常,开始向working memtable中写入数据,当memtable中Array空间不足时,向Array Pool申请新的Array。array pool判断是否有已向系统报备过的该类型array(即Buffered array)
- 如果有Buffered array,向Array Pool申请Buffered array并写入数据;
- 如果没有,则需要申请OOB(out of buffered)的数组
SG个数:M
系统为delay上报预留的内存大小:R
SG上报阈值: R/M
条件:
- available array:所有measurement都在memtable中有,且空间足够;
- available buffered array:B > B_u && 相应数据类型的array存在;
- 左下的update system info:G的增量大于R/M。
- 生成reject:S>=可写入内存(或达到一定比例,如90%)
- 左边的call for flush: S >=可写入内存*比例 (如50%)
当一个SG被拆分成多个时间分区时,将上文中的SG改为TSP。
当String类型的flush后,更新tsp中对应的内存统计量并上报给System
TsFile封口时,更新ChunkMetadata和Resource的内存统计并上报给System
文档原链接: https://shimo.im/docs/CWxXTDhvkRrHvXPx