/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
 */

package org.apache.pekko.dispatch

import com.typesafe.config.ConfigFactory

import org.apache.pekko
import pekko.actor.{ Actor, Props }
import pekko.testkit.{ ImplicitSender, PekkoSpec }

object ForkJoinPoolStarvationSpec {
  val config = ConfigFactory.parseString("""
      |actorhang {
      |  task-dispatcher {
      |    mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
      |    throughput = 5
      |    fork-join-executor {
      |      parallelism-factor = 2
      |      parallelism-max = 2
      |      parallelism-min = 2
      |    }
      |  }
      |}
    """.stripMargin)

  class SelfBusyActor extends Actor {
    self ! "tick"

    override def receive = {
      case "tick" =>
        self ! "tick"
    }
  }

  class InnocentActor extends Actor {

    override def receive = {
      case "ping" =>
        sender() ! "All fine"
    }
  }

}

class ForkJoinPoolStarvationSpec extends PekkoSpec(ForkJoinPoolStarvationSpec.config) with ImplicitSender {
  import ForkJoinPoolStarvationSpec._

  val Iterations = 1000

  "PekkoForkJoinPool" must {

    "not starve tasks arriving from external dispatchers under high internal traffic" in {
      // TODO issue #31117: starvation with JDK 17 FJP
      if (System.getProperty("java.specification.version") == "17")
        pending

      // Two busy actors that will occupy the threads of the dispatcher
      // Since they submit to the local task queue via fork, they can starve external submissions
      system.actorOf(Props(new SelfBusyActor).withDispatcher("actorhang.task-dispatcher"))
      system.actorOf(Props(new SelfBusyActor).withDispatcher("actorhang.task-dispatcher"))

      val innocentActor = system.actorOf(Props(new InnocentActor).withDispatcher("actorhang.task-dispatcher"))

      for (_ <- 1 to Iterations) {
        // External task submission via the default dispatcher
        innocentActor ! "ping"
        expectMsg("All fine")
      }
    }

  }
}
