fix: Ensure ActiveMessageCount is updated in a thread safe manner

[ashovlin]

Issue #, if available: N/A

Description of changes: @philasmar pointed out when working on PR #34 that if you call DefaultMessageManager.ProcessMessageAsync with a sufficient number of messages and then await all of the resulting tasks, ActiveMessageCount wouldn't arrive back at 0 as expected.

My PR #24 introduced a flaw: since the ++ and -- operators will call the getter then the setter, they'd release and reacquire the lock during the increment or decrement allowing another thread to update the value in the middle.

public int ActiveMessageCount 
{
        get
        {
            lock (_activeMessageCountLock)
            {
                return _activeMessageCount;
            }
        }
        set // Before, buggy
        {
            lock (_activeMessageCountLock)
            {                
                _activeMessageCount = value;
            }
        }
    }
}
public Task WaitAsync(TimeSpan timeout)
{
    ActiveMessageCount++;

    // do work

    ActiveMessageCount--;
}

This PR moves the setter to a private method to ensure the lock is held during the entire update. I also considered Interlocked.Increment and Decrement, but think that our internal team is more used to the lock pattern.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[slang25]

It might be worth comparing notes on this stuff, we had a similar poller implementation in JustSaying 5 and moved away from it. I did a write-up about the problems it had, some of it was due to the implementation, but in general an eager check of "how many free workers do we have" the moment one has finished leads to choppy message consumption when there's back-pressure to work through, as well as all of the complications you are facing about keeping count accurately (which is essential).

The write-up is here (that ones more about the shared throttling mechanism we had) and discussion here (this describes the choppiness issue which is one of the factors that lead to a Channels based multiplexer).

We switched out to a mechanism that pre-fetches 1 batch lazily, with the message visibility renewal process you have it might be a better approach. I see you have the possibility to allow replacing the poller, it would be handy to us if the abstraction didn't necessitate the presence of an ActiveMessageCount.

What we have by default now essentially means as soon as 1 message is being processed from a batch, the next batch is fetched. The downside is naturally that you might take a message earlier than you are ready to process it, but again the visibility renewal takes care of that.

I'd also suggest reviewing the default visibility timeout and renewal values. 20 seconds and 18 for renewal might be cutting it fine. 30 and 25 gives a bit more room for error.

[philasmar]

I am fine with what you are doing, but I'd prefer if we write this in a more reusable and user intuitive way. That being sad, I think it would be nice to make UpdateActiveMessageCount a more generic method that can update any int in a thread-safe way. Also it would be nice if we turn this into an extension method which would replace ActiveMessageCount++ and ActiveMessageCount-- by ActiveMessageCount.UpdateValue(1) and ActiveMessageCount.UpdateValue(-1) respectively.

[ashovlin]

@slang25 - thanks for taking a look. I agree that we've arrived at a similar initial approach to what you've described in justeattakeaway/JustSaying#422 (comment), we'll look over everything you shared.

[normj]

I'm good with the change. Obviously we might change things as we go through @slang25 links but that is a larger order change.

[ashovlin]

@slang25 - I agree with Norm's comment above, I'd rather merge this now to fix an issue in the current implementation so we can unblock work in other areas. I created #38 to track reviewing and discussing the poller design more holistically.

@philasmar - I'd prefer not to overload ++ and -- so that it's obvious we're doing extra work for concurrency, and am not too concerned with reusability since this is internal to the DefaultMessageManager and not exposed on the interface. Let me know if you still disagree.