[azservicebus] Memory leak when receiving messages

[gkampitakis]

Bug Report

• Package in question azure-sdk-for-go/sdk/messaging/azservicebus
• SDK version v1.5.0
• go version 1.21

• What happened?

We are trying to move away from the deprecated azure sdk for service bus https://github.com/Azure/azure-service-bus-go and use e the new in one of our services. This service is heavily reading on messages from a Service Bus Queue. We are noticing an increasing usage of memory.

Also from profiling data we can see that the Heap size increase is happening at go-ampq code

• What did you expect or want to happen?

I would expect the new SDK to have a different memory footprint (either higher or lower memory consumption) but after some time i was expecting the memory usage of the service to be stable as before and not increasing time over time.

• How can we reproduce it?

Not sure how easy it is, as we need to leave the service running for a couple of days to start seeing the increase in memory usage.

• Anything we should know about your environment.

Data shared above are from staging environment. I am expecting bigger usage on a prod environment with much higher traffic.

Our setup is we are using a number of worker goroutines each one receiving messages from a client.Receiver in parallel.

[jhendrixMSFT]

Looks related/due to Azure/go-amqp#314.

[gkampitakis]

Indeed it does looks related. Thanks for linking it 👍

[jhendrixMSFT]

Would you be able to share any additional info about your app? I've copied the questions from the linked issue.

• are there multiple senders/receivers?
• are the senders/receivers short or long lived?
• what are typical message sizes and how big is the largest message (best guess is fine)?

[gkampitakis]

• are there multiple senders/receivers?

Yes there are. We have a number of producer goroutines that are receiving messages and a number of consumer goroutines that process messages and communicate via a go channel. At the start of the service we create a pool of receivers client.NewReceiverForQueue for receiving messages in parallel.

• are the senders/receivers short or long lived?

They are long lived. At service start up we create one client and from that client we create the needed receivers and senders that are used throughout the life of the service.

• what are typical message sizes and how big is the largest message (best guess is fine)?

The average message size is ~3Kib and the max message size can be up to 256Kib.

[jhendrixMSFT]

Thanks this is helpful.

Is the number of senders/receivers dynamic or fixed? Sounds like it's fixed but I don't want to make assumptions. If it is dynamic, does the number of senders/receivers have a cap and does it ever shrink?

At the go-amqp level, each sender/receiver has its own message buffer, and the buffer size will grow to the largest sent/received message but will never shrink.

If the number of senders/receivers is fixed, then I would expect that the total buffer size would end up being "count senders/receivers * max message size" and level out. I'm thinking about this in the context of the linked issue. It could be that what you're observing is something different.

[gkampitakis]

Is the number of senders/receivers dynamic or fixed? Sounds like it's fixed but I don't want to make assumptions.

That is correct, it's a fixed number.

If the number of senders/receivers is fixed, then I would expect that the total buffer size would end up being "count senders/receivers * max message size" and level out. I'm thinking about this in the context of the linked issue. It could be that what you're observing is something different.

This is what I expected as well. Our service doesn't consume much memory so I would expect even if there was a memory increase at some point it will level out (given the service is receiving stable traffic).

But this is how our memory is trending in a period of 3 weeks (on staging env)

On the profile I shared above, addUnsettled looks like a potential point of leakage. Which just adds a key in a map if this map never get's empty can cause this type of slow leakage, unfortunately though I am not familiar with the go-ampq to understand the path of addUnsettled and deleteUnsettled.

[jhendrixMSFT]

Thanks for clarifying that point. I will take a look at this specific angle.

[richardpark-msft]

Interesting, possibly related to maps in Go never shrinking, even when keys are deleted?

golang/go#20135

[gkampitakis]

This was my 1st thought, but that should not be the issue. If keys where getting recycled in the map (adding and removing them in a steady flow), the map length should remain the same (grow until it reaches a stable point == the max number of keys inside the map at point in time), but here IF this is the issue and culprit, it looks like we are adding more keys than what we are removing right?

[jhendrixMSFT]

How do you configure the ReceiveMode for the receivers?

[gkampitakis]

How do you configure the ReceiveMode for the receivers?

Using the default ReceiveModePeekLock.

[jhendrixMSFT]

Thanks. From a go-amqp perspective, it means the receiver is using ReceiverSettleModeSecond, meaning that we don't remove the message from the unsettled map until we receive confirmation of settlement from Service Bus (that happens here).

Just to confirm, you're calling one of the message settlement methods as documented here right? I'm pretty sure you must be else the receiving link would run out of credit at some point and no more messages would be received (Richard can correct me if I'm wrong here).

[gkampitakis]

That is correct. We either Complete or Abandon the message.

Something to note though is we don't use the same receiver instance that called ReceiveMessages. So different receiver ReceiveMessages and different receiver "settles" them. Can that be the issue? Is the settled map a per receiver object?

[richardpark-msft]

That is correct. We either Complete or Abandon the message.

Something to note though is we don't use the same receiver instance that called ReceiveMessages. So different receiver ReceiveMessages and different receiver "settles" them. Can that be the issue? Is the settled map a per receiver object?

Ah, that could definitely be it. If you complete on a different Receiver than you received on the completion goes through a different path which means the original AMQP receiver won't be notified.

[richardpark-msft]

@gkampitakis, just curious why you complete on a different Receiver than you received on?

[jhendrixMSFT]

That is indeed the problem. The Receiver that received the message will add the Message to its unsettled map. So, unless you settle the message from the same Receiver it will never be removed.

@richardpark-msft we should document this requirement and at some point see if we can remove it.

[gkampitakis]

@richardpark-msft

We are using a pool of receivers to ReceiveMessages concurrently from multiple goroutines, process them and then complete them. So from the pool you pick a random (available) receiver to ReceiveMessages and then have a single receiver that settles the messages.

Thanks both for your time spent on this 🙇. I couldn't have figure it out. Will try to work around this and use the same receiver for both operations.

[jhendrixMSFT]

@gkampitakis FYI we'll be releasing an updated go-amqp (should be today) that fixes the cross-receiver message settlement memory leak.

[gkampitakis]

I have deployed my service with the latest go-ampq, but unfortunately I don't see any improvement 🤔

go-ampq has 55% of all allocated objects in my service.

and memory still trending upwards.

[jhendrixMSFT]

I've been doing a little reading up on how maps are implemented in Go, and I think this behavior is expected, although unwanted for our purposes.

In Receiver.unsettledMessages we use a message's delivery tag as the map's key. Since delivery tags are required to be unique per link, the number of buckets in a map will slowly increase as the keys will never be recycled. This growth is really easy to see in a benchmark.

func BenchmarkReceiverUnsettledCount(b *testing.B) {
	r := &Receiver{}
	r.unsettledMessages = map[string]struct{}{}
	deliveryTag := [32]byte{}
	fakeMsg := &Message{
		DeliveryTag: deliveryTag[:],
	}
	for i := 0; i < b.N; i++ {
		for unsettledCount := uint32(0); unsettledCount < 1000000; unsettledCount++ {
			// use the unsettled count as the delivery tag
			binary.BigEndian.PutUint32(deliveryTag[:], unsettledCount)
			r.addUnsettled(fakeMsg)
			r.deleteUnsettled(fakeMsg)
		}
	}
}

PS C:\git\Azure\go-amqp> go test -bench BenchmarkReceiverUnsettledCount -benchmem
goos: windows
goarch: amd64
pkg: github.com/Azure/go-amqp
cpu: AMD Ryzen 9 7950X3D 16-Core Processor
BenchmarkReceiverUnsettledCount-8             24          49150729 ns/op        32000029 B/op    1000000 allocs/op
PASS
ok      github.com/Azure/go-amqp        28.627s

We can fix this as described in Azure/go-amqp#317. There might be other options too.

[gkampitakis]

Ooops closed the issue by mistake.

Thanks for posting this benchmark it's helpful but, I am not sure that shows an issue on the go map increasing.

I just changed the code to have an intermediate step of casting to string

func (r *Receiver) addUnsettled(msg *Message) {
	r.unsettledMessagesLock.Lock()
	t := string(msg.DeliveryTag) // this changed
	r.unsettledMessages[t] = struct{}{}
	r.unsettledMessagesLock.Unlock()
}

func (r *Receiver) deleteUnsettled(msg *Message) {
	r.unsettledMessagesLock.Lock()
	t := string(msg.DeliveryTag) // this changed
	delete(r.unsettledMessages, t)
	r.unsettledMessagesLock.Unlock()
}

and this is shown from the profile.

go tool pprof mem.out  

ROUTINE ======================== github.com/Azure/go-amqp.(*Receiver).addUnsettled in /Users/george.kampitakis/Dev/go-amqp/receiver.go
    1.11GB     1.11GB (flat, cum) 90.91% of Total
         .          .    349:func (r *Receiver) addUnsettled(msg *Message) {
         .          .    350:   r.unsettledMessagesLock.Lock()
    1.11GB     1.11GB    351:   t := string(msg.DeliveryTag)
         .          .    352:   r.unsettledMessages[t] = struct{}{}
         .          .    353:   r.unsettledMessagesLock.Unlock()
         .          .    354:}
         .          .    355:
         .          .    356:func (r *Receiver) deleteUnsettled(msg *Message) {

and if i change the code a bit more to not allocate from byte to string

func (r *Receiver) addUnsettled(msg *Message) {
	r.unsettledMessagesLock.Lock()
	t := byteSliceToString(msg.DeliveryTag)
	r.unsettledMessages[t] = struct{}{}
	r.unsettledMessagesLock.Unlock()
}

func (r *Receiver) deleteUnsettled(msg *Message) {
	r.unsettledMessagesLock.Lock()
	t := byteSliceToString(msg.DeliveryTag)
	delete(r.unsettledMessages, t)
	r.unsettledMessagesLock.Unlock()
}

func byteSliceToString(b []byte) string {
	return unsafe.String(unsafe.SliceData(b), len(b))
}

I see this

➜  go-amqp git:(main) ✗ go test -bench BenchmarkReceiverUnsettledCount -memprofile mem.out
goos: darwin
goarch: arm64
pkg: github.com/Azure/go-amqp
BenchmarkReceiverUnsettledCount-10            31          38238093 ns/op               4 B/op          0 allocs/op
PASS
ok      github.com/Azure/go-amqp        18.973s

[jhendrixMSFT]

You're correct, great catch (I should have looked deeper).

I'm looking over the message settlement code and I see a few other possibilities.

• we fail to send the disposition frame (i.e. Receiver.AcceptMessage() et al returns an error)
• the Receiver never receives the acknowledging disposition frame

For the first case, do you ever see Complete/Abandon fail? If so, how do you handle that (retry etc)?

@richardpark-msft I just had a thought. If the receiver falls back to using the management link, will service bus send us an acknowledging disposition frame? If not, then that could also explain it. When the receiver is in mode second (which it is in this case), we'll remove the entry from the unsettled map upon receiving the ack'ing disposition frame. Is there an easy way to tell when settlement falls back to the management link?

[richardpark-msft]

I've been doing a little reading up on how maps are implemented in Go, and I think this behavior is expected, although unwanted for our purposes.

In Receiver.unsettledMessages we use a message's delivery tag as the map's key. Since delivery tags are required to be unique per link, the number of buckets in a map will slowly increase as the keys will never be recycled. This growth is really easy to see in a benchmark.

(I reference it here if anyone wants to read the Golang issue about it: #22157 (comment))

[richardpark-msft]

You're correct, great catch (I should have looked deeper).

I'm looking over the message settlement code and I see a few other possibilities.

• we fail to send the disposition frame (i.e. Receiver.AcceptMessage() et al returns an error)
• the Receiver never receives the acknowledging disposition frame

For the first case, do you ever see Complete/Abandon fail? If so, how do you handle that (retry etc)?

@richardpark-msft Richard Park (DEVDIV) FTE I just had a thought. If the receiver falls back to using the management link, will service bus send us an acknowledging disposition frame? If not, then that could also explain it. When the receiver is in mode second (which it is in this case), we'll remove the entry from the unsettled map upon receiving the ack'ing disposition frame. Is there an easy way to tell when settlement falls back to the management link?

The entire interaction, at that point, will occur over the management link. It seems that since the only reason we have the map is to do some logging that we're best just removing it and maybe just making it a counter instead. You only clear it when you get a successful disposition frame (assuming) so it seems like it'd still be accurate with what you've done with tracking the original receiver.

So I think that work was still worth it and doing it as a counter should eliminate memory concerns.

[jhendrixMSFT]

When the receiver is in mode second, we'd still decrement the unsettled count upon receiving the ack'ing disposition frame.

[jhendrixMSFT]

Richard and I spoke about this offline. The reason the go-amqp@v1.0.3 change didn't fix this is because cross-receiver settlement at the azservicebus level bypasses the go-amqp layer.

I've released go-amqp@v1.0.4 which replaces the unsettled messages map with a count which should fix this.

[gkampitakis]

I have deployed go-amqp 1.0.4 since morning and looks more promising. Thank you.

[jhendrixMSFT]

@gkampitakis thanks for your help with this and providing all the diagnostic info.