package scheduler

import (
	"gitee.com/h79/goutils/common/system"
	"sync/atomic"
)

type IdleHandler interface {
	Idle(num int32)
}

// IdleFunc idle working number
type IdleFunc func(num int32)

func (fn IdleFunc) Idle(num int32) {
	fn(num)
}

type Pool struct {
	jobs        chan *Job
	workers     chan *jobWorker
	stop        chan bool
	numWorkers  int32 //总数
	runWorkers  int32 //正常运行
	idleWorking int32 //空闲worker数
	idleHandler IdleHandler
	running     system.RunningCheck
}

func NewPool(numWorkers, jobQueueLen int) *Pool {
	if numWorkers <= 0 || jobQueueLen <= 0 {
		panic("length is zero")
	}
	var num = int32(numWorkers)
	pool := &Pool{
		jobs:        make(chan *Job, jobQueueLen),
		workers:     make(chan *jobWorker, numWorkers),
		stop:        make(chan bool),
		numWorkers:  num,
		runWorkers:  num,
		idleWorking: num,
	}
	pool.createWorkers(numWorkers)
	pool.Run()
	return pool
}

func (p *Pool) SetIdleHandler(fn IdleHandler) *Pool {
	p.idleHandler = fn
	return p
}

func (p *Pool) WithIdleFunc(fn func(num int32)) *Pool {
	return p.SetIdleHandler(IdleFunc(fn))
}

func (p *Pool) AddJob(job *Job) {
	if job == nil {
		return
	}
	ww := atomic.LoadInt32(&p.runWorkers)
	if ww <= p.numWorkers/2 {
		// 中间有原因，线程池中的线程，由于某些原因，退出了，需要再启动
		p.createWorkers(int(p.numWorkers - ww))
	}
	p.jobs <- job
}

func (p *Pool) Run() {
	p.running.TryGoRunning(p.dispatch)
}

func (p *Pool) HasWorking() bool {
	return atomic.LoadInt32(&p.idleWorking) != p.numWorkers
}

func (p *Pool) Release() {
	p.stop <- true
	<-p.stop
	p.idleHandle(p.numWorkers)
}

func (p *Pool) createWorkers(num int) {
	for i := 0; i < num; i++ {
		worker := newWorker(p)
		worker.start()
	}
}

func (p *Pool) idleHandle(idleWork int32) {
	if p.idleHandler != nil {
		p.idleHandler.Idle(idleWork)
	}
}

func (p *Pool) workerRunning() {
	atomic.AddInt32(&p.runWorkers, 1)
}

func (p *Pool) workerQuit() {
	atomic.AddInt32(&p.runWorkers, -1)
}

func (p *Pool) workerIdle() {
	p.idleHandle(atomic.AddInt32(&p.idleWorking, 1))
}

func (p *Pool) dispatch() {
	defer p.idleHandle(p.numWorkers)
	var exit = system.Exit()
	for {
		select {
		case <-exit.Done():
			p.stop <- true

		case job := <-p.jobs:
			worker := <-p.workers

			p.idleHandle(atomic.AddInt32(&p.idleWorking, -1))
			worker.timeout = job.Timeout
			worker.job <- job

		case <-p.stop:
			for i := 0; i < cap(p.workers); i++ {
				worker := <-p.workers
				worker.stop <- true
				<-worker.stop
			}
			atomic.StoreInt32(&p.idleWorking, p.numWorkers)
			p.stop <- true

			return

		case <-system.Closed():
			p.stop <- true
		}
	}
}