binlog-parser/vendor/b612.me/starmap/stack.go

package starmap

import (
	"errors"
	"fmt"
	"io"
	"os"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

type StarStack struct {
	datas   []interface{}
	pStart  uint64
	pEnd    uint64
	cap     uint64
	isClose atomic.Value
	rmu     sync.Mutex
	wmu     sync.Mutex
}

func NewStarStack(cap uint64) *StarStack {
	rtnBuffer := new(StarStack)
	rtnBuffer.cap = cap
	rtnBuffer.datas = make([]interface{}, cap)
	rtnBuffer.isClose.Store(false)
	return rtnBuffer
}

func (star *StarStack) init() {
	star.cap = 1024
	star.datas = make([]interface{}, star.cap)
	star.isClose.Store(false)
}

func (star *StarStack) Free() uint64 {
	return star.cap - star.Len()
}

func (star *StarStack) Cap() uint64 {
	return star.cap
}

func (star *StarStack) Len() uint64 {
	if star.pEnd >= star.pStart {
		return star.pEnd - star.pStart
	}
	return star.pEnd - star.pStart + star.cap
}

func (star *StarStack) PopNoWait() (interface{}, error) {
	if star.isClose.Load() == nil {
		star.init()
	}
	if star.isClose.Load().(bool) {
		return 0, io.EOF
	}
	if star.Len() == 0 {
		return 0, os.ErrNotExist
	}
	nowPtr := star.pStart
	nextPtr := star.pStart + 1
	if nextPtr >= star.cap {
		nextPtr = 0
	}
	data := star.datas[nowPtr]
	ok := atomic.CompareAndSwapUint64(&star.pStart, nowPtr, nextPtr)
	if !ok {
		return 0, os.ErrInvalid
	}
	return data, nil
}

func (star *StarStack) MustPop() interface{} {
	if star.isClose.Load() == nil {
		star.init()
	}
	data, err := star.Pop()
	if err != nil {
		return nil
	}
	return data
}

func (star *StarStack) Pop() (interface{}, error) {
	if star.isClose.Load() == nil {
		star.init()
	}
	for {
		if star.isClose.Load().(bool) {
			return 0, io.EOF
		}
		if star.Len() == 0 {
			return 0, os.ErrNotExist
		}
		nowPtr := star.pStart
		nextPtr := star.pStart + 1
		if nextPtr >= star.cap {
			nextPtr = 0
		}
		data := star.datas[nowPtr]
		ok := atomic.CompareAndSwapUint64(&star.pStart, nowPtr, nextPtr)
		if !ok {
			time.Sleep(time.Microsecond)
			runtime.Gosched()
			continue
		}
		return data, nil
	}
}

func (star *StarStack) Push(data interface{}) error {
	if star.isClose.Load() == nil {
		star.init()
	}
	if star.isClose.Load().(bool) {
		return io.EOF
	}
	nowPtr := star.pEnd
	kariEnd := nowPtr + 1
	if kariEnd == star.cap {
		kariEnd = 0
	}
	if kariEnd == atomic.LoadUint64(&star.pStart) {
		for {
			time.Sleep(time.Microsecond)
			runtime.Gosched()
			if kariEnd != atomic.LoadUint64(&star.pStart) {
				break
			}
		}
	}
	star.datas[nowPtr] = data
	if ok := atomic.CompareAndSwapUint64(&star.pEnd, nowPtr, kariEnd); !ok {
		return os.ErrInvalid
	}
	return nil
}

func (star *StarStack) Close() error {
	if star.isClose.Load() == nil {
		star.init()
	}
	star.isClose.Store(true)
	return nil
}
func (star *StarStack) Read(buf []interface{}) (int, error) {
	if star.isClose.Load() == nil {
		star.init()
	}
	if star.isClose.Load().(bool) {
		return 0, io.EOF
	}
	if buf == nil {
		return 0, errors.New("buffer is nil")
	}
	star.rmu.Lock()
	defer star.rmu.Unlock()
	var sum int = 0
	for i := 0; i < len(buf); i++ {
		data, err := star.PopNoWait()
		if err != nil {
			if err == io.EOF {
				return sum, err
			}
			if err == os.ErrNotExist {
				i--
				continue
			}
			return sum, nil
		}
		buf[i] = data
		sum++
	}
	return sum, nil
}

func (star *StarStack) Write(bts []byte) (int, error) {
	if star.isClose.Load() == nil {
		star.init()
	}
	if bts == nil || star.isClose.Load().(bool) {
		return 0, io.EOF
	}
	star.wmu.Lock()
	defer star.wmu.Unlock()
	var sum = 0
	for i := 0; i < len(bts); i++ {
		err := star.Push(bts[i])
		if err != nil {
			fmt.Println("Write bts err:", err)
			return sum, err
		}
		sum++
	}
	return sum, nil
}

type StarChanStack struct {
	data    chan interface{}
	cap     uint64
	current uint64
	isClose atomic.Value
}

func NewStarChanStack(cap uint64) *StarChanStack {
	rtnBuffer := new(StarChanStack)
	rtnBuffer.cap = cap
	rtnBuffer.isClose.Store(false)
	rtnBuffer.data = make(chan interface{}, cap)
	return rtnBuffer
}

func (star *StarChanStack) init() {
	star.cap = 1024
	star.data = make(chan interface{}, star.cap)
	star.isClose.Store(false)
}

func (star *StarChanStack) Free() uint64 {
	return star.cap - star.current
}

func (star *StarChanStack) Cap() uint64 {
	return star.cap
}

func (star *StarChanStack) Len() uint64 {
	return star.current
}

func (star *StarChanStack) Pop() (interface{}, error) {
	if star.isClose.Load() == nil {
		star.init()
	}
	if star.isClose.Load().(bool) {
		return 0, io.EOF
	}
	data, ok := <-star.data
	if !ok {
		star.isClose.Store(true)
		return 0, errors.New("channel read error")
	}
	for {
		current := atomic.LoadUint64(&star.current)
		if atomic.CompareAndSwapUint64(&star.current, current, current-1) {
			break
		}
	}
	return data, nil
}

func (star *StarChanStack) Push(data interface{}) error {
	defer func() {
		recover()
	}()
	if star.isClose.Load() == nil {
		star.init()
	}
	if star.isClose.Load().(bool) {
		return io.EOF
	}
	star.data <- data
	for {
		current := atomic.LoadUint64(&star.current)
		if atomic.CompareAndSwapUint64(&star.current, current, current+1) {
			break
		}
	}
	return nil
}

func (star *StarChanStack) Close() error {
	if star.isClose.Load() == nil {
		star.init()
	}
	star.isClose.Store(true)
	close(star.data)
	return nil
}