Comments (6)
To be concrete, given the following:
#r "../../../bin/FSharp.Control.AsyncSeq.dll"
open FSharp.Control
let printer() = printfn "allocate"; { new System.IDisposable with member __.Dispose() = printfn "disposed" }
let a =
asyncSeq { use _d = printer()
yield 1
yield 2 }
Then this prints "allocate" (without deallocate):
a |> AsyncSeq.take 1 |> AsyncSeq.iter(fun _ -> ()) |> Async.RunSynchronously
likewise this (which surprised me even more)
a |> AsyncSeq.take 2 |> AsyncSeq.iter(fun _ -> ()) |> Async.RunSynchronously
You only get a deallocate like this:
a |> AsyncSeq.take 3 |> AsyncSeq.iter(fun _ -> ()) |> Async.RunSynchronously
This problem is very intrinsic to the algebraic definition that's been used - the corresponding "iterator" definition would not have these problems.
from fsharp.control.asyncseq.
A type that can help address resource management is as follows (from issue and source code):
/// An async pipeline which consumes values of type 'i, produces
/// values of type 'o and completes with a result 'a or an error.
type AsyncPipe<'i, 'o, 'a> = Async<AsyncPipeStep<'i, 'o, 'a>>
/// An individual step in an async pipeline.
and AsyncPipeStep<'i, 'o, 'a> =
/// The pipeline completed with result 'a.
| Done of 'a
/// The pipeline is emitting a value of type 'o.
| Emit of 'o * AsyncPipe<'i, 'o, 'a>
/// The pipeline is consuming a value of type 'i.
| Await of ('i option -> AsyncPipe<'i, 'o, 'a>)
This is more generic than required by AsyncSeq
alone, but one of the things it can ensure is deterministic resource disposal. The idea would be to have the iterator be an async pipe which awaits input, and the async sequence be an async pipe which emits outputs. Then iteration is a fusion of these two pipes. In particular, fusion can ensure that an emitting pipe is drained. I'll come up with a sample over the weekend...
This is based on Iteratees
from fsharp.control.asyncseq.
One way to adapt this more directly to AsyncSeq
is to add an explicit compensation function:
type AsyncSeq<'T> = Async<AsyncSeqInner<'T> * (unit -> unit)>
and AsyncSeqInner<'T> =
| Nil
| Cons of 'T * AsyncSeq<'T>
type Async with
static member map f a = async.Bind(a, f >> async.Return)
static member bind f a = async.Bind(a, f)
let empty<'T> : AsyncSeq<'T> = async.Return(Nil,id)
let singleton (item:'T) : AsyncSeq<'T> =
async.Return(Cons(item,empty),id)
let rec tryFinally (s:AsyncSeq<'T>) (comp:unit -> unit) : AsyncSeq<'T> = async {
try
let! s = s
match s with
| Nil,comp' -> return Nil,(comp' >> comp)
| Cons(a,tl),comp' -> return Cons(a,tryFinally tl comp),(comp' >> comp)
with ex ->
comp()
return raise ex }
let rec append (f:AsyncSeq<'T>) (s:AsyncSeq<'T>) : AsyncSeq<'T> =
f |> Async.bind (function
| Nil,comp -> tryFinally s comp
| Cons(a,tl),comp -> (Cons(a, append tl s),comp) |> async.Return)
let rec collect (f:'T -> AsyncSeq<'U>) (s:AsyncSeq<'T>) : AsyncSeq<'U> =
s |> Async.bind (function
| Nil,comp -> (Nil,comp) |> async.Return
| Cons(a,tl),comp -> async {
return! append (f a) (collect f tl)
})
let rec iterAsync (f:'T -> Async<unit>) (s:AsyncSeq<'T>) : Async<unit> =
s |> Async.bind (function
| Nil,comp -> async { comp() }
| Cons(a,tl),comp -> async {
try
do! f a
return! iterAsync f tl
with ex ->
comp()
return raise ex
})
let rec ofList (ls:'T list) : AsyncSeq<'T> =
match ls with
| [] -> (Nil,id) |> async.Return
| hd::tl -> (Cons(hd, ofList tl),id) |> async.Return
let rec take (count:int) (s:AsyncSeq<'T>) : AsyncSeq<'T> =
s |> Async.map (function
| Nil,comp -> Nil,comp
| Cons(a,tl),comp ->
if count = 0 then Nil,comp
else Cons(a, take (count - 1) tl),comp)
type AsyncSeqBuilder() =
member x.Yield(v) = singleton v
member x.Return(()) = empty
member x.YieldFrom(s) = s
member x.Zero () = empty
member x.Bind (inp:Async<'T>, body : 'T -> AsyncSeq<'U>) : AsyncSeq<'U> =
async.Bind(inp, body)
member x.Combine (seq1:AsyncSeq<'T>,seq2:AsyncSeq<'T>) =
append seq1 seq2
member x.While (gd, seq:AsyncSeq<'T>) =
if gd() then x.Combine(seq,x.Delay(fun () -> x.While (gd, seq))) else x.Zero()
member x.Delay (f:unit -> AsyncSeq<'T>) =
async.Delay(f)
member x.Using (resource:#IDisposable, binder) =
tryFinally (binder resource) (fun () ->
if box resource <> null then resource.Dispose())
member x.TryFinally (body: AsyncSeq<'T>, compensation) =
tryFinally body compensation
let asyncSeq = new AsyncSeqBuilder()
let printer() = printfn "allocate"; { new System.IDisposable with member __.Dispose() = printfn "disposed" }
let a =
asyncSeq { use _d = printer()
yield 1
yield 2 }
a |> take 1 |> iterAsync (fun _ -> async.Return()) |> Async.RunSynchronously
which prints:
allocate
disposed
(I haven't yet verified that the desired compensation would be invoked in all cases with exceptions).
from fsharp.control.asyncseq.
Although it seems the compensation needs to be on the outside of the Async
, otherwise there is not way to run the compensation if the Async
call fails.
from fsharp.control.asyncseq.
Yeah... I've spent quite some time thinking about these issues.
In choice streams the streams are memoized for good reasons using lazy promises. I want choice streams to be consistent so that all consumers of a choice stream get the same sequence of results. I also don't want any operations embedded in choice streams to be repeated or cancelled each time a consumer requests an element or nondeterministically chooses to consume element from some other stream.
Choice stream builders do not provide a use
construct, because such functionality cannot be provided meaningfully given the semantics of choice streams. There is support for finalizers, but finalizers can't be used for deterministic resource management. They work fine for some things, however. I'm using finalizers to lazily remove Observable subscriptions, for example.
Good luck!
from fsharp.control.asyncseq.
FWIW there is an old port of the (ugly) implementation of the "Seq" combinators from FSharp.Core to get an IAsyncEnumerable builder. This supports "try/finally" correctly.
https://github.com/dsyme/FSharpDemoScripts/blob/master/extlib/AsyncSeq-0.1.fsx#L7
My intuition is that we will have to move to IAsyncEnumerable if we are to support try/finally. At least, I think I'd prefer an approach based on the conversion of known working synchronous Seq code.
from fsharp.control.asyncseq.
Related Issues (20)
- 3.x fails with fable HOT 6
- Best way to convert System.Collections.Generic.IAsyncEnumerable<'t> to AsyncSeq<'t>? HOT 6
- iterAsyncParallel and iterAsyncParallelThrottled may fail to cancel
- Add YieldFrom overload for seq<'a>? HOT 1
- Support .NET Standard 2.0 HOT 6
- Combination of groupBy and mapAsync involving access to the group causes deadlock. HOT 2
- [Feature] Can we have sort, sortBy, sortDescending and sortByDescending methods for AsyncSeq? HOT 3
- Update to use github actions
- netstandard2.1 only target forbids using the library in .net framework HOT 3
- asyncSeq not available HOT 6
- Currently broken with Fable HOT 7
- TaskSeq? HOT 6
- AsyncSeq deadlocks in Blazor WASM environment HOT 2
- AsyncSeq.mapAsyncParallel alternatives HOT 3
- Suggest to add a `chunkBy` function (to group and yield adjacent items sharing the same key) to the library HOT 4
- [General Question] HOT 2
- Deprecate `FSharp.Control.IAsyncEnumerable` in favor of the same type in BCL
- [Question] AsyncSeq.MapAsyncParallel HOT 2
- [Question] How to reduce Async<unit[]> HOT 1
- making AsyncSeq.mergeAll to accept array maybe
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from fsharp.control.asyncseq.