Method extractLocalShards

dsort/dsort.go:276–318

extractLocalShards iterates through files local to the current target and calls ExtractShard on matching files based on the given ParsedRequestSpec.

()

// extractLocalShards iterates through files local to the current target and
// calls ExtractShard on matching files based on the given ParsedRequestSpec.
func (m *Manager) extractLocalShards() (err error) {
	phaseInfo := &m.extractionPhase

	phaseInfo.adjuster.start()
	defer phaseInfo.adjuster.stop()

	// Metrics
	metrics := m.Metrics.Extraction
	metrics.begin()
	defer metrics.finish()

	metrics.mu.Lock()
	metrics.TotalCnt = m.rs.InputFormat.Template.Count()
	metrics.mu.Unlock()

	group, ctx := errgroup.WithContext(context.Background())
	pt := m.rs.InputFormat.Template
	pt.InitIter()

ExtractAllShards:
	for name, hasNext := pt.Next(); hasNext; name, hasNext = pt.Next() {
		select {
		case <-m.listenAborted():
			group.Wait()
			return newDSortAbortedError(m.ManagerUUID)
		case <-ctx.Done():
			break ExtractAllShards // context was canceled, therefore we have an error
		default:
		}

		phaseInfo.adjuster.acquireGoroutineSema()
		group.Go(m.extractShard(name, metrics))
	}
	if err := group.Wait(); err != nil {
		return err
	}

	// We will no longer reserve any memory
	m.dsorter.postExtraction()

	m.incrementRef(int64(m.recManager.Records.TotalObjectCount()))
	return nil
}

func (m *Manager) createShard(s *extract.Shard) (err error) {
	var (