Method startStreamDecoder

zstd/decoder.go:655–940

Create Decoder: ASYNC: Spawn 3 go routines. 0: Read frames and decode block literals. 1: Decode sequences. 2: Execute sequences, send to output.

(ctx context.Context, r io.Reader, output chan decodeOutput)

// 1: Decode sequences.
// 2: Execute sequences, send to output.
func (d *Decoder) startStreamDecoder(ctx context.Context, r io.Reader, output chan decodeOutput) {
	defer d.streamWg.Done()
	br := readerWrapper{r: r}

	var seqDecode = make(chan *blockDec, d.o.concurrent)
	var seqExecute = make(chan *blockDec, d.o.concurrent)

	// Async 1: Decode sequences...
	go func() {
		var hist history
		var hasErr bool

		for block := range seqDecode {
			if hasErr {
				if block != nil {
					seqExecute <- block
				}
				continue
			}
			if block.async.newHist != nil {
				if debugDecoder {
					println("Async 1: new history, recent:", block.async.newHist.recentOffsets)
				}
				hist.reset()
				hist.decoders = block.async.newHist.decoders
				hist.recentOffsets = block.async.newHist.recentOffsets
				hist.windowSize = block.async.newHist.windowSize
				if block.async.newHist.dict != nil {
					hist.setDict(block.async.newHist.dict)
				}
			}
			if block.err != nil || block.Type != blockTypeCompressed {
				hasErr = block.err != nil
				seqExecute <- block
				continue
			}

			hist.decoders.literals = block.async.literals
			block.err = block.prepareSequences(block.async.seqData, &hist)
			if debugDecoder && block.err != nil {
				println("prepareSequences returned:", block.err)
			}
			hasErr = block.err != nil
			if block.err == nil {
				block.err = block.decodeSequences(&hist)
				if debugDecoder && block.err != nil {
					println("decodeSequences returned:", block.err)
				}
				hasErr = block.err != nil
				//				block.async.sequence = hist.decoders.seq[:hist.decoders.nSeqs]
				block.async.seqSize = hist.decoders.seqSize
			}
			seqExecute <- block
		}
		close(seqExecute)
		hist.reset()
	}()