cubefs

1
// Copyright 2019+ Klaus Post. All rights reserved.
2
// License information can be found in the LICENSE file.
3
// Based on work by Yann Collet, released under BSD License.
4

5
package zstd
6

7
import (
8
	"bytes"
9
	"context"
10
	"encoding/binary"
11
	"io"
12
	"sync"
13

14
	"github.com/klauspost/compress/zstd/internal/xxhash"
15
)
16

17
// Decoder provides decoding of zstandard streams.
18
// The decoder has been designed to operate without allocations after a warmup.
19
// This means that you should store the decoder for best performance.
20
// To re-use a stream decoder, use the Reset(r io.Reader) error to switch to another stream.
21
// A decoder can safely be re-used even if the previous stream failed.
22
// To release the resources, you must call the Close() function on a decoder.
23
type Decoder struct {
24
	o decoderOptions
25

26
	// Unreferenced decoders, ready for use.
27
	decoders chan *blockDec
28

29
	// Current read position used for Reader functionality.
30
	current decoderState
31

32
	// sync stream decoding
33
	syncStream struct {
34
		decodedFrame uint64
35
		br           readerWrapper
36
		enabled      bool
37
		inFrame      bool
38
	}
39

40
	frame *frameDec
41

42
	// Custom dictionaries.
43
	// Always uses copies.
44
	dicts map[uint32]dict
45

46
	// streamWg is the waitgroup for all streams
47
	streamWg sync.WaitGroup
48
}
49

50
// decoderState is used for maintaining state when the decoder
51
// is used for streaming.
52
type decoderState struct {
53
	// current block being written to stream.
54
	decodeOutput
55

56
	// output in order to be written to stream.
57
	output chan decodeOutput
58

59
	// cancel remaining output.
60
	cancel context.CancelFunc
61

62
	// crc of current frame
63
	crc *xxhash.Digest
64

65
	flushed bool
66
}
67

68
var (
69
	// Check the interfaces we want to support.
70
	_ = io.WriterTo(&Decoder{})
71
	_ = io.Reader(&Decoder{})
72
)
73

74
// NewReader creates a new decoder.
75
// A nil Reader can be provided in which case Reset can be used to start a decode.
76
//
77
// A Decoder can be used in two modes:
78
//
79
// 1) As a stream, or
80
// 2) For stateless decoding using DecodeAll.
81
//
82
// Only a single stream can be decoded concurrently, but the same decoder
83
// can run multiple concurrent stateless decodes. It is even possible to
84
// use stateless decodes while a stream is being decoded.
85
//
86
// The Reset function can be used to initiate a new stream, which is will considerably
87
// reduce the allocations normally caused by NewReader.
88
func NewReader(r io.Reader, opts ...DOption) (*Decoder, error) {
89
	initPredefined()
90
	var d Decoder
91
	d.o.setDefault()
92
	for _, o := range opts {
93
		err := o(&d.o)
94
		if err != nil {
95
			return nil, err
96
		}
97
	}
98
	d.current.crc = xxhash.New()
99
	d.current.flushed = true
100

101
	if r == nil {
102
		d.current.err = ErrDecoderNilInput
103
	}
104

105
	// Transfer option dicts.
106
	d.dicts = make(map[uint32]dict, len(d.o.dicts))
107
	for _, dc := range d.o.dicts {
108
		d.dicts[dc.id] = dc
109
	}
110
	d.o.dicts = nil
111

112
	// Create decoders
113
	d.decoders = make(chan *blockDec, d.o.concurrent)
114
	for i := 0; i < d.o.concurrent; i++ {
115
		dec := newBlockDec(d.o.lowMem)
116
		dec.localFrame = newFrameDec(d.o)
117
		d.decoders <- dec
118
	}
119

120
	if r == nil {
121
		return &d, nil
122
	}
123
	return &d, d.Reset(r)
124
}
125

126
// Read bytes from the decompressed stream into p.
127
// Returns the number of bytes written and any error that occurred.
128
// When the stream is done, io.EOF will be returned.
129
func (d *Decoder) Read(p []byte) (int, error) {
130
	var n int
131
	for {
132
		if len(d.current.b) > 0 {
133
			filled := copy(p, d.current.b)
134
			p = p[filled:]
135
			d.current.b = d.current.b[filled:]
136
			n += filled
137
		}
138
		if len(p) == 0 {
139
			break
140
		}
141
		if len(d.current.b) == 0 {
142
			// We have an error and no more data
143
			if d.current.err != nil {
144
				break
145
			}
146
			if !d.nextBlock(n == 0) {
147
				return n, d.current.err
148
			}
149
		}
150
	}
151
	if len(d.current.b) > 0 {
152
		if debugDecoder {
153
			println("returning", n, "still bytes left:", len(d.current.b))
154
		}
155
		// Only return error at end of block
156
		return n, nil
157
	}
158
	if d.current.err != nil {
159
		d.drainOutput()
160
	}
161
	if debugDecoder {
162
		println("returning", n, d.current.err, len(d.decoders))
163
	}
164
	return n, d.current.err
165
}
166

167
// Reset will reset the decoder the supplied stream after the current has finished processing.
168
// Note that this functionality cannot be used after Close has been called.
169
// Reset can be called with a nil reader to release references to the previous reader.
170
// After being called with a nil reader, no other operations than Reset or DecodeAll or Close
171
// should be used.
172
func (d *Decoder) Reset(r io.Reader) error {
173
	if d.current.err == ErrDecoderClosed {
174
		return d.current.err
175
	}
176

177
	d.drainOutput()
178

179
	d.syncStream.br.r = nil
180
	if r == nil {
181
		d.current.err = ErrDecoderNilInput
182
		if len(d.current.b) > 0 {
183
			d.current.b = d.current.b[:0]
184
		}
185
		d.current.flushed = true
186
		return nil
187
	}
188

189
	// If bytes buffer and < 5MB, do sync decoding anyway.
190
	if bb, ok := r.(byter); ok && bb.Len() < 5<<20 {
191
		bb2 := bb
192
		if debugDecoder {
193
			println("*bytes.Buffer detected, doing sync decode, len:", bb.Len())
194
		}
195
		b := bb2.Bytes()
196
		var dst []byte
197
		if cap(d.current.b) > 0 {
198
			dst = d.current.b
199
		}
200

201
		dst, err := d.DecodeAll(b, dst[:0])
202
		if err == nil {
203
			err = io.EOF
204
		}
205
		d.current.b = dst
206
		d.current.err = err
207
		d.current.flushed = true
208
		if debugDecoder {
209
			println("sync decode to", len(dst), "bytes, err:", err)
210
		}
211
		return nil
212
	}
213
	// Remove current block.
214
	d.stashDecoder()
215
	d.current.decodeOutput = decodeOutput{}
216
	d.current.err = nil
217
	d.current.flushed = false
218
	d.current.d = nil
219

220
	// Ensure no-one else is still running...
221
	d.streamWg.Wait()
222
	if d.frame == nil {
223
		d.frame = newFrameDec(d.o)
224
	}
225

226
	if d.o.concurrent == 1 {
227
		return d.startSyncDecoder(r)
228
	}
229

230
	d.current.output = make(chan decodeOutput, d.o.concurrent)
231
	ctx, cancel := context.WithCancel(context.Background())
232
	d.current.cancel = cancel
233
	d.streamWg.Add(1)
234
	go d.startStreamDecoder(ctx, r, d.current.output)
235

236
	return nil
237
}
238

239
// drainOutput will drain the output until errEndOfStream is sent.
240
func (d *Decoder) drainOutput() {
241
	if d.current.cancel != nil {
242
		if debugDecoder {
243
			println("cancelling current")
244
		}
245
		d.current.cancel()
246
		d.current.cancel = nil
247
	}
248
	if d.current.d != nil {
249
		if debugDecoder {
250
			printf("re-adding current decoder %p, decoders: %d", d.current.d, len(d.decoders))
251
		}
252
		d.decoders <- d.current.d
253
		d.current.d = nil
254
		d.current.b = nil
255
	}
256
	if d.current.output == nil || d.current.flushed {
257
		println("current already flushed")
258
		return
259
	}
260
	for v := range d.current.output {
261
		if v.d != nil {
262
			if debugDecoder {
263
				printf("re-adding decoder %p", v.d)
264
			}
265
			d.decoders <- v.d
266
		}
267
	}
268
	d.current.output = nil
269
	d.current.flushed = true
270
}
271

272
// WriteTo writes data to w until there's no more data to write or when an error occurs.
273
// The return value n is the number of bytes written.
274
// Any error encountered during the write is also returned.
275
func (d *Decoder) WriteTo(w io.Writer) (int64, error) {
276
	var n int64
277
	for {
278
		if len(d.current.b) > 0 {
279
			n2, err2 := w.Write(d.current.b)
280
			n += int64(n2)
281
			if err2 != nil && (d.current.err == nil || d.current.err == io.EOF) {
282
				d.current.err = err2
283
			} else if n2 != len(d.current.b) {
284
				d.current.err = io.ErrShortWrite
285
			}
286
		}
287
		if d.current.err != nil {
288
			break
289
		}
290
		d.nextBlock(true)
291
	}
292
	err := d.current.err
293
	if err != nil {
294
		d.drainOutput()
295
	}
296
	if err == io.EOF {
297
		err = nil
298
	}
299
	return n, err
300
}
301

302
// DecodeAll allows stateless decoding of a blob of bytes.
303
// Output will be appended to dst, so if the destination size is known
304
// you can pre-allocate the destination slice to avoid allocations.
305
// DecodeAll can be used concurrently.
306
// The Decoder concurrency limits will be respected.
307
func (d *Decoder) DecodeAll(input, dst []byte) ([]byte, error) {
308
	if d.decoders == nil {
309
		return dst, ErrDecoderClosed
310
	}
311

312
	// Grab a block decoder and frame decoder.
313
	block := <-d.decoders
314
	frame := block.localFrame
315
	defer func() {
316
		if debugDecoder {
317
			printf("re-adding decoder: %p", block)
318
		}
319
		frame.rawInput = nil
320
		frame.bBuf = nil
321
		if frame.history.decoders.br != nil {
322
			frame.history.decoders.br.in = nil
323
		}
324
		d.decoders <- block
325
	}()
326
	frame.bBuf = input
327

328
	for {
329
		frame.history.reset()
330
		err := frame.reset(&frame.bBuf)
331
		if err != nil {
332
			if err == io.EOF {
333
				if debugDecoder {
334
					println("frame reset return EOF")
335
				}
336
				return dst, nil
337
			}
338
			return dst, err
339
		}
340
		if frame.DictionaryID != nil {
341
			dict, ok := d.dicts[*frame.DictionaryID]
342
			if !ok {
343
				return nil, ErrUnknownDictionary
344
			}
345
			if debugDecoder {
346
				println("setting dict", frame.DictionaryID)
347
			}
348
			frame.history.setDict(&dict)
349
		}
350

351
		if frame.FrameContentSize > d.o.maxDecodedSize-uint64(len(dst)) {
352
			return dst, ErrDecoderSizeExceeded
353
		}
354
		if frame.FrameContentSize > 0 && frame.FrameContentSize < 1<<30 {
355
			// Never preallocate more than 1 GB up front.
356
			if cap(dst)-len(dst) < int(frame.FrameContentSize) {
357
				dst2 := make([]byte, len(dst), len(dst)+int(frame.FrameContentSize))
358
				copy(dst2, dst)
359
				dst = dst2
360
			}
361
		}
362
		if cap(dst) == 0 {
363
			// Allocate len(input) * 2 by default if nothing is provided
364
			// and we didn't get frame content size.
365
			size := len(input) * 2
366
			// Cap to 1 MB.
367
			if size > 1<<20 {
368
				size = 1 << 20
369
			}
370
			if uint64(size) > d.o.maxDecodedSize {
371
				size = int(d.o.maxDecodedSize)
372
			}
373
			dst = make([]byte, 0, size)
374
		}
375

376
		dst, err = frame.runDecoder(dst, block)
377
		if err != nil {
378
			return dst, err
379
		}
380
		if len(frame.bBuf) == 0 {
381
			if debugDecoder {
382
				println("frame dbuf empty")
383
			}
384
			break
385
		}
386
	}
387
	return dst, nil
388
}
389

390
// nextBlock returns the next block.
391
// If an error occurs d.err will be set.
392
// Optionally the function can block for new output.
393
// If non-blocking mode is used the returned boolean will be false
394
// if no data was available without blocking.
395
func (d *Decoder) nextBlock(blocking bool) (ok bool) {
396
	if d.current.err != nil {
397
		// Keep error state.
398
		return false
399
	}
400
	d.current.b = d.current.b[:0]
401

402
	// SYNC:
403
	if d.syncStream.enabled {
404
		if !blocking {
405
			return false
406
		}
407
		ok = d.nextBlockSync()
408
		if !ok {
409
			d.stashDecoder()
410
		}
411
		return ok
412
	}
413

414
	//ASYNC:
415
	d.stashDecoder()
416
	if blocking {
417
		d.current.decodeOutput, ok = <-d.current.output
418
	} else {
419
		select {
420
		case d.current.decodeOutput, ok = <-d.current.output:
421
		default:
422
			return false
423
		}
424
	}
425
	if !ok {
426
		// This should not happen, so signal error state...
427
		d.current.err = io.ErrUnexpectedEOF
428
		return false
429
	}
430
	next := d.current.decodeOutput
431
	if next.d != nil && next.d.async.newHist != nil {
432
		d.current.crc.Reset()
433
	}
434
	if debugDecoder {
435
		var tmp [4]byte
436
		binary.LittleEndian.PutUint32(tmp[:], uint32(xxhash.Sum64(next.b)))
437
		println("got", len(d.current.b), "bytes, error:", d.current.err, "data crc:", tmp)
438
	}
439

440
	if len(next.b) > 0 {
441
		n, err := d.current.crc.Write(next.b)
442
		if err == nil {
443
			if n != len(next.b) {
444
				d.current.err = io.ErrShortWrite
445
			}
446
		}
447
	}
448
	if next.err == nil && next.d != nil && len(next.d.checkCRC) != 0 {
449
		got := d.current.crc.Sum64()
450
		var tmp [4]byte
451
		binary.LittleEndian.PutUint32(tmp[:], uint32(got))
452
		if !bytes.Equal(tmp[:], next.d.checkCRC) && !ignoreCRC {
453
			if debugDecoder {
454
				println("CRC Check Failed:", tmp[:], " (got) !=", next.d.checkCRC, "(on stream)")
455
			}
456
			d.current.err = ErrCRCMismatch
457
		} else {
458
			if debugDecoder {
459
				println("CRC ok", tmp[:])
460
			}
461
		}
462
	}
463

464
	return true
465
}
466

467
func (d *Decoder) nextBlockSync() (ok bool) {
468
	if d.current.d == nil {
469
		d.current.d = <-d.decoders
470
	}
471
	for len(d.current.b) == 0 {
472
		if !d.syncStream.inFrame {
473
			d.frame.history.reset()
474
			d.current.err = d.frame.reset(&d.syncStream.br)
475
			if d.current.err != nil {
476
				return false
477
			}
478
			if d.frame.DictionaryID != nil {
479
				dict, ok := d.dicts[*d.frame.DictionaryID]
480
				if !ok {
481
					d.current.err = ErrUnknownDictionary
482
					return false
483
				} else {
484
					d.frame.history.setDict(&dict)
485
				}
486
			}
487
			if d.frame.WindowSize > d.o.maxDecodedSize || d.frame.WindowSize > d.o.maxWindowSize {
488
				d.current.err = ErrDecoderSizeExceeded
489
				return false
490
			}
491

492
			d.syncStream.decodedFrame = 0
493
			d.syncStream.inFrame = true
494
		}
495
		d.current.err = d.frame.next(d.current.d)
496
		if d.current.err != nil {
497
			return false
498
		}
499
		d.frame.history.ensureBlock()
500
		if debugDecoder {
501
			println("History trimmed:", len(d.frame.history.b), "decoded already:", d.syncStream.decodedFrame)
502
		}
503
		histBefore := len(d.frame.history.b)
504
		d.current.err = d.current.d.decodeBuf(&d.frame.history)
505

506
		if d.current.err != nil {
507
			println("error after:", d.current.err)
508
			return false
509
		}
510
		d.current.b = d.frame.history.b[histBefore:]
511
		if debugDecoder {
512
			println("history after:", len(d.frame.history.b))
513
		}
514

515
		// Check frame size (before CRC)
516
		d.syncStream.decodedFrame += uint64(len(d.current.b))
517
		if d.frame.FrameContentSize > 0 && d.syncStream.decodedFrame > d.frame.FrameContentSize {
518
			if debugDecoder {
519
				printf("DecodedFrame (%d) > FrameContentSize (%d)\n", d.syncStream.decodedFrame, d.frame.FrameContentSize)
520
			}
521
			d.current.err = ErrFrameSizeExceeded
522
			return false
523
		}
524

525
		// Check FCS
526
		if d.current.d.Last && d.frame.FrameContentSize > 0 && d.syncStream.decodedFrame != d.frame.FrameContentSize {
527
			if debugDecoder {
528
				printf("DecodedFrame (%d) != FrameContentSize (%d)\n", d.syncStream.decodedFrame, d.frame.FrameContentSize)
529
			}
530
			d.current.err = ErrFrameSizeMismatch
531
			return false
532
		}
533

534
		// Update/Check CRC
535
		if d.frame.HasCheckSum {
536
			d.frame.crc.Write(d.current.b)
537
			if d.current.d.Last {
538
				d.current.err = d.frame.checkCRC()
539
				if d.current.err != nil {
540
					println("CRC error:", d.current.err)
541
					return false
542
				}
543
			}
544
		}
545
		d.syncStream.inFrame = !d.current.d.Last
546
	}
547
	return true
548
}
549

550
func (d *Decoder) stashDecoder() {
551
	if d.current.d != nil {
552
		if debugDecoder {
553
			printf("re-adding current decoder %p", d.current.d)
554
		}
555
		d.decoders <- d.current.d
556
		d.current.d = nil
557
	}
558
}
559

560
// Close will release all resources.
561
// It is NOT possible to reuse the decoder after this.
562
func (d *Decoder) Close() {
563
	if d.current.err == ErrDecoderClosed {
564
		return
565
	}
566
	d.drainOutput()
567
	if d.current.cancel != nil {
568
		d.current.cancel()
569
		d.streamWg.Wait()
570
		d.current.cancel = nil
571
	}
572
	if d.decoders != nil {
573
		close(d.decoders)
574
		for dec := range d.decoders {
575
			dec.Close()
576
		}
577
		d.decoders = nil
578
	}
579
	if d.current.d != nil {
580
		d.current.d.Close()
581
		d.current.d = nil
582
	}
583
	d.current.err = ErrDecoderClosed
584
}
585

586
// IOReadCloser returns the decoder as an io.ReadCloser for convenience.
587
// Any changes to the decoder will be reflected, so the returned ReadCloser
588
// can be reused along with the decoder.
589
// io.WriterTo is also supported by the returned ReadCloser.
590
func (d *Decoder) IOReadCloser() io.ReadCloser {
591
	return closeWrapper{d: d}
592
}
593

594
// closeWrapper wraps a function call as a closer.
595
type closeWrapper struct {
596
	d *Decoder
597
}
598

599
// WriteTo forwards WriteTo calls to the decoder.
600
func (c closeWrapper) WriteTo(w io.Writer) (n int64, err error) {
601
	return c.d.WriteTo(w)
602
}
603

604
// Read forwards read calls to the decoder.
605
func (c closeWrapper) Read(p []byte) (n int, err error) {
606
	return c.d.Read(p)
607
}
608

609
// Close closes the decoder.
610
func (c closeWrapper) Close() error {
611
	c.d.Close()
612
	return nil
613
}
614

615
type decodeOutput struct {
616
	d   *blockDec
617
	b   []byte
618
	err error
619
}
620

621
func (d *Decoder) startSyncDecoder(r io.Reader) error {
622
	d.frame.history.reset()
623
	d.syncStream.br = readerWrapper{r: r}
624
	d.syncStream.inFrame = false
625
	d.syncStream.enabled = true
626
	d.syncStream.decodedFrame = 0
627
	return nil
628
}
629

630
// Create Decoder:
631
// ASYNC:
632
// Spawn 4 go routines.
633
// 0: Read frames and decode blocks.
634
// 1: Decode block and literals. Receives hufftree and seqdecs, returns seqdecs and huff tree.
635
// 2: Wait for recentOffsets if needed. Decode sequences, send recentOffsets.
636
// 3: Wait for stream history, execute sequences, send stream history.
637
func (d *Decoder) startStreamDecoder(ctx context.Context, r io.Reader, output chan decodeOutput) {
638
	defer d.streamWg.Done()
639
	br := readerWrapper{r: r}
640

641
	var seqPrepare = make(chan *blockDec, d.o.concurrent)
642
	var seqDecode = make(chan *blockDec, d.o.concurrent)
643
	var seqExecute = make(chan *blockDec, d.o.concurrent)
644

645
	// Async 1: Prepare blocks...
646
	go func() {
647
		var hist history
648
		var hasErr bool
649
		for block := range seqPrepare {
650
			if hasErr {
651
				if block != nil {
652
					seqDecode <- block
653
				}
654
				continue
655
			}
656
			if block.async.newHist != nil {
657
				if debugDecoder {
658
					println("Async 1: new history")
659
				}
660
				hist.reset()
661
				if block.async.newHist.dict != nil {
662
					hist.setDict(block.async.newHist.dict)
663
				}
664
			}
665
			if block.err != nil || block.Type != blockTypeCompressed {
666
				hasErr = block.err != nil
667
				seqDecode <- block
668
				continue
669
			}
670

671
			remain, err := block.decodeLiterals(block.data, &hist)
672
			block.err = err
673
			hasErr = block.err != nil
674
			if err == nil {
675
				block.async.literals = hist.decoders.literals
676
				block.async.seqData = remain
677
			} else if debugDecoder {
678
				println("decodeLiterals error:", err)
679
			}
680
			seqDecode <- block
681
		}
682
		close(seqDecode)
683
	}()
684

685
	// Async 2: Decode sequences...
686
	go func() {
687
		var hist history
688
		var hasErr bool
689

690
		for block := range seqDecode {
691
			if hasErr {
692
				if block != nil {
693
					seqExecute <- block
694
				}
695
				continue
696
			}
697
			if block.async.newHist != nil {
698
				if debugDecoder {
699
					println("Async 2: new history, recent:", block.async.newHist.recentOffsets)
700
				}
701
				hist.decoders = block.async.newHist.decoders
702
				hist.recentOffsets = block.async.newHist.recentOffsets
703
				if block.async.newHist.dict != nil {
704
					hist.setDict(block.async.newHist.dict)
705
				}
706
			}
707
			if block.err != nil || block.Type != blockTypeCompressed {
708
				hasErr = block.err != nil
709
				seqExecute <- block
710
				continue
711
			}
712

713
			hist.decoders.literals = block.async.literals
714
			block.err = block.prepareSequences(block.async.seqData, &hist)
715
			if debugDecoder && block.err != nil {
716
				println("prepareSequences returned:", block.err)
717
			}
718
			hasErr = block.err != nil
719
			if block.err == nil {
720
				block.err = block.decodeSequences(&hist)
721
				if debugDecoder && block.err != nil {
722
					println("decodeSequences returned:", block.err)
723
				}
724
				hasErr = block.err != nil
725
				//				block.async.sequence = hist.decoders.seq[:hist.decoders.nSeqs]
726
				block.async.seqSize = hist.decoders.seqSize
727
			}
728
			seqExecute <- block
729
		}
730
		close(seqExecute)
731
	}()
732

733
	var wg sync.WaitGroup
734
	wg.Add(1)
735

736
	// Async 3: Execute sequences...
737
	frameHistCache := d.frame.history.b
738
	go func() {
739
		var hist history
740
		var decodedFrame uint64
741
		var fcs uint64
742
		var hasErr bool
743
		for block := range seqExecute {
744
			out := decodeOutput{err: block.err, d: block}
745
			if block.err != nil || hasErr {
746
				hasErr = true
747
				output <- out
748
				continue
749
			}
750
			if block.async.newHist != nil {
751
				if debugDecoder {
752
					println("Async 3: new history")
753
				}
754
				hist.windowSize = block.async.newHist.windowSize
755
				hist.allocFrameBuffer = block.async.newHist.allocFrameBuffer
756
				if block.async.newHist.dict != nil {
757
					hist.setDict(block.async.newHist.dict)
758
				}
759

760
				if cap(hist.b) < hist.allocFrameBuffer {
761
					if cap(frameHistCache) >= hist.allocFrameBuffer {
762
						hist.b = frameHistCache
763
					} else {
764
						hist.b = make([]byte, 0, hist.allocFrameBuffer)
765
						println("Alloc history sized", hist.allocFrameBuffer)
766
					}
767
				}
768
				hist.b = hist.b[:0]
769
				fcs = block.async.fcs
770
				decodedFrame = 0
771
			}
772
			do := decodeOutput{err: block.err, d: block}
773
			switch block.Type {
774
			case blockTypeRLE:
775
				if debugDecoder {
776
					println("add rle block length:", block.RLESize)
777
				}
778

779
				if cap(block.dst) < int(block.RLESize) {
780
					if block.lowMem {
781
						block.dst = make([]byte, block.RLESize)
782
					} else {
783
						block.dst = make([]byte, maxBlockSize)
784
					}
785
				}
786
				block.dst = block.dst[:block.RLESize]
787
				v := block.data[0]
788
				for i := range block.dst {
789
					block.dst[i] = v
790
				}
791
				hist.append(block.dst)
792
				do.b = block.dst
793
			case blockTypeRaw:
794
				if debugDecoder {
795
					println("add raw block length:", len(block.data))
796
				}
797
				hist.append(block.data)
798
				do.b = block.data
799
			case blockTypeCompressed:
800
				if debugDecoder {
801
					println("execute with history length:", len(hist.b), "window:", hist.windowSize)
802
				}
803
				hist.decoders.seqSize = block.async.seqSize
804
				hist.decoders.literals = block.async.literals
805
				do.err = block.executeSequences(&hist)
806
				hasErr = do.err != nil
807
				if debugDecoder && hasErr {
808
					println("executeSequences returned:", do.err)
809
				}
810
				do.b = block.dst
811
			}
812
			if !hasErr {
813
				decodedFrame += uint64(len(do.b))
814
				if fcs > 0 && decodedFrame > fcs {
815
					println("fcs exceeded", block.Last, fcs, decodedFrame)
816
					do.err = ErrFrameSizeExceeded
817
					hasErr = true
818
				} else if block.Last && fcs > 0 && decodedFrame != fcs {
819
					do.err = ErrFrameSizeMismatch
820
					hasErr = true
821
				} else {
822
					if debugDecoder {
823
						println("fcs ok", block.Last, fcs, decodedFrame)
824
					}
825
				}
826
			}
827
			output <- do
828
		}
829
		close(output)
830
		frameHistCache = hist.b
831
		wg.Done()
832
		if debugDecoder {
833
			println("decoder goroutines finished")
834
		}
835
	}()
836

837
decodeStream:
838
	for {
839
		frame := d.frame
840
		if debugDecoder {
841
			println("New frame...")
842
		}
843
		var historySent bool
844
		frame.history.reset()
845
		err := frame.reset(&br)
846
		if debugDecoder && err != nil {
847
			println("Frame decoder returned", err)
848
		}
849
		if err == nil && frame.DictionaryID != nil {
850
			dict, ok := d.dicts[*frame.DictionaryID]
851
			if !ok {
852
				err = ErrUnknownDictionary
853
			} else {
854
				frame.history.setDict(&dict)
855
			}
856
		}
857
		if err == nil && d.frame.WindowSize > d.o.maxWindowSize {
858
			err = ErrDecoderSizeExceeded
859
		}
860
		if err != nil {
861
			select {
862
			case <-ctx.Done():
863
			case dec := <-d.decoders:
864
				dec.sendErr(err)
865
				seqPrepare <- dec
866
			}
867
			break decodeStream
868
		}
869

870
		// Go through all blocks of the frame.
871
		for {
872
			var dec *blockDec
873
			select {
874
			case <-ctx.Done():
875
				break decodeStream
876
			case dec = <-d.decoders:
877
				// Once we have a decoder, we MUST return it.
878
			}
879
			err := frame.next(dec)
880
			if !historySent {
881
				h := frame.history
882
				if debugDecoder {
883
					println("Alloc History:", h.allocFrameBuffer)
884
				}
885
				dec.async.newHist = &h
886
				dec.async.fcs = frame.FrameContentSize
887
				historySent = true
888
			} else {
889
				dec.async.newHist = nil
890
			}
891
			if debugDecoder && err != nil {
892
				println("next block returned error:", err)
893
			}
894
			dec.err = err
895
			dec.checkCRC = nil
896
			if dec.Last && frame.HasCheckSum && err == nil {
897
				crc, err := frame.rawInput.readSmall(4)
898
				if err != nil {
899
					println("CRC missing?", err)
900
					dec.err = err
901
				}
902
				var tmp [4]byte
903
				copy(tmp[:], crc)
904
				dec.checkCRC = tmp[:]
905
				if debugDecoder {
906
					println("found crc to check:", dec.checkCRC)
907
				}
908
			}
909
			err = dec.err
910
			last := dec.Last
911
			seqPrepare <- dec
912
			if err != nil {
913
				break decodeStream
914
			}
915
			if last {
916
				break
917
			}
918
		}
919
	}
920
	close(seqPrepare)
921
	wg.Wait()
922
	d.frame.history.b = frameHistCache
923
}
924