This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-persistence-dynamodb.git
The following commit(s) were added to refs/heads/main by this push:
new 8624c8b refactor: replace Class.forName with MethodHandles in test
serializer (#336)
8624c8b is described below
commit 8624c8bda88aced08445af4356647c1f8e031e54
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 7 01:54:30 2026 +0800
refactor: replace Class.forName with MethodHandles in test serializer (#336)
Motivation:
Class.forName and reflective constructor calls bypass JIT inlining in the
test serializer used by the DynamoDB persistence plugin.
Modification:
- Replace Class.forName with MethodHandles.publicLookup().findConstructor
for serializer class loading
- Remove invalid 'final' modifier on local val in WriteThroughputBench
Result:
JIT-inlinable constructor access in the test serializer path.
Refs #336
---
.../dynamodb/journal/WriteThroughputBench.scala | 100 +++++++++++----------
.../dynamodb/journal/TestSerializer.scala | 9 +-
2 files changed, 59 insertions(+), 50 deletions(-)
diff --git
a/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
b/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
index 7023210..bdeedee 100644
---
a/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
+++
b/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
@@ -27,10 +27,52 @@ import java.util.concurrent.ThreadLocalRandom
import scala.concurrent.duration._
object WriteThroughputBench extends DynamoDBUtils {
+
+ private lazy val config =
+ ConfigFactory
+ .systemProperties()
+ .withFallback(
+ ConfigFactory
+ .parseString("""
+my-dynamodb-journal {
+ journal-table = "WriteThroughputBench"
+ endpoint = ${?AWS_DYNAMODB_ENDPOINT}
+ aws-access-key-id = ${?AWS_ACCESS_KEY_ID}
+ aws-secret-access-key = ${?AWS_SECRET_ACCESS_KEY}
+ aws-client-config {
+ max-connections = 100
+ }
+ plugin-dispatcher = "dispatcher"
+ replay-dispatcher = "dispatcher"
+ client-dispatcher = "dispatcher"
+}
+dispatcher {
+ type = Dispatcher
+ executor = "fork-join-executor"
+ fork-join-executor {
+ parallelism-min = 4
+ parallelism-max = 8
+ }
+}
+pekko.actor.default-dispatcher.fork-join-executor.parallelism-max = 4
+writers = 1000
+writer-dispatcher {
+ type = Dispatcher
+ executor = "fork-join-executor"
+ fork-join-executor {
+ parallelism-min = 2
+ parallelism-max = 2
+ }
+}
+""").resolve)
+ .withFallback(ConfigFactory.load())
+
+ lazy val system: ActorSystem = ActorSystem("WriteThroughputBench", config)
+
def main(args: Array[String]): Unit = {
def rnd = ThreadLocalRandom.current()
- final val oneBillion = 1000L * 1000 * 1000
+ val oneBillion = 1000L * 1000 * 1000
class H private (private val entries: Map[Int, Int]) {
def this(i: Int) = this(Map(i -> 1))
@@ -89,46 +131,6 @@ object WriteThroughputBench extends DynamoDBUtils {
def receiveRecover = Actor.emptyBehavior
}
- val config =
- ConfigFactory
- .systemProperties()
- .withFallback(
- ConfigFactory
- .parseString("""
-my-dynamodb-journal {
- journal-table = "WriteThroughputBench"
- endpoint = ${?AWS_DYNAMODB_ENDPOINT}
- aws-access-key-id = ${?AWS_ACCESS_KEY_ID}
- aws-secret-access-key = ${?AWS_SECRET_ACCESS_KEY}
- aws-client-config {
- max-connections = 100
- }
- plugin-dispatcher = "dispatcher"
- replay-dispatcher = "dispatcher"
- client-dispatcher = "dispatcher"
-}
-dispatcher {
- type = Dispatcher
- executor = "fork-join-executor"
- fork-join-executor {
- parallelism-min = 4
- parallelism-max = 8
- }
-}
-pekko.actor.default-dispatcher.fork-join-executor.parallelism-max = 4
-writers = 1000
-writer-dispatcher {
- type = Dispatcher
- executor = "fork-join-executor"
- fork-join-executor {
- parallelism-min = 2
- parallelism-max = 2
- }
-}
-""").resolve)
- .withFallback(ConfigFactory.load())
-
- implicit val system: ActorSystem = ActorSystem("WriteThroughputBench",
config)
implicit val materializer: Materializer = Materializer(system)
/*
@@ -169,6 +171,14 @@ writer-dispatcher {
.expand(Iterator.continually(_))
.withAttributes(Attributes.asyncBoundary)
+ def printStats(r: Report): Unit = {
+ def p(h: Histogram, pc: Double) = h.getValueAtPercentile(pc) / 1000000d
+ def perc(h: Histogram) =
+ f"50=${p(h, 0.5)}%5.1f 90=${p(h, 0.9)}%5.1f 99=${p(h, 0.99)}%5.1f
99.9=${p(h, 0.999)}%5.1f 99.99=${p(h, 0.9999)}%5.1f"
+ println(f"count ${r.endToEnd.getTotalCount}%6d/s percentiles:
endToEnd(${perc(r.endToEnd)}) calls(${perc(
+ r.calls)}) retries: ${r.retries}")
+ }
+
val (eRef, cRef) =
RunnableGraph
.fromGraph(GraphDSL.createGraph(endToEnd, calls)(Keep.both) { implicit
b => (e, c) =>
@@ -193,13 +203,5 @@ writer-dispatcher {
system.terminate()
dynamo.shutdown()
-
- def printStats(r: Report): Unit = {
- def p(h: Histogram, pc: Double) = h.getValueAtPercentile(pc) / 1000000d
- def perc(h: Histogram) =
- f"50=${p(h, 0.5)}%5.1f 90=${p(h, 0.9)}%5.1f 99=${p(h, 0.99)}%5.1f
99.9=${p(h, 0.999)}%5.1f 99.99=${p(h, 0.9999)}%5.1f"
- println(f"count ${r.endToEnd.getTotalCount}%6d/s percentiles:
endToEnd(${perc(r.endToEnd)}) calls(${perc(
- r.calls)}) retries: ${r.retries}")
- }
}
}
diff --git
a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/TestSerializer.scala
b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/TestSerializer.scala
index 766a836..ff16e6a 100644
---
a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/TestSerializer.scala
+++
b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/TestSerializer.scala
@@ -13,6 +13,8 @@
package org.apache.pekko.persistence.dynamodb.journal
+import java.lang.invoke.MethodHandles
+
import org.apache.pekko.actor.ExtendedActorSystem
import org.apache.pekko.serialization.{ AsyncSerializerWithStringManifest,
JavaSerializer }
@@ -29,7 +31,12 @@ class TestSerializer(system: ExtendedActorSystem) extends
AsyncSerializerWithStr
val javaSerializer: JavaSerializer = new JavaSerializer(system)
override def fromBinaryAsync(bytes: Array[Byte], manifest: String):
Future[AnyRef] =
- Future.successful(javaSerializer.fromBinary(bytes,
Class.forName(manifest)))
+ Future.successful {
+ val lookup = MethodHandles.lookup()
+ val clazz = lookup.findClass(manifest)
+ lookup.ensureInitialized(clazz)
+ javaSerializer.fromBinary(bytes, clazz)
+ }
override def toBinaryAsync(obj: AnyRef): Future[Array[Byte]] =
Future.successful(javaSerializer.toBinary(obj))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
