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

package zstd

import (
	
	rdebug 
	
)

type encJob struct {
	prefix []byte        // overlap from previous job (nil for first)
	input  []byte        // job's own input data (swapped from filling)
	last   bool          // last block of last job gets last=true
	output []byte        // compressed blocks (filled by worker)
	err    error         // encoding error
	done   chan struct{} // closed when complete
}

type jobState struct {
	jobSize     int
	overlapSize int
	filling     []byte // accumulates input up to jobSize
	nextPrefix  []byte // overlap prefix prepared for the next dispatched job

	jobSeq int // next job sequence number

	jobCh    chan *encJob // dispatch to workers
	resultCh chan *encJob // ordered results to flusher

	workerWg  sync.WaitGroup
	flusherWg sync.WaitGroup

	mu         sync.Mutex
	flushedSeq int // last flushed sequence number
	cond       *sync.Cond

	flusherErr error
	started    bool

	inputPool   sync.Pool // *[]byte buffers of jobSize cap
	outputPool  sync.Pool // *[]byte buffers for compressed output
	overlapPool sync.Pool // *[]byte buffers for overlap prefixes
}

func ( *Encoder) () {
	 := &.state.jobs
	 := .o.concurrent
	.jobCh = make(chan *encJob, )
	.resultCh = make(chan *encJob, )
	.flushedSeq = 0
	.cond = sync.NewCond(&.mu)

	// Workers borrow encoders from the shared e.encoders pool per-job.
	// Ensure the pool is initialized before any worker tries to borrow.
	.init.Do(.initialize)

	for range  {
		.workerWg.Add(1)
		go .jobWorker()
	}
	.flusherWg.Add(1)
	go .jobFlusher()
	.started = true
}

func ( *Encoder) () {
	 := &.state.jobs
	defer .workerWg.Done()
	for  := range .jobCh {
		 := <-.encoders
		.compressJob(, )
		.encoders <- 
		close(.done)
	}
}

func ( *Encoder) ( encoder,  *encJob) {
	defer func() {
		if  := recover();  != nil {
			.err = fmt.Errorf("panic in parallel job: %v", )
			rdebug.PrintStack()
		}
	}()

	if len(.prefix) > 0 {
		.ResetPrefix(.prefix)
	} else {
		.Reset(nil, false)
	}

	 := .input
	if len() == 0 && .last {
		 := .Block()
		.reset(nil)
		.last = true
		.encodeRaw(nil)
		.output = append(.output, .output...)
		return
	}

	 := .Block()
	for len() > 0 {
		 := 
		if len() > .o.blockSize {
			 = [:.o.blockSize]
		}
		 = [len():]

		.pushOffsets()
		.Encode(, )
		.last = len() == 0 && .last

		 := .encode(, .o.noEntropy, !.o.allLitEntropy)
		if  != nil {
			.err = 
			return
		}
		.output = append(.output, .output...)
		.reset(nil)
	}
}

func ( *jobState) ( int) []byte {
	if  := .inputPool.Get();  != nil {
		 := .(*[]byte)
		 := *
		if cap() >=  {
			return [:0]
		}
	}
	return make([]byte, 0, )
}

func ( *jobState) ( []byte) {
	if cap() > 0 {
		 = [:0]
		.inputPool.Put(&)
	}
}

func ( *jobState) ( int) []byte {
	if  := .outputPool.Get();  != nil {
		 := .(*[]byte)
		 := *
		if cap() >=  {
			return [:0]
		}
	}
	return make([]byte, 0, )
}

func ( *jobState) ( []byte) {
	if cap() > 0 {
		 = [:0]
		.outputPool.Put(&)
	}
}

func ( *jobState) ( int) []byte {
	if  := .overlapPool.Get();  != nil {
		 := .(*[]byte)
		 := *
		if cap() >=  {
			return [:]
		}
	}
	return make([]byte, )
}

func ( *jobState) ( []byte) {
	if cap() > 0 {
		 = [:0]
		.overlapPool.Put(&)
	}
}

func ( *Encoder) () {
	 := &.state.jobs
	defer .flusherWg.Done()
	for  := range .resultCh {
		<-.done
		// Worker has fully exited compressJob, so the prefix is no longer
		// in use. Return it to the pool regardless of outcome.
		if .prefix != nil {
			.putOverlapBuf(.prefix)
			.prefix = nil
		}
		if .err != nil {
			.mu.Lock()
			.flusherErr = .err
			.cond.Broadcast()
			.mu.Unlock()
			for range .resultCh {
			}
			return
		}
		if len(.output) > 0 {
			,  := .state.w.Write(.output)
			if  != nil {
				.mu.Lock()
				.flusherErr = 
				.cond.Broadcast()
				.mu.Unlock()
				for range .resultCh {
				}
				return
			}
			.state.nWritten += int64(len(.output))
		}
		// Return buffers to pools.
		.putInputBuf(.input)
		.putOutputBuf(.output)
		.input = nil
		.output = nil

		.mu.Lock()
		.flushedSeq++
		.cond.Broadcast()
		.mu.Unlock()
	}
}

func ( *Encoder) () {
	 := &.state.jobs
	if !.started {
		return
	}
	close(.jobCh)
	.workerWg.Wait()
	close(.resultCh)
	.flusherWg.Wait()
	.started = false
}

// waitAllJobs blocks until all dispatched jobs have been flushed.
func ( *Encoder) () {
	 := &.state.jobs
	if !.started {
		return
	}
	.mu.Lock()
	for .flushedSeq < .jobSeq && .flusherErr == nil {
		.cond.Wait()
	}
	.mu.Unlock()
}

func ( *Encoder) ( bool) error {
	 := &.state
	 := &.jobs

	.mu.Lock()
	 := .flusherErr
	.mu.Unlock()
	if  != nil {
		return 
	}

	if !.headerWritten {
		// Single-block optimization: fall through to encodeAll path.
		if  && len(.filling) > 0 && len(.filling) <= .o.blockSize {
			.current = .encodeAll(.encoder, .filling, .current[:0])
			var  int
			, .err = .w.Write(.current)
			if .err != nil {
				return .err
			}
			.nWritten += int64()
			.nInput += int64(len(.filling))
			.current = .current[:0]
			.filling = .filling[:0]
			.headerWritten = true
			.fullFrameWritten = true
			.eofWritten = true
			return nil
		}
		if  && len(.filling) == 0 && !.o.fullZero {
			.headerWritten = true
			.fullFrameWritten = true
			.eofWritten = true
			return nil
		}

		var  [maxHeaderSize]byte
		 := frameHeader{
			ContentSize:   uint64(.frameContentSize),
			WindowSize:    uint32(.encoder.WindowSize(.frameContentSize)),
			SingleSegment: false,
			Checksum:      .o.crc,
			DictID:        0,
		}
		 := .appendTo([:0])
		var  int
		, .err = .w.Write()
		if .err != nil {
			return .err
		}
		.nWritten += int64()
		.headerWritten = true
	}

	if len(.filling) == 0 && ! {
		return nil
	}

	if !.started {
		.startJobWorkers()
	}

	// Estimate output size for pooled buffer.
	 := max(len(.filling)/2, 512)

	 := &encJob{
		last:   ,
		done:   make(chan struct{}),
		output: .getOutputBuf(),
	}

	// Each job owns its prefix slice; the flusher returns it to the pool
	// after <-job.done, so workers and dispatch never share a buffer.
	if .nextPrefix != nil {
		.prefix = .nextPrefix
		.nextPrefix = nil
	}

	// Build the next job's prefix from the tail of this job's input.
	if ! && len(.filling) > 0 {
		 := min(.overlapSize, len(.filling))
		 := .getOverlapBuf()
		copy(, .filling[len(.filling)-:])
		.nextPrefix = 
	}

	// Swap filling buffer into job — zero-copy for the input data.
	.input = .filling
	.filling = .getInputBuf(.jobSize)

	.nInput += int64(len(.input))
	.jobSeq++

	if  {
		.eofWritten = true
	}

	.resultCh <- 
	.jobCh <- 

	return nil
}