functional-streams-for-scala/fs2

Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced

Andrey

@troolean

is there already a utility for batching based on arbitrary sizing of each element of the stream?

e.g., something that has the shape of:

def batcher[F[_], E](threshold: Int, sizeOf: E => Int): Pipe[F, E, NonEmptyVector[E]] = ???

which can then be used like this:

case class Group(id: Long, xs: NonEmptyVector[Symbol])

val groups: Stream[Pure, Group] = ???

val batched: Stream[Pure, NonEmptyVector[Group]] = groups.through(batcher(100, _.xs.length))

Fabio Labella

@SystemFw

not that I can recall, seems fairly specialised

Andrey

@troolean

ok. just wanted to check and make sure i'm not duplicating something that might already exist (likely, in better form)

Mark de Jong

@Fristi

Are there any write ups on performance tricks/improvements for fs2? I suspect I can make my avro rpc can be alot faster, but not sure what's the issue at the moment

Peter Aaser

@PeterAaser

What I can immediately tell you is to look at chunking

Mark de Jong

@Fristi

https://github.com/vectos/formulation/tree/rpc/rpc/src/main/scala/formulation/rpc

FS2 code is mainly in AvroServer/AvroClient/Example

It's mainly tcp server/client plus some pipes

Peter Aaser

@PeterAaser

Look at how large chunks are. It's especially bad if chunks are single element

I'd love to do some benchmarking though

Mark de Jong

@Fristi

I see the socket.reads(size) yields chunks, which are split up by scodec-stream

The frames are rather small

10-20 bytes

Ross A. Baker

@rossabaker

Some of the parboiled parsers for the http4s model are referenced on the companion objects, so private constructors can be used, so it's impossible to publicly create invalid ones.

Lopping off that parboiled dependency from a "kernel" would be difficult, but at least it's "our" dependency.

It might be possible to have a model without log4s.

The headers are due for a massive reform anyway, so their current state is less interesting to that discussion.

It's interesting that all our integrations give the appearance of hard dependencies. The core is actually fairly light.

Christopher Davenport

@ChristopherDavenport

An argument for multi-repo.

Ross A. Baker

@rossabaker

Yeah, well, when someone has a long weekend to devote to setting up umpteen Travis configurations... :smile:

Christopher Davenport

@ChristopherDavenport

@SystemFw or others who might time to look. Put a PR up establishing an fs2 module for atto. Comments appreciated tpolecat/atto#91

Pavel Chlupacek

@pchlupacek

@rossabaker would be excellent to see how we can move forward. I don’t see any reason why fs2-io backend couldn’t perform in pair with blaze-ds. If it does not, it means we still have work to do on perfromance, and this may be perhaps the focus of fs2-http to really make sure the core perfromance does not suffer

Then I love to have one single stack for Headers/Request/Response algebra, namely if you are in thoughts of revorking them.

I think pairboiled dependency is ok if we can prove its perfromance is far better than scodec one (btw could pairboiled be used with scalajs too?)

Michael Pilquist

@mpilquist

It would be nice to have a version of concurrently that took a stream of streams -- something like s.concurrently(Stream(u1, u2, u3))

Christopher Davenport

@ChristopherDavenport

Why not just join in the concurrently block?

Michael Pilquist

@mpilquist

I think that's the best we can do with the current API

Pavel Chlupacek

@pchlupacek

perhaps just as snytax-sugar concurrently may take proe streams to run and then perfrom join behind the scenes ?

Fabio Labella

@SystemFw

I wouldn't like a syntax-sugar only version

s.concurrently(SoS.join) is not onerous syntactically

it's just that it creates a queue and all that to collect values to SoS that you then throw away

but that being said, I moved away from really caring about that

Pavel Chlupacek

@pchlupacek

ah, so we actually speaking here for alias for Stream(s1,s2,s3…).map(_.drain).join right?

Michael Pilquist

@mpilquist

yeah I noticed some code like s.concurrently(u1).concurrently(u2) which seemed inefficient to me

Christopher Davenport

@ChristopherDavenport

It feels a bit odd adding extensions like this which are easily built using the primitives

I do that occasionally to be sure. :laughing:

Fabio Labella

@SystemFw

because in my code I have a lot of cases where I join streams in module A, join streams in module B, and then (moduleA, moduleB).join, and that is not fixable without destroying modularity

Michael Pilquist

@mpilquist

I didn't really mean an alias

Pavel Chlupacek

@pchlupacek

well we could have join, that does effects only :-)

Fabio Labella

@SystemFw

@mpilquist yeah, I guess you meant a version where you don't run join at all, just compile.drain.start + concurrently machinery

I guess the generalisation of this problem is what Pavel proposes

Michael Pilquist

@mpilquist

And Stream(s, u1.drain, u2.drain).parJoinUnbounded is bad b/c of different shutdown semantics

Fabio Labella

@SystemFw

a Stream[F, Stream[F, A]] => Stream[F, Unit]]

that mean you do nested join without allocating all the queues

Michael Pilquist

@mpilquist

right

Fabio Labella

@SystemFw

but you still have to allocate the deferreds and so on

Pavel Chlupacek

@pchlupacek

yeah. Lets open issue, i this it is worth of investigating ...

Fabio Labella

@SystemFw

this isn't fixable completely without destroying modularity

Pavel Chlupacek

@pchlupacek

@SystemFw I am not 100% sure on that. if bracketCase and cancel works now as expected, I think we may have few winnings there - a) we may not need interrupt anymore, (so no race at every eval) and b) we may just simple cancel running streams

Fabio Labella

@SystemFw

ah ok, that doesn't fix the general problem

but it reduces the amount of things you need to allocate in the first place

Where communities thrive

People

Repo info

Activity