go mod support

circle.go

@@ -4,7 +4,10 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"os"
+	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -13,79 +16,90 @@ type StarBuffer struct {
 	io.Writer
 	io.Closer
 	datas   []byte
-	pStart  int
+	pStart  uint64
-	pEnd    int
+	pEnd    uint64
-	cap     int
+	cap     uint64
-	isClose bool
+	isClose atomic.Value
-	isEnd   bool
+	isEnd   atomic.Value
 	rmu     sync.Mutex
 	wmu     sync.Mutex
 }
 
-func NewStarBuffer(cap int) *StarBuffer {
+func NewStarBuffer(cap uint64) *StarBuffer {
 	rtnBuffer := new(StarBuffer)
 	rtnBuffer.cap = cap
 	rtnBuffer.datas = make([]byte, cap)
+	rtnBuffer.isClose.Store(false)
+	rtnBuffer.isEnd.Store(false)
 	return rtnBuffer
 }
 
-func (star *StarBuffer) Free() int {
+func (star *StarBuffer) Free() uint64 {
 	return star.cap - star.Len()
 }
 
-func (star *StarBuffer) Cap() int {
+func (star *StarBuffer) Cap() uint64 {
 	return star.cap
 }
 
-func (star *StarBuffer) Len() int {
+func (star *StarBuffer) Len() uint64 {
-	length := star.pEnd - star.pStart
+	if star.pEnd >= star.pStart {
-	if length < 0 {
+		return star.pEnd - star.pStart
-		return star.cap + length - 1
 	}
-	return length
+	return star.pEnd - star.pStart + star.cap
 }
 
 func (star *StarBuffer) getByte() (byte, error) {
-	if star.isClose || (star.isEnd && star.Len() == 0) {
+	if star.isClose.Load().(bool) || (star.Len() == 0 && star.isEnd.Load().(bool)) {
 		return 0, io.EOF
 	}
 	if star.Len() == 0 {
-		return 0, errors.New("no byte available now")
+		return 0, os.ErrNotExist
 	}
-	data := star.datas[star.pStart]
+	nowPtr := star.pStart
-	star.pStart++
+	nextPtr := star.pStart + 1
-	if star.pStart == star.cap {
+	if nextPtr >= star.cap {
-		star.pStart = 0
+		nextPtr = 0
+	}
+	data := star.datas[nowPtr]
+	ok := atomic.CompareAndSwapUint64(&star.pStart, nowPtr, nextPtr)
+	if !ok {
+		return 0, os.ErrInvalid
 	}
 	return data, nil
 }
 
 func (star *StarBuffer) putByte(data byte) error {
-	if star.isClose || star.isEnd {
+	if star.isClose.Load().(bool) {
 		return io.EOF
 	}
-	kariEnd := star.pEnd + 1
+	nowPtr := star.pEnd
+	kariEnd := nowPtr + 1
 	if kariEnd == star.cap {
 		kariEnd = 0
 	}
-	if kariEnd == star.pStart {
+	if kariEnd == atomic.LoadUint64(&star.pStart) {
 		for {
 			time.Sleep(time.Microsecond)
-			if kariEnd != star.pStart {
+			runtime.Gosched()
+			if kariEnd != atomic.LoadUint64(&star.pStart) {
 				break
 			}
 		}
 	}
-	star.datas[star.pEnd] = data
+	star.datas[nowPtr] = data
-	star.pEnd = kariEnd
+	if ok := atomic.CompareAndSwapUint64(&star.pEnd, nowPtr, kariEnd); !ok {
+		return os.ErrInvalid
+	}
 	return nil
 }
+
 func (star *StarBuffer) Close() error {
-	star.isClose = true
+	star.isClose.Store(true)
 	return nil
 }
 func (star *StarBuffer) Read(buf []byte) (int, error) {
-	if star.isClose {
+	if star.isClose.Load().(bool) || (star.Len() == 0 && star.isEnd.Load().(bool)) {
 		return 0, io.EOF
 	}
 	if buf == nil {
@@ -100,6 +114,10 @@ func (star *StarBuffer) Read(buf []byte) (int, error) {
 			if err == io.EOF {
 				return sum, err
 			}
+			if err == os.ErrNotExist {
+				i--
+				continue
+			}
 			return sum, nil
 		}
 		buf[i] = data
@@ -109,8 +127,11 @@ func (star *StarBuffer) Read(buf []byte) (int, error) {
 }
 
 func (star *StarBuffer) Write(bts []byte) (int, error) {
-	if bts == nil || star.isClose {
+	if bts == nil && !star.isEnd.Load().(bool) {
-		star.isEnd = true
+		star.isEnd.Store(true)
+		return 0, nil
+	}
+	if bts == nil || star.isClose.Load().(bool) {
 		return 0, io.EOF
 	}
 	star.wmu.Lock()

circle_test.go

@@ -0,0 +1,86 @@
+package stario
+
+import (
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func Test_Circle(t *testing.T) {
+	buf := NewStarBuffer(2048)
+	go func() {
+		for {
+			//fmt.Println("write start")
+			buf.Write([]byte("中华人民共和国\n"))
+			//fmt.Println("write success")
+			time.Sleep(time.Millisecond * 50)
+		}
+	}()
+	cpp := ""
+	go func() {
+		time.Sleep(time.Second * 3)
+		for {
+			cache := make([]byte, 64)
+			ints, err := buf.Read(cache)
+			if err != nil {
+				fmt.Println("read error", err)
+				return
+			}
+			if ints != 0 {
+				cpp += string(cache[:ints])
+			}
+		}
+	}()
+	time.Sleep(time.Second * 13)
+	fmt.Println(cpp)
+}
+
+func Test_Circle_Speed(t *testing.T) {
+	buf := NewStarBuffer(1048976)
+	count := uint64(0)
+	for i := 1; i <= 10; i++ {
+		go func() {
+			for {
+				buf.putByte('a')
+			}
+		}()
+	}
+	for i := 1; i <= 10; i++ {
+		go func() {
+			for {
+				_, err := buf.getByte()
+				if err == nil {
+					atomic.AddUint64(&count, 1)
+				}
+			}
+		}()
+	}
+	time.Sleep(time.Second * 10)
+	fmt.Println(count)
+}
+
+func Test_Circle_Speed2(t *testing.T) {
+	buf := NewStarBuffer(8192)
+	count := uint64(0)
+	for i := 1; i <= 10; i++ {
+		go func() {
+			for {
+				buf.Write([]byte("hello world b612 hello world b612 b612 b612 b612 b612 b612"))
+			}
+		}()
+	}
+	for i := 1; i <= 10; i++ {
+		go func() {
+			for {
+				mybuf := make([]byte, 1024)
+				j, err := buf.Read(mybuf)
+				if err == nil {
+					atomic.AddUint64(&count, uint64(j))
+				}
+			}
+		}()
+	}
+	time.Sleep(time.Second * 10)
+	fmt.Println(float64(count) / 10 / 1024 / 1024)
+}

fn.go

@@ -0,0 +1,55 @@
+package stario
+
+import (
+	"errors"
+	"time"
+)
+
+var ERR_TIMEOUT = errors.New("TIME OUT")
+
+func WaitUntilTimeout(tm time.Duration, fn func(chan struct{}) error) error {
+	var err error
+	finished := make(chan struct{})
+	imout := make(chan struct{})
+	go func() {
+		err = fn(imout)
+		finished <- struct{}{}
+	}()
+	select {
+	case <-finished:
+		return err
+	case <-time.After(tm):
+		close(imout)
+		return ERR_TIMEOUT
+	}
+}
+
+func WaitUntilFinished(fn func() error) <-chan error {
+	finished := make(chan error)
+	go func() {
+		err := fn()
+		finished <- err
+	}()
+	return finished
+}
+
+func WaitUntilTimeoutFinished(tm time.Duration, fn func(chan struct{}) error) <-chan error {
+	var err error
+	finished := make(chan struct{})
+	result := make(chan error)
+	imout := make(chan struct{})
+	go func() {
+		err = fn(imout)
+		finished <- struct{}{}
+	}()
+	go func() {
+		select {
+		case <-finished:
+			result <- err
+		case <-time.After(tm):
+			close(imout)
+			result <- ERR_TIMEOUT
+		}
+	}()
+	return result
+}

go.mod

@@ -0,0 +1,5 @@
+module b612.me/stario
+
+go 1.16
+
+require golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000

go.sum

@@ -0,0 +1,11 @@
+golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000 h1:SL+8VVnkqyshUSz5iNnXtrBQzvFF2SkROm6t5RczFAE=
+golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 h1:SrN+KX8Art/Sf4HNj6Zcz06G7VEz+7w9tdXTPOZ7+l4=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

sync.go

@@ -0,0 +1,55 @@
+package stario
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type WaitGroup struct {
+	wg       *sync.WaitGroup
+	maxCount uint32
+	allCount uint32
+}
+
+func NewWaitGroup(maxCount int) WaitGroup {
+	return WaitGroup{wg: &sync.WaitGroup{}, maxCount: uint32(maxCount)}
+}
+
+func (swg *WaitGroup) Add(delta int) {
+	var Udelta uint32
+	if delta < 0 {
+		Udelta = uint32(-delta - 1)
+	} else {
+		Udelta = uint32(delta)
+	}
+	for {
+		allC := atomic.LoadUint32(&swg.allCount)
+		if atomic.LoadUint32(&swg.maxCount) == 0 || atomic.LoadUint32(&swg.maxCount) >= allC+uint32(delta) {
+			if delta < 0 {
+				atomic.AddUint32(&swg.allCount, ^uint32(Udelta))
+			} else {
+				atomic.AddUint32(&swg.allCount, uint32(Udelta))
+			}
+			break
+		}
+		time.Sleep(time.Microsecond)
+	}
+	swg.wg.Add(delta)
+}
+
+func (swg *WaitGroup) Done() {
+	swg.Add(-1)
+}
+
+func (swg *WaitGroup) Wait() {
+	swg.wg.Wait()
+}
+
+func (swg *WaitGroup) GetMaxWaitNum() int {
+	return int(atomic.LoadUint32(&swg.maxCount))
+}
+
+func (swg *WaitGroup) SetMaxWaitNum(num int) {
+	atomic.AddUint32(&swg.maxCount, uint32(num))
+}