Update http4s (#6040)

- Replace doobie with quill and async drivers.
- Update http4s, Scala, etc. to newest versions.
- Update to JDK 15.
- Improve JVM flags.
- Tweak thread pool sizes.

frameworks/Scala/http4s/build.sbt

@@ -2,7 +2,7 @@ name := "http4s"
 
 version := "1.0"
 
-scalaVersion := "2.13.1"
+scalaVersion := "2.13.3"
 
 scalacOptions ++= Seq(
   "-deprecation",
@@ -12,13 +12,18 @@ scalacOptions ++= Seq(
   "-unchecked",
   "-language:reflectiveCalls",
   "-Ywarn-numeric-widen",
+  "-target:11",
   "-Xlint"
 )
 
 enablePlugins(SbtTwirl)
 
-val http4sVersion = "0.21.3"
-val doobieVersion = "0.8.8"
+val http4sVersion = "0.21.7"
+
+assemblyMergeStrategy in assembly := {
+  case PathList(xs @ _*) if xs.last == "io.netty.versions.properties" => MergeStrategy.rename
+  case other => (assemblyMergeStrategy in assembly).value(other)
+}
 
 libraryDependencies ++= Seq(
   "org.http4s" %% "http4s-blaze-server" % http4sVersion,
@@ -27,8 +32,12 @@ libraryDependencies ++= Seq(
   "org.http4s" %% "http4s-circe" % http4sVersion,
   // Optional for auto-derivation of JSON codecs
   "io.circe" %% "circe-generic" % "0.13.0",
-  "org.tpolecat" %% "doobie-core" % doobieVersion,
-  "org.tpolecat" %% "doobie-hikari" % doobieVersion,
-  "org.tpolecat" %% "doobie-postgres" % doobieVersion,
+  "org.typelevel" %% "cats-effect" % "2.2.0",
+  "co.fs2" %% "fs2-core" % "2.4.4",
+  "co.fs2" %% "fs2-io" % "2.4.4",
+  "io.getquill" %% "quill-jasync-postgres" % "3.5.2",
+  "io.getquill" %% "quill-jasync" % "3.5.2",
   "ch.qos.logback" % "logback-classic" % "1.2.3"
 )
+
+addCompilerPlugin("com.olegpy" %% "better-monadic-for" % "0.3.1")

frameworks/Scala/http4s/http4s.dockerfile

@@ -1,4 +1,4 @@
-FROM openjdk:8 AS builder
+FROM openjdk:15 AS builder
 WORKDIR /http4s
 COPY project project
 COPY src src
@@ -11,7 +11,15 @@ RUN ./sbt assembly -batch && \
     rm -Rf ~/.sbt && \
     rm -Rf ~/.ivy2 && \
     rm -Rf /var/cache
-FROM openjdk:alpine
+FROM openjdk:15
 WORKDIR /http4s
 COPY --from=builder /http4s/http4s-assembly-1.0.jar /http4s/http4s-assembly-1.0.jar
-CMD ["java", "-server", "-Xms2g", "-Xmx2g", "-XX:NewSize=1g", "-XX:MaxNewSize=1g", "-XX:InitialCodeCacheSize=256m", "-XX:ReservedCodeCacheSize=256m", "-XX:+UseParallelGC", "-XX:+UseNUMA", "-XX:-UseBiasedLocking", "-XX:+AlwaysPreTouch", "-jar", "http4s-assembly-1.0.jar", "tfb-database"]
+CMD java \
+      -server \
+      -Xms2g \
+      -Xmx2g \
+      -XX:+AlwaysPreTouch \
+      -Dcats.effect.stackTracingMode=disabled \
+      -jar \
+      http4s-assembly-1.0.jar \
+      tfb-database

frameworks/Scala/http4s/project/build.properties

@@ -1 +1 @@
-sbt.version=1.3.9
+sbt.version=1.3.13

frameworks/Scala/http4s/project/plugins.sbt

@@ -1,3 +1,2 @@
 addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9")
-
 addSbtPlugin("com.typesafe.sbt" % "sbt-twirl" % "1.5.0")

frameworks/Scala/http4s/sbt

@@ -18,10 +18,10 @@ declare -r latest_28="2.8.2"
 
 declare -r buildProps="project/build.properties"
 
-declare -r sbt_launch_ivy_release_repo="http://repo.typesafe.com/typesafe/ivy-releases"
+declare -r sbt_launch_ivy_release_repo="https://repo.typesafe.com/typesafe/ivy-releases"
 declare -r sbt_launch_ivy_snapshot_repo="https://repo.scala-sbt.org/scalasbt/ivy-snapshots"
-declare -r sbt_launch_mvn_release_repo="http://repo.scala-sbt.org/scalasbt/maven-releases"
-declare -r sbt_launch_mvn_snapshot_repo="http://repo.scala-sbt.org/scalasbt/maven-snapshots"
+declare -r sbt_launch_mvn_release_repo="https://repo.scala-sbt.org/scalasbt/maven-releases"
+declare -r sbt_launch_mvn_snapshot_repo="https://repo.scala-sbt.org/scalasbt/maven-snapshots"
 
 declare -r default_jvm_opts_common="-Xms512m -Xss2m"
 declare -r noshare_opts="-Dsbt.global.base=project/.sbtboot -Dsbt.boot.directory=project/.boot -Dsbt.ivy.home=project/.ivy"

frameworks/Scala/http4s/src/main/resources/application.properties

@@ -0,0 +1,4 @@
+ctx.port=5432
+ctx.username=benchmarkdbuser
+ctx.password=benchmarkdbpass
+ctx.database=hello_world

frameworks/Scala/http4s/src/main/scala/http4s/techempower/benchmark/DatabaseService.scala

@@ -0,0 +1,73 @@
+package http4s.techempower.benchmark
+
+import java.util.concurrent.{Executor, ThreadLocalRandom}
+
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
+import cats.effect.{ContextShift, IO => CatsIO}
+import cats.syntax.all._
+import io.getquill._
+
+class DatabaseService(ctx: PostgresJAsyncContext[LowerCase.type], executor: Executor)(implicit
+    cs: ContextShift[CatsIO]
+) {
+  implicit val dbExecutionContext: ExecutionContextExecutor = ExecutionContext.fromExecutor(executor)
+  import ctx._
+
+  def close(): CatsIO[Unit] = {
+    CatsIO(ctx.close())
+  }
+
+  // Provide a random number between 1 and 10000 (inclusive)
+  private def randomWorldId() =
+    CatsIO(ThreadLocalRandom.current().nextInt(1, 10001))
+
+  // Update the randomNumber field with a random number
+  def updateRandomNumber(world: World): CatsIO[World] =
+    for {
+      randomId <- randomWorldId()
+    } yield world.copy(randomNumber = randomId)
+
+  // Select a World object from the database by ID
+  def selectWorld(id: Int): CatsIO[World] =
+    CatsIO.fromFuture(
+      CatsIO.delay(
+        ctx
+          .run(quote {
+            query[World].filter(_.id == lift(id))
+          })
+          .map(rq => rq.head)
+      )
+    )
+
+  // Select a random World object from the database
+  def selectRandomWorld(): CatsIO[World] =
+    for {
+      randomId <- randomWorldId()
+      world <- selectWorld(randomId)
+    } yield world
+
+  // Select a specified number of random World objects from the database
+  def getWorlds(numQueries: Int): CatsIO[List[World]] =
+    (0 until numQueries).toList.traverse(_ => selectRandomWorld())
+
+  // Update the randomNumber field with a new random number, for a list of World objects
+  def getNewWorlds(worlds: List[World]): CatsIO[List[World]] =
+    worlds.map(updateRandomNumber).sequence
+
+  // Update the randomNumber column in the database for a specified set of World objects,
+  // this uses a batch update SQL call.
+  def updateWorlds(newWorlds: List[World]): CatsIO[Int] = {
+    val u = quote {
+      liftQuery(newWorlds).foreach { world =>
+        query[World]
+          .filter(_.id == world.id)
+          .update(_.randomNumber -> world.randomNumber)
+      }
+    }
+    CatsIO.fromFuture(CatsIO.delay(ctx.run(u).map(_.length)))
+  }
+
+  // Retrieve all fortunes from the database
+  def getFortunes(): CatsIO[List[Fortune]] =
+    CatsIO.fromFuture(CatsIO.delay(ctx.run(query[Fortune]).map(_.toList)))
+}

frameworks/Scala/http4s/src/main/scala/http4s/techempower/benchmark/WebServer.scala

@@ -1,17 +1,15 @@
 package http4s.techempower.benchmark
 
-import java.util.concurrent.ThreadLocalRandom
+import java.util.concurrent.Executors
 
-import cats.effect._
-import cats.instances.list._
-import cats.syntax.parallel._
-import cats.syntax.traverse._
+import scala.concurrent.ExecutionContext
+import cats.effect.{ExitCode, IO, IOApp, Resource}
+import com.typesafe.config.ConfigValueFactory
 import io.circe.generic.auto._
 import io.circe.syntax._
-import doobie._
-import doobie.implicits._
-import doobie.hikari.HikariTransactor
-import doobie.util.ExecutionContexts
+import io.getquill.util.LoadConfig
+import io.getquill.LowerCase
+import io.getquill.PostgresJAsyncContext
 import org.http4s._
 import org.http4s.dsl._
 import org.http4s.circe._
@@ -20,9 +18,9 @@ import org.http4s.server.Router
 import org.http4s.server.blaze.BlazeServerBuilder
 import org.http4s.twirl._
 
-case class Message(message: String)
-case class World(id: Int, randomNumber: Int)
-case class Fortune(id: Int, message: String)
+final case class Message(message: String)
+final case class World(id: Int, randomNumber: Int)
+final case class Fortune(id: Int, message: String)
 
 // Extract queries parameter (with default and min/maxed)
 object Queries {
@@ -35,72 +33,25 @@ object Queries {
 }
 
 object WebServer extends IOApp with Http4sDsl[IO] {
-  def openDatabase(host: String,
-                   poolSize: Int): Resource[IO, HikariTransactor[IO]] =
+  def makeDatabaseService(
+      host: String,
+      poolSize: Int
+  ): Resource[IO, DatabaseService] = {
     for {
-      ce <- ExecutionContexts.fixedThreadPool[IO](32) // our connect EC
-      be <- Blocker[IO] // our blocking EC
-      xa <- HikariTransactor.newHikariTransactor[IO](
-        "org.postgresql.Driver",
-        s"jdbc:postgresql://$host/hello_world",
-        "benchmarkdbuser",
-        "benchmarkdbpass",
-        ce,
-        be
-      )
-      _ <- Resource.liftF(
-        xa.configure(
-          ds =>
-            IO {
-              ds.setMaximumPoolSize(poolSize)
-              ds.setMinimumIdle(poolSize)
-          }
-        )
-      )
-    } yield xa
-
-  // Provide a random number between 1 and 10000 (inclusive)
-  val randomWorldId: IO[Int] = IO(ThreadLocalRandom.current.nextInt(1, 10001))
-
-  // Update the randomNumber field with a random number
-  def updateRandomNumber(world: World): IO[World] =
-    randomWorldId.map(id => world.copy(randomNumber = id))
-
-  // Select a World object from the database by ID
-  def selectWorld(xa: Transactor[IO], id: Int): IO[World] =
-    sql"select id, randomNumber from World where id = $id"
-      .query[World]
-      .unique
-      .transact(xa)
-
-  // Select a random World object from the database
-  def selectRandomWorld(xa: Transactor[IO]): IO[World] =
-    randomWorldId flatMap { id =>
-      selectWorld(xa, id)
-    }
-
-  // Select a specified number of random World objects from the database
-  def getWorlds(xa: Transactor[IO], numQueries: Int): IO[List[World]] =
-    (0 until numQueries).toList.parTraverse(_ => selectRandomWorld(xa))
-
-  // Update the randomNumber field with a new random number, for a list of World objects
-  def getNewWorlds(worlds: List[World]): IO[List[World]] =
-    worlds.traverse(updateRandomNumber)
-
-  // Update the randomNumber column in the database for a specified set of World objects,
-  // this uses a batch update SQL call.
-  def updateWorlds(xa: Transactor[IO], newWorlds: List[World]): IO[Int] = {
-    val sql = "update World set randomNumber = ? where id = ?"
-    // Reason for sorting: https://github.com/TechEmpower/FrameworkBenchmarks/pull/4214#issuecomment-489358881
-    val update = Update[(Int, Int)](sql)
-      .updateMany(newWorlds.sortBy(_.id).map(w => (w.randomNumber, w.id)))
-    update.transact(xa)
-  }
-
-  // Retrieve all fortunes from the database
-  def getFortunes(xa: Transactor[IO]): IO[List[Fortune]] = {
-    val query = sql"select id, message from Fortune".query[Fortune]
-    query.to[List].transact(xa)
+      executor <- Resource(IO {
+        val pool = Executors.newFixedThreadPool(poolSize)
+        (pool, IO(pool.shutdown()))
+      })
+      ctx <- Resource.fromAutoCloseable(IO(new PostgresJAsyncContext(
+        LowerCase,
+        LoadConfig("ctx")
+          .withValue("host", ConfigValueFactory.fromAnyRef(host))
+          .withValue(
+            "maxActiveConnections",
+            ConfigValueFactory.fromAnyRef(poolSize)
+          )
+      )))
+    } yield new DatabaseService(ctx, executor)
   }
 
   // Add a new fortune to an existing list, and sort by message.
@@ -116,7 +67,7 @@ object WebServer extends IOApp with Http4sDsl[IO] {
     }
 
   // HTTP service definition
-  def service(xa: Transactor[IO]) =
+  def service(db: DatabaseService) =
     addServerHeader(HttpRoutes.of[IO] {
       case GET -> Root / "plaintext" =>
         Ok("Hello, World!")
@@ -125,41 +76,43 @@ object WebServer extends IOApp with Http4sDsl[IO] {
         Ok(Message("Hello, World!").asJson)
 
       case GET -> Root / "db" =>
-        Ok(selectRandomWorld(xa).map(_.asJson))
+        Ok(db.selectRandomWorld().map(_.asJson))
 
       case GET -> Root / "queries" :? Queries(numQueries) =>
-        Ok(getWorlds(xa, numQueries).map(_.asJson))
+        Ok(db.getWorlds(numQueries).map(_.asJson))
 
       case GET -> Root / "fortunes" =>
         Ok(for {
-          oldFortunes <- getFortunes(xa)
+          oldFortunes <- db.getFortunes()
           newFortunes = getSortedFortunes(oldFortunes)
         } yield html.index(newFortunes))
 
       case GET -> Root / "updates" :? Queries(numQueries) =>
         Ok(for {
-          worlds <- getWorlds(xa, numQueries)
-          newWorlds <- getNewWorlds(worlds)
-          _ <- updateWorlds(xa, newWorlds)
+          worlds <- db.getWorlds(numQueries)
+          newWorlds <- db.getNewWorlds(worlds)
+          _ <- db.updateWorlds(newWorlds)
         } yield newWorlds.asJson)
     })
 
+  val blazeEc = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(32))
+
   // Given a fully constructed HttpService, start the server and wait for completion
   def startServer(service: HttpRoutes[IO]) =
-    BlazeServerBuilder[IO]
+    BlazeServerBuilder[IO](blazeEc)
       .bindHttp(8080, "0.0.0.0")
       .withHttpApp(Router("/" -> service).orNotFound)
+      .withSocketKeepAlive(true)
       .resource
 
   // Entry point when starting service
   override def run(args: List[String]): IO[ExitCode] =
     (for {
-      db <- openDatabase(
+      db <- makeDatabaseService(
         args.headOption.getOrElse("localhost"),
-        sys.env.get("DB_POOL_SIZE").map(_.toInt).getOrElse(256)
+        sys.env.get("DB_POOL_SIZE").map(_.toInt).getOrElse(64)
       )
       server <- startServer(service(db))
     } yield server)
       .use(_ => IO.never)
-      .map(_ => ExitCode.Success)
 }