WIP: CE 3 #273

iRevive · 2021-04-04T15:58:15Z

Temporarily disabled modules: ~~odin-zio~~, odin-monix
Missing CE3 dependencies: ~~zio-cats~~, monix

odin-core depends on cats-effect-std.

Changes due to missing Monix dependency:

Replaced ConcurrentQueue from Monix with Queue from cats.effect.std
Replaced monix.Task with cats.effect.IO in tests
Removed monix.execution.Scheduler in favor of IORuntime

Once a compatible version of Monix will be released, I can revert these changes.

Benchmarks

I observed a performance degradation after the upgrade to CE3. Evaluating a task via .unsafeRunSync() in a for-loop is 3x slower comparing to the CE2. Using traverse instead of a for-loop leads to more clear results.

for-loop

The results below represent the evaluation of a logging effect in a for-loop. Example:

@Benchmark
@OperationsPerInvocation(1000)
def msg(): Unit = for (_ <- 1 to 1000) logger.info(message).unsafeRunSync()

Benchmark	Mode	Cnt	Score	Error	Units
FileLoggerBenchmarks.msg	avgt	25	21488.981	± 1462.785	ns/op
FileLoggerBenchmarks.msgAndCtx	avgt	25	20273.076	± 695.874	ns/op
FileLoggerBenchmarks.msgCtxThrowable	avgt	25	30110.791	± 1010.558	ns/op

AsyncLoggerBenchmark.msg	avgt	25	18097.440	± 4621.083	ns/op
AsyncLoggerBenchmark.msgAndCtx	avgt	25	14337.669	± 1383.534	ns/op
AsyncLoggerBenchmark.msgCtxThrowable	avgt	25	17652.329	± 1098.268	ns/op

ScribeBenchmark.asyncMsg	avgt	25	114.641	± 3.568	ns/op
ScribeBenchmark.asyncMsgCtx	avgt	25	131.083	± 2.123	ns/op
ScribeBenchmark.msg	avgt	25	1443.887	± 34.406	ns/op
ScribeBenchmark.msgAndCtx	avgt	25	1717.407	± 50.303	ns/op

traverse

@Benchmark
@OperationsPerInvocation(1000)
def msg(): Unit = (1 to 1000).toList.traverse(_ => logger.info(message)).unsafeRunSync()

Benchmark	Mode	Cnt	Score	Error	Units
FileLoggerBenchmarks.msg	avgt	25	7750.887	± 456.193	ns/op
FileLoggerBenchmarks.msgAndCtx	avgt	25	8385.711	± 585.243	ns/op
FileLoggerBenchmarks.msgCtxThrowable	avgt	25	21720.537	± 4569.168	ns/op

AsyncLoggerBenchmark.msg	avgt	25	1486.737	± 271.336	ns/op
AsyncLoggerBenchmark.msgAndCtx	avgt	25	1523.111	± 211.884	ns/op
AsyncLoggerBenchmark.msgCtxThrowable	avgt	25	1624.252	± 170.380	ns/op

AsyncLoggerBenchmark issue

From my point of view, the async logger benchmark implemented in a bit wrong way. And it does not measure the real throughput.

The key element of the AsyncLogger is a Queue. Logging a message, basically, an enqueue operation:

def submit(msg: LoggerMessage): F[Unit] = {
  queue.tryOffer(msg).void
}

In benchmarks, the size of a queue is 1_000_000 elements and the flush period is 1 millisecond. Since the JMH executes the code thousands of times, the queue is populated up to the limit almost immediately. Hence the tryOffer method does nothing during evaluation:

def tryOffer(a: A): F[Boolean] =
  state
    .modify {
      case State(queue, size, takers, offerers) if takers.nonEmpty =>
        val (taker, rest) = takers.dequeue
        State(queue, size, rest, offerers) -> taker.complete(a).as(true)
      case State(queue, size, takers, offerers) if size < capacity =>
        State(queue.enqueue(a), size + 1, takers, offerers) -> F.pure(true)
      case s => 
         s -> F.pure(false) <- the branch being evaluated when the queue is full
    }
    .flatten
    .uncancelable

To prove my assumption I changed the logic of the background fiber:

def runF: F[Fiber[F, Throwable, Unit]] = {
-  def drainLoop: F[Unit] = drain >> F.sleep(timeWindow) >> F.cede >> drainLoop
+  def drainLoop: F[Unit] = F.unit

  F.start(drainLoop).map { fiber =>
    new Fiber[F, Throwable, Unit] {
      override def cancel: F[Unit] = drain >> fiber.cancel
      override def join: F[Outcome[F, Throwable, Unit]] = fiber.join
    }
  }
}

The queue never being drained and tryOffer does nothing. And the measurements became similar to the CE2 version:

Benchmark	Mode	Cnt	Score	Error	Units
AsyncLoggerBenchmark.msg	avgt	25	996.862	± 393.928	ns/op
AsyncLoggerBenchmark.msgAndCtx	avgt	25	710.134	± 134.316	ns/op
AsyncLoggerBenchmark.msgCtxThrowable	avgt	25	741.075	± 195.111	ns/op

codecov · 2021-04-04T16:35:40Z

Codecov Report

Merging #273 (be287ec) into master (19c38e4) will decrease coverage by 1.70%.
The diff coverage is 76.40%.

@@            Coverage Diff             @@
##           master     #273      +/-   ##
==========================================
- Coverage   93.02%   91.32%   -1.71%     
==========================================
  Files          33       33              
  Lines         502      530      +28     
  Branches        9       14       +5     
==========================================
+ Hits          467      484      +17     
- Misses         35       46      +11

Flag	Coverage Δ
unittests	`91.32% <76.40%> (-1.71%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
benchmarks/src/main/scala/io/odin/Test.scala	`0.00% <0.00%> (ø)`
...src/main/scala/io/odin/config/DefaultBuilder.scala	`100.00% <ø> (ø)`
...c/main/scala/io/odin/config/EnclosureRouting.scala	`100.00% <ø> (ø)`
core/src/main/scala/io/odin/config/package.scala	`100.00% <ø> (ø)`
...src/main/scala/io/odin/loggers/ConsoleLogger.scala	`100.00% <ø> (ø)`
...ain/scala/io/odin/loggers/ConstContextLogger.scala	`100.00% <ø> (ø)`
.../main/scala/io/odin/loggers/ContextualLogger.scala	`100.00% <ø> (ø)`
...c/main/scala/io/odin/loggers/ContramapLogger.scala	`100.00% <ø> (ø)`
.../src/main/scala/io/odin/loggers/FilterLogger.scala	`100.00% <ø> (ø)`
...src/main/scala/io/odin/loggers/WriterTLogger.scala	`100.00% <ø> (ø)`
... and 16 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 19c38e4...be287ec. Read the comment docs.

iRevive · 2021-04-04T16:38:59Z

Outstanding items:

Fix flaky test: RollingFileLoggerSpec
Wait for compatible dependencies:
- monix
- zio-cats

iRevive · 2021-04-04T16:42:22Z

Evaluating a task via .unsafeRunSync() in a for-loop is 3x slower comparing to the CE2. Using traverse instead of a for-loop leads to more clear results.

I didn't investigate this issue yet.
@kubukoz perhaps you've observed the similar behavior in different libraries during upgrade to CE 3?

kubukoz · 2021-04-04T17:35:56Z

I haven't done any benchmarking with CE3.

@vasilmkd you might be interested

vasilmkd · 2021-04-04T18:01:24Z

Can confirm, unsafeRunSync on IO initially shifts to the IO compute execution context and there are some other bookkeeping actions taken to ensure safe execution and propagation of errors, which has a non-trivial overhead especially visible in benchmarks. It's basically not an apples to apples comparison anymore. Not sure if this helps...

kubukoz · 2021-04-04T18:16:13Z

README.md

-import cats.effect.{ContextShift, IO}
-import cats.effect.Clock
+import cats.effect.IO
+import cats.effect.unsafe.IORuntime


Suggested change

import cats.effect.unsafe.IORuntime

import cats.effect.unsafe.implicits._

and you won't need to define the implicit val for IORuntime.

kubukoz · 2021-04-04T18:18:21Z

README.md

-  implicit val F: Effect[IO] = IO.ioEffect
-  implicit val clock: Clock[IO] = Clock.create
+  implicit val F: Sync[IO] = IO.asyncForIO
+  implicit val dispatcher: Dispatcher[IO] = Dispatcher[IO].allocated.unsafeRunSync()._1


bad idea to do allocated on Dispatcher, can you use an IORuntime instead?

I guess it's not really possible since the interface is supposed to work on any F[_]...

kubukoz · 2021-04-04T18:19:09Z

build.sbt

@@ -97,7 +98,7 @@ lazy val sharedSettings = Seq(
 lazy val `odin-core` = (project in file("core"))
  .settings(sharedSettings)
  .settings(
-    libraryDependencies ++= (monix % Test) :: catsMtl :: sourcecode :: monixCatnap :: perfolation :: catsEffect :: cats
+    libraryDependencies ++= (catsEffect % Test) :: catsMtl :: sourcecode :: perfolation :: catsEffectStd :: cats


You don't need to add catsEffectStd if you have core already.

@kubukoz I want to have IO only in tests, since cats-effect-std is enough for the core.

kubukoz · 2021-04-04T18:20:02Z

core/src/main/scala/io/odin/loggers/AsyncLogger.scala

-    timer: Timer[F],
-    contextShift: ContextShift[F]
+case class AsyncLogger[F[_]](queue: Queue[F, LoggerMessage], timeWindow: FiniteDuration, inner: Logger[F])(
+    implicit F: Async[F]


Not sure if you need the full Async here

The runF method uses F.start under the hood. If we move the event consumer loop outside of the class, the constraints can be relaxed to Monad and Clock.
Related comment: #273 (comment)

start is from Spawn, Async is way more powerful than that ;)

kubukoz · 2021-04-04T18:21:05Z

core/src/main/scala/io/odin/loggers/AsyncLogger.scala

-      implicit F: ConcurrentEffect[F]
-  ): Logger[F] = F.toIO(withAsync(inner, timeWindow, maxBufferSize).allocated).unsafeRunSync()._1
+      implicit F: Async[F],
+      dispatcher: Dispatcher[F]


It's not recommended to pass Dispatcher implicitly, you might be better off creating one here and using allocated here.

Due to semantic of the withAsyncUnsafe method the Dispatcher cannot be instantiated:

def withAsyncUnsafe[F[_]]( inner: Logger[F], timeWindow: FiniteDuration, maxBufferSize: Option[Int] )( implicit F: Async[F] ): Logger[F] = { val dispatcher: F[(Dispatcher[F], F[Unit])] = Dispatcher[F].allocated <- still cannot run an effect and access the dispatcher }

It can work in a case of the different signature:

def withAsync[F[_]]: Resource[F, Logger[F]] = ??? - def withAsyncUnsafe[F[_]](...): Logger[F] = ??? + def withAsyncUnsafe[F[_]](...): F[Logger[F]] = ???

@sergeykolbasov what do you think?

I'd say if someone is feeling adventurous enough to deal with unsafe API, let them do it on their own by providing a custom dispatcher implicitly. If the users want to deal with unsafety, they should know better how to deal with it on their side.

I'd take it explicitly, but I agree about having to manage it on the user's side :)

kubukoz · 2021-04-04T18:22:02Z

core/src/main/scala/io/odin/loggers/DefaultLogger.scala

@@ -28,7 +26,7 @@ abstract class DefaultLogger[F[_]](val minLevel: Level)(implicit clock: Clock[F]
          exception = t,
          position = position,
          threadName = Thread.currentThread().getName,


random thought, this should be suspended

kubukoz · 2021-04-04T18:22:55Z

core/src/main/scala/io/odin/loggers/RollingFileLogger.scala

-              _ <- timer.sleep(100.millis)
-              _ <- cs.shift
+              _ <- F.sleep(100.millis)
+              _ <- F.cede


There's no need to cede after a sleep

Didn't know this, thanks!

kubukoz · 2021-04-04T18:24:32Z

core/src/test/scala/io/odin/loggers/ConstContextLoggerSpec.scala


  checkAll(
    "ContextualLogger",
    LoggerTests[F](
      new WriterTLogger[IO].withConstContext(Map.empty),
-      _.written.unsafeRunSync()
+      _.written.evalOn(singleThreadCtx).unsafeRunSync()


Why do you need these evalOns?

IO executes effects on different threads more often comparing to CE2. Therefore loggerMessageEq returns false since message.threadName is different. Executing effects on a single thread prevents such an issue.

On the other hand, evalOn feels more a bandaid than a proper fix. Perhaps I should ignore threadName field in the Eq logic.

sergeykolbasov · 2021-04-05T09:43:13Z

core/src/main/scala/io/odin/loggers/AsyncLogger.scala

+  def runF: F[Fiber[F, Throwable, Unit]] = {
+    def drainOnce: F[Unit] = drain >> F.sleep(timeWindow) >> F.cede
+
+    F.start(drainOnce.foreverM[Unit]).map { fiber =>


there is a Supervisor abstraction in CE3:

https://github.com/typelevel/cats-effect/blob/series/3.x/std/shared/src/main/scala/cats/effect/std/Supervisor.scala

Would it fit better here?

Also https://github.com/typelevel/cats-effect/blob/series/3.x/kernel/shared/src/main/scala/cats/effect/kernel/GenSpawn.scala

I'm not sure to be honest. runF already is a part of the AsyncLogger lifecycle.

Btw, should runF even be public or part of the class? The event consumer loop should be started only once.

At this case, the definition of the AsyncLogger can be simplified:

case class AsyncLogger[F: Monad: Clock] private (...) extends DefaultLogger(...) { def submit(msg: LoggerMessage): F[Unit] = ... private def drain: F[Unit] = ... } object AsyncLogger { def withAsync[F[_]: Async](inner: Logger[F], timeWindow: FiniteDuration, maxBufferSize: Option[Int]): Resource[F, Logger[F]] = { val createQueue = ... def backgroundConsumer(logger: AsyncLogger[F]): Resource[F, Unit] = { def drainLoop: F[Unit] = F.andWait(logger.drain, timeWindow).foreverM[Unit] // cannot use F.background due to a custom cancellation logic Resource.make(F.start(drainLoop))(fiber => logger.drain >> fiber.cancel).void } for { queue <- Resource.eval(createQueue) logger <- Resource.pure(AsyncLogger(queue, timeWindow, inner)) _ <- backgroundConsumer(logger) } yield logger } }

sounds good to me

sergeykolbasov · 2021-04-05T09:50:35Z

Thanks for the effort @iRevive

That RollingFileLogger spec is indeed annoying, however I couldn't manage the timer mock for that specific case back in the days. I guess it's related to the internal rolling loop, but gave up tracing it down to the root cause

sergeykolbasov · 2021-04-06T21:42:33Z

@iRevive do you think using Hotswap would make a rolling file logger cleaner, and those fix the test?

kubukoz · 2021-04-06T22:42:49Z

fs2-io is using Hotswap internally to implement something similar, for reference: https://github.com/typelevel/fs2/blob/24370abb527147da78b93d59a5be60e1079fdfbe/io/src/main/scala/fs2/io/file/Files.scala#L507-L555

iRevive · 2021-04-07T11:56:52Z

@iRevive do you think using Hotswap would make a rolling file logger cleaner, and those fix the test?

Can be useful. I will try to use the Hotswap

iRevive · 2021-04-07T13:43:14Z

Switched to Hotswap. RollingFileLoggerSpec is not failing locally and on CI.

kubukoz · 2021-04-07T13:52:19Z

core/src/main/scala/io/odin/loggers/AsyncLogger.scala

+      * Run internal loop of consuming events from the queue and push them down the chain
+      */
+    def backgroundConsumer(logger: AsyncLogger[F]): Resource[F, Unit] = {
+      def drainLoop: F[Unit] = F.andWait(logger.drain, timeWindow).foreverM[Unit]


Seeing andWait being actually used makes me happy.

kubukoz · 2021-04-07T13:53:05Z

I can't promise I'll make it before the weekend, but I'm planning to go through this again and search for any potential fiber leaks etc. :)

iRevive · 2021-04-20T19:16:57Z

zio/src/main/scala/io/odin/zio/package.scala

@@ -28,13 +30,18 @@ package object zio {
      fileName: String,
      formatter: Formatter = Formatter.default,
      minLevel: Level = Level.Trace
-  ): Managed[LoggerError, Logger[IO[LoggerError, *]]] =
+  ): ZManaged[Clock & CBlocking, LoggerError, Logger[IO[LoggerError, *]]] =
    ZManaged


@kubukoz A gut feeling says there should be a more elegant combinator. Something similar to Resource.suspend.

If there is one, I'm not aware of it.

kubukoz · 2021-05-13T15:27:57Z

Can we get this merged? :)

sergeykolbasov · 2021-05-14T11:40:34Z

Thanks, @iRevive & @kubukoz

I'll drop a release soon

vasilmkd · 2021-05-15T01:20:54Z

@iRevive Do you mind repeating the unsafeRunSync benchmarks against the latest Cats Effect 3.1.1 release? We released several performance optimizations right on this code path that should be quite noticeable in benchmarks. Thank you and sorry for the trouble.

iRevive · 2021-05-16T08:33:47Z

@vasilmkd sure, I will give it a try.

kubukoz · 2021-06-11T14:58:09Z

hey @sergeykolbasov, I don't see a release in maven, did it fail or did you just not have a chance to do it yet?

Just FYI, I checked a local snapshot of this and it seems to work :) a little bummer about having to do Dispatcher[IO].allocated.unsafeRunSync()._1 but maybe it'll be better if we get typelevel/cats-effect#1791.

iRevive added 7 commits April 4, 2021 17:11

Update cats-effect to 3.0.1

99e1786

Fix examples/SimpleAppSpec

ea55927

Update tests

ab6419e

Use traverse in benchmarks

c4210e1

Run scalafmtAll

7b00e01

Drop unused dependencies

7044c41

Fix compilation issue in odin-slf4j module

01c54f9

iRevive changed the title ~~CE 3~~ WIP: CE 3 Apr 4, 2021

kubukoz reviewed Apr 4, 2021

View reviewed changes

sergeykolbasov reviewed Apr 5, 2021

View reviewed changes

iRevive added 2 commits April 5, 2021 13:06

Drop unnecessary F.cede

4c1dce4

Update documentation: mention import cats.effect.unsafe.implicits._

7d514ac

iRevive force-pushed the cats-effect-3 branch from 9775764 to 7d514ac Compare April 5, 2021 10:26

Move background consumer outside of the AsyncLogger. Relax constraints

3ce915f

Use Hotswap in RollingFileLogger internals

616cd27

kubukoz reviewed Apr 7, 2021

View reviewed changes

kubukoz approved these changes Apr 14, 2021

View reviewed changes

Update zio-interop-cats to 3.0.2.0

be287ec

kubukoz mentioned this pull request Apr 21, 2021

Cats Effect 3 support #271

Closed

kubukoz mentioned this pull request Apr 21, 2021

Ecosystem upgrade megathread typelevel/cats-effect#1330

Closed

iRevive commented Apr 21, 2021

View reviewed changes

kubukoz mentioned this pull request Apr 21, 2021

Cats Effect 3 migration checklist polyvariant/pitgull#179

Closed

5 tasks

sergeykolbasov merged commit 276ba00 into valskalla:master May 14, 2021

kubukoz mentioned this pull request Jun 11, 2021

Add IORuntime#dispatcher? typelevel/cats-effect#1791

Closed

	import cats.effect.unsafe.IORuntime
	import cats.effect.unsafe.implicits._

WIP: CE 3 #273

WIP: CE 3 #273

Conversation

iRevive commented Apr 4, 2021 • edited Loading

Benchmarks

for-loop

traverse

AsyncLoggerBenchmark issue

codecov bot commented Apr 4, 2021 • edited Loading

Codecov Report

iRevive commented Apr 4, 2021 • edited Loading

iRevive commented Apr 4, 2021

kubukoz commented Apr 4, 2021

vasilmkd commented Apr 4, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iRevive Apr 5, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iRevive Apr 5, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov commented Apr 5, 2021

sergeykolbasov commented Apr 6, 2021

kubukoz commented Apr 6, 2021

iRevive commented Apr 7, 2021

iRevive commented Apr 7, 2021

Choose a reason for hiding this comment

kubukoz commented Apr 7, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kubukoz commented May 13, 2021

sergeykolbasov commented May 14, 2021

vasilmkd commented May 15, 2021

iRevive commented May 16, 2021

kubukoz commented Jun 11, 2021

iRevive commented Apr 4, 2021 •

edited

Loading

codecov bot commented Apr 4, 2021 •

edited

Loading

iRevive commented Apr 4, 2021 •

edited

Loading

iRevive Apr 5, 2021 •

edited

Loading

iRevive Apr 5, 2021 •

edited

Loading