From a3d8cf4fad819c059a1155af44b64368cf2afc34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Thu, 9 Jul 2026 02:31:19 +0800 Subject: [PATCH] Optimize reflective hot paths with method handles Motivation: Reduce repeated reflective lookup and per-call adaptation in actor creation, serialization, and JVM integration points while preserving reflection-compatible behavior on modular JDKs. Modification: Cache pre-adapted MethodHandles and VarHandles at component or class lifetime, use exact invocation on hot paths, and retain safe test migrations. Preserve JPMS caller lookup, class initialization, varargs, and exception behavior with focused regression tests. Result: Hot paths avoid repeated lookup and adaptation. Actor creation JMH showed no regression and a small non-conclusive improvement signal from 9.640 +/- 0.314 to 9.341 +/- 0.279 us/op. Tests: - JDK 17 focused actor reflection, dynamic access, PekkoException, protobuf serializer, and file descriptor metrics tests passed after the HashCode split: 36 tests - Earlier focused Scala 3.3.8, JDK 21, and JDK 25 reflection/virtual-thread validation passed on the same MethodHandle changes - sbt +headerCheckAll passed - scalafmt --list --mode diff-ref=origin/main passed - git diff --check passed - qodercli formal stdout review found no must-fix findings - GitHub Binary Compatibility, headers, code style, scalafmt, Scala 2.13/3/next tests, classic remoting, and Java 21 checks passed before the scope-only split References: Refs #3300 --- .../util/ReflectionExceptionFixtures.java | 46 ++++ .../org/apache/pekko/PekkoExceptionSpec.scala | 4 +- .../pekko/actor/DynamicAccessSpec.scala | 37 ++++ .../apache/pekko/util/ByteIteratorSpec.scala | 10 +- .../util/ByteStringInitializationSpec.scala | 9 +- .../org/apache/pekko/util/ReflectSpec.scala | 69 +++++- .../actor/typed/internal/ExtensionsImpl.scala | 7 +- .../apache/pekko/io/ByteBufferCleaner.java | 11 +- .../pekko/actor/IndirectActorProducer.scala | 17 +- .../pekko/actor/ReflectiveDynamicAccess.scala | 87 ++++++-- .../pekko/dispatch/VirtualThreadSupport.scala | 91 +++++--- .../org/apache/pekko/io/dns/DnsSettings.scala | 39 +++- .../org/apache/pekko/util/LineNumbers.scala | 15 +- .../scala/org/apache/pekko/util/Reflect.scala | 198 ++++++++++++++---- .../cluster/sharding/FlightRecording.scala | 18 +- .../DurableStateExceptionsSpec.scala | 4 +- .../serialization/ProtobufSerializer.scala | 111 ++++++---- .../FailingInitializerMessage.java | 30 +++ .../ProtobufSerializerSpec.scala | 43 +++- .../metrics/FileDescriptorMetricSet.scala | 32 ++- .../metrics/FileDescriptorMetricSetSpec.scala | 42 ++++ 21 files changed, 744 insertions(+), 176 deletions(-) create mode 100644 actor-tests/src/test/java/org/apache/pekko/util/ReflectionExceptionFixtures.java create mode 100644 remote/src/test/java/org/apache/pekko/remote/serialization/FailingInitializerMessage.java create mode 100644 testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSetSpec.scala diff --git a/actor-tests/src/test/java/org/apache/pekko/util/ReflectionExceptionFixtures.java b/actor-tests/src/test/java/org/apache/pekko/util/ReflectionExceptionFixtures.java new file mode 100644 index 00000000000..96e5f1a17bd --- /dev/null +++ b/actor-tests/src/test/java/org/apache/pekko/util/ReflectionExceptionFixtures.java @@ -0,0 +1,46 @@ +/* + * 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.util; + +final class FailingInitializerConstructor { + private static final Object INITIALIZED = failInitialization(); + + FailingInitializerConstructor() {} + + private static Object failInitialization() { + throw new IllegalStateException("constructor-initializer-bug"); + } +} + +final class StaticMethodExceptionFixture { + static Object failInBody() { + throw new IllegalArgumentException("static-method-user-bug"); + } +} + +final class FailingInitializerStaticMethod { + private static final Object INITIALIZED = failInitialization(); + + static Object getInstance() { + return new Object(); + } + + private static Object failInitialization() { + throw new IllegalStateException("static-method-initializer-bug"); + } +} diff --git a/actor-tests/src/test/scala/org/apache/pekko/PekkoExceptionSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/PekkoExceptionSpec.scala index 1c178ceb94a..384f8ce1b4f 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/PekkoExceptionSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/PekkoExceptionSpec.scala @@ -13,6 +13,8 @@ package org.apache.pekko +import java.lang.invoke.{ MethodHandles, MethodType } + import org.apache.pekko.actor._ import org.scalatest.matchers.should.Matchers @@ -37,6 +39,6 @@ class PekkoExceptionSpec extends AnyWordSpec with Matchers { } def verify(clazz: java.lang.Class[?]): Unit = { - clazz.getConstructor(Array(classOf[String]): _*) + MethodHandles.publicLookup().findConstructor(clazz, MethodType.methodType(Void.TYPE, classOf[String])) } } diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/DynamicAccessSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/DynamicAccessSpec.scala index 4517af50c21..4e04b213d33 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/actor/DynamicAccessSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/actor/DynamicAccessSpec.scala @@ -30,6 +30,20 @@ class TestClassWithStringConstructor(val name: String) extends TestSuperclass class TestClassWithDefaultConstructor extends TestSuperclass { override def name = "default" } +class TestClassWithPrivateConstructor private () extends TestSuperclass { + override def name = "private" +} +object TestClassWithPrivateConstructor { + def direct(): TestClassWithPrivateConstructor = new TestClassWithPrivateConstructor() +} +class TestClassWithThrowingConstructor(@scala.annotation.nowarn("msg=never used") value: String) + extends TestSuperclass { + throw new IllegalArgumentException("user-bug") + override def name = "unreachable" +} +object TestDynamicAccessObject extends TestSuperclass { + override def name = "object" +} class DynamicAccessSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll { val system = ActorSystem() @@ -74,6 +88,29 @@ class DynamicAccessSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll dynamicAccess.classIsOnClasspath("org.apache.pekko.actor.Actor") should ===(true) } + "reuse constructor handles without changing private constructor access" in { + val clazz = classOf[TestClassWithPrivateConstructor] + dynamicAccess.createInstanceFor[TestClassWithPrivateConstructor](clazz, Nil).get.name should ===("private") + dynamicAccess.createInstanceFor[TestClassWithPrivateConstructor](clazz, Nil).get.name should ===("private") + } + + "preserve the target exception from a failing constructor" in { + val result = dynamicAccess.createInstanceFor[TestClassWithThrowingConstructor]( + classOf[TestClassWithThrowingConstructor], + immutable.Seq(classOf[String] -> "ignored")) + val exception = result.failed.get + exception shouldBe a[IllegalArgumentException] + exception.getMessage should ===("user-bug") + } + + "reuse the Scala object field handle" in { + val withoutSuffix = + dynamicAccess.getObjectFor[TestSuperclass]("org.apache.pekko.actor.TestDynamicAccessObject").get + val withSuffix = dynamicAccess.getObjectFor[TestSuperclass]("org.apache.pekko.actor.TestDynamicAccessObject$").get + (withoutSuffix should be).theSameInstanceAs(TestDynamicAccessObject) + (withSuffix should be).theSameInstanceAs(TestDynamicAccessObject) + } + def instantiateWithDefaultOrStringCtor(fqcn: String): Try[TestSuperclass] = // recoverWith doesn't work with scala 2.13.0-M5 // https://github.com/scala/bug/issues/11242 diff --git a/actor-tests/src/test/scala/org/apache/pekko/util/ByteIteratorSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/util/ByteIteratorSpec.scala index 4539117a238..2f3a4815061 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/util/ByteIteratorSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/util/ByteIteratorSpec.scala @@ -35,11 +35,13 @@ class ByteIteratorSpec extends AnyWordSpec with Matchers { freshIterator().indexOf(0x10, 3) should be(5) // There is also an indexOf with another signature, which is hard to invoke :D + val otherIndexOfHandle = + java.lang.invoke.MethodHandles.publicLookup().findVirtual( + classOf[ByteIterator], + "indexOf", + java.lang.invoke.MethodType.methodType(classOf[Int], classOf[Byte], classOf[Int])) def otherIndexOf(iterator: ByteIterator, byte: Byte, from: Int): Int = - classOf[ByteIterator] - .getMethod("indexOf", classOf[Byte], classOf[Int]) - .invoke(iterator, byte.asInstanceOf[Object], from.asInstanceOf[Object]) - .asInstanceOf[Int] + otherIndexOfHandle.invoke(iterator, byte, from).asInstanceOf[Int] otherIndexOf(freshIterator(), 0x20, 1) should be(1) otherIndexOf(freshIterator(), 0x10, 1) should be(2) diff --git a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInitializationSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInitializationSpec.scala index 071fbea89c4..d68aef41285 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInitializationSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringInitializationSpec.scala @@ -46,10 +46,11 @@ class ByteStringInitializationSpec extends AnyWordSpec with Matchers { } } - cleanCl - .loadClass("org.apache.pekko.util.ByteStringInitTest") - .getDeclaredConstructor() - .newInstance() + val clazz = cleanCl.loadClass("org.apache.pekko.util.ByteStringInitTest") + java.lang.invoke.MethodHandles + .privateLookupIn(clazz, java.lang.invoke.MethodHandles.lookup()) + .findConstructor(clazz, java.lang.invoke.MethodType.methodType(Void.TYPE)) + .invokeWithArguments() .asInstanceOf[Runnable] .run() } diff --git a/actor-tests/src/test/scala/org/apache/pekko/util/ReflectSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/util/ReflectSpec.scala index d786578cef3..6dbc152031e 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/util/ReflectSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/util/ReflectSpec.scala @@ -13,6 +13,10 @@ package org.apache.pekko.util +import java.lang.invoke.MethodHandles +import java.lang.reflect.InvocationTargetException +import java.util.concurrent.TimeoutException + import scala.annotation.nowarn import scala.collection.immutable @@ -30,6 +34,15 @@ object ReflectSpec { def this(a: A) = this(a, null) def this(b: B) = this(null, b) } + + final class StringOnly(val value: String) + final class PrivateConstructor private (val value: String) + object PrivateConstructor { + def direct(value: String): PrivateConstructor = new PrivateConstructor(value) + } + final class ThrowingConstructor(@nowarn("msg=never used") value: String) { + throw new IllegalArgumentException("user-bug") + } } class ReflectSpec extends AnyWordSpec with Matchers { @@ -50,7 +63,7 @@ class ReflectSpec extends AnyWordSpec with Matchers { } "deal with `null` in 1 matching case" in { val constructor = Reflect.findConstructor(classOf[One], immutable.Seq(null)) - constructor.newInstance(null) + constructor.invoke(null.asInstanceOf[AnyRef]) } "deal with multiple constructors" in { Reflect.findConstructor(classOf[MultipleOne], immutable.Seq(new A)) @@ -62,6 +75,60 @@ class ReflectSpec extends AnyWordSpec with Matchers { Reflect.findConstructor(classOf[MultipleOne], immutable.Seq(null)) } } + "use public lookup for public JDK constructors in non-open modules" in { + val constructor = Reflect.findConstructor(classOf[TimeoutException], immutable.Seq("err")) + val instance = Reflect.instantiate[TimeoutException](constructor, immutable.Seq("err")) + instance.getMessage should ===("err") + } + "not confuse null with an Object argument across lookups" in { + Reflect.instantiate(classOf[StringOnly], immutable.Seq(null)).value shouldBe null + intercept[IllegalArgumentException] { + Reflect.findConstructor(classOf[StringOnly], immutable.Seq(new Object)) + }.getMessage should include("no matching constructor") + } + "access private constructors when the package is open" in { + Reflect.instantiate(classOf[PrivateConstructor], immutable.Seq("private")).value should ===("private") + } + "preserve InvocationTargetException for constructor failures" in { + val exception = intercept[InvocationTargetException] { + Reflect.instantiate(classOf[ThrowingConstructor], immutable.Seq("ignored")) + } + exception.getCause shouldBe a[IllegalArgumentException] + exception.getCause.getMessage should ===("user-bug") + } + "not wrap constructor class-initialization failures" in { + val exception = intercept[ExceptionInInitializerError] { + Reflect.instantiate(classOf[FailingInitializerConstructor]) + } + exception.getCause.getMessage should ===("constructor-initializer-bug") + } + "invoke varargs constructors as fixed arity" in { + val processBuilder = + Reflect.instantiate(classOf[ProcessBuilder], immutable.Seq(Array("echo", "fixed-arity"))) + processBuilder.command().toArray should ===(Array[AnyRef]("echo", "fixed-arity")) + } + } + + "Reflect#invokeStaticNoArg" must { + "preserve InvocationTargetException for target failures" in { + val clazz = classOf[StaticMethodExceptionFixture] + val callerLookup = MethodHandles.lookup() + val method = Reflect.findStaticNoArgMethod(clazz, "failInBody", callerLookup) + val exception = intercept[InvocationTargetException] { + Reflect.invokeStaticNoArg[AnyRef](clazz, method, callerLookup) + } + exception.getCause.getMessage should ===("static-method-user-bug") + } + + "not wrap static method class-initialization failures" in { + val clazz = classOf[FailingInitializerStaticMethod] + val callerLookup = MethodHandles.lookup() + val method = Reflect.findStaticNoArgMethod(clazz, "getInstance", callerLookup) + val exception = intercept[ExceptionInInitializerError] { + Reflect.invokeStaticNoArg[AnyRef](clazz, method, callerLookup) + } + exception.getCause.getMessage should ===("static-method-initializer-bug") + } } "Reflect#getCallerClass" must { diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ExtensionsImpl.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ExtensionsImpl.scala index 53395451256..bfb75f180c8 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ExtensionsImpl.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ExtensionsImpl.scala @@ -13,6 +13,7 @@ package org.apache.pekko.actor.typed.internal +import java.lang.invoke.MethodHandles import java.util.concurrent.{ ConcurrentHashMap, CountDownLatch } import scala.annotation.tailrec @@ -23,6 +24,7 @@ import org.apache.pekko import pekko.actor.typed.{ ActorSystem, Extension, ExtensionId, Extensions } import pekko.actor.typed.ExtensionSetup import pekko.annotation.InternalApi +import pekko.util.Reflect /** * INTERNAL API @@ -33,6 +35,7 @@ import pekko.annotation.InternalApi private[pekko] trait ExtensionsImpl extends Extensions { self: ActorSystem[?] with InternalRecipientRef[?] => private val extensions = new ConcurrentHashMap[ExtensionId[?], AnyRef] + private val lookup = MethodHandles.lookup() /** * Hook for ActorSystem to load extensions on startup @@ -70,8 +73,8 @@ private[pekko] trait ExtensionsImpl extends Extensions { self: ActorSystem[?] wi dynamicAccess.getClassFor[ExtensionId[Extension]](extensionIdFQCN).flatMap[ExtensionId[Extension]] { (clazz: Class[?]) => Try { - val singletonAccessor = clazz.getDeclaredMethod("getInstance") - singletonAccessor.invoke(null).asInstanceOf[ExtensionId[Extension]] + val handle = Reflect.findStaticNoArgMethod(clazz, "getInstance", lookup) + Reflect.invokeStaticNoArg[ExtensionId[Extension]](clazz, handle, lookup) } } diff --git a/actor/src/main/java/org/apache/pekko/io/ByteBufferCleaner.java b/actor/src/main/java/org/apache/pekko/io/ByteBufferCleaner.java index aab02e45ab7..47b722166c2 100644 --- a/actor/src/main/java/org/apache/pekko/io/ByteBufferCleaner.java +++ b/actor/src/main/java/org/apache/pekko/io/ByteBufferCleaner.java @@ -21,7 +21,7 @@ import java.lang.invoke.MethodHandle; import java.lang.invoke.MethodHandles; -import java.lang.reflect.Field; +import java.lang.invoke.VarHandle; import java.nio.ByteBuffer; /** @@ -50,11 +50,12 @@ private static final class Java9Cleaner implements Cleaner { private final MethodHandle invokeCleaner; private Java9Cleaner() throws ReflectiveOperationException { - final Class unsafeClass = Class.forName("sun.misc.Unsafe"); - final Field field = unsafeClass.getDeclaredField("theUnsafe"); - field.setAccessible(true); - final Object theUnsafe = field.get(null); MethodHandles.Lookup lookup = MethodHandles.lookup(); + final Class unsafeClass = lookup.findClass("sun.misc.Unsafe"); + final VarHandle unsafeHandle = + MethodHandles.privateLookupIn(unsafeClass, lookup) + .findStaticVarHandle(unsafeClass, "theUnsafe", unsafeClass); + final Object theUnsafe = unsafeHandle.get(); MethodHandle invokeCleaner = lookup.findVirtual( unsafeClass, "invokeCleaner", methodType(void.class, ByteBuffer.class)); diff --git a/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala b/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala index 39dec1dae9d..0c011c293ba 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala @@ -105,16 +105,25 @@ private[pekko] class TypedCreatorFunctionConsumer(clz: Class[? <: Actor], creato */ private[pekko] class ArgsReflectConstructor(clz: Class[? <: Actor], args: immutable.Seq[Any]) extends IndirectActorProducer { - private val constructor = Reflect.findConstructor(clz, args) + private val constructor = Reflect.constructorInvoker(Reflect.findConstructor(clz, args)) + private val arguments = Reflect.argumentArray(args) + private lazy val initialized: Unit = Reflect.ensureInitialized(clz) override def actorClass = clz - override def produce() = Reflect.instantiate(constructor, args) + override def produce() = { + initialized + Reflect.instantiate[Actor](constructor, arguments) + } } /** * INTERNAL API */ private[pekko] class NoArgsReflectConstructor(clz: Class[? <: Actor]) extends IndirectActorProducer { - Reflect.findConstructor(clz, List.empty) + private val constructor = Reflect.noArgsConstructorInvoker(Reflect.findConstructor(clz, Nil)) + private lazy val initialized: Unit = Reflect.ensureInitialized(clz) override def actorClass = clz - override def produce() = Reflect.instantiate(clz) + override def produce() = { + initialized + Reflect.instantiate[Actor](constructor) + } } diff --git a/actor/src/main/scala/org/apache/pekko/actor/ReflectiveDynamicAccess.scala b/actor/src/main/scala/org/apache/pekko/actor/ReflectiveDynamicAccess.scala index a35feafce5c..14d1aeb2116 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/ReflectiveDynamicAccess.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/ReflectiveDynamicAccess.scala @@ -13,7 +13,8 @@ package org.apache.pekko.actor -import java.lang.reflect.InvocationTargetException +import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType, VarHandle } +import java.util.concurrent.ConcurrentHashMap import scala.collection.immutable import scala.reflect.ClassTag @@ -26,13 +27,18 @@ import pekko.annotation.DoNotInherit /** * This is the default [[pekko.actor.DynamicAccess]] implementation used by [[pekko.actor.ExtendedActorSystem]] * unless overridden. It uses reflection to turn fully-qualified class names into `Class[_]` objects - * and creates instances from there using `getDeclaredConstructor()` and invoking that. The class loader + * and creates instances from there using constructor method handles. The class loader * to be used for all this is determined by the actor system’s class loader by default. * * Not for user extension or construction */ @DoNotInherit class ReflectiveDynamicAccess(val classLoader: ClassLoader) extends DynamicAccess { + private val lookup = MethodHandles.lookup() + private val publicLookup = MethodHandles.publicLookup() + private val genericConstructorType = MethodType.methodType(classOf[AnyRef], classOf[Array[AnyRef]]) + private val constructorHandles = new ConcurrentHashMap[Class[?], ConcurrentHashMap[MethodType, MethodHandle]] + private val moduleHandles = new ConcurrentHashMap[Class[?], VarHandle] override def getClassFor[T: ClassTag](fqcn: String): Try[Class[? <: T]] = Try[Class[? <: T]] { @@ -43,15 +49,23 @@ class ReflectiveDynamicAccess(val classLoader: ClassLoader) extends DynamicAcces override def createInstanceFor[T: ClassTag](clazz: Class[?], args: immutable.Seq[(Class[?], AnyRef)]): Try[T] = Try { - val types = args.map(_._1).toArray - val values = args.map(_._2).toArray - val constructor = clazz.getDeclaredConstructor(types: _*) - constructor.setAccessible(true) - val obj = constructor.newInstance(values: _*) + val types = new Array[Class[?]](args.size) + val values = new Array[AnyRef](args.size) + val iterator = args.iterator + var index = 0 + while (iterator.hasNext) { + val argument: (Class[?], AnyRef) = iterator.next() + types(index) = argument._1 + values(index) = argument._2 + index += 1 + } + val methodType = MethodType.methodType(Void.TYPE, types) + val handle = constructorHandle(clazz, methodType) + val obj: AnyRef = handle.invokeExact(values) val t = implicitly[ClassTag[T]].runtimeClass if (t.isInstance(obj)) obj.asInstanceOf[T] else throw new ClassCastException(clazz.getName + " is not a subtype of " + t) - }.recover { case i: InvocationTargetException if i.getTargetException ne null => throw i.getTargetException } + } override def createInstanceFor[T: ClassTag](fqcn: String, args: immutable.Seq[(Class[?], AnyRef)]): Try[T] = getClassFor(fqcn).flatMap { c => @@ -72,17 +86,66 @@ class ReflectiveDynamicAccess(val classLoader: ClassLoader) extends DynamicAcces else getClassFor(fqcn + "$").recoverWith { case _ => getClassFor(fqcn) } classTry.flatMap { c => Try { - val module = c.getDeclaredField("MODULE$") - module.setAccessible(true) + val moduleHandle = moduleHandleFor(c) val t = implicitly[ClassTag[T]].runtimeClass - module.get(null) match { + moduleHandle.get() match { case null => throw new NullPointerException case x if !t.isInstance(x) => throw new ClassCastException(fqcn + " is not a subtype of " + t) case x: T => x case unexpected => throw new IllegalArgumentException(s"Unexpected module field: $unexpected") // will not happen, for exhaustiveness check } - }.recover { case i: InvocationTargetException if i.getTargetException ne null => throw i.getTargetException } + } + } + } + + private def constructorHandle(clazz: Class[?], methodType: MethodType): MethodHandle = { + val existingByType = constructorHandles.get(clazz) + val byType = + if (existingByType ne null) existingByType + else { + val fresh = new ConcurrentHashMap[MethodType, MethodHandle] + val existing = constructorHandles.putIfAbsent(clazz, fresh) + if (existing eq null) fresh else existing + } + val cached = byType.get(methodType) + if (cached ne null) cached + else { + val rawHandle = + try lookup.findConstructor(clazz, methodType) + catch { + case _: IllegalAccessException => + try publicLookup.findConstructor(clazz, methodType) + catch { + case _: IllegalAccessException => + MethodHandles.privateLookupIn(clazz, lookup).findConstructor(clazz, methodType) + } + } + val handle = rawHandle + .asFixedArity() + .asSpreader(classOf[Array[AnyRef]], methodType.parameterCount()) + .asType(genericConstructorType) + val existing = byType.putIfAbsent(methodType, handle) + if (existing eq null) handle else existing + } + } + + private def moduleHandleFor(clazz: Class[?]): VarHandle = { + val cached = moduleHandles.get(clazz) + if (cached ne null) cached + else { + val handle = + try lookup.findStaticVarHandle(clazz, "MODULE$", clazz) + catch { + case _: IllegalAccessException => + try publicLookup.findStaticVarHandle(clazz, "MODULE$", clazz) + catch { + case _: IllegalAccessException => + MethodHandles.privateLookupIn(clazz, lookup).findStaticVarHandle(clazz, "MODULE$", clazz) + } + } + val existing = moduleHandles.putIfAbsent(clazz, handle) + if (existing eq null) handle else existing } } } diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala index 4d2595bce6d..f0953134629 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala @@ -17,8 +17,8 @@ package org.apache.pekko.dispatch -import java.lang.invoke.{ MethodHandles, MethodType } -import java.util.concurrent.{ ExecutorService, ForkJoinPool, ForkJoinWorkerThread, ThreadFactory } +import java.lang.invoke.{ MethodHandles, MethodType, VarHandle } +import java.util.concurrent.{ Executor, ExecutorService, ForkJoinPool, ForkJoinWorkerThread, ThreadFactory } import scala.util.control.NonFatal @@ -30,6 +30,50 @@ import pekko.util.JavaVersion private[dispatch] object VirtualThreadSupport { private val zero = java.lang.Long.valueOf(0L) private val lookup = MethodHandles.publicLookup() + private val privateLookup = MethodHandles.lookup() + private val newThreadPerTaskExecutorMethodType = + MethodType.methodType(classOf[ExecutorService], classOf[ThreadFactory]) + private val threadFactoryMethodType = MethodType.methodType(classOf[ThreadFactory]) + private val carrierThreadConstructorMethodType = MethodType.methodType(Void.TYPE, classOf[ForkJoinPool]) + + // Cached method/var handles to avoid repeated lookups on every invocation + private lazy val newThreadPerTaskExecutorHandle = { + val executorsClazz = ClassLoader.getSystemClassLoader.loadClass("java.util.concurrent.Executors") + lookup.findStatic(executorsClazz, "newThreadPerTaskExecutor", newThreadPerTaskExecutorMethodType) + } + + private lazy val threadBuilderClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder") + private lazy val threadBuilderOfVirtualClass = + ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder$OfVirtual") + + private lazy val ofVirtualHandle = + lookup.findStatic(classOf[Thread], "ofVirtual", MethodType.methodType(threadBuilderOfVirtualClass)) + + private lazy val ofVirtualNameHandle = + lookup.findVirtual(threadBuilderOfVirtualClass, "name", + MethodType.methodType(threadBuilderOfVirtualClass, classOf[String])) + + private lazy val ofVirtualNameWithStartHandle = + lookup.findVirtual(threadBuilderOfVirtualClass, "name", + MethodType.methodType(threadBuilderOfVirtualClass, classOf[String], java.lang.Long.TYPE)) + + private lazy val builderFactoryHandle = + lookup.findVirtual(threadBuilderClass, "factory", threadFactoryMethodType) + + private lazy val virtualThreadClass = + ClassLoader.getSystemClassLoader.loadClass("java.lang.VirtualThread") + + private lazy val defaultSchedulerHandle = + MethodHandles + .privateLookupIn(virtualThreadClass, privateLookup) + .findStaticVarHandle(virtualThreadClass, "DEFAULT_SCHEDULER", classOf[ForkJoinPool]) + + private val schedulerHandles = new ClassValue[VarHandle] { + override def computeValue(clazz: Class[?]): VarHandle = + MethodHandles + .privateLookupIn(clazz, privateLookup) + .findVarHandle(clazz, "scheduler", classOf[Executor]) + } /** * Is virtual thread supported @@ -42,12 +86,7 @@ private[dispatch] object VirtualThreadSupport { def newThreadPerTaskExecutor(threadFactory: ThreadFactory): ExecutorService = { require(threadFactory != null, "threadFactory should not be null.") try { - val executorsClazz = ClassLoader.getSystemClassLoader.loadClass("java.util.concurrent.Executors") - val newThreadPerTaskExecutorMethod = lookup.findStatic( - executorsClazz, - "newThreadPerTaskExecutor", - MethodType.methodType(classOf[ExecutorService], classOf[ThreadFactory])) - newThreadPerTaskExecutorMethod.invoke(threadFactory).asInstanceOf[ExecutorService] + newThreadPerTaskExecutorHandle.invoke(threadFactory).asInstanceOf[ExecutorService] } catch { case NonFatal(e) => // --add-opens java.base/java.lang=ALL-UNNAMED @@ -84,30 +123,18 @@ private[dispatch] object VirtualThreadSupport { require(isSupported, "Virtual thread is not supported.") require(prefix != null && prefix.nonEmpty, "prefix should not be null or empty.") try { - val builderClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder") - val ofVirtualClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder$OfVirtual") - val ofVirtualMethod = lookup.findStatic(classOf[Thread], "ofVirtual", MethodType.methodType(ofVirtualClass)) - var builder = ofVirtualMethod.invoke() + var builder = ofVirtualHandle.invoke() // set the name if (start <= -1) { - val nameMethod = lookup.findVirtual(ofVirtualClass, "name", - MethodType.methodType(ofVirtualClass, classOf[String])) - builder = nameMethod.invoke(builder, prefix + "-virtual-thread") + builder = ofVirtualNameHandle.invoke(builder, prefix + "-virtual-thread") } else { - val nameMethod = lookup.findVirtual(ofVirtualClass, "name", - MethodType.methodType(ofVirtualClass, classOf[String], classOf[Long])) - builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", zero) + builder = ofVirtualNameWithStartHandle.invoke(builder, prefix + "-virtual-thread-", zero) } // set the scheduler if (executor ne null) { - // Use reflection here, method handle is stricter on access control - val clazz = builder.getClass - val field = clazz.getDeclaredField("scheduler") - field.setAccessible(true) - field.set(builder, executor) + schedulerHandles.get(builder.getClass).set(builder, executor) } - val factoryMethod = lookup.findVirtual(builderClass, "factory", MethodType.methodType(classOf[ThreadFactory])) - factoryMethod.invoke(builder).asInstanceOf[ThreadFactory] + builderFactoryHandle.invoke(builder).asInstanceOf[ThreadFactory] } catch { case NonFatal(e) => // --add-opens java.base/java.lang=ALL-UNNAMED @@ -119,10 +146,12 @@ private[dispatch] object VirtualThreadSupport { // --add-opens java.base/java.lang=ALL-UNNAMED // --add-opens java.base/jdk.internal.misc=ALL-UNNAMED private val clazz = ClassLoader.getSystemClassLoader.loadClass("jdk.internal.misc.CarrierThread") - // TODO lookup.findClass is only available in Java 9 - private val constructor = clazz.getDeclaredConstructor(classOf[ForkJoinPool]) + private val constructorHandle = + MethodHandles + .privateLookupIn(clazz, privateLookup) + .findConstructor(clazz, carrierThreadConstructorMethodType) override def newThread(pool: ForkJoinPool): ForkJoinWorkerThread = { - constructor.newInstance(pool).asInstanceOf[ForkJoinWorkerThread] + constructorHandle.invoke(pool).asInstanceOf[ForkJoinWorkerThread] } } @@ -132,11 +161,7 @@ private[dispatch] object VirtualThreadSupport { def getVirtualThreadDefaultScheduler: ForkJoinPool = try { require(isSupported, "Virtual thread is not supported.") - val clazz = ClassLoader.getSystemClassLoader.loadClass("java.lang.VirtualThread") - val fieldName = "DEFAULT_SCHEDULER" - val field = clazz.getDeclaredField(fieldName) - field.setAccessible(true) - field.get(null).asInstanceOf[ForkJoinPool] + defaultSchedulerHandle.get().asInstanceOf[ForkJoinPool] } catch { case NonFatal(e) => // --add-opens java.base/java.lang=ALL-UNNAMED diff --git a/actor/src/main/scala/org/apache/pekko/io/dns/DnsSettings.scala b/actor/src/main/scala/org/apache/pekko/io/dns/DnsSettings.scala index 273d2f15963..fde89e6c816 100644 --- a/actor/src/main/scala/org/apache/pekko/io/dns/DnsSettings.scala +++ b/actor/src/main/scala/org/apache/pekko/io/dns/DnsSettings.scala @@ -30,6 +30,7 @@ package org.apache.pekko.io.dns import java.io.File +import java.lang.invoke.{ MethodHandles, MethodType } import java.net.{ InetSocketAddress, URI } import java.util @@ -160,7 +161,8 @@ private[dns] final class DnsSettings(system: ExtendedActorSystem, c: Config) { def failUnableToDetermineDefaultNameservers = throw new IllegalStateException( - "Unable to obtain default nameservers from JNDI or via reflection. " + + "Unable to obtain default nameservers from JNDI or via the JDK internal ResolverConfiguration fallback. " + + "On modular JDKs, that fallback requires access to java.base/sun.net.dns. " + "Please set `pekko.io.dns.async-dns.nameservers` explicitly in order to be able to resolve domain names. ") } @@ -169,6 +171,9 @@ object DnsSettings { private final val DnsFallbackPort = 53 private val inetSocketAddress = """(.*?)(?::(\d+))?""".r + private val publicLookup = MethodHandles.publicLookup() + private val lookup = MethodHandles.lookup() + private val nameserversMethodType = MethodType.methodType(classOf[util.List[?]]) /** * INTERNAL API @@ -186,7 +191,8 @@ object DnsSettings { * Find out the default search lists that Java would use normally, e.g. when using InetAddress to resolve domains. * * The default nameservers are attempted to be obtained from: jndi-dns and from `sun.net.dnsResolverConfiguration` - * as a fallback (which is expected to fail though when running on JDK9+ due to the module encapsulation of sun packages). + * as a fallback. That JDK-internal fallback is expected to fail when running on JDK9+ unless java.base/sun.net.dns + * is exported or opened to Pekko. * * Based on: https://github.com/netty/netty/blob/4.1/resolver-dns/src/main/java/io/netty/resolver/dns/DefaultDnsServerAddressStreamProvider.java#L58-L146 */ @@ -229,24 +235,37 @@ object DnsSettings { } } - // this method is used as a fallback in case JNDI results in an empty list - // this method will not work when running modularised of course since it needs access to internal sun classes - def getNameserversUsingReflection: Try[List[InetSocketAddress]] = { + // This method is used as a best-effort fallback in case JNDI results in an empty list. It needs access to + // JDK-internal sun classes, so it will not work on modular JDKs unless java.base/sun.net.dns is exported or opened. + def getNameserversUsingMethodHandles: Try[List[InetSocketAddress]] = { system.dynamicAccess.getClassFor[Any]("sun.net.dns.ResolverConfiguration").flatMap { c => Try { - val open = c.getMethod("open") - val nameservers = c.getMethod("nameservers") - val instance = open.invoke(null) + val effectiveLookup: MethodHandles.Lookup = + try { + lookup.findStatic(c, "open", MethodType.methodType(c)) + lookup + } catch { + case _: IllegalAccessException => + try { + publicLookup.findStatic(c, "open", MethodType.methodType(c)) + publicLookup + } catch { + case _: IllegalAccessException => MethodHandles.privateLookupIn(c, lookup) + } + } + val open = effectiveLookup.findStatic(c, "open", MethodType.methodType(c)) + val nameservers = effectiveLookup.findVirtual(c, "nameservers", nameserversMethodType) + val instance = open.invoke() val ns = nameservers.invoke(instance).asInstanceOf[util.List[String]] val res = if (ns.isEmpty) throw new IllegalStateException( - "Empty nameservers list discovered using reflection. Consider configuring default nameservers manually!") + "Empty nameservers list discovered using method handles. Consider configuring default nameservers manually!") else ns.asScala.toList res.flatMap(s => asInetSocketAddress(s).toOption) } } } - getNameserversUsingJNDI.orElse(getNameserversUsingReflection) + getNameserversUsingJNDI.orElse(getNameserversUsingMethodHandles) } } diff --git a/actor/src/main/scala/org/apache/pekko/util/LineNumbers.scala b/actor/src/main/scala/org/apache/pekko/util/LineNumbers.scala index 424941490c1..b70c7ca0f12 100644 --- a/actor/src/main/scala/org/apache/pekko/util/LineNumbers.scala +++ b/actor/src/main/scala/org/apache/pekko/util/LineNumbers.scala @@ -14,7 +14,7 @@ package org.apache.pekko.util import java.io.{ DataInputStream, InputStream } -import java.lang.invoke.SerializedLambda +import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType, SerializedLambda } import scala.annotation.switch import scala.util.control.NonFatal @@ -30,6 +30,15 @@ import scala.util.control.NonFatal */ object LineNumbers { + private val lookup = MethodHandles.lookup() + private val writeReplaceMethodType = MethodType.methodType(classOf[Object]) + private val writeReplaceHandles = new ClassValue[MethodHandle] { + override def computeValue(clazz: Class[?]): MethodHandle = + MethodHandles + .privateLookupIn(clazz, lookup) + .findVirtual(clazz, "writeReplace", writeReplaceMethodType) + } + sealed trait Result case object NoSourceInfo extends Result final case class UnknownSourceFormat(explanation: String) extends Result @@ -207,9 +216,7 @@ object LineNumbers { private def getStreamForLambda(l: AnyRef): Option[(InputStream, Some[String])] = try { val c = l.getClass - val writeReplace = c.getDeclaredMethod("writeReplace") - writeReplace.setAccessible(true) - writeReplace.invoke(l) match { + writeReplaceHandles.get(c).invoke(l) match { case serialized: SerializedLambda => if (debug) println(s"LNB: found Lambda implemented in ${serialized.getImplClass}:${serialized.getImplMethodName}") diff --git a/actor/src/main/scala/org/apache/pekko/util/Reflect.scala b/actor/src/main/scala/org/apache/pekko/util/Reflect.scala index 45203e718bc..ea22aa9ad17 100644 --- a/actor/src/main/scala/org/apache/pekko/util/Reflect.scala +++ b/actor/src/main/scala/org/apache/pekko/util/Reflect.scala @@ -13,13 +13,11 @@ package org.apache.pekko.util import java.lang.StackWalker -import java.lang.reflect.Constructor -import java.lang.reflect.ParameterizedType -import java.lang.reflect.Type +import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType } +import java.lang.reflect.{ Constructor, InvocationTargetException, Modifier, ParameterizedType, Type } import scala.annotation.tailrec import scala.collection.immutable -import scala.util.Try import scala.util.control.NonFatal import org.apache.pekko.annotation.InternalApi @@ -32,6 +30,14 @@ import org.apache.pekko.annotation.InternalApi */ @InternalApi private[pekko] object Reflect { + private val lookup = MethodHandles.lookup() + private val publicLookup = MethodHandles.publicLookup() + private val noArgsInvokerType = MethodType.methodType(classOf[AnyRef]) + private val genericConstructorType = MethodType.methodType(classOf[AnyRef], classOf[Array[AnyRef]]) + private val invocationTargetExceptionConstructor = + lookup.findConstructor( + classOf[InvocationTargetException], + MethodType.methodType(Void.TYPE, classOf[Throwable])) /** * This optionally holds a function which looks N levels above itself @@ -59,13 +65,9 @@ private[pekko] object Reflect { * @return a new instance from the default constructor of the given class */ private[pekko] def instantiate[T](clazz: Class[T]): T = { - val ctor = clazz.getDeclaredConstructor() - try ctor.newInstance() - catch { - case _: IllegalAccessException => - ctor.setAccessible(true) - ctor.newInstance() - } + val constructor = findConstructor(clazz, Nil) + ensureInitialized(clazz) + instantiate(noArgsConstructorInvoker(constructor)) } /** @@ -73,57 +75,96 @@ private[pekko] object Reflect { * Calls findConstructor and invokes it with the given arguments. */ private[pekko] def instantiate[T](clazz: Class[T], args: immutable.Seq[Any]): T = { - instantiate(findConstructor(clazz, args), args) + val constructor = findConstructor(clazz, args) + ensureInitialized(clazz) + instantiate(constructorInvoker(constructor), argumentArray(args)) } /** * INTERNAL API * Invokes the constructor with the given arguments. */ - private[pekko] def instantiate[T](constructor: Constructor[T], args: immutable.Seq[Any]): T = { - constructor.setAccessible(true) - try constructor.newInstance(args.asInstanceOf[Seq[AnyRef]]: _*) - catch { - case e: IllegalArgumentException => - val argString = args.map(safeGetClass).mkString("[", ", ", "]") - throw new IllegalArgumentException(s"constructor $constructor is incompatible with arguments $argString", e) - } + private[pekko] def instantiate[T](constructor: MethodHandle, args: immutable.Seq[Any]): T = { + ensureInitialized(constructor.`type`().returnType()) + instantiate(constructorInvoker(constructor), argumentArray(args)) } + /** INTERNAL API */ + private[pekko] def instantiate[T](constructorInvoker: MethodHandle, args: Array[AnyRef]): T = { + val instance: AnyRef = constructorInvoker.invokeExact(args) + instance.asInstanceOf[T] + } + + /** INTERNAL API */ + private[pekko] def instantiate[T](constructorInvoker: MethodHandle): T = { + val instance: AnyRef = constructorInvoker.invokeExact() + instance.asInstanceOf[T] + } + + /** INTERNAL API */ + private[pekko] def noArgsConstructorInvoker(constructor: MethodHandle): MethodHandle = + wrapTargetException(constructor).asFixedArity().asType(noArgsInvokerType) + + /** INTERNAL API */ + private[pekko] def invokeStaticNoArg[T]( + clazz: Class[?], + method: MethodHandle, + callerLookup: MethodHandles.Lookup): T = { + ensureInitialized(clazz, callerLookup) + val invoker = wrapTargetException(method).asFixedArity().asType(noArgsInvokerType) + val result: AnyRef = invoker.invokeExact() + result.asInstanceOf[T] + } + + /** INTERNAL API */ + private[pekko] def constructorInvoker(constructor: MethodHandle): MethodHandle = + wrapTargetException(constructor) + .asFixedArity() + .asSpreader(classOf[Array[AnyRef]], constructor.`type`().parameterCount()) + .asType(genericConstructorType) + + /** INTERNAL API */ + private[pekko] def argumentArray(args: immutable.Seq[Any]): Array[AnyRef] = + args.iterator.map(_.asInstanceOf[AnyRef]).toArray + /** * INTERNAL API * Implements a primitive form of overload resolution a.k.a. finding the * right constructor. */ - private[pekko] def findConstructor[T](clazz: Class[T], args: immutable.Seq[Any]): Constructor[T] = { + private[pekko] def findConstructor[T](clazz: Class[T], args: immutable.Seq[Any]): MethodHandle = { def error(msg: String): Nothing = { val argClasses = args.map(safeGetClass).mkString(", ") throw new IllegalArgumentException(s"$msg found on $clazz for arguments [$argClasses]") } - val constructor: Constructor[T] = - if (args.isEmpty) Try { clazz.getDeclaredConstructor() }.getOrElse(null) - else { - val length = args.length - val candidates = - clazz.getDeclaredConstructors.asInstanceOf[Array[Constructor[T]]].iterator.filter { c => - val parameterTypes = c.getParameterTypes - parameterTypes.length == length && - (parameterTypes.iterator.zip(args.iterator).forall { - case (found, required) => - found.isInstance(required) || BoxedType(found).isInstance(required) || - (required == null && !found.isPrimitive) - }) - } - if (candidates.hasNext) { - val cstrtr = candidates.next() - if (candidates.hasNext) error("multiple matching constructors") - else cstrtr - } else null + val selectedConstructorType = + findConstructorMethodTypeFromClassMetadata(clazz, args).getOrElse(error("no matching constructor")) + findConstructorHandle(clazz, selectedConstructorType) + } + + private def findConstructorMethodTypeFromClassMetadata[T](clazz: Class[T], args: immutable.Seq[Any]) + : Option[MethodType] = { + def matches(parameterTypes: Array[Class[?]]): Boolean = + parameterTypes.length == args.length && + parameterTypes.iterator.zip(args.iterator).forall { + case (found, required) => + found.isInstance(required) || BoxedType(found).isInstance(required) || + (required == null && !found.isPrimitive) } - if (constructor eq null) error("no matching constructor") - else constructor + val candidates = clazz.getDeclaredConstructors.iterator + .map(constructorMethodType) + .filter(methodType => matches(methodType.parameterArray())) + .toList + candidates match { + case single :: Nil => Some(single) + case Nil => None + case _ => + val argClasses = args.map(safeGetClass).mkString(", ") + throw new IllegalArgumentException( + s"multiple matching constructors found on $clazz for arguments [$argClasses]") + } } private def safeGetClass(a: Any): Class[?] = @@ -134,7 +175,15 @@ private[pekko] object Reflect { * @param clazz the class which to instantiate an instance of * @return a function which when applied will create a new instance from the default constructor of the given class */ - private[pekko] def instantiator[T](clazz: Class[T]): () => T = () => instantiate(clazz) + private[pekko] def instantiator[T](clazz: Class[T]): () => T = { + lazy val constructor = noArgsConstructorInvoker(findConstructor(clazz, Nil)) + lazy val initialized: Unit = ensureInitialized(clazz) + () => { + val invoker = constructor + initialized + instantiate(invoker) + } + } def findMarker(root: Class[?], marker: Class[?]): Type = { @tailrec def rec(curr: Class[?]): Type = { @@ -153,6 +202,69 @@ private[pekko] object Reflect { rec(root) } + private def constructorMethodType(constructor: Constructor[?]): MethodType = + MethodType.methodType(Void.TYPE, constructor.getParameterTypes) + + private def findConstructorHandle[T](clazz: Class[T], methodType: MethodType): MethodHandle = { + try lookup.findConstructor(clazz, methodType) + catch { + case _: IllegalAccessException => + try publicLookup.findConstructor(clazz, methodType) + catch { + case _: IllegalAccessException => + MethodHandles.privateLookupIn(clazz, lookup).findConstructor(clazz, methodType) + } + } + } + + private def wrapTargetException(target: MethodHandle): MethodHandle = { + val exceptionThrower = + MethodHandles.throwException(target.`type`().returnType(), classOf[InvocationTargetException]) + val handler = MethodHandles.filterArguments(exceptionThrower, 0, invocationTargetExceptionConstructor) + val handlerWithArguments = MethodHandles.dropArguments(handler, 1, target.`type`().parameterList()) + MethodHandles.catchException(target, classOf[Throwable], handlerWithArguments) + } + + private[pekko] def ensureInitialized(clazz: Class[?]): Unit = + ensureInitialized(clazz, lookup) + + private def ensureInitialized(clazz: Class[?], callerLookup: MethodHandles.Lookup): Unit = { + try callerLookup.ensureInitialized(clazz) + catch { + case _: IllegalAccessException => + try publicLookup.ensureInitialized(clazz) + catch { + case _: IllegalAccessException => + MethodHandles.privateLookupIn(clazz, callerLookup).ensureInitialized(clazz) + } + } + } + + private[pekko] def findStaticNoArgMethod( + clazz: Class[?], + methodName: String, + callerLookup: MethodHandles.Lookup): MethodHandle = + clazz.getDeclaredMethods.iterator + .filter(method => + method.getName == methodName && Modifier.isStatic(method.getModifiers) && method.getParameterCount == 0) + .toList match { + case method :: Nil => + val methodType = MethodType.methodType(method.getReturnType) + try callerLookup.findStatic(clazz, methodName, methodType) + catch { + case _: IllegalAccessException => + try publicLookup.findStatic(clazz, methodName, methodType) + catch { + case _: IllegalAccessException => + MethodHandles.privateLookupIn(clazz, callerLookup).findStatic(clazz, methodName, methodType) + } + } + case Nil => + throw new NoSuchMethodException(s"${clazz.getName}.$methodName()") + case _ => + throw new IllegalArgumentException(s"multiple static no-arg methods named [$methodName] found on $clazz") + } + /** * INTERNAL API */ diff --git a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/FlightRecording.scala b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/FlightRecording.scala index d1628667c03..a46ee637a9a 100644 --- a/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/FlightRecording.scala +++ b/cluster-sharding-typed/src/test/scala/org/apache/pekko/cluster/sharding/FlightRecording.scala @@ -12,6 +12,7 @@ */ package org.apache.pekko.cluster.sharding +import java.lang.invoke.{ MethodHandles, MethodType } import java.nio.file.Files import java.nio.file.Path @@ -25,19 +26,26 @@ import org.apache.pekko.actor.{ ActorSystem, ExtendedActorSystem } */ class FlightRecording(system: ActorSystem) { + private val lookup = MethodHandles.publicLookup() + private val noArgVoidMethodType = MethodType.methodType(Void.TYPE) + private val dumpMethodType = MethodType.methodType(Void.TYPE, classOf[Path]) + private val dynamic = system.asInstanceOf[ExtendedActorSystem].dynamicAccess private val recording = dynamic.createInstanceFor[AnyRef]("jdk.jfr.Recording", Nil).toOption private val clazz = recording.map(_.getClass) - private val startMethod = clazz.map(_.getDeclaredMethod("start")) - private val stopMethod = clazz.map(_.getDeclaredMethod("stop")) - private val dumpMethod = clazz.map(_.getDeclaredMethod("dump", classOf[Path])) + private val startMethod = + clazz.map(lookup.findVirtual(_, "start", noArgVoidMethodType)) + private val stopMethod = + clazz.map(lookup.findVirtual(_, "stop", noArgVoidMethodType)) + private val dumpMethod = + clazz.map(lookup.findVirtual(_, "dump", dumpMethodType)) def start() = { for { r <- recording - m <- startMethod - } yield m.invoke(r) + handle <- startMethod + } yield handle.invoke(r) } def endAndDump(location: Path) = { diff --git a/persistence/src/test/scala/org/apache/pekko/persistence/state/exception/DurableStateExceptionsSpec.scala b/persistence/src/test/scala/org/apache/pekko/persistence/state/exception/DurableStateExceptionsSpec.scala index 1a37d0c4744..66fadb3bcfa 100644 --- a/persistence/src/test/scala/org/apache/pekko/persistence/state/exception/DurableStateExceptionsSpec.scala +++ b/persistence/src/test/scala/org/apache/pekko/persistence/state/exception/DurableStateExceptionsSpec.scala @@ -39,8 +39,8 @@ class DurableStateExceptionsSpec extends AnyWordSpecLike "DurableStateException support" must { "allow creating DeleteRevisionException using MethodHandle" in { - val exceptionClassOpt: Option[Class[?]] = Try(Class.forName( - "org.apache.pekko.persistence.state.exception.DeleteRevisionException")).toOption + val exceptionClassOpt: Option[Class[?]] = + Try(Class.forName("org.apache.pekko.persistence.state.exception.DeleteRevisionException")).toOption exceptionClassOpt should not be empty val constructorOpt = exceptionClassOpt.map { clz => val mt = MethodType.methodType(classOf[Unit], classOf[String]) diff --git a/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala b/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala index c11a4d43f01..e2503d11bc9 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala @@ -13,10 +13,10 @@ package org.apache.pekko.remote.serialization -import java.lang.reflect.Method -import java.util.concurrent.atomic.AtomicReference +import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType } +import java.lang.reflect.InvocationTargetException +import java.util.concurrent.ConcurrentHashMap -import scala.annotation.tailrec import scala.util.control.NonFatal import org.apache.pekko @@ -28,7 +28,54 @@ import pekko.serialization.{ BaseSerializer, Serialization } import pekko.serialization.SerializationExtension object ProtobufSerializer { - private val ARRAY_OF_BYTE_ARRAY = Array[Class[?]](classOf[Array[Byte]]) + private val lookup = MethodHandles.lookup() + private val publicLookup = MethodHandles.publicLookup() + private val toByteArrayMethodType = MethodType.methodType(classOf[Array[Byte]]) + private val parsingInvokerType = MethodType.methodType(classOf[AnyRef], classOf[Array[Byte]]) + private val toByteArrayInvokerType = MethodType.methodType(classOf[AnyRef], classOf[AnyRef]) + private val invocationTargetExceptionConstructor = + lookup.findConstructor( + classOf[InvocationTargetException], + MethodType.methodType(Void.TYPE, classOf[Throwable])) + + private def parsingHandle(clazz: Class[?]): MethodHandle = { + val methodType = MethodType.methodType(clazz, classOf[Array[Byte]]) + val handle = + try lookup.findStatic(clazz, "parseFrom", methodType) + catch { + case _: IllegalAccessException => publicLookup.findStatic(clazz, "parseFrom", methodType) + } + val invoker = wrapTargetException(handle).asType(parsingInvokerType) + ensureInitialized(clazz) + invoker + } + + private def toByteArrayHandle(clazz: Class[?]): MethodHandle = { + val handle = + try lookup.findVirtual(clazz, "toByteArray", toByteArrayMethodType) + catch { + case _: IllegalAccessException => publicLookup.findVirtual(clazz, "toByteArray", toByteArrayMethodType) + } + val invoker = wrapTargetException(handle).asType(toByteArrayInvokerType) + ensureInitialized(clazz) + invoker + } + + private def wrapTargetException(target: MethodHandle): MethodHandle = { + val exceptionThrower = + MethodHandles.throwException(target.`type`().returnType(), classOf[InvocationTargetException]) + val handler = MethodHandles.filterArguments(exceptionThrower, 0, invocationTargetExceptionConstructor) + val handlerWithArguments = MethodHandles.dropArguments(handler, 1, target.`type`().parameterList()) + MethodHandles.catchException(target, classOf[Throwable], handlerWithArguments) + } + + private def ensureInitialized(clazz: Class[?]): Unit = { + try lookup.ensureInitialized(clazz) + catch { + case _: IllegalAccessException => + publicLookup.ensureInitialized(clazz) + } + } /** * Helper to serialize a [[pekko.actor.ActorRef]] to Pekko's @@ -56,8 +103,8 @@ object ProtobufSerializer { */ class ProtobufSerializer(val system: ExtendedActorSystem) extends BaseSerializer { - private val parsingMethodBindingRef = new AtomicReference[Map[Class[?], Method]](Map.empty) - private val toByteArrayMethodBindingRef = new AtomicReference[Map[Class[?], Method]](Map.empty) + private val parsingMethodBindings = new ConcurrentHashMap[Class[?], MethodHandle] + private val toByteArrayMethodBindings = new ConcurrentHashMap[Class[?], MethodHandle] private val allowedClassNames: Set[String] = { import scala.jdk.CollectionConverters._ @@ -74,25 +121,18 @@ class ProtobufSerializer(val system: ExtendedActorSystem) extends BaseSerializer override def fromBinary(bytes: Array[Byte], manifest: Option[Class[?]]): AnyRef = { manifest match { case Some(clazz) => - @tailrec - def parsingMethod(method: Method = null): Method = { - val parsingMethodBinding = parsingMethodBindingRef.get() - parsingMethodBinding.get(clazz) match { - case Some(cachedParsingMethod) => cachedParsingMethod - case None => - checkAllowedClass(clazz) - val unCachedParsingMethod = - if (method eq null) clazz.getDeclaredMethod("parseFrom", ProtobufSerializer.ARRAY_OF_BYTE_ARRAY: _*) - else method - if (parsingMethodBindingRef.compareAndSet( - parsingMethodBinding, - parsingMethodBinding.updated(clazz, unCachedParsingMethod))) - unCachedParsingMethod - else - parsingMethod(unCachedParsingMethod) + def parsingMethod(): MethodHandle = { + val cached = parsingMethodBindings.get(clazz) + if (cached ne null) cached + else { + checkAllowedClass(clazz) + val handle = ProtobufSerializer.parsingHandle(clazz) + val existing = parsingMethodBindings.putIfAbsent(clazz, handle) + if (existing eq null) handle else existing } } - parsingMethod().invoke(null, bytes) + val result: AnyRef = parsingMethod().invokeExact(bytes) + result case None => throw new IllegalArgumentException("Need a protobuf message class to be able to serialize bytes using protobuf") @@ -101,24 +141,17 @@ class ProtobufSerializer(val system: ExtendedActorSystem) extends BaseSerializer override def toBinary(obj: AnyRef): Array[Byte] = { val clazz = obj.getClass - @tailrec - def toByteArrayMethod(method: Method = null): Method = { - val toByteArrayMethodBinding = toByteArrayMethodBindingRef.get() - toByteArrayMethodBinding.get(clazz) match { - case Some(cachedtoByteArrayMethod) => cachedtoByteArrayMethod - case None => - val unCachedtoByteArrayMethod = - if (method eq null) clazz.getMethod("toByteArray") - else method - if (toByteArrayMethodBindingRef.compareAndSet( - toByteArrayMethodBinding, - toByteArrayMethodBinding.updated(clazz, unCachedtoByteArrayMethod))) - unCachedtoByteArrayMethod - else - toByteArrayMethod(unCachedtoByteArrayMethod) + def toByteArrayMethod(): MethodHandle = { + val cached = toByteArrayMethodBindings.get(clazz) + if (cached ne null) cached + else { + val handle = ProtobufSerializer.toByteArrayHandle(clazz) + val existing = toByteArrayMethodBindings.putIfAbsent(clazz, handle) + if (existing eq null) handle else existing } } - toByteArrayMethod().invoke(obj).asInstanceOf[Array[Byte]] + val result: AnyRef = toByteArrayMethod().invokeExact(obj) + result.asInstanceOf[Array[Byte]] } private def checkAllowedClass(clazz: Class[?]): Unit = { diff --git a/remote/src/test/java/org/apache/pekko/remote/serialization/FailingInitializerMessage.java b/remote/src/test/java/org/apache/pekko/remote/serialization/FailingInitializerMessage.java new file mode 100644 index 00000000000..90a5e7aa66a --- /dev/null +++ b/remote/src/test/java/org/apache/pekko/remote/serialization/FailingInitializerMessage.java @@ -0,0 +1,30 @@ +/* + * 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.remote.serialization; + +final class FailingInitializerMessage { + private static final Object INITIALIZED = failInitialization(); + + static FailingInitializerMessage parseFrom(byte[] bytes) { + return new FailingInitializerMessage(); + } + + private static Object failInitialization() { + throw new IllegalStateException("protobuf-initializer-bug"); + } +} diff --git a/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala index d9d83c36943..3b9885398e2 100644 --- a/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala +++ b/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala @@ -13,6 +13,8 @@ package org.apache.pekko.remote.serialization +import java.lang.reflect.InvocationTargetException + import scala.annotation.nowarn import org.apache.pekko @@ -27,6 +29,16 @@ import pekko.testkit.PekkoSpec // those must be defined as top level classes, to have static parseFrom case class MaliciousMessage() {} +object ThrowingParseMessage { + def parseFrom(@nowarn("msg=never used") bytes: Array[Byte]): ThrowingParseMessage = + throw new IllegalArgumentException("parse-user-bug") +} +case class ThrowingParseMessage() {} + +class ThrowingToByteArrayMessage { + def toByteArray(): Array[Byte] = throw new IllegalArgumentException("serialize-user-bug") +} + object ProtobufSerializerSpec { trait AnotherInterface abstract class AnotherBase @@ -62,6 +74,8 @@ class ProtobufSerializerSpec extends PekkoSpec(s""" "scalapb.GeneratedMessageCompanion", "org.apache.pekko.protobufv3.internal.GeneratedMessage", "${classOf[AnotherMessage].getName}", + "${classOf[FailingInitializerMessage].getName}", + "${classOf[ThrowingParseMessage].getName}", "${classOf[ProtobufSerializerSpec.AnotherInterface].getName}", "${classOf[ProtobufSerializerSpec.AnotherBase].getName}" ] @@ -89,12 +103,16 @@ class ProtobufSerializerSpec extends PekkoSpec(s""" deserialized.getMessage should ===(protobufMessage.getMessage) // same "hello" } - "work for a serialized protobuf v3 message" in { + "work for a serialized protobuf v3 message and reuse cached handles" in { val protobufV3Message: MyMessageV3 = MyMessageV3.newBuilder().setQuery("query1").setPageNumber(1).setResultPerPage(2).build() val bytes = ser.serialize(protobufV3Message).get val deserialized: MyMessageV3 = ser.deserialize(bytes, protobufV3Message.getClass).get + val cachedBytes = ser.serialize(protobufV3Message).get + val cachedDeserialized: MyMessageV3 = ser.deserialize(cachedBytes, protobufV3Message.getClass).get protobufV3Message should ===(deserialized) + cachedBytes should ===(bytes) + protobufV3Message should ===(cachedDeserialized) } "disallow deserialization of classes that are not in bindings and not in configured allowed classes" in { @@ -129,5 +147,28 @@ class ProtobufSerializerSpec extends PekkoSpec(s""" deserialized.getClass should ===(classOf[AnotherMessage3]) } + "preserve reflection-compatible target exception wrapping" in { + val serializer = ser.serializerFor(classOf[MyMessage]).asInstanceOf[ProtobufSerializer] + + val parseException = intercept[InvocationTargetException] { + serializer.fromBinary(Array.emptyByteArray, Some(classOf[ThrowingParseMessage])) + } + parseException.getCause.getMessage should ===("parse-user-bug") + + val serializeException = intercept[InvocationTargetException] { + serializer.toBinary(new ThrowingToByteArrayMessage) + } + serializeException.getCause.getMessage should ===("serialize-user-bug") + } + + "not wrap protobuf class-initialization failures" in { + val serializer = ser.serializerFor(classOf[MyMessage]).asInstanceOf[ProtobufSerializer] + + val exception = intercept[ExceptionInInitializerError] { + serializer.fromBinary(Array.emptyByteArray, Some(classOf[FailingInitializerMessage])) + } + exception.getCause.getMessage should ===("protobuf-initializer-bug") + } + } } diff --git a/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSet.scala b/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSet.scala index 0cb8f9179db..271b44281ff 100644 --- a/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSet.scala +++ b/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSet.scala @@ -13,6 +13,7 @@ package org.apache.pekko.testkit.metrics +import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType } import java.lang.management.{ ManagementFactory, OperatingSystemMXBean } import java.util @@ -22,6 +23,8 @@ import com.codahale.metrics.{ Gauge, Metric, MetricSet } import com.codahale.metrics.MetricRegistry._ import com.codahale.metrics.jvm.FileDescriptorRatioGauge +import com.sun.management.UnixOperatingSystemMXBean + /** * MetricSet exposing number of open and maximum file descriptors used by the JVM process. */ @@ -30,16 +33,33 @@ private[pekko] class FileDescriptorMetricSet(os: OperatingSystemMXBean = Managem override def getMetrics: util.Map[String, Metric] = { Map[String, Metric](name("file-descriptors", "open") -> new Gauge[Long] { - override def getValue: Long = invoke("getOpenFileDescriptorCount") + override def getValue: Long = FileDescriptorMetricSet.openFileDescriptorCount(os) }, name("file-descriptors", "max") -> new Gauge[Long] { - override def getValue: Long = invoke("getMaxFileDescriptorCount") + override def getValue: Long = FileDescriptorMetricSet.maxFileDescriptorCount(os) }, name("file-descriptors", "ratio") -> new FileDescriptorRatioGauge(os)).asJava } +} + +private object FileDescriptorMetricSet { + private val longMethodType = MethodType.methodType(java.lang.Long.TYPE) + private val invokerType = MethodType.methodType(classOf[AnyRef], classOf[AnyRef]) + private val lookup = MethodHandles.publicLookup() + private val openFileDescriptorCountHandle = lookup + .findVirtual(classOf[UnixOperatingSystemMXBean], "getOpenFileDescriptorCount", longMethodType) + .asType(invokerType) + private val maxFileDescriptorCountHandle = lookup + .findVirtual(classOf[UnixOperatingSystemMXBean], "getMaxFileDescriptorCount", longMethodType) + .asType(invokerType) + + private def openFileDescriptorCount(os: OperatingSystemMXBean): Long = + invoke(openFileDescriptorCountHandle, os) + + private def maxFileDescriptorCount(os: OperatingSystemMXBean): Long = + invoke(maxFileDescriptorCountHandle, os) - private def invoke(name: String): Long = { - val method = os.getClass.getDeclaredMethod(name) - method.setAccessible(true) - method.invoke(os).asInstanceOf[Long] + private def invoke(handle: MethodHandle, os: OperatingSystemMXBean): Long = { + val result: AnyRef = handle.invokeExact(os.asInstanceOf[AnyRef]) + result.asInstanceOf[Long] } } diff --git a/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSetSpec.scala b/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSetSpec.scala new file mode 100644 index 00000000000..308016240f4 --- /dev/null +++ b/testkit/src/test/scala/org/apache/pekko/testkit/metrics/FileDescriptorMetricSetSpec.scala @@ -0,0 +1,42 @@ +/* + * 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.testkit.metrics + +import java.lang.management.ManagementFactory + +import com.codahale.metrics.Gauge +import com.sun.management.UnixOperatingSystemMXBean +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class FileDescriptorMetricSetSpec extends AnyWordSpec with Matchers { + + "FileDescriptorMetricSet" must { + "read file descriptor counts through the public management interface" in { + val os = ManagementFactory.getOperatingSystemMXBean + if (!os.isInstanceOf[UnixOperatingSystemMXBean]) cancel("UnixOperatingSystemMXBean is not available") + + val metrics = new FileDescriptorMetricSet(os).getMetrics + val open = metrics.get("file-descriptors.open").asInstanceOf[Gauge[Long]].getValue + val max = metrics.get("file-descriptors.max").asInstanceOf[Gauge[Long]].getValue + + open should be >= 0L + max should be >= open + } + } +}