Method extractShard

dsort/dsort.go:160–272

(name string, metrics *LocalExtraction)

}

func (m *Manager) extractShard(name string, metrics *LocalExtraction) func() error {
	return func() error {
		var (
			warnPossibleOOM          bool
			estimateTotalRecordsSize uint64
			phaseInfo                = &m.extractionPhase
		)

		defer phaseInfo.adjuster.releaseGoroutineSema()

		shardName := name + m.rs.Extension
		lom := cluster.AllocLOM(shardName)
		defer cluster.FreeLOM(lom)
		if err := lom.InitBck(&m.rs.Bck); err != nil {
			return err
		}
		_, local, err := lom.HrwTarget(m.smap)
		if err != nil {
			return err
		}
		if !local {
			return nil
		}
		if err = lom.Load(false /*cache it*/, false /*locked*/); err != nil {
			if cmn.IsErrObjNought(err) {
				msg := fmt.Sprintf("shard %q does not exist (is missing)", shardName)
				return m.react(m.rs.MissingShards, msg)
			}
			return err
		}

		phaseInfo.adjuster.acquireSema(lom.MpathInfo())
		if m.aborted() {
			phaseInfo.adjuster.releaseSema(lom.MpathInfo())
			return newDSortAbortedError(m.ManagerUUID)
		}
		//
		// FIXME: check capacity *prior* to starting
		//
		if cs := fs.GetCapStatus(); cs.Err != nil {
			phaseInfo.adjuster.releaseSema(lom.MpathInfo())
			return cs.Err
		}

		lom.Lock(false)
		f, err := os.Open(lom.FQN)
		if err != nil {
			phaseInfo.adjuster.releaseSema(lom.MpathInfo())
			lom.Unlock(false)
			return errors.Errorf("unable to open local file, err: %v", err)
		}
		var compressedSize int64
		if m.extractCreator.UsingCompression() {
			compressedSize = lom.SizeBytes()
		}

		expectedUncompressedSize := uint64(float64(lom.SizeBytes()) / m.avgCompressionRatio())
		toDisk := m.dsorter.preShardExtraction(expectedUncompressedSize)