golang源码阅读：VictoriaMetrics中的协程优先级的处理方式

2022-01-26 19:05:28 阅读：154 来源： 互联网

标签：insert 协程写入 storage VictoriaMetrics golang 源码 CPU select

在阅读VictoriaMetrics的源码的时候，读到了那么平平无奇的一段：

// AddRows adds the given mrs to s.
func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
	if len(mrs) == 0 {
		return nil
	}

	// Limit the number of concurrent goroutines that may add rows to the storage.
	// This should prevent from out of memory errors and CPU trashing when too many
	// goroutines call AddRows.
	select {
	case addRowsConcurrencyCh <- struct{}{}:
	default:
		// Sleep for a while until giving up
		atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1)
		t := timerpool.Get(addRowsTimeout)

		// Prioritize data ingestion over concurrent searches.
		storagepacelimiter.Search.Inc()

		select {
		case addRowsConcurrencyCh <- struct{}{}:
			timerpool.Put(t)
			storagepacelimiter.Search.Dec()
		case <-t.C:
			timerpool.Put(t)
			storagepacelimiter.Search.Dec()
			atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1)
			atomic.AddUint64(&s.addRowsConcurrencyDroppedRows, uint64(len(mrs)))
			return fmt.Errorf("cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers; add more CPUs or reduce load",
				len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh))
		}
	}

仔细看了以后，真是不得了。这感觉就像——

1.背景

在vm-storage这个组件中，作为存储节点，它一边要负责数据的写入，一边要负责数据的查询。很明显，数据写入的工作很重要，而查询的优先级相比写入就要低一些。

遇到这种问题，我的第一反应就是：把写入的协程数设置得比查询的协程数多不就行了吗？想要多高的优先级就设置为多大的比例。

太天真了！

物理核是性能真正的限制。无论你有多少协程，理论上N个核就最多只有N个协程处于执行状态。
协程调度并非没有成本，协程越多，就会有越多的CPU时间花在协程调度上。对于CPU密集型的业务，计算的协程数超过物理核的个数的部分都是白瞎。
假设写的协程数是读的协程数的2倍，概率上看调度到写的次数是读的次数的2倍；但是读和写的计算量并不是对等的，假设某个查询的数据量较大，就会导致读协程总体的CPU时间多于写协程，最终可能会导致写入超时失败。正确的办法是通过机制来让读协程主动让出CPU资源。

所以这里我直接总结vm-storage在协程控制的处理思路，再逐段分析源码：

区分IO协程和计算协程。

IO协程收到数据后，通过channel转给计算协程。

计算协程的数量与核的数量相等。

处理insert操作的协程数等于CPU的核数，且接收任务的channel的长度也等于CPU核数。
处理query_range等查询操作的协程数是CPU核数的2倍，猜测这里是因为部分读操作可能导致mmap区域内存产生缺页中断，继而引发IO阻塞。但不管怎么样，对协程数仍然是很克制。

insert协程执行业务逻辑前，在一个排队channel里面写入一个struct{}，这个排队channel的长度与CPU核数相等。写入成功，证明同时进行的写操作小于核数，允许继续写入。

写入队列失败，就证明某个insert协程没有被及时调度，就需要通知select协程主动让出CPU资源。
每当有一个insert操作被阻塞，就会通过原子操作累加计数。这个计数代表了有多少个insert操作处于等待。

如果insert操作排队成功，计数器就会减一。当计数器为0时，通过条件变量来发起 broadcast()，唤醒在等待的select操作。
select协程中，每扫描4095个block就会检查一次是否有insert操作在等待。如果有，调用条件变量 cond.Wait()进入等待，让出协程调度。

(源码位于：https://github.com/VictoriaMetrics/VictoriaMetrics)

2. insert操作源码分析

2.1 工作协程的创建

lib/protoparser/common/unmarshal_work.go:24

// StartUnmarshalWorkers starts unmarshal workers.
func StartUnmarshalWorkers() {
	if unmarshalWorkCh != nil {
		logger.Panicf("BUG: it looks like startUnmarshalWorkers() has been alread called without stopUnmarshalWorkers()")
	}
	gomaxprocs := cgroup.AvailableCPUs()   //获取物理核的个数
	unmarshalWorkCh = make(chan UnmarshalWork, gomaxprocs)  //创建一个channel,长度与核数相等
	unmarshalWorkersWG.Add(gomaxprocs)
	for i := 0; i < gomaxprocs; i++ {
		go func() {  // 启动N个协程，数量与核数相等
			defer unmarshalWorkersWG.Done()
			for uw := range unmarshalWorkCh {
				uw.Unmarshal()  // 这里调用具体的业务处理函数
			}
		}()
	}
}

IO协程获取数据后，把请求丢到unmarshalWorkCh中：

// ScheduleUnmarshalWork schedules uw to run in the worker pool.
//
// It is expected that StartUnmarshalWorkers is already called.
func ScheduleUnmarshalWork(uw UnmarshalWork) {
	unmarshalWorkCh <- uw
}

2.2 insert协程的并发检查

lib/storage/storage.go:1617

首先创建了一个用于管理写入并发的channel:

var (
	// Limit the concurrency for data ingestion to GOMAXPROCS, since this operation
	// is CPU bound, so there is no sense in running more than GOMAXPROCS concurrent
	// goroutines on data ingestion path.
	addRowsConcurrencyCh = make(chan struct{}, cgroup.AvailableCPUs())
	addRowsTimeout       = 30 * time.Second
)

队列的长度是CPU核数。假设有10个核，则写入操作最多10个并发。

下面是对于写入并发的处理：lib/storage/storage.go:1529

// AddRows adds the given mrs to s.
func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
	if len(mrs) == 0 {
		return nil
	}

	// Limit the number of concurrent goroutines that may add rows to the storage.
	// This should prevent from out of memory errors and CPU trashing when too many
	// goroutines call AddRows.
	select {
	case addRowsConcurrencyCh <- struct{}{}:    //如果写入channel成功，说明并发小于最大核数。然后走到插入逻辑去。
	default:  //如果插入channel失败，说明某个insert操作的协程被阻塞。这时需要通知select协程去让出。
		// Sleep for a while until giving up
		atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1)
		t := timerpool.Get(addRowsTimeout)

		// Prioritize data ingestion over concurrent searches.
		storagepacelimiter.Search.Inc() // pacelimiter（步长限制器）中有个原子累加的变量，说明有多少个insert操作在等待

		select {
		case addRowsConcurrencyCh <- struct{}{}:  //在超时的时间内，等待入队成功的事件。
			timerpool.Put(t)  //把timer放回对象池，减少GC
			storagepacelimiter.Search.Dec()  // insert操作可以顺利调度了，等待的数量原子减一。
            // 等待数量为0的时候，调用 cond.Broadcast() 来通知select协程开始工作。
		case <-t.C:  //等待30秒
			timerpool.Put(t)
			storagepacelimiter.Search.Dec()
			atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1)
			atomic.AddUint64(&s.addRowsConcurrencyDroppedRows, uint64(len(mrs)))
			return fmt.Errorf("cannot add %d rows to storage in %s, since it is overloaded with %d concurrent writers; add more CPUs or reduce load",
				len(mrs), addRowsTimeout, cap(addRowsConcurrencyCh))
            // 等待了30秒仍然没有CPU资源，只能报错
		}
	}
    // 这里以下是具体的插入逻辑...
    <-addRowsConcurrencyCh  // insert逻辑执行完成后，出队

	return firstErr
}

3. select操作源码分析

select请求没有区分IO协程和计算协程，因为查询请求通常不多且包很小。

3.1 用于查询并发限制的channel

lib/storage/storage.go:1097

var (
	// Limit the concurrency for TSID searches to GOMAXPROCS*2, since this operation
	// is CPU bound and sometimes disk IO bound, so there is no sense in running more
	// than GOMAXPROCS*2 concurrent goroutines for TSID searches.
	searchTSIDsConcurrencyCh = make(chan struct{}, cgroup.AvailableCPUs()*2)
)

查询的并发数限制为CPU核的两倍。

查询限制的处理代码如下：lib/storage/storage.go:1056

// searchTSIDs returns sorted TSIDs for the given tfss and the given tr.
func (s *Storage) searchTSIDs(tfss []*TagFilters, tr TimeRange, maxMetrics int, deadline uint64) ([]TSID, error) {
	// Do not cache tfss -> tsids here, since the caching is performed
	// on idb level.

	// Limit the number of concurrent goroutines that may search TSIDS in the storage.
	// This should prevent from out of memory errors and CPU trashing when too many
	// goroutines call searchTSIDs.
	select {
	case searchTSIDsConcurrencyCh <- struct{}{}:  //处理思路上与insert并发限制一样。入队成功才允许进入查询逻辑
	default:
		// Sleep for a while until giving up
		atomic.AddUint64(&s.searchTSIDsConcurrencyLimitReached, 1)
		currentTime := fasttime.UnixTimestamp()
		timeoutSecs := uint64(0)
		if currentTime < deadline {
			timeoutSecs = deadline - currentTime  //与insert的超时处理不同，每个查询可能与不同的查询超时时间
		}
		timeout := time.Second * time.Duration(timeoutSecs)
		t := timerpool.Get(timeout)
		select {
		case searchTSIDsConcurrencyCh <- struct{}{}:
			timerpool.Put(t)
		case <-t.C:
			timerpool.Put(t)
			atomic.AddUint64(&s.searchTSIDsConcurrencyLimitTimeout, 1)
			return nil, fmt.Errorf("cannot search for tsids, since more than %d concurrent searches are performed during %.3f secs; add more CPUs or reduce query load",
				cap(searchTSIDsConcurrencyCh), timeout.Seconds())
		}
	}

3.2 select协程主动让出的实现

lib/storage/search.go:188

// NextMetricBlock proceeds to the next MetricBlockRef.
func (s *Search) NextMetricBlock() bool {
	if s.err != nil {
		return false
	}
	for s.ts.NextBlock() {
		if s.loops&paceLimiterSlowIterationsMask == 0 {  //每执行4095次后，检查是否有insert协程在等待
			if err := checkSearchDeadlineAndPace(s.deadline); err != nil {
                // 如果有insert协程等待，在WaitIfNeeded()方法中用条件变量阻塞： cond.Wait()
				s.err = err
				return false
			}
		}
		s.loops++
        //...
    }
    //...
}

WaitIfNeeded()方法的实现细节：lib/pacelimiter/pacelimiter.go:43

// WaitIfNeeded blocks while the number of Inc calls is bigger than the number of Dec calls.
func (pl *PaceLimiter) WaitIfNeeded() {
	if atomic.LoadInt32(&pl.n) <= 0 {
		// Fast path - there is no need in lock.
		return
	}
	// Slow path - wait until Dec is called.
	pl.mu.Lock()
	for atomic.LoadInt32(&pl.n) > 0 {  // n代表了高优先级协程等到的个数
		pl.delaysTotal++
		pl.cond.Wait()   // 当n==0时，触发 pl.cond.Broadcast()，让低优先级的协程重新调度
	}
	pl.mu.Unlock()
}

4. 总结

关键的计算协程的数量，围绕可用的物理CPU核的数量展开。超过物理核数的协程，CPU资源只会白白浪费在协程调度器上。
区分高优先级和低优先级的协程，低优先级的协程要能够主动让出。
用一个队列来代表被调度的关键协程的数量，队列被阻塞就证明有关键协程处于未被调度的状态，这时就需要触发对应的协调机制。感觉就像在golang调度器的基础上又封装了部分能力。

不管怎么样，感谢valyala大神，后面我们就可以直接import这些代码来抄作业了。

标签：insert,协程,写入,storage,VictoriaMetrics,golang,源码,CPU,select
来源： https://www.cnblogs.com/ahfuzhang/p/15847860.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9