Add throttle(while:on:) #58

sharplet · 2016-10-06T16:26:36Z

There are times where you'd like to throttle values whenever some condition is met, rather than based on a time interval.

This is inspired by ReactiveViewModel's -throttleSignalWhileInactive:.

Basic usage might involve a simple MutableProperty to toggle throttled values:

let isActive = MutableProperty(false)

label.reactive.text <~ somethingChanged
  .throttle(while: isActive.map(!), on: QueueScheduler())
  .flatMap(.latest, transform: doSomethingExpensive)
  .observeOn(UIScheduler())

// later

isActive.value = true

At some point it would be great to be able to throttle events while the view has disappeared:

somethingChanged
  .throttle(while: reactive.viewHasDisappeared, on: UIScheduler())
  .startWithValues {
    // expensive layout stuff
  }

I'd love some feedback on this, especially potential thread safety issues! For example, I'm a little concerned that shouldThrottle could technically have changed between reading shouldThrottle.value and starting shouldThrottle.producer.uniqueValues { $0 }.

sharplet · 2016-10-06T16:27:25Z

There are also no tests yet, which I'm planning to rectify. I should be able to cover the main semantics, although I don't trust I can write tests to verify this code is thread-safe.

mdiep

Here are some preliminary thoughts. The idea makes sense to me overall. 👍

mdiep · 2016-10-07T01:28:18Z

Sources/Signal.swift

+	///   - scheduler: A scheduler to deliver events on.
+	///
+	/// - returns: A signal that sends values only while `shouldThrottle` is false.
+	func throttle<P: PropertyProtocol>(while shouldThrottle: P, on scheduler: SchedulerProtocol) -> Signal<Value, Error>


This seems like it should take a Signal, not a Property. 😕

It should be easy to go from the property to the signal, but it's not easy to go the other way.

Hmm, I think I kinda see it the other way. If this was implemented in terms of Signal, we'd have to pick a default starting state for the throttle, probably .resumed so that events are passed through. Using Property means we don't have to make that assumption.

From a slightly different perspective, Property is more powerful than Signal in that you could implement a signal-based overload of this method by composing it with an initial value and calling through to the property-based implementation.

I pushed 4dcd2bf to demonstrate what I mean. If you think those overloads would be useful I'll add some docs and merge those changes in here.

mdiep · 2016-10-07T01:30:00Z

Sources/Signal.swift

+			disposable += schedulerDisposable
+			disposable += { _ = shouldThrottle }
+
+			disposable += shouldThrottle.producer.uniqueValues { $0 }.startWithValues { shouldThrottle in


You probably meant skipRepeats here, not uniqueValues?

Also, this should be formatted according to the RAC style.

mdiep · 2016-10-07T01:32:07Z

Sources/Signal.swift

+					} else {
+						if case .throttled(let value?) = state {
+							schedulerDisposable.innerDisposable = scheduler.schedule {
+								observer.send(value: value)


I think this value should get sent outside of the Atomic's lock to prevent deadlock.

Even though it's in a schedule block, that can execute synchronously if, e.g., you're on the main queue and you're using the UIScheduler.

Same goes for the send below.

Made this change in e2dd090, let me know what you think.

mdiep · 2016-10-07T01:34:17Z

Sources/Signal.swift

+		} else {
+			return false
+		}
+	}


This could just be return self == .terminated.

That requires explicit Equatable conformance of ThrottleWhileState since case throttled has an associated value.

Stylistically, this could also become a switch statement if you prefer, which would add exhaustively checking.

which would add exhaustively checking.

And fail to compile if one adds a new case, which is a good thing

ikesyo · 2016-10-07T14:17:09Z

Sources/Signal.swift

+							}
+							return .resumed
+						}
+					}


The returned value from an action of Atomic.modify is not a new value for the Atomic instance since ReactiveCocoa/ReactiveCocoa#2984. So this should be written as follows:

let valueToSend = state.modify { state -> Value? in guard !state.isTerminated else { return nil } if shouldThrottle { state = .throttled(nil) return nil } else { state = .resumed if case let .throttled(value) = state { return value } else { return nil } } }

Good call! I've pushed 58e5bed. I ended up having a single return nil after the conditional block, but if everyone prefers this style (a return at every branch) I can change it.

I ended up having a single return nil after the conditional block

👍

ikesyo · 2016-10-07T14:18:20Z

Sources/Signal.swift

+						eventToSend = event
+						return .terminated
+					}
+				}


Same as https://github.com/ReactiveCocoa/ReactiveSwift/pull/58/files#r82397528:

let eventToSend = state.modify { state -> Event<Value, Error>? in switch event { case let .value(value): switch state { case .throttled: state = .throttled(value) return nil case .resumed: return event case .terminated: return nil } case .completed, .interrupted, .failed: state = .terminated return event } }

ikesyo · 2016-10-07T14:41:31Z

Sources/Signal.swift

+						} else {
+							state = .resumed
+
+							if case .throttled(let value?) = state {


You don't need to unwrap the bound value with ? since the return type is Value?: case .throttled(let value).

We should consistently place let right after the case: if case let .throttled(value).

I've moved the let outside 👍

You don't need to unwrap the bound value with ? since the return type is Value?

While that's technically true, I think there's value in being precise with exactly the conditions where this branch should be followed. The idea is that we should only send value event if a value was actually throttled, and I feel like this version communicates that more clearly. What do you think?

ikesyo · 2016-10-07T14:43:42Z

Sources/Signal.swift

+			disposable += self.observe { event in
+				let eventToSend = state.modify { state -> Event<Value, Error>? in
+					switch event {
+					case .value(let value):


Same as https://github.com/ReactiveCocoa/ReactiveSwift/pull/58/files#r82402862.

ikesyo · 2016-10-07T14:45:59Z

Sources/Signal.swift

+							state = .throttled(value)
+							return nil
+						case .resumed:
+							return .value(value)


Could return the original event here?

sharplet · 2016-10-12T14:31:54Z

Ok, I think I've addressed everyone's feedback, thank you! I've also pushed some tests.

@mdiep Do you think it's worth adding the Signal and SignalProducer overloads like I mentioned in #58 (comment)?

mdiep · 2016-11-06T16:17:45Z

Do you think it's worth adding the Signal and SignalProducer overloads like I mentioned in #58

I think we can go without them for now.

Mind merging master in? That should fix up the SwiftPM tests.

sharplet · 2016-11-06T23:54:00Z

And we're green 👍

andersio · 2016-11-07T02:05:13Z

Sources/Signal.swift

+		where P.Value == Bool
+	{
+		return Signal { observer in
+			let initial: ThrottleWhileState<Value> = shouldThrottle.value ? .throttled(nil) : .resumed


This initial check seems unnecessary, since shouldThrottle.producer below would emit the current value.

sharplet · 2016-11-16T22:01:11Z

@andersio Is this kind of thing what you had in mind? I'm trying to work out how to safely get the initial value, and also not miss any changes to shouldThrottle...

diff --git a/Sources/Signal.swift b/Sources/Signal.swift
index cb3adbf..941bf79 100644
--- a/Sources/Signal.swift
+++ b/Sources/Signal.swift
@@ -1515,7 +1515,8 @@ extension SignalProtocol {
        where P.Value == Bool
    {
        return Signal { observer in
-           let initial: ThrottleWhileState<Value> = shouldThrottle.value ? .throttled(nil) : .resumed
+           let shouldThrottleProducer = shouldThrottle.producer.replayLazily(upTo: Int.max)
+           let initial: ThrottleWhileState<Value> = shouldThrottleProducer.first()!.value! ? .throttled(nil) : .resumed
            let state = Atomic(initial)
            let schedulerDisposable = SerialDisposable()

@@ -1523,7 +1524,8 @@ extension SignalProtocol {
            disposable += schedulerDisposable
            disposable += { _ = shouldThrottle }

-           disposable += shouldThrottle.producer
+           disposable += shouldThrottleProducer
+               .skip(first: 1)
                .skipRepeats()
                .startWithValues { shouldThrottle in
                    let valueToSend = state.modify { state -> Value? in

andersio · 2016-11-16T22:29:43Z

@sharplet I mean the reverse of this. initial can be simply .resumed. Starting shouldThrottle.producer would always give the current value of the property synchronously, so there is no need to grab it twice.

sharplet · 2016-11-16T22:59:59Z

Oh, that totally clears things up. Fixed in f757e53 (and rebased).

andersio · 2016-11-17T00:58:08Z

Sources/Signal.swift

+
+			let disposable = CompositeDisposable()
+			disposable += schedulerDisposable
+			disposable += { _ = shouldThrottle }


@sharplet Hmm, may I ask what's the specific reason to retain it? I don't think the resulting signal should own shouldThrottle in any way.

What if it terminates the resulting signal when shouldThrottle completes, gaining the take(during:) semantic?

In my mind it's the same reason that Action captures a strong reference to its enabledIf property. So to me it seems convenient, not having to maintain an extra strong reference to that property.

But that doesn't help me to understand what the problem with that strong reference is.

What if it terminates the resulting signal when shouldThrottle completes, gaining the take(during:) semantic?

Regardless of the current behaviour, are you saying that if shouldThrottle completes, the returned signal should or should not also complete? I think it would make sense to add some tests for these semantics so it's clear what we want the behaviour to be.

An observation to the push-based Signal should not have any effect on the event stream. If we treat operator as specialised observations, and Propertys as specialised Signals with stronger guarantees, by the logic a signal operator that takes a Property should not cause any side effect on its sources.

Now retaining the property violates this, as the retainment influences the termination of the stream (the property) that is supposed to just be observed.

IIRC this would be the first ever signal operator that has a side effect upon the upstream signal if merged.

Action is a different story, since it is not a stream of value itself, but a state machine that owns a few streams.

Regardless of the current behaviour, are you saying that if shouldThrottle completes, the returned signal should or should not also complete?

Either would do, but completing with shouldThrottle's completion sounds a fit. Say for a Swift equivalent of ReactiveViewModel, I think it makes sense to have these throttled signals gone together with the supplied shouldThrottle property (which is owned by the view model).

It seems to me like the options are:

When shouldThrottle completes, the signal may get stuck in the throttled state (i.e., whichever state was last)

When shouldThrottle completes, the state gets switched back to .resumed

When shouldThrottle completes, the signal completes

(2) feels plain wrong to me, which makes it a toss-up between (1) and (3). So I feel like it comes down to which option is more surprising, and which option is easier to see/debug. For that reason I'm leaning towards tying the lifetime of the returned signal to the lifetime of shouldThrottle.

@mdiep What do you think?

Or anyone else who's been involved in the thread of course! 😅

(1) seems like the right thing to me, although it's a tough call.

That seems like the least surprising thing, and most in line with other operators. (c.f. sample, which only completes when both signals have completed)

We should definitely add tests for this—and probably for an interrupt on the signal as well.

sharplet · 2016-11-24T16:20:13Z

Sources/SignalProducer.swift

+	public func throttle<P: PropertyProtocol>(while shouldThrottle: P, on scheduler: SchedulerProtocol) -> SignalProducer<Value, Error>
+		where P.Value == Bool
+	{
+		return lift { $0.throttle(while: shouldThrottle, on: scheduler) }


This keeps a strong reference to shouldThrottle for SignalProducers. @andersio does this need to change as well?

The first thing I tried was to wrap shouldThrottle in an existential:

let shouldThrottle = Property(shouldThrottle) return lift { $0.throttle(while: shouldThrottle, on: scheduler) }

But that also increases the lifetime of the underlying property. Is there any way to capture a reference to this property that won't extend its lifetime? It looks like composing properties fundamentally alters the lifetime of the source property, extending it to be the union of all transitive property lifetimes. Is that intentional?

That is intentional IIRC. All composed properties derived from a property is considered just a reduced/transformed view of that property, so that the behaviour is in line with existentials which retain the wrapped property (like the one in your example).

That's said it might be worth discussing if the source capturing should be dropped. It could be an oversight (especially with the arguments I made in #58 (comment)... ehm), since in ReactiveCocoa/ReactiveCocoa#2922 @mdiep and I focused just on how signal and producer should behave.

Edit: Edited many times.

Edit 2: It seems the capturing is originated from ReactiveCocoa/ReactiveCocoa#2788. The argument to capture was different though, and it was actually invalidated by ReactiveCocoa/ReactiveCocoa#2922...

Edit 3: Opened a PR. #117

#117 is merged. ~~Now this is the last stop before RC1 (hopefully). 🎉~~

The semantics of `Atomic.modify` have changed such that the return value of the closure is no longer the old state, and the state is instead modified by assigning to the `inout` parameter.

When `shouldThrottle.producer` is started, the observer will be executed synchronously first time. This makes the initial check redundant, because the value will be updated as soon as `shouldThrottle` is observed.

sharplet · 2016-11-28T22:36:09Z

Sources/Signal.swift

+	///
+	/// - note: If `shouldThrottle` completes before the receiver, and its last
+	///         value is `true`, the returned signal will remain in the throttled
+	///         state, emitting no further values until it terminates.


Is 4 notes excessive? 😅

Perhaps just two notes (shouldThrottle + termination behavior), or no notes at all?

sharplet · 2016-11-28T22:37:07Z

That should be everything, please take a look at the latest 2 commits!

sharplet · 2016-11-29T00:30:26Z

Streamlined to 3 notes in the doc comments.

sharplet · 2016-11-29T04:31:19Z

Green 👍

andersio · 2016-11-29T04:41:25Z

sharplet · 2016-11-29T14:29:52Z

🎉

mdiep requested changes Oct 7, 2016

View reviewed changes

sharplet force-pushed the as-throttle-while branch from 7643d65 to e2dd090 Compare October 7, 2016 13:48

ikesyo reviewed Oct 7, 2016

View reviewed changes

sharplet force-pushed the as-throttle-while branch 2 times, most recently from ee9260f to c0367c3 Compare October 12, 2016 14:26

mdiep approved these changes Nov 6, 2016

View reviewed changes

sharplet force-pushed the as-throttle-while branch from c0367c3 to d2d61b1 Compare November 6, 2016 22:03

andersio requested changes Nov 7, 2016

View reviewed changes

sharplet force-pushed the as-throttle-while branch from 5dc0ed2 to f757e53 Compare November 16, 2016 22:58

andersio approved these changes Nov 16, 2016

View reviewed changes

mdiep approved these changes Nov 17, 2016

View reviewed changes

andersio requested changes Nov 17, 2016

View reviewed changes

andersio added this to the 1.0 milestone Nov 18, 2016

sharplet commented Nov 24, 2016

View reviewed changes

andersio mentioned this pull request Nov 24, 2016

Remove Property source capturing. #117

Merged

sharplet added 4 commits November 28, 2016 17:19

Add throttle(while:on:)

73e8b06

Schedule throttled events outside the lock

f5ab55c

Fix issue where throttle state was never modified

a88287c

The semantics of `Atomic.modify` have changed such that the return value of the closure is no longer the old state, and the state is instead modified by assigning to the `inout` parameter.

Add tests for throttle(while:)

9c0aec4

sharplet added 3 commits November 28, 2016 17:19

Remove redundant check in throttle(while:on:)

27cefb7

When `shouldThrottle.producer` is started, the observer will be executed synchronously first time. This makes the initial check redundant, because the value will be updated as soon as `shouldThrottle` is observed.

Avoid capturing a strong reference to throttle property

3ea05c7

Add tests for the throttle property completing before its signal

1cc110f

sharplet force-pushed the as-throttle-while branch from f757e53 to 1cc110f Compare November 28, 2016 22:35

sharplet commented Nov 28, 2016

View reviewed changes

Combine some notes in docs for throttle(while:on:)

ec56cda

andersio approved these changes Nov 29, 2016

View reviewed changes

mdiep merged commit be6f0ad into master Nov 29, 2016

mdiep deleted the as-throttle-while branch November 29, 2016 13:10

sharplet mentioned this pull request Jun 27, 2017

pause(on:..., until:...) functional #466

Closed

Add throttle(while:on:) #58

Add throttle(while:on:) #58

Conversation

sharplet commented Oct 6, 2016 • edited Loading

sharplet commented Oct 6, 2016 • edited Loading

mdiep left a comment

Choose a reason for hiding this comment

mdiep Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdiep Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

NachoSoto Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

ikesyo Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ikesyo Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ikesyo Oct 7, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sharplet commented Oct 12, 2016

mdiep commented Nov 6, 2016

sharplet commented Nov 6, 2016

Choose a reason for hiding this comment

sharplet commented Nov 16, 2016

andersio commented Nov 16, 2016 • edited Loading

sharplet commented Nov 16, 2016

andersio Nov 17, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Nov 17, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Nov 24, 2016 • edited Loading

Choose a reason for hiding this comment

andersio Nov 28, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sharplet commented Nov 28, 2016

sharplet commented Nov 29, 2016

sharplet commented Nov 29, 2016

andersio commented Nov 29, 2016

sharplet commented Nov 29, 2016

sharplet commented Oct 6, 2016 •

edited

Loading

sharplet commented Oct 6, 2016 •

edited

Loading

mdiep Oct 7, 2016 •

edited

Loading

mdiep Oct 7, 2016 •

edited

Loading

NachoSoto Oct 7, 2016 •

edited

Loading

ikesyo Oct 7, 2016 •

edited

Loading

ikesyo Oct 7, 2016 •

edited

Loading

ikesyo Oct 7, 2016 •

edited

Loading

andersio commented Nov 16, 2016 •

edited

Loading

andersio Nov 17, 2016 •

edited

Loading

andersio Nov 17, 2016 •

edited

Loading

andersio Nov 24, 2016 •

edited

Loading

andersio Nov 28, 2016 •

edited

Loading