Re-introduce fixed limiter to streaming calls to mitigate OOM issues.

wetware · Jan 12, 2023 · 9f8ddf3 · 9f8ddf3
1 parent b6c72a7
commit 9f8ddf3
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 2 deletions.
diff --git a/pkg/csp/chan.go b/pkg/csp/chan.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 
 	"capnproto.org/go/capnp/v3"
+	"capnproto.org/go/capnp/v3/flowcontrol"
 	casm "github.com/wetware/casm/pkg"
 	"github.com/wetware/casm/pkg/util/stream"
 	"github.com/wetware/ww/internal/api/channel"
@@ -207,7 +208,7 @@ func (s Sender) Send(ctx context.Context, v Value) (casm.Future, capnp.ReleaseFu
 // for a given sender.
 func (s Sender) NewStream(ctx context.Context) SendStream {
 	sender := channel.Sender(s)
-	// TODO:  use BBR once scheduler bug is fixed
+	sender.SetFlowLimiter(flowcontrol.NewFixedLimiter(1024)) // TODO:  use BBR once scheduler bug is fixed
 
 	return SendStream{
 		ctx:    ctx,

diff --git a/pkg/pubsub/topic.go b/pkg/pubsub/topic.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"capnproto.org/go/capnp/v3"
+	"capnproto.org/go/capnp/v3/flowcontrol"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/lthibault/log"
 
@@ -56,6 +57,7 @@ func (t Topic) Publish(ctx context.Context, b []byte) error {
 // FlowLimiter.
 func (t Topic) NewStream(ctx context.Context) Stream {
 	// TODO:  use BBR once scheduler bug is fixed
+	api.Topic(t).SetFlowLimiter(flowcontrol.NewFixedLimiter(1024))
 
 	cherr := make(chan error, 1)
 	done := make(chan struct{})
@@ -233,7 +235,7 @@ func (t topicServer) Subscribe(ctx context.Context, call MethodSubscribe) error
 	defer sub.Cancel()
 
 	sender := call.Args().Chan()
-	// TODO:  use BBR once scheduler bug is fixed
+	sender.SetFlowLimiter(flowcontrol.NewFixedLimiter(1024)) // TODO:  use BBR once scheduler bug is fixed
 
 	t.log.Debug("registered subscription handler")
 	defer t.log.Debug("unregistered subscription handler")