{"action":"closed","issue":{"url":"https://api.github.com/repos/dotnet/reactive/issues/19","repository_url":"https://api.github.com/repos/dotnet/reactive","labels_url":"https://api.github.com/repos/dotnet/reactive/issues/19/labels{/name}","comments_url":"https://api.github.com/repos/dotnet/reactive/issues/19/comments","events_url":"https://api.github.com/repos/dotnet/reactive/issues/19/events","html_url":"https://github.com/dotnet/reactive/issues/19","id":45656170,"node_id":"MDU6SXNzdWU0NTY1NjE3MA==","number":19,"title":"Backpressure and Unbounded Queues","user":{"login":"mattpodwysocki","id":49051,"node_id":"MDQ6VXNlcjQ5MDUx","avatar_url":"https://avatars2.githubusercontent.com/u/49051?v=4","gravatar_id":"","url":"https://api.github.com/users/mattpodwysocki","html_url":"https://github.com/mattpodwysocki","followers_url":"https://api.github.com/users/mattpodwysocki/followers","following_url":"https://api.github.com/users/mattpodwysocki/following{/other_user}","gists_url":"https://api.github.com/users/mattpodwysocki/gists{/gist_id}","starred_url":"https://api.github.com/users/mattpodwysocki/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mattpodwysocki/subscriptions","organizations_url":"https://api.github.com/users/mattpodwysocki/orgs","repos_url":"https://api.github.com/users/mattpodwysocki/repos","events_url":"https://api.github.com/users/mattpodwysocki/events{/privacy}","received_events_url":"https://api.github.com/users/mattpodwysocki/received_events","type":"User","site_admin":false},"labels":[{"id":394130112,"node_id":"MDU6TGFiZWwzOTQxMzAxMTI=","url":"https://api.github.com/repos/dotnet/reactive/labels/[area]%20Ix","name":"[area] Ix","color":"1d76db","default":false},{"id":394130570,"node_id":"MDU6TGFiZWwzOTQxMzA1NzA=","url":"https://api.github.com/repos/dotnet/reactive/labels/[area]%20Rx","name":"[area] Rx","color":"1d76db","default":false},{"id":54865812,"node_id":"MDU6TGFiZWw1NDg2NTgxMg==","url":"https://api.github.com/repos/dotnet/reactive/labels/enhancement","name":"enhancement","color":"84b6eb","default":true}],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":43,"created_at":"2014-10-13T15:43:55Z","updated_at":"2019-11-13T22:57:34Z","closed_at":"2019-11-13T22:57:34Z","author_association":"CONTRIBUTOR","body":"Copied from https://rx.codeplex.com/workitem/82\n\nThe problems of backpressure (fast producer, slow consumer) and unbounded queues in native operators (e.g., Zip, GroupBy, Merge) appear to be solved in a general way in RxJava by incorporating a **Producer** interface as a first-class citizen.\n\nhttps://speakerdeck.com/benjchristensen/reactive-streams-with-rx-at-javaone-2014?slide=67\nhttps://github.com/reactive-streams/reactive-streams\n\nIt's worth considering whether it makes sense for - and how - it could be implemented in Rx.NET.\n\nHowever, my initial impression is that it may not be an optimal approach.  For example, how does an observer know what number to pass to the producer's request method? What happens when it's wrong?\n\nIt seems to me that simply adding overloads to problematic operators to accept a maximum queue size avoids the unbounded problem in a simple and effective way.  Furthermore, accepting an IObservable<int> instead would make it reactive (similar to [this work item](https://rx.codeplex.com/workitem/44)). Ensuring that the supplied observable pushes notifications in the same context as the producer allows for synchronized change to the queue size, presumably similar to how RxJava's Producer works.  For example: \n\n```\nIO<R> Zip<F,S,R>(IO<F> f, IO<S> s, Func<F,S,R> sel, IObservable<int> maxQueueSize)\n```\n\nAs for backpressure, in Rxx, I've implemented a pattern that I call [Introspection](https://rxx.codeplex.com/SourceControl/changeset/view/74914#Main/Source/Rxx/System/Reactive/Linq/Observable2%20-%20Introspection.cs) to solve a similar problem with backpressure, though it _does not_ solve the problem of unbounded queues in a general way.\n\n```\nIObservable<IList<TSource>> WindowIntrospective<TSource>(this IObservable<TSource> source, IScheduler scheduler)\n```\n\nThe **WindowIntrospective** operator (pooly named) in Rxx works close to the metal in Rx to dynamically produce heterogeneously-sized buffers, to accommodate slow observers, as follows: \n- The operator is applied to a presumably fast observable(*)\n- A concurrency-introducing scheduler argument is supplied.  It could be, and often is DispatcherScheduler, but it doesn't matter as long as the source is pushing values in a different context.(**)\n- All calls to OnNext are on the specified scheduler.\n- OnNext calls are serialized (Rx contract), which means that a slow observer would block a normal observable.\n- WindowIntrospective handles the situation differently: it buffers notifications while OnNext is blocking, and only while OnNext is blocking.\n- When OnNext finally returns, WindowIntrospective immediately pushes the next buffer to the observer.\n\n*i.e., we presume it's faster than the observer, but it need not be.  In fact, the observer's speed can vary dynamically.  It's one of the benefits over having an explicit _Producer.request(#)_ method, from which an observer must pull.  The introspective operators push heterogeneously-sized buffers that vary dynamically based on the dynamically varying performance of the observer.\n\n*\\* Often the next operator applied is **ObserveOnDispatcher** anyway, which adds an unbounded queue. The introspective operator has the same semantics without an unbounded queue -- it only assumes that the observer will not block OnNext indefinitely.\n"}}
{"actor":{"display_login":"akarnokd"}}