锁的种类
insertLock in StorageGroupProcessor
用以保护StorageGroupProcessor中的内存结构,如当前的workingTsFileProcessor、latestTimeForEachDevice、partitionLatestFlushedTimeForEachDevice、newlyFlushedPartitionLatestFlushedTimeForEachDevice、globalLatestFlushedTimeForEachDevice、partitionMaxFileVersions等等
compactionMergeLock in TsFileManagement
用以保护TsFileManagement中管理的顺序和乱序的TsFileResource的List,所有对TsFileResource的List的写操作(比如写入时,创建新的TsFileResource,compact时删除旧的TsFileResource,并产生新的TsFileResource)需要获得其写锁,所有对TsFileResource的List的读操作(如查询时获得满足此次查询条件的所有TsFileResource)需要获得其读锁
tsFileLock in TsFileResource
用以保证此TsFileResource对应的单个Tsfile的一致性,删除tsfile需要获得其写锁,查询期间,涉及到的所有tsfile文件,需要一直持有他们的读锁
以上三个锁的获取顺序
因为常常在某一操作流程中,为了保证正确性,需要获得这三个锁中的两个或者全部,所以为了避免产生死锁,所以以上三个锁的获得需要有一定的顺序约束,现在约定俗成的顺序是:
insertLock → compactionMergeLock → tsFileLock
查询获取和释放锁的流程
- 在各种查询 Executor 的初始化方法中
- 调用StorageEngine.mergeLock
- 根据每个查询涉及的序列,获得这些序列所属的所有StorageGroupProcessor,按照StorageGroupProcessor的name的字母序排序
- 对于每个查询涉及的 StorageGroupProcessor
- 对 insertLock 加读锁,对 compactionMergeLock 加读锁
- 对每个涉及的序列所在的 StorageGroupProsessor
- 对 insertLock 加读锁,对 compactionMergeLock 加读锁
- 筛选待查的 TsFileResource 列表,对于待查的每个 TsFileResource
- 对 TsFileLock 加读锁
- 释放 insertLock 和 compactionMergeLock 的读锁
- 调用StorageEngine.mergeUnlock,对每个涉及的序列所在的 StorageGroupProsessor
- 释放 insertLock 和 compactionMergeLock 的读锁
- 直到查询结束,释放所有 TsFileLock 的读锁
- 查询正常结束,则会在 fetchResult() 没有下一组数据时,调用 releaseQueryResourceNoExceptions() 方法释放读锁
- 查询正常进行,但未取到最后一组数据时,dataSet 被关闭,也会调用 releaseQueryResourceNoExceptions() 方法释放读锁
- 查询正常进行,此时 session 连接调用 close()方法,会对此连接执行的每个未被释放查询,调用 releaseQueryResourceNoExceptions() 方法释放读锁
- 查询遇到异常,则会在异常捕获内调用 releaseQueryResourceNoExceptions() 释放读锁
写入获取和释放锁的流程
写入获取锁的流程较为简单,在StorageGroupProcessor中,每一个insert接口都会直接获取该SGP的insert lock的写锁,直至写入完成,返回StorageEngine后再释放写锁,具体写入接口如下:
public void insert(InsertRowPlan insertRowPlan)
public void insert(InsertRowsOfOneDevicePlan insertRowsOfOneDevicePlan)
public void insertTablet(InsertTabletPlan insertTabletPlan)
另外一个获取锁的地方是flush,当内存控制需要将某个tsfile刷入磁盘时,会先拿tsFileLock的写锁,再进行刷盘,具体刷盘接口如下:
public void asyncFlush()
public void asyncClose()
- 写入过程中,新建一个文件,并把 TsFileResource 加到 list 时
- compactionMergeLock.writeLock()
- 修改resourceList
- compactionMergeLock.writeUnlock()
合并获取和释放锁的流程
- 选择文件时
- compactionMergeLock.readLock()
- 拷贝 resourceList
- compactionMergeLock.readUnlock()
- 合并文件结束时
- compactionMergeLock.writeLock()
- 修改 resourceList,移除要删除的 TsFileResource
- compactionMergeLock.readUnlock()
- 对于每个待删除的 TsFileResource
- 拿 TsFileLock 的写锁
- 删文件
- 释放 TsFileLock 的写锁
提交合并任务:
1、resourceListLock.readLock()
2、提交合并任务
3、resourceListLock.readUnlock()
合并完对文件列表和实际文件进行删除和新增:
1、resourceListLock.writeLock(long timeout)
2、然后进行文件列表的旧文件删除和新文件的增加
3、resourceListLock.writeUnlock()
4、循环要删除的每一个TsFile文件,对于每一个TsFile文件进行如下操作:
1、获取TsFileResource的tsFileLock写锁
2、删除文件
3、释放TsFileResource的tsFileLock写锁
数据删除操作
在StorageGroupProcessor中,每一个删除接口都会直接获取该SGP的insert lock的写锁,直至删除完成、或者捕获异常后释放写锁。
deleteFolder(String systemDir) // 删除data/system下的sg文件夹
delete(PartialPath path, long startTime, long endTime, long planIndex) // 删除时间序列数据
syncDeleteDataFiles() // 删除这个sg里的所有数据文件
removePartitions(TimePartitionFilter filter) //删除时间分区
删除时间分区还会拿resourceListLock:
resourceListLock.writeLock()
等待此分区的所有合并任务结束
delete 此分区所有文件
resourceListLock.writeUnlock()
1 Comment
jaquan rice
Good