Points count in SeriesBuffer can change independent of actual stored points

[KevinWMatthews]

Is it possible to document the preconditions for accessing SeriesBuffer safely across threads? It looks to me that the stored data and the description of the stored data can be updated across threads independently, leaving the risk of them getting out of sync.

In more detail, the number of data points in a SeriesBuffer can be changed independent of the actual points:

struct SeriesBuffer {
    points: Mutex<HashMap<ChannelDescriptor, PointsType>>,
    count: AtomicUsize,
    // ..

even across threads. Put another way, a race on individual struct members is not possible, but a race between them is. With just the right (well, wrong) implementation and execution conditions, the count can be left inaccurate.

It appears that the SeriesBufferGuard is designed to ameliorate this:

struct SeriesBufferGuard<'sb> {
    sb: MutexGuard<'sb, HashMap<ChannelDescriptor, PointsType>>,
    count: &'sb AtomicUsize,
}

as it provides a means to update count while the mutex is locked. This seems reasonable to me (and a clever idea!), provided that count is only updated through the guard. This condition is met in SeriesBufferGuard::extend():

impl SeriesBufferGuard<'_> {
    fn extend(/* .. */) {
        // ..
        self.count.fetch_add(new_point_count, Ordering::Release);
    }
}

but it is not explicitly met in SeriesBuffer::take():

impl SeriesBuffer {
    fn take(&self) -> (/* .. */) {
        let mut points = self.lock(); // <-- lock taken
        let result = points // <-- guard used
            .sb
            .drain()
            .map(/* .. */)
            .collect();

        let result_count = self // <-- guard not explicitly held
            .count
            .fetch_update(Ordering::Release, Ordering::Acquire, |_| Some(0))
            .unwrap();
        (result_count, result)
    }
}

I think that this implementation works in practice due to the drop scope of guard points:

• the guard is held until the end of the function
• the guard prevents SeriesBufferGuard::extend() from being called (because another SeriesBufferGuard) can not exist

This is implicit behavior, and leaves a foot-gun for future developers if the drop scope is changed:

impl SeriesBuffer {
    fn take(&self) -> (usize, Vec<Series>) {
        let result = self.lock() // <-- lock taken, guard used and immediately dropped
            .sb
            .drain()
            .map(/* snip */)
            .collect();

        let result_count = self // <-- guard no longer exists, SeriesBufferGuard::extend() can be called
            .count
            .fetch_update(Ordering::Release, Ordering::Acquire, |_| Some(0))
            .unwrap();
        (result_count, result)
    }
}

As this precondition isn't enforced by the type system or hinted at in documentation, I could see a future developer not noticing this pitfall and making this or a similar mistake.

Are any of these solutions feasible?

1. move count and points into a new data type so they can be locked together (not implemented; not performant)
2. always update count through the SeriesBufferGuard (see fix: update point count while MutexGuard is held #140)
3. move count and points into a new data type that enforces thread-safe read and write (see refactor: add SeriesBufferInner type #144)

[KevinWMatthews]

I will post PRs for the two remaining proposals soon - I am out of time for this morning!

[alxhill]

Good catch! It's been a while since I dug into this code, but I think your read is correct that this is a risk, and that it currently works because of the lock guard drop scope.

For context, tracking count outside the mutex is done to allow checking capacity without needing to lock the buffer / block the processing threads:

fn has_capacity(&self, new_points_count: usize) -> bool {
    let count = self.count.load(Ordering::Acquire);
    count == 0 || count + new_points_count <= self.max_capacity
}

we use that function in enqueue:

if self.primary_buffer.has_capacity(new_count) {
    debug!("adding {} points to primary buffer", new_count);
    callback(self.primary_buffer.lock());
}

so ideally we can fix the gap here without needing to lock the full buffer just to check capacity

[KevinWMatthews]

tracking count outside the mutex is done to allow checking capacity without needing to lock the buffer / block the processing thread

Thank you! I was wondering about this; it seemed like a tricky optimization to make unless locking the mutex actually is a bottleneck in practice.

ideally we can fix the gap here without needing to lock the full buffer just to check capacity

This is a variant of a read-write lock, it seems? I am interested to see if I can find an ergonomic solution to this that is enforced by Rust's type system. If something nice emerges from my musings, I will post it and see what you think.

[KevinWMatthews]

Ok! The core of the idea of proposal 3 above ("move count and points into a new data type") is available in #144.

While working on this, I realized that some clarity can be gained by rearranging the implementation of take(). This is is proposed in #145.

I also posted an optimistic / opinionated refactoring, in #146. This asks for forgiveness (not permission) so feedback is welcome.