Fix goroutines

* Fix goroutines count because of history dispatch

Author: antonbashir

.vscode/launch.json

@@ -33,7 +33,6 @@
         "MERCURE_EXTRA_DIRECTIVES": "anonymous\nwrite_timeout 10s",
         "MERCURE_REDIS_ADDRESS": "localhost:6379",
         "MERCURE_REDIS_SUBSCRIBERS_SIZE": "10000",
-        "MERCURE_REDIS_DISPATCHER_POOL_SIZE": "16",
         "MERCURE_REDIS_CHANNEL": "mercure",
         "GLOBAL_OPTIONS": "debug",
         "SERVER_NAME": ":1234",

caddy/redis.go

@@ -15,12 +15,11 @@ func init() {
 }
 
 type Redis struct {
-	Address            string `json:"address,omitempty"`
-	Username           string `json:"username,omitempty"`
-	Password           string `json:"password,omitempty"`
-	SubscribersSize    int    `json:"subscribers_size,omitempty"`
-	DispatcherPoolSize int    `json:"dispatcher_pool_size,omitempty"`
-	RedisChannel       string `json:"redis_channel,omitempty"`
+	Address         string `json:"address,omitempty"`
+	Username        string `json:"username,omitempty"`
+	Password        string `json:"password,omitempty"`
+	SubscribersSize int    `json:"subscribers_size,omitempty"`
+	RedisChannel    string `json:"redis_channel,omitempty"`
 
 	transport    *mercure.RedisTransport
 	transportKey string
@@ -47,7 +46,7 @@ func (r *Redis) Provision(ctx caddy.Context) error {
 	r.transportKey = key.String()
 
 	destructor, _, err := TransportUsagePool.LoadOrNew(r.transportKey, func() (caddy.Destructor, error) {
-		t, err := mercure.NewRedisTransport(ctx.Logger(), r.Address, r.Username, r.Password, r.SubscribersSize, r.DispatcherPoolSize, r.RedisChannel)
+		t, err := mercure.NewRedisTransport(ctx.Logger(), r.Address, r.Username, r.Password, r.SubscribersSize, r.RedisChannel)
 		if err != nil {
 			return nil, err
 		}
@@ -108,18 +107,6 @@ func (r *Redis) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 
 				r.SubscribersSize = s
 
-			case "dispatcher_pool_size":
-				if !d.NextArg() {
-					return d.ArgErr()
-				}
-
-				s, e := strconv.Atoi(replacer.ReplaceKnown(d.Val(), ""))
-				if e != nil {
-					return e
-				}
-
-				r.DispatcherPoolSize = s
-
 			case "redis_channel":
 				if !d.NextArg() {
 					return d.ArgErr()

charts/mercure/templates/deployment.yaml

@@ -56,10 +56,6 @@ spec:
             - name: MERCURE_REDIS_SUBSCRIBERS_SIZE
               value: "{{ .Values.redis.subscribers_size }}"
             {{- end }}
-            {{- if .Values.redis.dispatcher_pool_size }}
-            - name: MERCURE_REDIS_DISPATCHER_POOL_SIZE
-              value: "{{ .Values.redis.dispatcher_pool_size }}"
-            {{- end }}
             {{- if .Values.redis.redis_channel }}
             - name: MERCURE_REDIS_CHANNEL
               value: "{{ .Values.redis.redis_channel }}"

charts/mercure/values.yaml

@@ -223,7 +223,6 @@ redis:
   password: ~
   dispatch_timer: ~
   subscribers_size: ~
-  dispatcher_pool_size: ~
   redis_channel: mercure
 
 customLabels: ~

redis.Caddyfile

@@ -31,7 +31,6 @@
       username {env.MERCURE_REDIS_USERNAME}
       password {env.MERCURE_REDIS_PASSWORD}
       subscribers_size {env.MERCURE_REDIS_SUBSCRIBERS_SIZE}
-      dispatcher_pool_size {env.MERCURE_REDIS_DISPATCHER_POOL_SIZE}
       redis_channel {env.MERCURE_REDIS_CHANNEL}
     }
 	}

redis.dev.Caddyfile

@@ -37,7 +37,6 @@
       username {env.MERCURE_REDIS_USERNAME}
       password {env.MERCURE_REDIS_PASSWORD}
       subscribers_size {env.MERCURE_REDIS_SUBSCRIBERS_SIZE}
-      dispatcher_pool_size {env.MERCURE_REDIS_DISPATCHER_POOL_SIZE}
       redis_channel {env.MERCURE_REDIS_CHANNEL}
     }
 	}

redis.go

@@ -21,20 +21,13 @@ const (
 
 type RedisTransport struct {
 	sync.RWMutex
-	logger             Logger
-	client             *redis.Client
-	subscribers        *SubscriberList
-	dispatcherPoolSize int
-	dispatcher         chan SubscriberPayload
-	closed             chan any
-	publishScript      *redis.Script
-	closedOnce         sync.Once
-	redisChannel       string
-}
-
-type SubscriberPayload struct {
-	subscriber *LocalSubscriber
-	payload    Update
+	logger        Logger
+	client        *redis.Client
+	subscribers   *SubscriberList
+	closed        chan any
+	publishScript *redis.Script
+	closedOnce    sync.Once
+	redisChannel  string
 }
 
 func NewRedisTransport(
@@ -43,7 +36,6 @@ func NewRedisTransport(
 	username string,
 	password string,
 	subscribersSize int,
-	dispatcherPoolSize int,
 	redisChannel string,
 ) (*RedisTransport, error) {
 	client := redis.NewClient(&redis.Options{
@@ -56,29 +48,26 @@ func NewRedisTransport(
 		return nil, fmt.Errorf("failed to connect to Redis: %w", pong.Err())
 	}
 
-	return NewRedisTransportInstance(logger, client, subscribersSize, dispatcherPoolSize, redisChannel)
+	return NewRedisTransportInstance(logger, client, subscribersSize, redisChannel)
 }
 
 func NewRedisTransportInstance(
 	logger Logger,
 	client *redis.Client,
 	subscribersSize int,
-	dispatcherPoolSize int,
 	redisChannel string,
 ) (*RedisTransport, error) {
 	subscriber := client.PSubscribe(context.Background(), redisChannel)
 
 	subscribeCtx, subscribeCancel := context.WithCancel(context.Background())
 
 	transport := &RedisTransport{
-		logger:             logger,
-		client:             client,
-		subscribers:        NewSubscriberList(subscribersSize),
-		dispatcherPoolSize: dispatcherPoolSize,
-		publishScript:      redis.NewScript(publishScript),
-		dispatcher:         make(chan SubscriberPayload),
-		closed:             make(chan any),
-		redisChannel:       redisChannel,
+		logger:        logger,
+		client:        client,
+		subscribers:   NewSubscriberList(subscribersSize),
+		publishScript: redis.NewScript(publishScript),
+		closed:        make(chan any),
+		redisChannel:  redisChannel,
 	}
 
 	go func() {
@@ -103,19 +92,6 @@ func NewRedisTransportInstance(
 		transport.subscribe(subscribeCtx, subscribeCancel, subscriber)
 	}()
 
-	wg.Add(dispatcherPoolSize)
-	for range dispatcherPoolSize {
-		go func() {
-			defer wg.Done()
-			transport.dispatch()
-		}()
-	}
-
-	go func() {
-		wg.Wait()
-		close(transport.dispatcher)
-	}()
-
 	return transport, nil
 }
 
@@ -154,10 +130,13 @@ func (t *RedisTransport) AddSubscriber(s *LocalSubscriber) error {
 		return ErrClosedTransport
 	default:
 	}
+
 	t.Lock()
 	t.subscribers.Add(s)
 	t.Unlock()
-
+	if s.RequestLastEventID != "" {
+		s.HistoryDispatched(EarliestLastEventID)
+	}
 	s.Ready()
 
 	return nil
@@ -232,24 +211,12 @@ func (t *RedisTransport) subscribe(ctx context.Context, cancel context.CancelFun
 		t.Lock()
 		for _, subscriber := range t.subscribers.MatchAny(&update) {
 			update.Topics = topics
-			t.dispatcher <- SubscriberPayload{subscriber, update}
+			subscriber.Dispatch(&update, false)
 		}
 		t.Unlock()
 	}
 }
 
-func (t *RedisTransport) dispatch() {
-	for {
-		select {
-		case message := <-t.dispatcher:
-			message.subscriber.Dispatch(&message.payload, false)
-		case <-t.closed:
-
-			return
-		}
-	}
-}
-
 var (
 	_ Transport            = (*RedisTransport)(nil)
 	_ TransportSubscribers = (*RedisTransport)(nil)

redis_test.go

@@ -11,14 +11,13 @@ import (
 )
 
 const (
-	redisHost             = "localhost:6379"
-	redisSubscriberSize   = 100000
-	redisDispatchPoolSize = 16
-	redisChannel          = "channel"
+	redisHost           = "localhost:6379"
+	redisSubscriberSize = 100000
+	redisChannel        = "channel"
 )
 
 func initialize() *RedisTransport {
-	transport, _ := NewRedisTransport(zap.NewNop(), redisHost, "", "", redisSubscriberSize, redisDispatchPoolSize, redisChannel)
+	transport, _ := NewRedisTransport(zap.NewNop(), redisHost, "", "", redisSubscriberSize, redisChannel)
 
 	return transport
 }