[RSDK-8666] Use Stoppable Workers #408

bashar-515 · 2025-01-22T20:06:00Z

WIP

This PR is a continuation of #402 re: this comment from @benjirewis (i.e., it updates rpc/wrtc_server.go, rpc/wrtc_call_queue_memory.go, and rpc/wrtc_call_queue_mongodb.go to use StoppableWorkers - and some other files as a side effect).

rpc/wrtc_server.go
rpc/wrtc_call_queue_memory.go
rpc/wrtc_call_queue_mongodb.go

bashar-515 · 2025-01-22T20:06:53Z

rpc/wrtc_call_queue_memory.go

-			}
-			select {
-			case <-cancelCtx.Done():
+			if ctx.Err() != nil {


Removing this check results in a deadlock/hang when running tests in the rpc package.

Yeah we'll still want to check context. This looks correct. Ticker case should be removed as you've done, too. Just see my comment about not needing the for loop surrounding this at all.

bashar-515 · 2025-01-22T20:15:09Z

rpc/wrtc_server_channel.go

@@ -37,7 +37,7 @@ func newWebRTCServerChannel(
 	logger utils.ZapCompatibleLogger,
 ) *webrtcServerChannel {
 	base := newBaseChannel(
-		server.ctx,
+		server.processHeadersWorkers.Context(),


An artifact of no longer storing a context.Context inside of the webrtcServer struct.

Gotcha; this makes sense to me. There's a hierarchy here of WebRTC server owns WebRTC server channels owns WebRTC server streams. We want the context cancelations to propagate in that order. So, I think what you have here is correct. Once Stop() is called on the WebRTC server, we should cancel all the way down.

Agreed this will do the right thing. It would be more customary to just have newWebRTCServerChannel accept a context to bound its lifetime, but no need to change how things are structured right now.

bashar-515 · 2025-01-22T20:15:44Z

rpc/wrtc_server_stream.go

-	if err := s.ch.server.ctx.Err(); err != nil {
-		s.ch.server.processHeadersWorkers.Done()
+	// in which case we return.
+	if err := s.ch.server.processHeadersWorkers.Context().Err(); err != nil {


Also an artifact of removing the webrtcServer struct's context.Context. See above.

@dgottlieb do you remember why we need this context check before even taking a ticket?

I'm not seeing anything arguing the ordering is important. I think the ordering was this was simply more to make the relationship between Adding and the context canceling explicit. Because Stop both cancels the context and waits on the waitgroup.

My expectation is that by moving to StoppableWorkers, we no longer need this if-block. StoppableWorkers does that for us before calling its worker function.

Cool; that's my thought, too. Wanna remove this whole if block @bashar-515 ?

benjirewis · 2025-01-27T15:49:42Z

rpc/wrtc_call_queue_mongodb.go

This file is pretty critical to the functioning of distributed signaling for Viam. I'm ok not even making changes to it, since we haven't yet seen any issues in its usage of wait groups/contexts. If you and @dgottlieb want to get stoppable workers into this file, though, then I think we should test a local version of app + MongoDB to ensure signaling still works. Can also probably do it with an app PR that creates a temporary staging environment.

benjirewis · 2025-01-27T15:50:10Z

rpc/wrtc_server_stream.go

-	if err := s.ch.server.ctx.Err(); err != nil {
-		s.ch.server.processHeadersWorkers.Done()
+	// in which case we return.
+	if err := s.ch.server.processHeadersWorkers.Context().Err(); err != nil {


@dgottlieb do you remember why we need this context check before even taking a ticket?

dgottlieb · 2025-01-28T14:35:59Z

rpc/wrtc_call_queue_memory.go

-						delete(hostQueue.activeOffers, offerID)
-					}
+	queue.activeBackgroundWorkers = utils.NewStoppableWorkerWithTicker(5*time.Second, func(ctx context.Context) {
+		if ctx.Err() != nil {


Before, this was necessary to exit the loop. But this is no longer serving a purpose. The stoppable workers caller checks the state/context before calling this method.

dgottlieb · 2025-01-28T14:47:02Z

rpc/wrtc_call_queue_memory.go

-	queue.activeBackgroundWorkers.Add(1)
-	utils.PanicCapturingGo(func() {
-		queue.activeBackgroundWorkers.Done()
+	queue.activeBackgroundWorkers.Add(func(ctx context.Context) {
 		select {
 		case <-sendCtx.Done():
 		case <-ctx.Done():


This is no longer the same context. Before this ctx was the one passed into SendOfferInit. Now we've shadowed that with the activeBackgroundWorkers context. Which is already covered by the sendCtx.

Taking a step back. In the existing code, I'm not sure what the relationship is between the caller's context and the queue.cancelCtx. I think* the SendOfferInit ctx is the gRPC request's context. In which case we'd want to revert to using that one.

nice catch totally missed this

dgottlieb · 2025-01-28T14:53:49Z

rpc/wrtc_call_queue_mongodb.go


 		csStateUpdates:        make(chan changeStreamStateUpdate),
 		callExchangeSubs:      map[string]map[*mongodbCallExchange]struct{}{},
 		waitingForNewCallSubs: map[string]map[*mongodbNewCallEventHandler]struct{}{},
 		activeAnswerersfunc:   &activeAnswerersfunc,
 	}

-	queue.activeBackgroundWorkers.Add(2)
+	// TODO(RSDK-8666): using StoppableWorkers is causing a data race


Did we find any more detail about what was happening here?

dgottlieb · 2025-01-28T14:55:32Z

rpc/wrtc_call_queue_mongodb.go

 		defer queue.csManagerSeq.Add(1) // helpful on panicked restart
 		select {
-		case <-queue.cancelCtx.Done():
+		case <-queue.activeStoppableWorkers.Context().Done():


I would expect this to be replaced with the input ctx.

dgottlieb · 2025-01-28T14:58:15Z

rpc/wrtc_call_queue_mongodb.go

-	utils.ManagedGo(queue.operatorLivenessLoop, queue.activeBackgroundWorkers.Done)
-	utils.ManagedGo(queue.changeStreamManager, queue.activeBackgroundWorkers.Done)
+	// TODO(RSDK-8666): using StoppableWorkers is causing a data race
+	queue.activeStoppableWorkers.Add(func(ctx context.Context) { queue.operatorLivenessLoop() })


We should change the queue.operatorLivenessLoop and queue.changeStreamManager functions to accept the context. And change their select ... queue.activeStoppableWorkers.Context() to instead select on the function input ctx.

dgottlieb · 2025-01-28T16:00:04Z

rpc/wrtc_call_queue_mongodb.go

@@ -344,7 +339,7 @@ func (queue *mongoDBWebRTCCallQueue) operatorLivenessLoop() {
 	ticker := time.NewTicker(operatorStateUpdateInterval)
 	defer ticker.Stop()
 	for {
-		if !utils.SelectContextOrWaitChan(queue.cancelCtx, ticker.C) {
+		if !utils.SelectContextOrWaitChan(queue.activeStoppableWorkers.Context(), ticker.C) {


As per the comment on having operatorLivenessLoop accepting a ctx, this is the line that caught my attention. We should just be checking the input context and not need to do this lookup on the stoppable workers itself.

dgottlieb · 2025-01-28T16:01:21Z

rpc/wrtc_call_queue_mongodb.go

@@ -500,11 +496,11 @@ func (queue *mongoDBWebRTCCallQueue) changeStreamManager() {
 		queue.csStateMu.Unlock()
 		activeHosts.Set(queue.operatorID, int64(len(hosts)))

-		nextCSCtx, nextCSCtxCancel := context.WithCancel(queue.cancelCtx)
+		nextCSCtx, nextCSCtxCancel := context.WithCancel(queue.activeStoppableWorkers.Context())


dgottlieb · 2025-01-28T16:04:04Z

rpc/wrtc_call_queue_mongodb.go

@@ -235,11 +230,11 @@ func NewMongoDBWebRTCCallQueue(
 			})
 			queue.subscriptionManager(newState.ChangeStream)


Also pass the ctx into subscriptionManager

dgottlieb · 2025-01-28T16:04:32Z

rpc/wrtc_call_queue_mongodb.go

@@ -677,14 +673,14 @@ func (queue *mongoDBWebRTCCallQueue) processNextSubscriptionEvent(next mongoutil
 func (queue *mongoDBWebRTCCallQueue) subscriptionManager(currentCS <-chan mongoutils.ChangeEventResult) {
 	var waitForNextCS bool
 	for {
-		if queue.cancelCtx.Err() != nil {
+		if queue.activeStoppableWorkers.Context().Err() != nil {


As per above, if we add in the ctx to this call, we can swap out accessing activeStoppableWorkers here.

dgottlieb · 2025-01-28T16:10:35Z

rpc/wrtc_call_queue_mongodb.go

@@ -889,7 +885,7 @@ func (queue *mongoDBWebRTCCallQueue) SendOfferInit(
 	sendCtx, sendCtxCancel := context.WithDeadline(ctx, offerDeadline)

 	// need to watch before insertion to avoid a race
-	sendAndQueueCtx, sendAndQueueCtxCancel := utils.MergeContext(sendCtx, queue.cancelCtx)
+	sendAndQueueCtx, sendAndQueueCtxCancel := utils.MergeContext(sendCtx, queue.activeStoppableWorkers.Context())


I suppose we can keep this to minimize the changes as part of this refactor. But this feels clunky. We're just merging here to reduce the number of cases inside the below selects by one.

And I think there are only two select statements.

dgottlieb · 2025-01-28T16:15:10Z

rpc/wrtc_server_channel.go

@@ -37,7 +37,7 @@ func newWebRTCServerChannel(
 	logger utils.ZapCompatibleLogger,
 ) *webrtcServerChannel {
 	base := newBaseChannel(
-		server.ctx,
+		server.processHeadersWorkers.Context(),


Agreed this will do the right thing. It would be more customary to just have newWebRTCServerChannel accept a context to bound its lifetime, but no need to change how things are structured right now.

bashar-515 · 2025-01-28T18:43:33Z

No longer upgrading rpc/wrtc_call_queue_mongodb.go. Ticket tracking it here.

bashar-515 added 9 commits January 22, 2025 11:35

add TODO comments

5a92489

lint

99c5ece

use stoppable workers in constructor with ticker

c8a02cf

use stoppable workers when initiating offer

80eb5df

replace activeBackgroundWorkers

2fd4869

no longer store context

7a7b604

remove lock in constructor

b59f413

use stoppable workers in webRTC server

7a86540

no longer store context and cancel function in webRTC server

e57158a

viambot added the safe to test Mark as safe to test label Jan 22, 2025

bashar-515 commented Jan 22, 2025

View reviewed changes

remove context comments

642c943

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 22, 2025

remove TODO comment

d46bd50

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 22, 2025

bashar-515 commented Jan 22, 2025

View reviewed changes

add stoppable workers to mongoDB webRTC call queue

0b5c310

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 22, 2025

use only stoppable workers throughout constructor

41309b6

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 22, 2025

use only stoppable workers in SendOfferInit

54731bc

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 22, 2025

viambot added the safe to test Mark as safe to test label Jan 24, 2025

no longer store context in struct

0840eba

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 24, 2025

lint

7eb91ef

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 24, 2025

bashar-515 requested review from benjirewis and dgottlieb January 24, 2025 21:51

benjirewis reviewed Jan 27, 2025

View reviewed changes

dgottlieb reviewed Jan 28, 2025

View reviewed changes

no longer check for context error

6a5d752

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 28, 2025

restore rpc/wrtc_call_queue_mongodb.go file

8241b32

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 28, 2025

no longer check parent context in ticker

16e2f4b

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 28, 2025

use caller context

ae45b2b

viambot added safe to test Mark as safe to test and removed safe to test Mark as safe to test labels Jan 28, 2025

bashar-515 requested review from benjirewis and dgottlieb January 28, 2025 18:25

dgottlieb approved these changes Jan 28, 2025

View reviewed changes

bashar-515 merged commit cf6c56e into viamrobotics:main Jan 28, 2025
6 checks passed

bashar-515 deleted the RSDK-8666-part-2 branch January 28, 2025 18:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[RSDK-8666] Use Stoppable Workers #408

[RSDK-8666] Use Stoppable Workers #408

bashar-515 commented Jan 22, 2025 •

edited

Loading

bashar-515 Jan 22, 2025

benjirewis Jan 23, 2025

bashar-515 Jan 22, 2025

benjirewis Jan 23, 2025

dgottlieb Jan 28, 2025

bashar-515 Jan 22, 2025 •

edited

Loading

benjirewis Jan 27, 2025

dgottlieb Jan 27, 2025

benjirewis Jan 27, 2025

benjirewis Jan 27, 2025

benjirewis Jan 27, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

bashar-515 Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

dgottlieb Jan 28, 2025

bashar-515 commented Jan 28, 2025

		@@ -235,11 +230,11 @@ func NewMongoDBWebRTCCallQueue(
		})
		queue.subscriptionManager(newState.ChangeStream)

[RSDK-8666] Use Stoppable Workers #408

[RSDK-8666] Use Stoppable Workers #408

Conversation

bashar-515 commented Jan 22, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bashar-515 Jan 22, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bashar-515 commented Jan 28, 2025

bashar-515 commented Jan 22, 2025 •

edited

Loading

bashar-515 Jan 22, 2025 •

edited

Loading