Refactor general case into push! and popfirst!

tpgillam · tpgillam · commit eda83b9993a1 · 2022-08-08T23:08:10.000+01:00
diff --git a/docs/src/reference/base.md b/docs/src/reference/base.md
@@ -1,10 +1,11 @@
 # General window
 
-A state to represent a window of aribtrarily variable capacity.
+A state to represent a window of arbitrarily variable capacity.
 
-Every time a new value is pushed onto the end of the window, we must specify how many values are removed from the front of the window.
+We push new values onto the window with [`Base.push!`](@ref).
+We can remove old values from the window with [`Base.popfirst!`](@ref).
 
 ```@autodocs
-Modules = [AssociativeWindowAggregation]
+Modules = [AssociativeWindowAggregation, Base]
 Pages = ["base.jl"]
 ```
diff --git a/src/base.jl b/src/base.jl
@@ -72,96 +72,20 @@ end
 WindowedAssociativeOp{T,Op}() where {T,Op} = WindowedAssociativeOp{T,Op,Op}()
 
 """
-    update_state!(
-        state::WindowedAssociativeOp,
-        value,
-        num_dropped_from_window::Integer
-    ) -> state
+    push!(state::WindowedAssociativeOp{T}, value) -> state
 
-Add the specified value to the state, drop some number of elements from the start of the
-window, and return `state` (which will have been mutated).
+Push `value` onto the end of `state`.
 
 # Arguments
 - `state::WindowedAssociativeOp`: The state to update (will be mutated).
 - `value`: The value to add to the end of the window - must be convertible to a `T`.
-- `num_dropped_from_window::Integer`: The number of elements to remove from the front of
-    the window.
-
-# Returns
-- The instance `state` that was passed in.
 """
-Base.@propagate_inbounds function update_state!(
-    state::WindowedAssociativeOp{T,Op,Op!}, value, num_dropped_from_window::Integer
-) where {T,Op,Op!}
-    # Our index into previous_cumsum is advanced by the number of values we drop from the
-    # window.
-    # TODO This could be dangerous, since if we get an error later the state will be
-    #   invalid.
-    state.ri_previous_cumsum += num_dropped_from_window
-
-    # If this has taken us out-of-range of the _previous_cumsum values, we must recompute
-    # them.
-    elements_remaining = length(state.previous_cumsum) - state.ri_previous_cumsum
-
-    if elements_remaining < 0
-        # We have overshot the end of previous_cumsum.
-
-        # We may also need to discard elements from values. This could happen if
-        # num_dropped_from_window > 1.
-        num_values_to_remove = -elements_remaining
-        @boundscheck if num_values_to_remove > length(state.values)
-            throw(
-                ArgumentError(
-                    "num_dropped_from_window = $num_dropped_from_window is out of range"
-                ),
-            )
-        end
-
-        # We now generate the partial sum, and set our index back to zero. values is also
-        # emptied, since its information is now reflected in previous_cumsum.
-        # NOTE: We need to take care here in the case of non-commutation. In accumulate, we
-        # will be getting:
-        #
-        #    (x0, op(x0, x1), op(op(x0, x1), x2), ...)
-        #
-        # but we actually want:
-        #
-        #    (x0, op(x1, x0), op(x2, op(x1, x0)), ...)
-
-        # Conceptually the following code is equivalent to the following, however avoids
-        # unnecessary allocations:
-        # trimmed_reversed_values = state.values[end:-1:1 + num_values_to_remove]
-        # state.previous_cumsum = accumulate(
-        #     (x, y) -> Op(y, x),
-        #     trimmed_reversed_values
-        # )
-
-        empty!(state.previous_cumsum)
-        upper = length(state.values)
-        lower = 1 + num_values_to_remove
-        @inbounds if (upper - lower) >= 0  # i.e. we have a non-zero range
-            i = upper
-            accumulation = state.values[i]
-            while true
-                push!(state.previous_cumsum, accumulation)
-                i -= 1
-                i >= lower || break
-                # If we were to use a mutating operation here, then we'd have to introduce
-                # a copy above. So there is no drawback to using the non-mutating Op.
-                accumulation = Op(state.values[i], accumulation)
-            end
-        end
-
-        state.ri_previous_cumsum = 0
-        empty!(state.values)
-        # state.sum is now garbage, but we are not going to use it before we recompute it.
-    end
-
+function Base.push!(state::WindowedAssociativeOp{T,Op,Op!}, value) where {T,Op,Op!}
     # We want to make sure that we never actually mutate any value object that we pass to
-    # `update_state!`.
+    # `push!`.
     # Note that when initialising `state.sum`, we should take a copy if our mutating `Op!`
     # is different to `Op`. This is because we know that `Op` is necessarily non-mutating.
-    _copy(x) = (Op == Op!) ? x : deepcopy(x)
+    _copy(x) = (Op === Op!) ? x : deepcopy(x)
 
     # Include the new value in sum and values.
     state.sum = isempty(state.values) ? _copy(value) : Op!(state.sum, value)
@@ -170,29 +94,24 @@ Base.@propagate_inbounds function update_state!(
 end
 
 """
-    drop_values!(
-        state::WindowedAssociativeOp{T,Op,Op!}, n::Integer
-    ) where {T,Op,Op!}
+    popfirst!(state::WindowedAssociativeOp, n::Integer=1) -> state
 
-Drop `n` values from the leading edge of the window.
+Drop `n` values from the start of `state`.
 """
-function drop_values!(state::WindowedAssociativeOp{T,Op,Op!}, n::Integer) where {T,Op,Op!}
-    n > 0 || throw(ArgumentError("n must be positive, got n=$n"))
-
-    # FIXME At best this will be hacky
-    # FIXME At best this will be hacky
-    # FIXME At best this will be hacky
-    # FIXME At best this will be hacky
+Base.@propagate_inbounds function Base.popfirst!(
+    state::WindowedAssociativeOp{T,Op}, n::Integer=1
+) where {T,Op}
+    n == 0 && return state
+    n >= 0 || throw(ArgumentError("n must be non-negative, got n=$n"))
 
-    # Our index into previous_cumsum is advanced by the number of values we drop from the
-    # window.
-    # TODO This could be dangerous, since if we get an error later the state will be
-    #   invalid.
-    state.ri_previous_cumsum += n
+    # Our index into previous_cumsum is advanced by the number of values we drop.
+    # Note that we do not assign to `ri_previous_cumsum` immediately, since this would make
+    # `state` invalid were we to throw an exception before returning.
+    new_ri_previous_cumsum = state.ri_previous_cumsum + n
 
     # If this has taken us out-of-range of the _previous_cumsum values, we must recompute
     # them.
-    elements_remaining = length(state.previous_cumsum) - state.ri_previous_cumsum
+    elements_remaining = length(state.previous_cumsum) - new_ri_previous_cumsum
 
     if elements_remaining < 0
         # We have overshot the end of previous_cumsum.
@@ -225,37 +144,27 @@ function drop_values!(state::WindowedAssociativeOp{T,Op,Op!}, n::Integer) where
         empty!(state.previous_cumsum)
         upper = length(state.values)
         lower = 1 + num_values_to_remove
-        if (upper - lower) >= 0  # i.e. we have a non-zero range
+        @inbounds if (upper - lower) >= 0  # i.e. we have a non-zero range
             i = upper
-            accumulation = @inbounds(state.values[i])
+            accumulation = state.values[i]
             while true
                 push!(state.previous_cumsum, accumulation)
                 i -= 1
                 i >= lower || break
                 # If we were to use a mutating operation here, then we'd have to introduce
                 # a copy above. So there is no drawback to using the non-mutating Op.
-                accumulation = Op(@inbounds(state.values[i]), accumulation)
+                accumulation = Op(state.values[i], accumulation)
             end
         end
 
         state.ri_previous_cumsum = 0
         empty!(state.values)
-        # FIXME state.sum is now garbage.
-        # This matters...
+        # WARNING! `state.sum` is now garbage.
+        #   This means that we have to be careful to not use it if `state.values` is empty.
+    else
+        state.ri_previous_cumsum = new_ri_previous_cumsum
     end
 
-
-    # # We want to make sure that we never actually mutate any value object that we pass to
-    # # `update_state!`.
-    # # Note that when initialising `state.sum`, we should take a copy if our mutating `Op!`
-    # # is different to `Op`. This is because we know that `Op` is necessarily non-mutating.
-    # _copy(x) = (Op == Op!) ? x : deepcopy(x)
-
-    # # Include the new value in sum and values.
-    # state.sum = isempty(state.values) ? _copy(value) : Op!(state.sum, value)
-    # state.sum =
-    # push!(state.values, value)
-
     return state
 end
 
@@ -280,7 +189,7 @@ function window_value(state::WindowedAssociativeOp{T,Op})::T where {T,Op}
         # We aren't using the A buffer, either because values is full or the A buffer has
         # not yet been populated.
         state.sum
-    elseif length(state.values) == 0
+    elseif isempty(state.values)
         # We are using the A buffer, but not the B buffer.
         # `state.sum` has an uninitialised value.
         @inbounds(state.previous_cumsum[index])
diff --git a/src/fixed.jl b/src/fixed.jl
@@ -16,9 +16,7 @@ mutable struct FixedWindowAssociativeOp{T,Op,Op!}
     window_state::WindowedAssociativeOp{T,Op,Op!,CircularBuffer{T}}
     remaining_window::Int
 
-    function FixedWindowAssociativeOp{T,Op,Op!}(
-        window::Integer
-    ) where {T,Op,Op!}
+    function FixedWindowAssociativeOp{T,Op,Op!}(window::Integer) where {T,Op,Op!}
         window < 1 && throw(ArgumentError("Got window $window, but it must be positive."))
         window_state = WindowedAssociativeOp{T,Op,Op!,CircularBuffer{T}}(
             CircularBuffer{T}(window - 1), CircularBuffer{T}(window)
@@ -50,7 +48,8 @@ function update_state!(state::FixedWindowAssociativeOp, value)
 
     # With @inbounds, we assert that num_dropped_from_window will never exceed the size of
     # the window.
-    @inbounds update_state!(state.window_state, value, num_dropped_from_window)
+    @inbounds popfirst!(state.window_state, num_dropped_from_window)
+    push!(state.window_state, value)
     return state
 end
 
diff --git a/src/time.jl b/src/time.jl
@@ -84,7 +84,8 @@ function update_state!(state::TimeWindowAssociativeOp, time, value)
         num_dropped_from_window += 1
     end
 
-    update_state!(state.window_state, value, num_dropped_from_window)
+    @inbounds popfirst!(state.window_state, num_dropped_from_window)
+    push!(state.window_state, value)
 
     if !state.window_full && num_dropped_from_window > 0
         # The window has now filled; record this for use in `window_full` below.
diff --git a/test/base.jl b/test/base.jl
@@ -9,4 +9,32 @@
             end
         end
     end
+
+    @testset "simple" begin
+        state = WindowedAssociativeOp{Int64,+}()
+        @test window_size(state) == 0
+
+        push!(state, 1)
+        @test window_size(state) == 1
+        @test window_value(state) == 1
+
+        push!(state, 2)
+        push!(state, 3, 4)
+        @test window_size(state) == 4
+        @test window_value(state) == 10
+
+        popfirst!(state)
+        @test window_size(state) == 3
+        @test window_value(state) == 9
+
+        popfirst!(state, 2)
+        @test window_size(state) == 1
+        @test window_value(state) == 4
+        @test_throws ArgumentError popfirst!(state, 2)
+        @test window_size(state) == 1
+        @test window_value(state) == 4
+
+        popfirst!(state, 1)
+        @test window_size(state) == 0
+    end
 end
