深入浅出 InfluxDB — 存储引擎

func (e *Engine) compactCache() {
	t := time.NewTicker(time.Second)
	defer t.Stop()
	for {
		e.mu.RLock()
		quit := e.snapDone
		e.mu.RUnlock()

		select {
		case <-quit:
			return
		// 每秒钟检查一次
		case <-t.C:
			// ...
			// 检查是否满足快照条件
			if e.ShouldCompactCache(time.Now()) {
				start := time.Now()
				err := e.WriteSnapshot() // 对 cache 的快照写入 TSM files
				// 错误处理
			}
		}
	}
}

func (e *Engine) WriteSnapshot() (err error) {
	started := time.Now()

	closedFiles, snapshot, err := func() (segments []string, snapshot *Cache, err error) {
		e.mu.Lock()
		defer e.mu.Unlock()

		if e.WALEnabled {
			// 关闭当前 WAL 的 segment file 并打开新文件
			if err = e.WAL.CloseSegment(); err != nil {
				return
			}
			// 返回已关闭的 segment
			segments, err = e.WAL.ClosedSegments()
			if err != nil {
				return
			}
		}
		// 对 cache 进行快照，本质上是创建新环并和已有旧哈希环交换指针
		snapshot, err = e.Cache.Snapshot()
		if err != nil {
			return
		}

		return
	}()

	if err != nil {
		return err
	}
	dedup := time.Now()
	// 在写入前需要对 snapshot 进行去重和稳定排序
	snapshot.Deduplicate()

	return e.writeSnapshotAndCommit(log, closedFiles, snapshot)
}

func (e *Engine) writeSnapshotAndCommit(log *zap.Logger, closedFiles []string, snapshot *Cache) (err error) {
	// ...
	// 创建一个过多个 TSM file，由 Compactor 将快照 cache 分片并发写入
	newFiles, err := e.Compactor.WriteSnapshot(snapshot) // 
	// 更新 FileStore
	if err := e.FileStore.Replace(nil, newFiles); err != nil {
		// 如果出错就将新文件删除
		for _, file := range newFiles {
			err := os.Remove(file)
		}
		return err
	}
	// 快照已落盘，清理 cache 和 WAL 释放空间
	e.Cache.ClearSnapshot(true)
	e.WAL.Remove(closedFiles)
	return nil
}

func (e *Engine) compact(wg *sync.WaitGroup) {
	t := time.NewTicker(time.Second)
	defer t.Stop()

	for {
		e.mu.RLock()
		quit := e.done
		e.mu.RUnlock()

		select {
		case <-quit:
			return

		case <-t.C:

			// 逐层寻找相邻的 generation 并将它们进行分组
			level1Groups := e.CompactionPlan.PlanLevel(1)
			level2Groups := e.CompactionPlan.PlanLevel(2)
			level3Groups := e.CompactionPlan.PlanLevel(3)
			level4Groups := e.CompactionPlan.Plan(e.LastModified())
			atomic.StoreInt64(&e.stats.TSMOptimizeCompactionsQueue, int64(len(level4Groups)))

			// 不需要全量合并，尝试进行查询优化
			if len(level4Groups) == 0 {
				level4Groups = e.CompactionPlan.PlanOptimize()
			}

			// 设定每一层级用于合并的队列的长度
			e.scheduler.setDepth(1, len(level1Groups))
			e.scheduler.setDepth(2, len(level2Groups))
			e.scheduler.setDepth(3, len(level3Groups))
			e.scheduler.setDepth(4, len(level4Groups))

			// 由调度器进行判断哪个 level 的合并操作是可以执行的
			if level, runnable := e.scheduler.next(); runnable {
				switch level {
				case 1:
					if e.compactHiPriorityLevel(level1Groups[0], 1, false, wg) {
						level1Groups = level1Groups[1:]
					}
				case 2:
					if e.compactHiPriorityLevel(level2Groups[0], 2, false, wg) {
						level2Groups = level2Groups[1:]
					}
				case 3:
					if e.compactLoPriorityLevel(level3Groups[0], 3, true, wg) {
						level3Groups = level3Groups[1:]
					}
				case 4:
					if e.compactFull(level4Groups[0], wg) {
						level4Groups = level4Groups[1:]
					}
				}
			}

			// 每次只取分组中队头的那一组进行合并，其他的需要手动解除避免锁定
			e.CompactionPlan.Release(level1Groups)
			e.CompactionPlan.Release(level2Groups)
			e.CompactionPlan.Release(level3Groups)
			e.CompactionPlan.Release(level4Groups)
		}
	}
}

// 根据 tsm 文件的 seq 划分 level
func (t *tsmGeneration) level() int {
	_, seq, _ := t.parseFileName(t.files[0].Path)
	if seq < 4 {
		return seq
	}
	// seq 如果超过4则一律当作L4
	return 4
}

const partitions = 16

type ring struct {
	keysHint int64 // 对 key 进行计数
	// 环的底层数组
	partitions []*partition
}

type partition struct {
	mu    sync.RWMutex
	store map[string]*entry
}

// 构造哈希环，n必须是2的指数次幂，最大不超过256
func newring(n int) (*ring, error) {
	if n <= 0 || n > partitions {
		return nil, fmt.Errorf("invalid number of paritions: %d", n)
	}

	r := ring{
		partitions: make([]*partition, n),
	}

	for i := 0; i < len(r.partitions); i++ {
		r.partitions[i] = &partition{
			store: make(map[string]*entry),
		}
	}
	return &r, nil
}

// cache 的快速分片的具体实现
func (r *ring) split(n int) []storer {
	var keys int
	storers := make([]storer, n)
	for i := 0; i < n; i++ {
		storers[i], _ = newring(len(r.partitions))
	}

	for i, p := range r.partitions {
		r := storers[i%n].(*ring)
		r.partitions[i] = p
		keys += len(p.store)
	}
	return storers
}

// 快速找到 key 所在的分组
func (r *ring) getPartition(key []byte) *partition {
	return r.partitions[int(xxhash.Sum64(key)%uint64(len(r.partitions)))]
}

// 同一个 key 先找到分组，再找分组中的 entry
func (r *ring) entry(key []byte) *entry {
	return r.getPartition(key).entry(key)
}

// 正常的映射
func (p *partition) entry(key []byte) *entry {
	p.mu.RLock()
	e := p.store[string(key)]
	p.mu.RUnlock()
	return e
}