Method DecodeConcurrent

s2/reader.go:408–662

DecodeConcurrent will decode the full stream to w. This function should not be combined with reading, seeking or other operations. Up to 'concurrent' goroutines will be used. If <= 0, runtime.NumCPU will be used. On success the number of bytes decompressed nil and is returned. This is mainly intende

(w io.Writer, concurrent int)

// On success the number of bytes decompressed nil and is returned.
// This is mainly intended for bigger streams.
func (r *Reader) DecodeConcurrent(w io.Writer, concurrent int) (written int64, err error) {
	if r.i > 0 || r.j > 0 || r.blockStart > 0 {
		return 0, errors.New("DecodeConcurrent called after ")
	}
	if concurrent <= 0 {
		concurrent = runtime.NumCPU()
	}

	// Write to output
	var errMu sync.Mutex
	var aErr error
	setErr := func(e error) (ok bool) {
		errMu.Lock()
		defer errMu.Unlock()
		if e == nil {
			return aErr == nil
		}
		if aErr == nil {
			aErr = e
		}
		return false
	}
	hasErr := func() (ok bool) {
		errMu.Lock()
		v := aErr != nil
		errMu.Unlock()
		return v
	}

	var aWritten int64
	toRead := make(chan []byte, concurrent)
	writtenBlocks := make(chan []byte, concurrent)
	queue := make(chan chan []byte, concurrent)
	reUse := make(chan chan []byte, concurrent)
	for i := 0; i < concurrent; i++ {
		toRead <- make([]byte, 0, r.maxBufSize)
		writtenBlocks <- make([]byte, 0, r.maxBufSize)
		reUse <- make(chan []byte, 1)
	}
	// Writer
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for toWrite := range queue {
			entry := <-toWrite
			reUse <- toWrite
			if hasErr() || entry == nil {
				if entry != nil {
					writtenBlocks <- entry
				}
				continue
			}
			if hasErr() {
				writtenBlocks <- entry
				continue
			}
			n, err := w.Write(entry)