Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix goroutine leak on closing #259

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Fix goroutine leak on closing #259

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8e87ca0
Fix goroutine leak on closing
ItalyPaleAle Jun 3, 2022
bf5e6d2
Merge branch 'master' into patch-1
jhendrixMSFT Jul 27, 2022
02450e9
doc that calling Close() is required
jhendrixMSFT Aug 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion eph/eph.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,7 @@ func (h *EventProcessorHost) UnregisterHandler(ctx context.Context, id HandlerID
}

// Start begins processing of messages for registered handlers on the EventHostProcessor. The call is blocking.
// You MUST call Close() when the event processor host is no longer required.
func (h *EventProcessorHost) Start(ctx context.Context) error {
span, ctx := startConsumerSpanFromContext(ctx, "eph.EventProcessorHost.Start")
defer span.End()
Expand Down Expand Up @@ -329,7 +330,8 @@ func (h *EventProcessorHost) Start(ctx context.Context) error {
return h.Close(ctx)
}

// StartNonBlocking begins processing of messages for registered handlers
// StartNonBlocking begins processing of messages for registered handlers.
// You MUST call Close() when the event processor host is no longer required.
func (h *EventProcessorHost) StartNonBlocking(ctx context.Context) error {
span, ctx := startConsumerSpanFromContext(ctx, "eph.EventProcessorHost.StartNonBlocking")
defer span.End()
Expand Down
6 changes: 6 additions & 0 deletions eph/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ type (
done func()
leaseRenewalInterval time.Duration
receiverMu sync.Mutex
close chan struct{}
}

ownerCount struct {
Expand All @@ -69,6 +70,7 @@ func newScheduler(eventHostProcessor *EventProcessorHost) *scheduler {
processor: eventHostProcessor,
receivers: make(map[string]*leasedReceiver),
leaseRenewalInterval: DefaultLeaseRenewalInterval,
close: make(chan struct{}),
}
}

Expand All @@ -83,6 +85,9 @@ func (s *scheduler) Run(ctx context.Context) {
case <-ctx.Done():
s.dlog(ctx, "shutting down scan")
return
case <-s.close:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In line 188, we actually call the s.done() function, which cancels the context we used to spin up this entire thing. So I think this s.close is not needed.

s.dlog(ctx, "shutting down scan")
return
default:
s.scan(ctx)
skew := time.Duration(rand.Intn(1000)-500) * time.Millisecond
Expand Down Expand Up @@ -193,6 +198,7 @@ func (s *scheduler) Stop(ctx context.Context) error {
_, _ = s.processor.leaser.ReleaseLease(ctx, lr.lease.GetPartitionID())
}

close(s.close)
return lastErr
}

Expand Down