{"action":"created","issue":{"url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/19","repository_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET","labels_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/19/labels{/name}","comments_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/19/comments","events_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/19/events","html_url":"https://github.com/Reactive-Extensions/Rx.NET/issues/19","id":45656170,"number":19,"title":"Backpressure and Unbounded Queues","user":{"login":"mattpodwysocki","id":49051,"avatar_url":"https://avatars.githubusercontent.com/u/49051?v=3","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":[{"url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/labels/[area]%20Rx","name":"[area] Rx","color":"1d76db"}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":28,"created_at":"2014-10-13T15:43:55Z","updated_at":"2016-07-06T03:54:39Z","closed_at":null,"body":"Copied from https://rx.codeplex.com/workitem/82\r\n\r\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.\r\n\r\nhttps://speakerdeck.com/benjchristensen/reactive-streams-with-rx-at-javaone-2014?slide=67\r\nhttps://github.com/reactive-streams/reactive-streams\r\n\r\nIt's worth considering whether it makes sense for - and how - it could be implemented in Rx.NET.\r\n\r\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?\r\n\r\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: \r\n\r\n```\r\nIO<R> Zip<F,S,R>(IO<F> f, IO<S> s, Func<F,S,R> sel, IObservable<int> maxQueueSize)\r\n```\r\n\r\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.\r\n```\r\nIObservable<IList<TSource>> WindowIntrospective<TSource>(this IObservable<TSource> source, IScheduler scheduler)\r\n```\r\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: \r\n* The operator is applied to a presumably fast observable(*)\r\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.(**)\r\n* All calls to OnNext are on the specified scheduler.\r\n* OnNext calls are serialized (Rx contract), which means that a slow observer would block a normal observable.\r\n* WindowIntrospective handles the situation differently: it buffers notifications while OnNext is blocking, and only while OnNext is blocking.\r\n* When OnNext finally returns, WindowIntrospective immediately pushes the next buffer to the observer.\r\n\r\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.\r\n\r\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."},"comment":{"url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/comments/230665148","html_url":"https://github.com/Reactive-Extensions/Rx.NET/issues/19#issuecomment-230665148","issue_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/19","id":230665148,"user":{"login":"masaeedu","id":3674056,"avatar_url":"https://avatars.githubusercontent.com/u/3674056?v=3","gravatar_id":"","url":"https://api.github.com/users/masaeedu","html_url":"https://github.com/masaeedu","followers_url":"https://api.github.com/users/masaeedu/followers","following_url":"https://api.github.com/users/masaeedu/following{/other_user}","gists_url":"https://api.github.com/users/masaeedu/gists{/gist_id}","starred_url":"https://api.github.com/users/masaeedu/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/masaeedu/subscriptions","organizations_url":"https://api.github.com/users/masaeedu/orgs","repos_url":"https://api.github.com/users/masaeedu/repos","events_url":"https://api.github.com/users/masaeedu/events{/privacy}","received_events_url":"https://api.github.com/users/masaeedu/received_events","type":"User","site_admin":false},"created_at":"2016-07-06T03:54:39Z","updated_at":"2016-07-06T03:54:39Z","body":"@Igorbek I know you posted your original snippet a long time ago, but would an extension method like this solve your issue?\r\n\r\n```\r\npublic static IObservable<U> SelectAsync<T, U>(this IObservable<T> source, Func<T, Task<U>> selector, int limit)\r\n{\r\n    var sem = new SemaphoreSlim(limit);\r\n    sem.Wait();\r\n\r\n    return source\r\n        .Select(async x =>\r\n        {\r\n            var result = await selector(x);\r\n            sem.Release();\r\n\r\n            return result;\r\n        })\r\n        .Do(_ => sem.Wait())\r\n        .Concat();\r\n}\r\n```\r\n\r\nTaking your original code (with stuff added for logging):\r\n\r\n```\r\nObservable.Range(1, 10000)\r\n    .Do(x => WriteLine($\"Produced {x}\"))\r\n    .Select(i => Observable.Timer(TimeSpan.FromSeconds(1)).Select(_ => i))\r\n    .Concat()\r\n    .Do(x => WriteLine($\"Consumed {x}\"));\r\n```\r\n\r\nYou would change it to this:\r\n\r\n```\r\nObservable.Range(1, 20)\r\n    .Do(x => WriteLine($\"Produced {x}\"))\r\n    .SelectAsync(async i => await Observable.Timer(TimeSpan.FromSeconds(1)).Select(_ => i), 5)\r\n    .Do(x => WriteLine($\"Consumed {x}\"));\r\n```\r\n\r\nImagining that `Observable.Range` produced some really memory intensive objects instead of just integers, the `5` passed as the `limit` argument of `SelectAsync` would now ensure that no more than 5 are pending consumption at any time."}}
{"actor":{"display_login":"masaeedu"}}