raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

simplify forkjoin code (#1975)

Browse source 
* simplify forkjoin code

* Create simplify-forkjoin.excludes
This commit is contained in:
PJ Fanning 2025-07-30 16:14:33 +01:00 • committed by GitHub
parent d73899ed77
commit b6849b9564
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 25 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

19
actor/src/main/mima-filters/2.0.x.backwards.excludes/simplify-forkjoin.excludes Normal file
View file

@ -0,0 +1,19 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you 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 distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# No longer need special code to support forkjoin dispatcher on JDK 8
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.ForkJoinExecutorConfigurator#PekkoForkJoinPool.this")

51
actor/src/main/scala-jdk-9/org/apache/pekko/dispatch/PekkoJdk9ForkJoinPool.scala
View file

@ -1,51 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.pekko.dispatch
import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.dispatch.ForkJoinExecutorConfigurator.PekkoForkJoinTask
import java.util.concurrent.{ ForkJoinPool, ForkJoinTask, TimeUnit }
/**
* INTERNAL PEKKO USAGE ONLY
*
* An alternative version of [[ForkJoinExecutorConfigurator.PekkoForkJoinPool]]
* that supports the `maximumPoolSize` feature available in [[java.util.concurrent.ForkJoinPool]] in JDK9+.
*/
@InternalApi
private[dispatch] final class PekkoJdk9ForkJoinPool(
parallelism: Int,
threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory,
maximumPoolSize: Int,
unhandledExceptionHandler: Thread.UncaughtExceptionHandler,
asyncMode: Boolean)
extends ForkJoinPool(parallelism, threadFactory, unhandledExceptionHandler, asyncMode,
0, maximumPoolSize, 1, null, ForkJoinPoolConstants.DefaultKeepAliveMillis, TimeUnit.MILLISECONDS)
with LoadMetrics {
override def execute(r: Runnable): Unit =
if (r ne null)
super.execute(
(if (r.isInstanceOf[ForkJoinTask[_]]) r else new PekkoForkJoinTask(r)).asInstanceOf[ForkJoinTask[Any]])
else
throw new NullPointerException("Runnable was null")
def atFullThrottle(): Boolean = this.getActiveThreadCount() >= this.getParallelism()
}

39
actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
View file

@ -18,9 +18,7 @@ import org.apache.pekko
import pekko.dispatch.VirtualThreadSupport.newVirtualThreadFactory
import pekko.util.JavaVersion
import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType }
import java.util.concurrent.{ Executor, ExecutorService, ForkJoinPool, ForkJoinTask, ThreadFactory }
import scala.util.Try
import java.util.concurrent.{ Executor, ExecutorService, ForkJoinPool, ForkJoinTask, ThreadFactory, TimeUnit }
object ForkJoinExecutorConfigurator {
@ -30,15 +28,12 @@ object ForkJoinExecutorConfigurator {
final class PekkoForkJoinPool(
parallelism: Int,
threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory,
maximumPoolSize: Int,
unhandledExceptionHandler: Thread.UncaughtExceptionHandler,
asyncMode: Boolean)
extends ForkJoinPool(parallelism, threadFactory, unhandledExceptionHandler, asyncMode)
extends ForkJoinPool(parallelism, threadFactory, unhandledExceptionHandler, asyncMode,
0, maximumPoolSize, 1, null, ForkJoinPoolConstants.DefaultKeepAliveMillis, TimeUnit.MILLISECONDS)
with LoadMetrics {
def this(
parallelism: Int,
threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory,
unhandledExceptionHandler: Thread.UncaughtExceptionHandler) =
this(parallelism, threadFactory, unhandledExceptionHandler, asyncMode = true)
override def execute(r: Runnable): Unit =
if (r ne null)
@ -110,23 +105,6 @@ class ForkJoinExecutorConfigurator(config: Config, prerequisites: DispatcherPrer
asyncMode: Boolean,
maxPoolSize: Int) = this(threadFactory, parallelism, asyncMode, maxPoolSize, false)
private def pekkoJdk9ForkJoinPoolClassOpt: Option[Class[_]] =
Try(Class.forName("org.apache.pekko.dispatch.PekkoJdk9ForkJoinPool")).toOption
private lazy val pekkoJdk9ForkJoinPoolHandleOpt: Option[MethodHandle] = {
if (JavaVersion.majorVersion == 8) {
None
} else {
pekkoJdk9ForkJoinPoolClassOpt.map { clz =>
val methodHandleLookup = MethodHandles.lookup()
val mt = MethodType.methodType(classOf[Unit], classOf[Int],
classOf[ForkJoinPool.ForkJoinWorkerThreadFactory],
classOf[Int], classOf[Thread.UncaughtExceptionHandler], classOf[Boolean])
methodHandleLookup.findConstructor(clz, mt)
}
}
}
def this(threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory, parallelism: Int) =
this(threadFactory, parallelism, asyncMode = true)
@ -139,14 +117,7 @@ class ForkJoinExecutorConfigurator(config: Config, prerequisites: DispatcherPrer
}
} else threadFactory
val pool = pekkoJdk9ForkJoinPoolHandleOpt match {
case Some(handle) =>
// carrier Thread only exists in JDK 17+
handle.invoke(parallelism, tf, maxPoolSize, MonitorableThreadFactory.doNothing, asyncMode)
.asInstanceOf[ExecutorService with LoadMetrics]
case _ =>
new PekkoForkJoinPool(parallelism, tf, MonitorableThreadFactory.doNothing, asyncMode)
}
val pool = new PekkoForkJoinPool(parallelism, tf, maxPoolSize, MonitorableThreadFactory.doNothing, asyncMode)
if (virtualize && JavaVersion.majorVersion >= 21) {
// when virtualized, we need enhanced thread factory

2
build.sbt
View file

@ -121,7 +121,7 @@ lazy val actor = pekkoModule("actor")
.settings(AddMetaInfLicenseFiles.actorSettings)
.settings(VersionGenerator.settings)
.settings(serialversionRemoverPluginSettings)
.enablePlugins(BoilerplatePlugin, SbtOsgi, Jdk9)
.enablePlugins(BoilerplatePlugin, SbtOsgi)
lazy val actorTests = pekkoModule("actor-tests")
.configs(Jdk9.TestJdk9)