Fix a bug with reduce & ReduceIf & withprogress (#177)

This is a bug introduced while implementing generic `Reduced` handling
#172.  If `b` is `Reduced`, all the "private" states of transducers
are stripped off.  So, `combine` should be called only for the bottom
reducing function.  This is implemented in `combine_right_reduced`.

src/reduce.jl

@@ -158,12 +158,14 @@ function _reduce(ctx, rf, init, reducible::Reducible)
         b0 = fetch(task)
         a = @return_if_reduced a0
         should_abort(ctx) && return a  # slight optimization
-        b = unreduced(b0)
-        b0 isa Reduced && return reduced(combine(rf, a, b))
-        return combine(rf, a, b)
+        b0 isa Reduced && return combine_right_reduced(rf, a, b0)
+        return combine(rf, a, b0)
     end
 end
 
+combine_right_reduced(rf, a, b0::Reduced) =
+    reduced(combine(_realbottomrf(rf), a, unreduced(b0)))
+
 function _reduce_threads_for(rf, init, reducible::SizedReducible{<:AbstractArray})
     arr = reducible.reducible
     basesize = reducible.basesize
@@ -200,9 +202,8 @@ end
 combine_step(rf) =
     asmonoid() do a0, b0
         a = @return_if_reduced a0
-        b = unreduced(b0)
-        b0 isa Reduced && return reduced(combine(rf, a, b))
-        return combine(rf, a, b)
+        b0 isa Reduced && return combine_right_reduced(rf, a, b0)
+        return combine(rf, a, b0)
     end
 
 # AbstractArray for disambiguation

test/test_parallel_reduce.jl

@@ -167,6 +167,10 @@ end
     end
     @test reduce(right, xf2, withprogress(1:100; interval=0); basesize=1) == 5050
     @test reduce(right, xf2, withprogress(1:100; interval=0); basesize=1, simd=true) == 5050
+
+    xf3 = ReduceIf(x -> x == 100)
+    @test reduce(right, xf3, withprogress(1:1000; interval=0); basesize=1) == 100
+    @test reduce(right, xf3, withprogress(1:1000; interval=0); basesize=1, simd=true) == 100
 end
 
 end  # module