Method triggerCompaction

internal/dmap/compaction.go:72–105

()

}

func (s *Service) triggerCompaction() {
	var wg sync.WaitGroup

	// NumCPU returns the number of logical CPUs usable by the current process.
	//
	// The set of available CPUs is checked by querying the operating system
	// at process startup. Changes to operating system CPU allocation after
	// process startup are not reflected.
	numWorkers := runtime.NumCPU()
	sem := semaphore.NewWeighted(int64(numWorkers))
	for partID := uint64(0); partID < s.config.PartitionCount; partID++ {
		select {
		case <-s.ctx.Done():
			break
		default:
		}

		if err := sem.Acquire(s.ctx, 1); err != nil {
			if err != context.Canceled {
				s.log.V(3).Printf("[ERROR] Failed to acquire semaphore for DMap compaction: %v", err)
			}
			continue
		}

		wg.Add(1)
		go func(id uint64) {
			defer wg.Done()
			defer sem.Release(1)
			s.doCompaction(id)
		}(partID)
	}

	wg.Wait()
}

func (s *Service) compactionWorker() {
	defer s.wg.Done()