[feat] Support memory limit for producer.

[shibd]

Master Issue: #927

Motivation

#927

Modifications

• Added channel_cond class to enhance sync.cond to support wait that can accept context.
• Added the memory_limit_controller class.
• Producer adds memory limit relate logic.

Verifying this change

• Added channel_cond_test to cover channel_cond
• Added memory_limit_controller_test to cover mem_mory_limit_controller
• Added TestMemLimitRejectProducerMessages and TestMemLimitContextCancel to cover producer relate logic.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (yes)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (GoDocs)

[BewareMyPower]

And I don't get why do we have to wrap a conditional variable? We can achieve the same goal with the channel itself, or just reuse the Semaphore here

[shibd]

And I don't get why do we have to wrap a conditional variable?

The native cond does not support timeout wait, and mainly wants to control the timeout or cancel the wait through context.

https://github.com/apache/pulsar-client-go/pull/955/files#diff-d4362cb0b0046577b68820bba2bde37b97ca5d7db7237e4e4b5eac54d7aa8bc3R89-R120

We can achieve the same goal with the channel itself,

Do you want mem_controller_limit all the methods into the channel first, and then select to read the signal and process it? Similar to #957 ackGroupingTracker ?

or just reuse the Semaphore here

This semaphore can only support acquire or release 1 signal.

[BewareMyPower]

This semaphore can only support acquire or release 1 signal.

Yes. But we can adopt a similar way to handle this case.

For example, if we add a ch chan bool field to memoryLimitController, we can implement blocking ReserveMemory method like:

func (m *memoryLimitController) ReserveMemory(ctx context.Context, size int64) bool {
        // NOTE: maybe we need to check if m.currentUsage > m.limit first
        currentUsage := atomic.AddInt64(&m.currentUsage, size)
        for currentUsage > m.limit {
                select {
                case <-m.ch:
                        currentUsage = atomic.LoadInt64(&m.currentUsage)
                case <-ctx.Done(): // NOTE: Not sure if we need to reset some fields here
                        return false
                }
        }
        return true
}

func (m *memoryLimitController) ReleaseMemory(size int64) {
        newUsage := atomic.AddInt64(&m.currentUsage, -size)
        // newUsage+size > m.limit means m was blocked in ReserveMemory method
        if newUsage+size > m.limit && newUsage <= m.limit {
                m.ch <- true
        }
}

The code above is not verified yet. But with the channel the code looks more simple and clear.

[shibd]

This semaphore can only support acquire or release 1 signal.

Yes. But we can adopt a similar way to handle this case.

For example, if we add a ch chan bool field to memoryLimitController, we can implement blocking ReserveMemory method like:

func (m *memoryLimitController) ReserveMemory(ctx context.Context, size int64) bool {
        // NOTE: maybe we need to check if m.currentUsage > m.limit first
        currentUsage := atomic.AddInt64(&m.currentUsage, size)
        for currentUsage > m.limit {
                select {
                case <-m.ch:
                        currentUsage = atomic.LoadInt64(&m.currentUsage)
                case <-ctx.Done(): // NOTE: Not sure if we need to reset some fields here
                        return false
                }
        }
        return true
}

func (m *memoryLimitController) ReleaseMemory(size int64) {
        newUsage := atomic.AddInt64(&m.currentUsage, -size)
        // newUsage+size > m.limit means m was blocked in ReserveMemory method
        if newUsage+size > m.limit && newUsage <= m.limit {
                m.ch <- true
        }
}

The code above is not verified yet. But with the channel the code looks more simple and clear.

This implementation has a problem, and the broadcasting cannot be implemented using channels.

For example:

1. currentUsage = 100, limit = 100
2. goroutine 1 call ReseveMemory(10), it will blocked.
3. goroutine 2 call ReseveMemory(10), it will blocked.
4. goroutine 3 call ReleaseMemory(20), it only wake one goroutine(1 or 2). The expectation is that both are woken up and return true.

// NOTE: maybe we need to check if m.currentUsage > m.limit first

Like this note. If we want to handle this case, we may need to introduce a variable waitNum. So I feel like it's complicated to implement mem_controller_limit, and it's better to introduce channel_cond.

[BewareMyPower]

If we want to handle this case, we may need to introduce a variable waitNum

I see. Currently let's go head with the conditional variable, but don't expose this struct to users.

[RobertIndie]

Like this note. If we want to handle this case, we may need to introduce a variable waitNum. So I feel like it's complicated to implement mem_controller_limit, and it's better to introduce channel_cond.

@shibd For this case, another workaround is to close the channel instead of sending value. But it's actually what this PR does.