现有问题:
- 用户设置活跃的partition比较麻烦
- 有无乱序情况的最佳配置不一样
- PrimitiveArrayPool内存占用没考虑,容易爆内存
- 开启动态参数后创建时间序列经常失败
- 动态参数计算出的memtable偏小,chunk较小,影响查询性能
- 对象内存估计不准确
- 内存中一个时间序列点数过多,上1万,拷贝排序较慢
新策略:
目标(解决1-5)
尽量保证创建元数据成功,为保证内存不爆,可以拒绝写入
尽量在有无乱序情况下,iotdb参数不需要变化都能很好适应
让chunk大小最大化
尽量不阻塞写入
尽量不加入固定参数
RPC模块:
- 一次请求的大小受限制 b.(防止许用户一条SQL写入1亿个点等场景;或者写了一个大于2GB的bytes[]).
- 并发数受限制 c。
内存写入模块:
优点:
- 所有SG共享内存,,因此创建序列(或今后改为序列活跃情况变化)时也不需要再更新SG;好处是内存利用率可以很高;
缺点:
- 部分步骤需要全局锁;目前看,假设array为k,SG info 写x延迟上报,则个memTable写入16MB后,会拿一次全局锁更新全局内存情况。
核心思想:
- Schema和历史resource单独分配大小;下文仅考虑其余写数据部分大小。
- 每个SG统计自身的chunk_metadata和unseal_resource大小;
- 全局ArrayPool统计buffered和out of buffer的array大小
- 系统统计总的大小
写入线程:
- 如果是非空的写入线程
- 在 StorageEngine 中检查SystemInfo是否为reject状态;如果是,则该写入线程循环sleep 100ms(等待flush线程释放内存,system置回正常状态)再进行写入;如果等待6000ms后仍为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 内存占用 < 总写入内存 * flush_proportion,返回 true。
- 如果 总写入内存 * flush_proportion ≤ SystemInfo 内存占用 < 总写入内存 * reject_proportion, 执行 写入线程的提交flush流程,返回 true。
- 如果 总写入内存 * reject_proportion ≤ SystemInfo 内存占用, SystemInfo 置为 reject 状态, 执行 写入线程的提交flush流程,记返回值为 flag
- 如果 flag = true
- 如果 SystemInfo 内存占用 < 总写入内存,则返回 true
- 如果 SystemInfo 内存占用 ≥ 总写入内存,直接抛 写入Reject 异常
- 如果 flag = false,则返回 false
}
- 判断 向SystemInfo上报 的返回结果
- 如果返回 false,则该写入线程循环 writeLock.condition.await(50ms) ,检查 SystemInfo 的 reject 状态如果不 reject,执行正常写入。如果等待 max_waiting_time_when_insert_blocked 后仍为reject状态,抛出写入异常
- 如果返回 true,则执行正常写入
- 如果捕获到 写入Reject 异常,reset SystemInfo,并继续向上抛
- 在 StorageGroupProcessor 里检查 shouldFlush,如果为true,进行异步flush过程
- StorageGroupProsessor. 释放writeLock
- 如果是空的写入线程
- 进入对应的 StorageGroupProcessor,获取 writeLock
- 获取对应分区的 TsFileProcessor:如果(其 workingMemtable 不为空且 shouldFlush 为 true),则提交 flush 任务;否则直接返回。
- StorageGroupProsessor. 释放writeLock
Flush流程:
flush 分为两种 一种为正常flush,为写入过程中触发,在insertPlan写入完成后正式开始进行异步flush(正常写入流程中flush);另一种为异步即时flush,即触发后立刻开始flush(flush释放内存后,若System总内存仍处于flush阈值之上且当前没有flush任务时触发)
写入线程的提交flush流程:
- 使用 PriorityQueue 对当前系统所有 memtable 按占用内存由大到小排序
- boolean flag = false
- 当前活跃内存 = SystemInfo 总内存 - SystemInfo flush内存
- 对 PriorityQueue 的每个 workingMemtable 逐个标记 shouldFlush,(直到标记的这些 TsFileProcessor 刷盘后 当前活跃内存 能降到 flush 阈值之下)
- 提交一个异步的空的写入线程(写入被标记的 Memtable 中)
- 判断此 workingMemtable 是否属于当前 TsFileProcessor,如果属于,flag = true
- 返回 flag
Flush 流程:
- 先更新 SystemInfo flush 内存
- 将 workingMemtable 移到 flushingMemtables 中
关闭TsFile文件逻辑:
文件封口触发逻辑与现有master版本相同,都为一个insertPlan写入完成后,检查该TSP是否需要flush,如果需要,再检查是否TsFile大小超过阈值,如果超过,flush memtable后将文件封口。
TsFile关闭完成后,清空该TSPInfo,向对应SGInfo重置状态并向SystemInfo报告重置后SGInfo
- 如果此时SystemInfo 为reject状态 且 `SystemInfo中统计的总内存 < 总写入内存 * 80%`,将SystemInfo 置于正常状态
- 如果`SystemInfo中统计的总内存 >= 总写入内存 * 50%`,触发即时flush
MTree内存控制:
注册时间序列时,如果总时间序列个数*300 > 总内存*0.1,此时拒绝注册,抛出异常。
细节:关于Array Pool中分类型的数组如何管理?
解法: :
a. 按序列注册比例置换:
各类型数据在pool中的比例由schema中序列的数据类型比例决定。当pool中buffer已满,但是各类型的array的比例还未调整至schema中的比例时,申请(那些比例应该很高、但是buffer中还很少的)数据类型的size时,先当做OOP,待归还时逐渐将比例调好。另,为了加速比例调整,可以在此处触发flush;
按使用频度置换:
LRU缺点:有类型写入很快、有类型很慢,则抖动太厉害(慢速的加入池中,会踢出快类型的,然后还没被复用,就又被踢走了);LFU:负载变化时适应太慢;2Q。
细节:为什么当SG的info变化超过一定阈值才向Sys汇报?
解法:为了减少全局锁。假设Sys预留1GB出来,有n个SG,那么每个SG info每增长1/n GB数据时,才需要跟全局同步一次。
细节: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, SysInfo,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
Historical Resource 设计
采用二级索引的方式来降低TsFileResource常驻内存的数量,进而控制内存使用。
rootFile: 记录是<device, [starttime, endTime, IndexFile]>,是TsFileResource的根索引,常驻内存
IndexFile:记录是<device, [starttime, endTime, TsFileRecource]>, 是TsFileResource的中间索引,固定大小,按需load进内存
TsFileResource:记录是<device, starttime, endTime>, 是TsFile的索引,按需load进内存
写入
TsFile关闭时,unsealed_resource刷入磁盘;维护IndexFile,将涉及的各device的tsfile resource都进行记录。
如果IndexFile文件大于阈值(1GB)了,就进行关闭持久化到磁盘上,并开启一个新的IndexFile用于记录下一批TsFileResource中间索引。IndexFile关闭时,往rootFile里继续写一批记录,将涉及的各device的 index file都进行记录。
细节:rootFile也过大怎么办?
解法:多级索引(缺点,一次查询过多次访问磁盘);或者在rootFile中忽略一些device(缺点是这些设备的查询需要逐个去扫描indexFile)
估算下一个rootFile可以索引多少个TsFileResource。 假设: 1个storage group,每个设备100个测点,每个storage group 51200个设备, 也就是5.12e6个测点 内存128GB Tsfile 512MB IndexFile是1GB 每条ResourceIndex 100B,一个IndexFile是1GB,可以记录1GB/100B=10240000条记录,也就是200个TsFileResource。 每个rootFile在一个IndexFile关闭时会记录一下,也是51200*100B=5.12MB 也就是200个TsFileResource会产1GB的IndexFile和5.12MB的rootFile 100T的磁盘空间,有100T/512MB=100000个TsFile,需要100000/200 * 1GB = 500GB IndexFile,100000/200*5.12MB= 2.56GB内存 假设一个storage group A个设备,一个设备D个测点, 总共就是A * D个测点 假设每次TsFile刷写会造成所有设备的索引更新,就会产生A * 100B 的IndexFile记录 一个IndexFile假设是 C GB, 那一个IndexFile能支持 C * 1000,000,000 / A * 100 = C * 10,000,000/A 个TsFile的刷写 IndexFile关闭时,也会在RootFile里记录下A * 100B的rootFile记录 也就是C * 10,000,000/A 个TsFile的刷写,会有C GB的IndexFile产生, 还有A * 100B的rootFile记录 假设整个数据库有S个 Storage group,那么常驻内存的就是 S * (rootfile + C GB) A 平均一个storage group 设备数 D 平均一个设备测点数 C IndexFile文件大小,单位是GB 512MB TsFile大小 rootFile = C * 10,000,000/A * 100B IndexFile= C GB 磁盘空间= C * 10,000,000/A * 512MB 测点=A * D 该估算有个问题就是对于TsFileName长度可以进行优化,可以只消耗Tsfile个数的TsFilename长度内存占用,而不是device * [starttime, endTime] * tsFile个数来估算,因此其内存占用估算放大了很多倍。 这个优化的前提是java的string 常量池,可以保证多个string公用一个字符串常量。 因此,限制一个StorageGroup的设备数是可以做到只使用(2.56GB+1GB)/128GB~ 3%的内存索引100TB的磁盘空间。 |
读取流程
- 查找常驻内存的rootFile,找到对应的device的记录,根据startTime和endTime,找到对应的IndexFile。
- load IndexFile进内存,找到对应的device的记录,根据startTime和endTime,找到对应的TsFile。
- load TsFile在内存中构造TsFileResource
merge流程
TsFile会定期的跟乱序文件进行合并,因此其名字和元数据也会进行更新。
对于新生成的TsFile,也需要往IndexFile里写入记录即<device, ResourceIndexs>, 但是这个会造成有多个indexFile里某条device的记录区间有重叠的情况。
例如有IndexFile1记录的device, 10, 200, Tsfile1, merge后,新的indexFile2记录device, 10,400, tsFile2。
这两条记录都会在rootFile里进行记录,查询的时候需要读取两个IndexFile,但是IndexFile1里的TsFile1找不到了,那就不再查找。只找IndexFile2里的TsFile2.
这里会多读一次文件,为了减少这种无效的索引干扰,可以在merge 数据结束后进行一下相关IndexFile的合并操作,尽量保证一个device的一个时间区间在一个IndexFile里。
cache优化
因为最近的TsFile会被读到概率更大,因此可以再加个TsFileResource的cache,用来存放TsFileResource。可以按时间周期来进行cache替换,最近写入的TsFileResource都高优先级存到这个cache里。
文档原链接: https://shimo.im/docs/CWxXTDhvkRrHvXPx