use cursor to fix [b[[loopSaveCandle function

core/core.go

@@ -0,0 +1,35 @@
+package core
+
+import (
+	"context"
+)
+
+type Core struct {
+	// Add any necessary fields here
+}
+
+func (c *Core) GetScoreList(count int) []string {
+	// Implementation of GetScoreList
+}
+
+func (c *Core) GetScoreListWithContext(ctx context.Context, count int) ([]string, error) {
+	done := make(chan []string, 1)
+	errChan := make(chan error, 1)
+
+	go func() {
+		list := c.GetScoreList(count)
+		select {
+		case done <- list:
+		case <-ctx.Done():
+		}
+	}()
+
+	select {
+	case list := <-done:
+		return list, nil
+	case err := <-errChan:
+		return nil, err
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/phyer/core v0.1.90
 	github.com/phyer/v5sdkgo v0.1.4
 	github.com/sirupsen/logrus v1.9.3
+	golang.org/x/sync v0.10.0
 )
 
 require (

go.sum

@@ -81,7 +81,10 @@ golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
 golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

main.go

@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+
 	// "fmt"
 	"math/rand"
 	"os"
@@ -12,6 +14,7 @@ import (
 
 	"github.com/phyer/v5sdkgo/rest"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
 
 	// "v5sdk_go/ws"
 	// "v5sdk_go/ws/wImpl"
@@ -169,64 +172,126 @@ func ShowSysTime(cr *core.Core) {
 // range: 随机的范围，从0开始到range个周期，作为查询的after值，也就是随机n个周期，去取之前的记录,对于2D，5D等数据，可以用来补全数据, range值越大，随机散点的范围越大, 越失焦
 
 func LoopAllCoinsList(period int64, delay int64, mdura int, barPeriod string, onceCount int, rge int) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	cr := core.Core{}
 	cr.Init()
-	allScoreChan := make(chan []string)
-	// logrus.Info("start LoopAllCoinsList: period: ", period, " delay: ", delay, " mdura:", mdura, " barPeriod: ", barPeriod, " onceCount: ", onceCount, " rge:", rge)
-	per1 := 1 * time.Minute
-	ticker := time.NewTicker(per1)
-	go func() {
+
+	// Increase buffer size to accommodate multiple updates
+	// Buffer size should be large enough to handle typical batch size
+	// but not so large that it consumes too much memory
+	const channelBufferSize = 100
+	allScoreChan := make(chan []string, channelBufferSize)
+
+	// Add channel overflow protection
+	sendWithTimeout := func(scores []string) {
+		select {
+		case allScoreChan <- scores:
+			// Successfully sent
+		case <-time.After(5 * time.Second):
+			logrus.Warn("Failed to send scores to channel - buffer full")
+		case <-ctx.Done():
+			return
+		}
+	}
+
+	mainTicker := time.NewTicker(time.Duration(period) * time.Second)
+	defer mainTicker.Stop()
+
+	var eg errgroup.Group
+
+	// Producer goroutine
+	eg.Go(func() error {
 		for {
-			tsi := time.Now().Unix()
-			//logrus.Info("tsi, period, delay, tsi%(period): ", tsi, period, delay, tsi%(period))
-			if tsi%(period) != delay {
-				time.Sleep(1 * time.Second)
-				continue
-			}
 			select {
-			case <-ticker.C:
-				go func() {
-					// -1 是获取全部coin列表
-					list := cr.GetScoreList(-1)
-					// logrus.Info("allCoins3", list)
-					allScoreChan <- list
-				}()
+			case <-ctx.Done():
+				return nil
+			case <-mainTicker.C:
+				if time.Now().Unix()%period != delay {
+					continue
+				}
+
+				_, cancel := context.WithTimeout(ctx, 5*time.Second)
+				list := cr.GetScoreList(-1)
+				cancel()
+
+				// Use the new send function with timeout
+				if len(list) > 0 {
+					sendWithTimeout(list)
+				}
 			}
 		}
-	}()
-	for {
-		allScore, _ := <-allScoreChan
-		logrus.Debug("allCoins allScore", allScore)
-		if len(allScore) == 0 {
-			continue
+	})
+
+	// Consumer goroutine
+	eg.Go(func() error {
+		for {
+			select {
+			case <-ctx.Done():
+				return nil
+			case allScore := <-allScoreChan:
+				if len(allScore) == 0 {
+					continue
+				}
+
+				// Process in batches to avoid overwhelming the system
+				if err := processScores(&cr, allScore, mdura, barPeriod, onceCount, rge); err != nil {
+					logrus.WithError(err).Error("Failed to process scores")
+					// Consider implementing exponential backoff retry here
+				}
+			}
+		}
+	})
+
+	if err := eg.Wait(); err != nil {
+		logrus.WithError(err).Error("LoopAllCoinsList failed")
+	}
+}
+
+func processScores(cr *core.Core, allScore []string, mdura int, barPeriod string, onceCount int, rge int) error {
+	utils.TickerWrapper(time.Duration(mdura)*time.Second, allScore, func(i int, ary []string) error {
+		nw := time.Now()
+		// 使用更安全的随机数生成
+		randSource := rand.NewSource(nw.UnixNano())
+		randGen := rand.New(randSource)
+		ct := randGen.Intn(rge)
+
+		minutes, err := cr.PeriodToMinutes(barPeriod)
+		if err != nil {
+			return fmt.Errorf("failed to get period minutes: %w", err)
 		}
 
-		utils.TickerWrapper(time.Duration(mdura)*time.Second, allScore, func(i int, ary []string) error {
-			nw := time.Now()
-			rand.Seed(nw.UnixNano())
-			ct := rand.Intn(rge)
-			minutes, _ := cr.PeriodToMinutes(barPeriod)
-			tmi := nw.UnixMilli()
-			tmi = tmi - tmi%60000
-			tmi = tmi - (int64(ct) * minutes * 60000)
-			lm := strconv.Itoa(onceCount)
-			// logrus.Info("instId: ", ary[i], " limit: ", lm, " onceCount:", onceCount)
-			if lm == "0" {
-				lm = "100"
-			}
-			restQ := core.RestQueue{
-				InstId: ary[i],
-				Bar:    barPeriod,
-				WithWs: false,
-				Limit:  lm,
-				After:  tmi,
-			}
-			js, err := json.Marshal(restQ)
-			logrus.Debug("allCoins lpush js:", string(js))
-			cr.RedisLocalCli.LPush("restQueue", js)
-			return err
-		})
-	}
+		tmi := nw.UnixMilli()
+		tmi = tmi - tmi%60000
+		tmi = tmi - (int64(ct) * minutes * 60000)
+
+		limit := onceCount
+		if limit == 0 {
+			limit = 100
+		}
+
+		restQ := core.RestQueue{
+			InstId: ary[i],
+			Bar:    barPeriod,
+			WithWs: false,
+			Limit:  strconv.Itoa(limit),
+			After:  tmi,
+		}
+
+		js, err := json.Marshal(restQ)
+		if err != nil {
+			return fmt.Errorf("failed to marshal RestQueue: %w", err)
+		}
+		logrus.WithFields(logrus.Fields{
+			"instId": ary[i],
+			"bar":    barPeriod,
+			"limit":  limit,
+		}).Debug("Pushing to restQueue")
+
+		return cr.RedisLocalCli.LPush("restQueue", js).Err()
+	})
+	return nil
 }
 
 func main() {