Method init

dsort/manager.go:172–249

init initializes all necessary fields. PRECONDITION: `m.mu` must be locked.

(rs *ParsedRequestSpec)

//
// PRECONDITION: `m.mu` must be locked.
func (m *Manager) init(rs *ParsedRequestSpec) error {
	debug.AssertMutexLocked(&m.mu)

	m.ctx = ctx
	m.smap = m.ctx.smapOwner.Get()

	targetCount := m.smap.CountActiveTargets()

	m.rs = rs
	m.Metrics = newMetrics(rs.Description, rs.ExtendedMetrics)
	m.startShardCreation = make(chan struct{}, 1)

	m.ctx.smapOwner.Listeners().Reg(m)

	if err := m.setDSorter(); err != nil {
		return err
	}

	if err := m.dsorter.init(); err != nil {
		return err
	}

	// Set extract creator depending on extension provided by the user
	if err := m.setExtractCreator(); err != nil {
		return err
	}

	// NOTE: Total size of the records metadata can sometimes be large
	// and so this is why we need such a long timeout.
	config := cmn.GCO.Get()
	m.client = cmn.NewClient(cmn.TransportArgs{
		DialTimeout: 5 * time.Minute,
		Timeout:     30 * time.Minute,
		UseHTTPS:    config.Net.HTTP.UseHTTPS,
		SkipVerify:  config.Net.HTTP.SkipVerify,
	})

	m.fileExtension = rs.Extension
	m.received.ch = make(chan int32, 10)

	// By default we want avg compression ratio to be equal to 1
	m.compression.compressed = *atomic.NewInt64(1)
	m.compression.uncompressed = *atomic.NewInt64(1)

	// Concurrency

	// Number of goroutines should be larger than number of concurrency limit
	// but it should not be:
	// * too small - we don't want to artificially bottleneck the phases.
	// * too large - we don't want too much goroutines in the system, it can cause
	//               too much overhead on context switching and managing the goroutines.
	//               Also for large workloads goroutines can take a lot of memory.
	//
	// Coefficient for extraction should be larger and depends on target count
	// because we will skip a lot shards (which do not belong to us).
	m.extractionPhase.adjuster = newConcAdjuster(
		rs.ExtractConcMaxLimit,
		2*targetCount, /*goroutineLimitCoef*/