Fs2: Add `Stream#rechunkRandomly`

[mpilquist]

/**
  * Rechunks the stream such that output chunks are within [inputChunk.size * minFactor, inputChunk.size * maxFactor].
  */
def rechunkRandomlyWithSeed(minFactor: Double, maxFactor: Double, seed: Long): Stream[F, O] = ???

def rechunkRandomly(minFactor: Double = 0.1, maxFactor: Double = 2.0): Stream[F, O] =
  Stream.suspend(this.rechunkRandomlyWithSeed(minFactor, maxFactor, System.currentTimeMillis))

[djspiewak]

Dare I ask… why randomly? Fairness?

[mpilquist]

For test purposes only :)

On Jul 14, 2018, at 4:58 PM, Daniel Spiewak [e-mail] wrote:

Dare I ask… why randomly? Fairness?

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub, or mute the thread.

[Daenyth]

If it's for test purposes only, perhaps add it as a pipe to the test suite?

[mpilquist]

@Daenyth The fs2 tests aren't part of our API and come with no compatibility guarantees. I don't recommend anyone add a dependency on the fs2 tests jar(s).

[Daenyth]

Right, that's why I was suggesting it be added there instead of the public api, unless anyone has a production use case

[mpilquist]

I wanted rechunkRandomly to be part of the public API though, as it's very useful for anyone writing chunk aware pipes. :)

[pchlupacek]

I think the randomness may not be usefull only in tests. There are many real world use cases that will benefits from random sampling an variance. so +1 to make this part of API.

[monadplus]

@mpilquist Can I work on this method ? It seems doable for a novice like me.

[mpilquist]

Go for it! Thanks for helping out!

[mpilquist]

BTW, it might be good to implement these using some combination of Stream.{random, randomSeeded} and zip. Not sure though.

[monadplus]

For me the hardest part is to combine the stream of random factors with the one which is pulling from this in N sized chunks. I came up with this horrible looking implementation. So I'd appreciate some guidance.

def rechunkRandomlyWithSeed[F2[x] >: F[x]: Concurrent](minFactor: Double,
                                                         maxFactor: Double,
                                                         seed: Long): Stream[F2, O] = {
    assert(maxFactor >= minFactor, "maxFactor should be greater or equal to minFactor")
    Stream.eval(Queue.bounded[F2, Double](1)).flatMap { queue =>
      val factors =
        Stream
          .randomSeeded(seed)
          .covary[F]
          .map(x => Math.abs(x) % (maxFactor - minFactor) + minFactor)
          .through(queue.enqueue)

      def go(s: Stream[F2, O]): Pull[F2, O, Unit] =
        s.pull.peek.flatMap {
          case Some((c, _)) =>
            Pull.eval(queue.dequeue1).flatMap { factor =>
              val size = Math.max(1, factor * c.size).toInt
              s.pull.unconsN(size, true).flatMap {
                case Some((hd, tl)) =>
                  Pull.output(hd) >> go(tl)
                case None =>
                  Pull.done
              }
            }
          case None =>
            Pull.done
        }

      go(this.covary[F2]).stream.concurrently(factors)
    }
  }

[monadplus]

I'll try, during this week, to come up with a better approach.
How should I proceed with testing ?

[monadplus]

This is the secondd try.

this.chunks
      .zip(factors)
      .pull
      .uncons1
      .flatMap {
        case Some(((chunk, factor), tl)) =>
          val size = Math.max(1, factor * chunk.size).toInt
          // TODO: go does something very similar to this logic
          //       Unify both.
          if (chunk.size < size) go(chunk, Some(size), tl)
          else if (chunk.size == size) Pull.output(chunk) >> go(Chunk.empty, None, tl)
          else {
            val (out, remaining) = chunk.splitAt(size - 1)
            Pull.output(out) >> go(remaining, None, tl)
          }
        case None =>
          Pull.done
      }
      .stream

[mpilquist]

The second attempt looks promising. Definitely don't want to use a Queue for performance reasons. Basing the implementation on zip will give better performance. You should be able to restructure that a bit so that the recursive go function contains all the chunk handling logic. Something like this:

def go(acc: Chunk.ChunkQueue[O], size: Option[Long], s: Stream[F2, (Chunk[O], Long)]): Pull[F2, O, Unit] =
  s.pull.uncons1.flatMap {
    case Some(((hd, factor), tl)) =>
      // If size is defined, we need to add elements to the `ChunkQueue` until it reaches the defined size,
      // so check if the new hd should be added in total or split. If we output a chunk, or if size is not
      // defined, compute a new size using the new factor and see if the remainder satisfies it.
      // Repeat until unsatisfied, then recurse.
    case None =>
      // Output the accumulated value as a chunk
  }

go(Chunk.ChunkQueue.empty, None, chunks.zip(factors))

One downside to this implementation is that each chunk in the original stream results in one new random number generation -- which will be ignored often if the original chunk sizes are small and the factors are large. The general way to improve this is avoiding the use of zip and instead, defining the recursive function to take two Stream.StepLeg arguments, one for the chunks stream and one for the factors stream.

Another implementation, which is the most performant and probably the simplest to build, is avoiding creating a factors stream totally and instead, "inlining" the "next factor" logic in to your recursive function.

[monadplus]

@mpilquist There is my proposal. I chose the "inline" implementation. It is the simplest implementation and probably the most performant.

[monadplus]

Out of curiosity, what's the difference between uncons and stepLeg for a final user ? Looking at the code, both have support for similar operations. The only difference I can see is stepLeg exposes the scopeId which may be used to control the execution but that's implementation detail.

[mpilquist]

Fixed in #1428