thrasher-corp · shazbert · Nov 19, 2024 · Nov 19, 2024 · Nov 19, 2024 · Nov 24, 2024
diff --git a/common/common.go b/common/common.go
@@ -647,3 +647,38 @@ func (c *Counter) IncrementAndGet() int64 {
 	}
 	return newID
 }
+
+// ElementProcessor defines the function signature for processing an individual element with its index.
+type ElementProcessor[E any] func(index int, element E) error
+
+// ProcessElementsByBatch takes a slice of elements and processes them in batches of `batchSize` concurrently.
+// For example, if batchSize = 10 and list has 100 elements, 10 goroutines will process 10 elements concurrently
+// in each batch. Each batch completes before the next batch begins.
+// `process` is a function called for each individual element with its index and value.
+func ProcessElementsByBatch[S ~[]E, E any](batchSize int, list S, process ElementProcessor[E]) error {
+	var errs error
+	for i, s := range Batch(list, batchSize) {
+		err := CollectErrors(len(s))
+		for j, e := range s {
+			go func(index int, element E) { err.C <- process(index, element); err.Wg.Done() }((i*batchSize)+j, e)
+		}
+		errs = AppendError(errs, err.Collect())
+	}
+	return errs
+}
-// ElementProcessor defines the function signature for processing an individual element with its index.
-type ElementProcessor[E any] func(index int, element E) error
-
-// ProcessElementsByBatch takes a slice of elements and processes them in batches of `batchSize` concurrently.
-// For example, if batchSize = 10 and list has 100 elements, 10 goroutines will process 10 elements concurrently
-// in each batch. Each batch completes before the next batch begins.
-// `process` is a function called for each individual element with its index and value.
-func ProcessElementsByBatch[S ~[]E, E any](batchSize int, list S, process ElementProcessor[E]) error {
-	var errs error
-	for i, s := range Batch(list, batchSize) {
-		err := CollectErrors(len(s))
-		for j, e := range s {
-			go func(index int, element E) { err.C <- process(index, element); err.Wg.Done() }((i*batchSize)+j, e)
-		}
-		errs = AppendError(errs, err.Collect())
-	}
-	return errs
-}
+// PipelineProcessor defines the function signature for processing an individual elements
+type PipelineProcessor[E any] func(element E) error
+
+// ThrottledPipeline processes a slice concurrently with a throttled number of workers
+func ThrottledPipeline[S ~[]E, E any](workers int, list S, process PipelineProcessor[E]) error {
+	q := make(chan E, len(list))
+	for _, s := range list {
+		q <- s
+	}
+	err := CollectErrors(len(list))
+	for range workers {
+		go func() {
+			defer err.Wg.Done()
+			for {
+				select {
+				case s := <-q:
+					err.C <- process(s)
+				default:
+					return
+				}
+			}
+		}()
+	}
+	return err.Collect()
+}
-// ElementProcessor defines the function signature for processing an individual element with its index.
-type ElementProcessor[E any] func(index int, element E) error
-
-// ProcessElementsByBatch takes a slice of elements and processes them in batches of `batchSize` concurrently.
-// For example, if batchSize = 10 and list has 100 elements, 10 goroutines will process 10 elements concurrently
-// in each batch. Each batch completes before the next batch begins.
-// `process` is a function called for each individual element with its index and value.
-func ProcessElementsByBatch[S ~[]E, E any](batchSize int, list S, process ElementProcessor[E]) error {
-	var errs error
-	for i, s := range Batch(list, batchSize) {
-		err := CollectErrors(len(s))
-		for j, e := range s {
-			go func(index int, element E) { err.C <- process(index, element); err.Wg.Done() }((i*batchSize)+j, e)
-		}
-		errs = AppendError(errs, err.Collect())
-	}
-	return errs
-}
+// PipelineProcessor defines the function signature for processing an individual elements
+type PipelineProcessor[E any] func(element E) error
+
+// ThrottledPipeline processes a slice concurrently with a throttled number of workers
+func ThrottledPipeline[S ~[]E, E any](workers int, list S, process PipelineProcessor[E]) error {
+	q := make(chan E, len(list))
+	for _, s := range list {
+		q <- s
+	}
+	err := CollectErrors(len(list))
+	for range workers {
+		go func() {
+			defer err.Wg.Done()
+			for {
+				select {
+				case s := <-q:
+					err.C <- process(s)
+				default:
+					return
+				}
+			}
+		}()
+	}
+	return err.Collect()
+}
+
+// BatchProcessor defines the function signature for processing a batch of elements.
+type BatchProcessor[S ~[]E, E any] func(batch S) error
+
+// ProcessBatches processes `S` concurrently in batches of `batchSize`
+// For example, if batchSize = 10 and list has 100 elements, 10 goroutines will be created to process
+// 10 batches. Each batch is processed sequentially by the `process` function, and batches are processed
+// in parallel.
+func ProcessBatches[S ~[]E, E any](batchSize int, list S, process BatchProcessor[S, E]) error {
+	batches := Batch(list, batchSize)
+	errs := CollectErrors(len(batches))
+	for _, s := range batches {
+		go func(s S) { errs.C <- process(s); errs.Wg.Done() }(s)
+	}
+	return errs.Collect()
+}
diff --git a/common/common_test.go b/common/common_test.go
@@ -12,6 +12,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -793,3 +794,60 @@ func BenchmarkCounter(b *testing.B) {
 		c.IncrementAndGet()
 	}
 }
+
+func TestProcessElementsByBatch(t *testing.T) {
+	t.Parallel()
+	testSlice := make([]int, 0, 100)
+	for i := range 100 {
+		testSlice = append(testSlice, i)
+	}
+
+	trackIndex := map[int]bool{}
+	m := sync.Mutex{}
+
+	ch := make(chan int, len(testSlice))
+	require.NoError(t, ProcessElementsByBatch(10, testSlice, func(_ int, v int) error {
+		m.Lock()
+		defer m.Unlock()
+		if trackIndex[v] {
+			return errors.New("duplicate index")
+		}
+		trackIndex[v] = true
+		ch <- v
+		return nil
+	}))
+
+	require.Len(t, trackIndex, len(testSlice))
+
+	close(ch)
+	for v := range ch {
+		assert.Contains(t, testSlice, v)
+	}
+
+	expected := errors.New("test error")
+	require.ErrorIs(t, ProcessElementsByBatch(10, testSlice, func(int, int) error { return expected }), expected)
+}
+
+func TestProcessBatches(t *testing.T) {
+	t.Parallel()
+	testSlice := make([]int, 0, 100)
+	for i := range 100 {
+		testSlice = append(testSlice, i)
+	}
+
+	ch := make(chan int, len(testSlice))
+	require.NoError(t, ProcessBatches(10, testSlice, func(v []int) error {
+		for _, i := range v {
+			ch <- i
+		}
+		return nil
+	}))
+
+	close(ch)
+	for v := range ch {
+		assert.Contains(t, testSlice, v)
+	}
+
+	expected := errors.New("test error")
+	require.ErrorIs(t, ProcessBatches(10, testSlice, func([]int) error { return expected }), expected)
+}
diff --git a/exchanges/exchange.go b/exchanges/exchange.go
@@ -1810,28 +1810,7 @@ func (b *Base) GetOpenInterest(context.Context, ...key.PairAsset) ([]futures.Ope
 
 // ParallelChanOp performs a single method call in parallel across streams and waits to return any errors
 func (b *Base) ParallelChanOp(channels subscription.List, m func(subscription.List) error, batchSize int) error {
-	wg := sync.WaitGroup{}
-	errC := make(chan error, len(channels))
-
-	for _, b := range common.Batch(channels, batchSize) {
-		wg.Add(1)
-		go func(c subscription.List) {
-			defer wg.Done()
-			if err := m(c); err != nil {
-				errC <- err
-			}
-		}(b)
-	}
-
-	wg.Wait()
-	close(errC)
-
-	var errs error
-	for err := range errC {
-		errs = common.AppendError(errs, err)
-	}
-
-	return errs
+	return common.ProcessBatches(batchSize, channels, m)
 }
 
 // Bootstrap function allows for exchange authors to supplement or override common startup actions

diff --git a/exchanges/gateio/gateio_test.go b/exchanges/gateio/gateio_test.go
@@ -2845,7 +2845,7 @@ func TestGenerateSubscriptionsSpot(t *testing.T) {
 					case subscription.CandlesChannel:
 						s.QualifiedChannel = "5m," + pairs[i].String()
 					case subscription.OrderbookChannel:
-						s.QualifiedChannel = pairs[i].String() + ",100ms"
+						s.QualifiedChannel = pairs[i].String() + ",100,100ms"
 					case spotOrderbookChannel:
 						s.QualifiedChannel = pairs[i].String() + ",5,1000ms"
 					}

diff --git a/exchanges/gateio/gateio_websocket.go b/exchanges/gateio/gateio_websocket.go
@@ -51,12 +51,15 @@ const (
 
 	subscribeEvent   = "subscribe"
 	unsubscribeEvent = "unsubscribe"
+
+	// subscriptionBatchCount is the number of subscriptions to send in a single batch
+	subscriptionBatchCount = 200
 )
 
 var defaultSubscriptions = subscription.List{
 	{Enabled: true, Channel: subscription.TickerChannel, Asset: asset.Spot},
 	{Enabled: true, Channel: subscription.CandlesChannel, Asset: asset.Spot, Interval: kline.FiveMin},
-	{Enabled: true, Channel: subscription.OrderbookChannel, Asset: asset.Spot, Interval: kline.HundredMilliseconds},
+	{Enabled: true, Channel: subscription.OrderbookChannel, Asset: asset.Spot, Interval: kline.HundredMilliseconds, Levels: 100},
 	{Enabled: true, Channel: spotBalancesChannel, Asset: asset.Spot, Authenticated: true},
 	{Enabled: true, Channel: crossMarginBalanceChannel, Asset: asset.CrossMargin, Authenticated: true},
 	{Enabled: true, Channel: marginBalancesChannel, Asset: asset.Margin, Authenticated: true},
@@ -66,8 +69,10 @@ var defaultSubscriptions = subscription.List{
 var fetchedCurrencyPairSnapshotOrderbook = make(map[string]bool)
 
 var subscriptionNames = map[string]string{
-	subscription.TickerChannel:    spotTickerChannel,
-	subscription.OrderbookChannel: spotOrderbookUpdateChannel,
+	subscription.TickerChannel: spotTickerChannel,
+	// NOTE: Opted for `spot.order_book` as snapshot to ensure long term stability over higher processing cost. Also there
+	// is no latency difference between this and `spot.order_book_update` both being 100ms.
+	subscription.OrderbookChannel: spotOrderbookChannel,
 	subscription.CandlesChannel:   spotCandlesticksChannel,
 	subscription.AllTradesChannel: spotTradesChannel,
 }
@@ -579,32 +584,36 @@ func (g *Gateio) manageSubs(ctx context.Context, event string, conn stream.Conne
 		return errs
 	}
 
-	for _, s := range subs {
-		if err := func() error {
-			msg, err := g.manageSubReq(ctx, event, conn, s)
-			if err != nil {
-				return err
-			}
-			result, err := conn.SendMessageReturnResponse(ctx, websocketRateLimitNotNeededEPL, msg.ID, msg)
-			if err != nil {
-				return err
-			}
-			var resp WsEventResponse
-			if err := json.Unmarshal(result, &resp); err != nil {
-				return err
-			}
-			if resp.Error != nil && resp.Error.Code != 0 {
-				return fmt.Errorf("(%d) %s", resp.Error.Code, resp.Error.Message)
-			}
-			if event == "unsubscribe" {
-				return g.Websocket.RemoveSubscriptions(conn, s)
-			}
-			return g.Websocket.AddSuccessfulSubscriptions(conn, s)
-		}(); err != nil {
-			errs = common.AppendError(errs, fmt.Errorf("%s %s %s: %w", s.Channel, s.Asset, s.Pairs, err))
+	// This will batch the subscriptions into groups of subscriptionBatchCount then concurrently subscribe to them.
+	// This will decrease the amount of time it takes to subscribe to a large number of subscriptions.
+	return common.ProcessElementsByBatch(subscriptionBatchCount, subs, func(_ int, s *subscription.Subscription) error {
+		if err := g.manageSubPayload(ctx, conn, event, s); err != nil {
+			return fmt.Errorf("%s %s %s: %w", s.Channel, s.Asset, s.Pairs, err)
 		}
+		return nil
+	})
+}
+
+func (g *Gateio) manageSubPayload(ctx context.Context, conn stream.Connection, event string, s *subscription.Subscription) error {
+	msg, err := g.manageSubReq(ctx, event, conn, s)
+	if err != nil {
+		return err
+	}
+	result, err := conn.SendMessageReturnResponse(ctx, websocketRateLimitNotNeededEPL, msg.ID, msg)
+	if err != nil {
+		return err
 	}
-	return errs
+	var resp WsEventResponse
+	if err := json.Unmarshal(result, &resp); err != nil {
+		return err
+	}
+	if resp.Error != nil && resp.Error.Code != 0 {
+		return fmt.Errorf("(%d) %s", resp.Error.Code, resp.Error.Message)
+	}
+	if event == "unsubscribe" {
+		return g.Websocket.RemoveSubscriptions(conn, s)
+	}
+	return g.Websocket.AddSuccessfulSubscriptions(conn, s)
 }
 
 // manageSubReq constructs the subscription management message for a subscription
@@ -694,27 +703,31 @@ func (g *Gateio) handleSubscription(ctx context.Context, conn stream.Connection,
 	if err != nil {
 		return err
 	}
-	var errs error
-	for k := range payloads {
-		result, err := conn.SendMessageReturnResponse(ctx, websocketRateLimitNotNeededEPL, payloads[k].ID, payloads[k])
-		if err != nil {
-			errs = common.AppendError(errs, err)
-			continue
-		}
-		var resp WsEventResponse
-		if err = json.Unmarshal(result, &resp); err != nil {
-			errs = common.AppendError(errs, err)
-		} else {
-			if resp.Error != nil && resp.Error.Code != 0 {
-				errs = common.AppendError(errs, fmt.Errorf("error while %s to channel %s error code: %d message: %s", payloads[k].Event, payloads[k].Channel, resp.Error.Code, resp.Error.Message))
-				continue
-			}
-			if event == subscribeEvent {
-				errs = common.AppendError(errs, g.Websocket.AddSuccessfulSubscriptions(conn, channelsToSubscribe[k]))
-			} else {
-				errs = common.AppendError(errs, g.Websocket.RemoveSubscriptions(conn, channelsToSubscribe[k]))
-			}
+
+	// This will batch the subscriptions into groups of subscriptionBatchCount then concurrently subscribe to them.
+	// This will decrease the amount of time it takes to subscribe to a large number of subscriptions.
+	return common.ProcessElementsByBatch(subscriptionBatchCount, payloads, func(index int, payload WsInput) error {
+		if err := g.sendSubPayload(ctx, conn, event, channelsToSubscribe[index], &payload); err != nil {
+			return fmt.Errorf("%s %s %s: %w", channelsToSubscribe[index].Channel, channelsToSubscribe[index].Asset, channelsToSubscribe[index].Pairs, err)
 		}
+		return nil
+	})
+}
+
+func (g *Gateio) sendSubPayload(ctx context.Context, conn stream.Connection, event string, s *subscription.Subscription, payload *WsInput) error {
+	result, err := conn.SendMessageReturnResponse(ctx, websocketRateLimitNotNeededEPL, payload.ID, payload)
+	if err != nil {
+		return err
+	}
+	var resp WsEventResponse
+	if err := json.Unmarshal(result, &resp); err != nil {
+		return err
+	}
+	if resp.Error != nil && resp.Error.Code != 0 {
+		return fmt.Errorf("error while %s to channel %s error code: %d message: %s", event, payload.Channel, resp.Error.Code, resp.Error.Message)
+	}
+	if event == subscribeEvent {
+		return g.Websocket.AddSuccessfulSubscriptions(conn, s)
 	}
-	return errs
+	return g.Websocket.RemoveSubscriptions(conn, s)
 }
diff --git a/exchanges/gateio/gateio_ws_delivery_futures.go b/exchanges/gateio/gateio_ws_delivery_futures.go
@@ -13,7 +13,6 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/account"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 )
@@ -55,7 +54,7 @@ func (g *Gateio) WsDeliveryFuturesConnect(ctx context.Context, conn stream.Conne
 	if err != nil {
 		return err
 	}
-	conn.SetupPingHandler(request.Unset, stream.PingHandler{
+	conn.SetupPingHandler(websocketRateLimitNotNeededEPL, stream.PingHandler{
 		Websocket:   true,
 		Delay:       time.Second * 5,
 		MessageType: websocket.PingMessage,

diff --git a/exchanges/gateio/gateio_ws_futures.go b/exchanges/gateio/gateio_ws_futures.go
@@ -19,7 +19,6 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/order"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
@@ -76,7 +75,7 @@ func (g *Gateio) WsFuturesConnect(ctx context.Context, conn stream.Connection) e
 	if err != nil {
 		return err
 	}
-	conn.SetupPingHandler(request.Unset, stream.PingHandler{
+	conn.SetupPingHandler(websocketRateLimitNotNeededEPL, stream.PingHandler{
 		Websocket:   true,
 		MessageType: websocket.PingMessage,
 		Delay:       time.Second * 15,

diff --git a/exchanges/gateio/gateio_ws_option.go b/exchanges/gateio/gateio_ws_option.go
@@ -18,7 +18,6 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/order"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
@@ -85,7 +84,7 @@ func (g *Gateio) WsOptionsConnect(ctx context.Context, conn stream.Connection) e
 	if err != nil {
 		return err
 	}
-	conn.SetupPingHandler(request.Unset, stream.PingHandler{
+	conn.SetupPingHandler(websocketRateLimitNotNeededEPL, stream.PingHandler{
 		Websocket:   true,
 		Delay:       time.Second * 5,
 		MessageType: websocket.PingMessage,

diff --git a/exchanges/gateio/ratelimiter.go b/exchanges/gateio/ratelimiter.go
@@ -379,7 +379,7 @@ var packageRateLimits = request.RateLimitDefinitions{
 
 	privateUnifiedSpotEPL: standardRateLimit(),
 
-	websocketRateLimitNotNeededEPL: nil, // no rate limit for certain websocket functions
+	websocketRateLimitNotNeededEPL: request.RateLimitNotRequired, // no rate limit for certain websocket functions
 }
 
 func standardRateLimit() *request.RateLimiterWithWeight {

diff --git a/exchanges/request/limit.go b/exchanges/request/limit.go
@@ -20,6 +20,9 @@ var (
 	errSpecificRateLimiterIsNil = errors.New("specific rate limiter is nil")
 )
 
+// RateLimitNotRequired is used for when an endpoint does not require rate limiting
+var RateLimitNotRequired *RateLimiterWithWeight
+
 // Const here define individual functionality sub types for rate limiting
 const (
 	Unset EndpointLimit = iota