This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch refactor/influxdb-methodhandle-varhandle in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
commit a82c64187f9c2fc3df869095ecf94bb24ec8fce7 Author: 虎鸣 <[email protected]> AuthorDate: Mon Jul 6 16:27:44 2026 +0800 refactor: replace java.lang.reflect with VarHandle/MethodHandle for improved JIT optimization Motivation: java.lang.reflect Field/Constructor invocation bypasses JIT inlining, causing unnecessary overhead in hot paths like InfluxDB result mapping and HDFS sequence file reading. Modification: - InfluxDB mapper: replace Field cache with VarHandle-based ColumnInfo cache; replace Constructor.newInstance with cached MethodHandle.invoke; move caches to companion object for cross-instance sharing - HDFS source: replace getDeclaredConstructor().newInstance() in iterator hot loop with cached MethodHandle.invoke - Jakarta JMS test: replace getDeclaredFields().filter pattern with VarHandle.get via getDeclaredField + unreflectVarHandle Result: VarHandle and MethodHandle enable JIT inlining of field access and constructor invocation, reducing overhead in data-mapping hot paths. Tests: - sbt "influxdb / compile" - success - sbt "hdfs / compile" - success - sbt "jakartams / Test / compile" - success References: None - internal refactoring --- .../connectors/hdfs/scaladsl/HdfsSource.scala | 23 ++- .../impl/PekkoConnectorsResultMapperHelper.scala | 162 ++++++++++++--------- .../scala/docs/scaladsl/JmsConnectorsSpec.scala | 19 ++- 3 files changed, 124 insertions(+), 80 deletions(-) diff --git a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala index 53cc8cfee..ca68a3c9f 100644 --- a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala +++ b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/scaladsl/HdfsSource.scala @@ -13,6 +13,9 @@ package org.apache.pekko.stream.connectors.hdfs.scaladsl +import java.lang.invoke.{ MethodHandle, MethodHandles } +import java.util.concurrent.ConcurrentHashMap + import org.apache.pekko import pekko.NotUsed import pekko.stream.ActorAttributes.IODispatcher @@ -27,6 +30,20 @@ import scala.concurrent.Future object HdfsSource { + private val constructorCache = new ConcurrentHashMap[Class[?], MethodHandle]() + + private def getNoArgConstructor[T](clazz: Class[T]): MethodHandle = { + var handle = constructorCache.get(clazz) + if (handle == null) { + val ctor = clazz.getDeclaredConstructor() + ctor.setAccessible(true) + handle = MethodHandles.lookup().unreflectConstructor(ctor) + val existing = constructorCache.putIfAbsent(clazz, handle) + if (existing != null) handle = existing + } + handle + } + /** * Scala API: creates a `Source` that consumes as `ByteString` * @@ -69,10 +86,12 @@ object HdfsSource { classK: Class[K], classV: Class[V]): Source[(K, V), NotUsed] = { val reader: SequenceFile.Reader = new SequenceFile.Reader(fs.getConf, SequenceFile.Reader.file(path)) + val keyHandle = getNoArgConstructor(classK) + val valueHandle = getNoArgConstructor(classV) val it = Iterator .continually { - val key = classK.getDeclaredConstructor().newInstance() - val value = classV.getDeclaredConstructor().newInstance() + val key = keyHandle.invoke().asInstanceOf[K] + val value = valueHandle.invoke().asInstanceOf[V] val hasCurrent = reader.next(key, value) (hasCurrent, (key, value)) } diff --git a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala index 681ed63dd..91cf57758 100644 --- a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala +++ b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/PekkoConnectorsResultMapperHelper.scala @@ -13,7 +13,7 @@ package org.apache.pekko.stream.connectors.influxdb.impl -import java.lang.reflect.Field +import java.lang.invoke.{ MethodHandle, MethodHandles, VarHandle } import java.time.Instant import java.time.format.DateTimeFormatterBuilder import java.time.temporal.ChronoField @@ -35,7 +35,7 @@ import scala.jdk.CollectionConverters._ @InternalApi private[impl] class PekkoConnectorsResultMapperHelper { - val CLASS_FIELD_CACHE: ConcurrentHashMap[String, ConcurrentMap[String, Field]] = new ConcurrentHashMap(); + import PekkoConnectorsResultMapperHelper.{ CLASS_COLUMN_CACHE, CONSTRUCTOR_CACHE, ColumnInfo } private val FRACTION_MIN_WIDTH = 0 private val FRACTION_MAX_WIDTH = 9 @@ -57,7 +57,7 @@ private[impl] class PekkoConnectorsResultMapperHelper { throwExceptionIfMissingAnnotation(model.getClass) cacheClassFields(model.getClass) - val colNameAndFieldMap: ConcurrentMap[String, Field] = CLASS_FIELD_CACHE.get(model.getClass.getName) + val colMap: ConcurrentMap[String, ColumnInfo] = CLASS_COLUMN_CACHE.get(model.getClass.getName) try { val modelType = model.getClass(); @@ -66,27 +66,18 @@ private[impl] class PekkoConnectorsResultMapperHelper { val time = timeUnit.convert(System.currentTimeMillis(), TimeUnit.MILLISECONDS); val pointBuilder: Point.Builder = Point.measurement(measurement).time(time, timeUnit); - for (key <- colNameAndFieldMap.keySet().asScala) { - val field = colNameAndFieldMap.get(key) - val column = field.getAnnotation(classOf[Column]) - val columnName: String = column.name() - val fieldType: Class[?] = field.getType() + for (key <- colMap.keySet().asScala) { + val colInfo = colMap.get(key) + val value = colInfo.varHandle.get(model); - val isAccessible = field.isAccessible() - if (!isAccessible) { - field.setAccessible(true); - } - - val value = field.get(model); - - if (column.tag()) { - pointBuilder.tag(columnName, value.toString()); - } else if ("time".equals(columnName)) { + if (colInfo.isTag) { + pointBuilder.tag(colInfo.columnName, value.toString()); + } else if ("time".equals(colInfo.columnName)) { if (value != null) { - setTime(pointBuilder, fieldType, timeUnit, value); + setTime(pointBuilder, colInfo.fieldType, timeUnit, value); } } else { - setField(pointBuilder, fieldType, columnName, value); + setField(pointBuilder, colInfo.fieldType, colInfo.columnName, value); } } @@ -96,13 +87,13 @@ private[impl] class PekkoConnectorsResultMapperHelper { } } - private[impl] def cacheClassFields(clazz: Class[?]) = - if (!CLASS_FIELD_CACHE.containsKey(clazz.getName)) { - val initialMap: ConcurrentMap[String, Field] = new ConcurrentHashMap() - var influxColumnAndFieldMap = CLASS_FIELD_CACHE.putIfAbsent(clazz.getName, initialMap) + private[impl] def cacheClassFields(clazz: Class[?]): Unit = + if (!CLASS_COLUMN_CACHE.containsKey(clazz.getName)) { + val initialMap: ConcurrentMap[String, ColumnInfo] = new ConcurrentHashMap() + var columnMap = CLASS_COLUMN_CACHE.putIfAbsent(clazz.getName, initialMap) - if (influxColumnAndFieldMap == null) { - influxColumnAndFieldMap = initialMap; + if (columnMap == null) { + columnMap = initialMap; } var c = clazz; @@ -111,7 +102,15 @@ private[impl] class PekkoConnectorsResultMapperHelper { for (field <- c.getDeclaredFields()) { val colAnnotation = field.getAnnotation(classOf[Column]); if (colAnnotation != null) { - influxColumnAndFieldMap.put(colAnnotation.name(), field); + field.setAccessible(true) + val varHandle = MethodHandles.lookup().unreflectVarHandle(field) + val colInfo = new ColumnInfo( + colAnnotation.name(), + colAnnotation.tag(), + field.getType, + varHandle, + field.getName) + columnMap.put(colAnnotation.name(), colInfo); } } c = c.getSuperclass(); @@ -155,111 +154,119 @@ private[impl] class PekkoConnectorsResultMapperHelper { throw new IllegalArgumentException( "Class " + clazz.getName + " is not annotated with @" + classOf[Measurement].getSimpleName) + private def getOrCreateConstructorHandle(clazz: Class[?]): MethodHandle = { + var handle = CONSTRUCTOR_CACHE.get(clazz) + if (handle == null) { + val constructor = clazz.getDeclaredConstructor() + constructor.setAccessible(true) + handle = MethodHandles.lookup().unreflectConstructor(constructor) + val existing = CONSTRUCTOR_CACHE.putIfAbsent(clazz, handle) + if (existing != null) handle = existing + } + handle + } + private def parseRowAs[T](clazz: Class[T], columns: java.util.List[String], values: java.util.List[AnyRef], precision: TimeUnit): T = try { - val fieldMap = CLASS_FIELD_CACHE.get(clazz.getName) + val colMap = CLASS_COLUMN_CACHE.get(clazz.getName) - val obj: T = clazz.getDeclaredConstructor().newInstance() + val handle = getOrCreateConstructorHandle(clazz) + val obj: T = handle.invoke().asInstanceOf[T] for (i <- 0 until columns.size()) { - val correspondingField = fieldMap.get(columns.get(i)) - if (correspondingField != null) { - setFieldValue(obj, correspondingField, values.get(i), precision) + val colInfo = colMap.get(columns.get(i)) + if (colInfo != null) { + setFieldValue(obj, colInfo, values.get(i), precision) } } obj } catch { - case e @ (_: InstantiationException | _: IllegalAccessException) => + case e: InfluxDBMapperException => throw e + case e: Exception => throw new InfluxDBMapperException(e) } - @throws[IllegalArgumentException] - @throws[IllegalAccessException] - private def setFieldValue[T](obj: T, field: Field, value: Any, precision: TimeUnit): Unit = { + private def setFieldValue[T](obj: T, colInfo: ColumnInfo, value: Any, precision: TimeUnit): Unit = { if (value == null) return - val fieldType = field.getType + val fieldType = colInfo.fieldType try { - val isAccessible = field.isAccessible() - if (!isAccessible) field.setAccessible(true) - if (fieldValueModified(fieldType, field, obj, value, precision) || fieldValueForPrimitivesModified( - fieldType, - field, - obj, - value) || fieldValueForPrimitiveWrappersModified(fieldType, field, obj, value)) return + if (fieldValueModified(fieldType, colInfo, obj, value, precision) || + fieldValueForPrimitivesModified(fieldType, colInfo, obj, value) || + fieldValueForPrimitiveWrappersModified(fieldType, colInfo, obj, value)) return val msg = - s"""Class '${obj.getClass.getName}' field '${field.getName}' is from an unsupported type '${field.getType}'.""" + s"""Class '${obj.getClass.getName}' field '${colInfo.fieldName}' is from an unsupported type '${colInfo.fieldType}'.""" throw new InfluxDBMapperException(msg) } catch { case e: ClassCastException => val msg = - s"""Class '${obj.getClass.getName}' field '${field.getName}' was defined with a different field type and caused a ClassCastException. + s"""Class '${obj.getClass.getName}' field '${colInfo.fieldName}' was defined with a different field type and caused a ClassCastException. |The correct type is '${value.getClass.getName}' (current field value: '${value}')""".stripMargin throw new InfluxDBMapperException(msg) } } - @throws[IllegalArgumentException] - @throws[IllegalAccessException] - private def fieldValueForPrimitivesModified[T](fieldType: Class[?], field: Field, obj: T, value: Any): Boolean = + private def fieldValueForPrimitivesModified[T]( + fieldType: Class[?], + colInfo: ColumnInfo, + obj: T, + value: Any): Boolean = if (classOf[Double].isAssignableFrom(fieldType)) { - field.setDouble(obj, value.asInstanceOf[Double].doubleValue) + colInfo.varHandle.set(obj, value.asInstanceOf[Double].doubleValue) true } else if (classOf[Long].isAssignableFrom(fieldType)) { - field.setLong(obj, value.asInstanceOf[Double].longValue) + colInfo.varHandle.set(obj, value.asInstanceOf[Double].longValue) true } else if (classOf[Int].isAssignableFrom(fieldType)) { - field.setInt(obj, value.asInstanceOf[Double].intValue) + colInfo.varHandle.set(obj, value.asInstanceOf[Double].intValue) true } else if (classOf[Boolean].isAssignableFrom(fieldType)) { - field.setBoolean(obj, String.valueOf(value).toBoolean) + colInfo.varHandle.set(obj, String.valueOf(value).toBoolean) true } else { false } - @throws[IllegalArgumentException] - @throws[IllegalAccessException] - private def fieldValueForPrimitiveWrappersModified[T](fieldType: Class[?], - field: Field, + private def fieldValueForPrimitiveWrappersModified[T]( + fieldType: Class[?], + colInfo: ColumnInfo, obj: T, value: Any): Boolean = if (classOf[java.lang.Double].isAssignableFrom(fieldType)) { - field.set(obj, value) + colInfo.varHandle.set(obj, value) true } else if (classOf[java.lang.Long].isAssignableFrom(fieldType)) { - field.set(obj, value.asInstanceOf[Double].longValue()) + colInfo.varHandle.set(obj, value.asInstanceOf[Double].longValue()) true } else if (classOf[Integer].isAssignableFrom(fieldType)) { - field.set(obj, value.asInstanceOf[java.lang.Integer]) + colInfo.varHandle.set(obj, value.asInstanceOf[java.lang.Integer]) true } else if (classOf[java.lang.Boolean].isAssignableFrom(fieldType)) { - field.set(obj, value.asInstanceOf[java.lang.Boolean]) + colInfo.varHandle.set(obj, value.asInstanceOf[java.lang.Boolean]) true } else { false } - @throws[IllegalArgumentException] - @throws[IllegalAccessException] - private def fieldValueModified[T](fieldType: Class[?], - field: Field, + private def fieldValueModified[T]( + fieldType: Class[?], + colInfo: ColumnInfo, obj: T, value: Any, precision: TimeUnit): Boolean = if (classOf[String].isAssignableFrom(fieldType)) { - field.set(obj, String.valueOf(value)) + colInfo.varHandle.set(obj, String.valueOf(value)) true } else if (classOf[Instant].isAssignableFrom(fieldType)) { - val instant: Instant = getInstant(field, value, precision) - field.set(obj, instant) + val instant: Instant = getInstant(colInfo, value, precision) + colInfo.varHandle.set(obj, instant) true } else { false } - private def getInstant(field: Field, value: Any, precision: TimeUnit): Instant = + private def getInstant(colInfo: ColumnInfo, value: Any, precision: TimeUnit): Instant = if (value.isInstanceOf[String]) Instant.from(RFC3339_FORMATTER.parse(String.valueOf(value))) else if (value.isInstanceOf[java.lang.Long]) Instant.ofEpochMilli(toMillis(value.asInstanceOf[Long], precision)) else if (value.isInstanceOf[java.lang.Double]) @@ -267,9 +274,24 @@ private[impl] class PekkoConnectorsResultMapperHelper { else if (value.isInstanceOf[java.lang.Integer]) Instant.ofEpochMilli(toMillis(value.asInstanceOf[Integer].longValue, precision)) else { - throw new InfluxDBMapperException(s"""Unsupported type ${field.getClass} for field ${field.getName}""") + throw new InfluxDBMapperException(s"""Unsupported type for field ${colInfo.fieldName}""") } private def toMillis(value: Long, precision: TimeUnit) = TimeUnit.MILLISECONDS.convert(value, precision) } + +@InternalApi +private[impl] object PekkoConnectorsResultMapperHelper { + private val CLASS_COLUMN_CACHE: ConcurrentHashMap[String, ConcurrentMap[String, ColumnInfo]] = + new ConcurrentHashMap() + + private val CONSTRUCTOR_CACHE: ConcurrentHashMap[Class[?], MethodHandle] = new ConcurrentHashMap() + + final class ColumnInfo( + val columnName: String, + val isTag: Boolean, + val fieldType: Class[?], + val varHandle: VarHandle, + val fieldName: String) +} diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala index c4cbc94cd..b550083ca 100644 --- a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala +++ b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala @@ -26,6 +26,7 @@ import org.mockito.Mockito._ import org.mockito.invocation.InvocationOnMock import org.mockito.stubbing.Answer +import java.lang.invoke.MethodHandles import java.nio.charset.StandardCharsets import java.util.UUID import java.util.concurrent.atomic.AtomicInteger @@ -511,11 +512,12 @@ class JmsConnectorsSpec extends JmsSpec { // make sure connection was closed eventually { - val isClosedField = - connectionFactory.cachedConnection.getClass.getDeclaredFields.filter(x => x.getName == "closed").head - isClosedField.setAccessible(true) + val conn = connectionFactory.cachedConnection + val closedField = conn.getClass.getDeclaredField("closed") + closedField.setAccessible(true) + val isClosedHandle = MethodHandles.lookup().unreflectVarHandle(closedField) - val isClosedValue = isClosedField.get(connectionFactory.cachedConnection) + val isClosedValue = isClosedHandle.get(conn) isClosedValue shouldBe true } } @@ -535,11 +537,12 @@ class JmsConnectorsSpec extends JmsSpec { completionFuture.failed.futureValue shouldBe a[RuntimeException] // make sure connection was closed eventually { - val isClosedField = - connectionFactory.cachedConnection.getClass.getDeclaredFields.filter(x => x.getName == "closed").head - isClosedField.setAccessible(true) + val conn = connectionFactory.cachedConnection + val closedField = conn.getClass.getDeclaredField("closed") + closedField.setAccessible(true) + val isClosedHandle = MethodHandles.lookup().unreflectVarHandle(closedField) - val isClosedValue = isClosedField.get(connectionFactory.cachedConnection) + val isClosedValue = isClosedHandle.get(conn) isClosedValue shouldBe true } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
