pchlupacek

{
  "ignition": {
    "config": {},
    "security": {
      "tls": {}
    },
    "timeouts": {},
    "version": "2.2.0"
  },
  "networkd": {},

pchlupacek

package fs2.benchmark

import java.util.concurrent.TimeUnit

import cats.implicits._
import cats.effect.{Concurrent, ContextShift, IO}
import org.openjdk.jmh.annotations.{Benchmark, BenchmarkMode, Mode, OutputTimeUnit, Scope, State}
import fs2._
import fs2.concurrent.{SignallingRef, Topic}

pchlupacek

/* related to case https://github.com/functional-streams-for-scala/fs2/issues/1397 **/

package fs2.concurrent

import java.util.concurrent.Executors

import cats.Traverse
import cats.implicits._
import cats.effect.{ExitCode, IO, IOApp}

pchlupacek

/*
 * Copyright (c) 2017-2018 The Typelevel Cats-effect Project Developers
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software

pchlupacek

 val refSync = async.syncRefOf[IO, Long](0l).unsafeRunSync()
  val refAsync = async.refOf[IO, Long](0l).unsafeRunSync()

  val count = 10000000

  def time[A](s: String)(f: IO[A]): A = {
    val start = System.currentTimeMillis()
    val a = f.unsafeRunSync()
    val took = System.currentTimeMillis() - start
    println(s"Execution of : $s took ${took.toFloat/1000} s")

pchlupacek

// source of uris, may be for example file, or another http server, db ...
val sourceOfUris: Stream[F, Uri] = ??? 

// max requests to process in parallel 
val maxConcurrent: Int = ???

val responses : Stream[F, Stream[F, HttpResponse[F]]] = 
  http.client[Task]().flatMap { client =>
    sourceOfUris map { uri => client.request(HttpRequest.get[F](uri))}
  }

pchlupacek

pub enum Trampoline<A> {
    Return(A)
    , Suspend(Box<Fn() -> Trampoline<A>>)
}

pchlupacek

package fs2.benchmark

import java.util.concurrent.CountDownLatch
import java.util.concurrent.atomic.AtomicInteger

import fs2.{Scope => _, _}
import org.openjdk.jmh.annotations._
import org.openjdk.jmh.infra.Blackhole

import scala.concurrent.duration._

pchlupacek

> jmh:run -i 3 -wi 3 -f1 -t1 -gc true -jvmArgs "-Xmx4G -XX:+UseG1GC"  .RealTimeBenchmark
[info] Compiling 1 Scala source to /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/classes...
Processing 262 classes from /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/classes with "reflection" generator
Writing out Java source to /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/src_managed/jmh and resources to /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/resource_managed/jmh
[info] Compiling 4 Java sources to /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/classes...
[info] Compiling 34 Java sources to /Users/pach/work/repo/n/scalaz-stream/benchmark/target/scala-2.12/classes...
[info] Running org.openjdk.jmh.Main -i 3 -wi 3 -f1 -t1 -gc true -jvmArgs -Xmx4G -XX:+UseG1GC .RealTimeBenchmark
[info] # JMH 1.14.1 (released 94 days ago)
[info] # VM version: JDK 1.8.0_40, VM 25.40-b25
[info] # VM invoker: /Library/Java/JavaVirtualMachin

pchlupacek

/**
    * Runs supplied process1 `process` in scope of resulting pipe.
    */
  def liftProcess1[F[_],I,O](process:Process1[I,O]):Pipe[F,I,O] = {
    def go(p1:Process1[I,O]):Handle[F,I] => Pull[F,O,Unit] = {
      _.receive1 {
        case (evt,h) => p1.feed1(evt).unemit match {
          case (out, np1) => np1 match {
            case Halt(End | Kill) => Pull.output(Chunk.seq(out))
            case Halt(Error(rsn)) => Pull.output(Chunk.seq(out)) >> Pull.fail(rsn)