Method start

dsort/dsort_mem.go:201–246

()

}

func (ds *dsorterMem) start() error {
	// Requests are usually small packets, no more 1KB that is why we want to
	// utilize intraControl network.
	config := cmn.GCO.Get()
	reqNetwork := cmn.NetIntraControl
	// Responses to the other targets are objects that is why we want to use
	// intraData network.
	respNetwork := cmn.NetIntraData

	client := transport.NewIntraDataClient()

	streamMultiplier := config.DSort.SbundleMult
	if ds.m.rs.StreamMultiplier != 0 {
		streamMultiplier = ds.m.rs.StreamMultiplier
	}
	trname := fmt.Sprintf(recvReqStreamNameFmt, ds.m.ManagerUUID)
	reqSbArgs := bundle.Args{
		Multiplier: streamMultiplier,
		Net:        reqNetwork,
		Trname:     trname,
		Ntype:      cluster.Targets,
	}
	if err := transport.HandleObjStream(trname, ds.makeRecvRequestFunc()); err != nil {
		return errors.WithStack(err)
	}

	trname = fmt.Sprintf(recvRespStreamNameFmt, ds.m.ManagerUUID)
	respSbArgs := bundle.Args{
		Multiplier: streamMultiplier,
		Net:        respNetwork,
		Trname:     trname,
		Ntype:      cluster.Targets,
		Extra: &transport.Extra{
			Compression: config.DSort.Compression,
			Config:      config,
			MMSA:        mm,
		},
	}
	if err := transport.HandleObjStream(trname, ds.makeRecvResponseFunc()); err != nil {
		return errors.WithStack(err)
	}

	ds.streams.builder = bundle.NewStreams(ds.m.ctx.smapOwner, ds.m.ctx.node, client, reqSbArgs)
	ds.streams.records = bundle.NewStreams(ds.m.ctx.smapOwner, ds.m.ctx.node, client, respSbArgs)
	return nil
}

func (ds *dsorterMem) cleanupStreams() (err error) {
	if !ds.streams.cleanupDone.CAS(false, true) {