Add enchanced-double-hash-random for id-generator-policy

Co-authored-by: Claude Warren <claude.warren@aiven.io>
Co-authored-by: PJ Fanning <pjfanning@users.noreply.github.com>

actor/src/main/resources/reference.conf

@@ -1145,11 +1145,13 @@ pekko {
         # other platforms, will default to 1).
         ndots = default
 
-        # The policy used to generate dns transaction ids. Options are thread-local-random or secure-random.
-        # Defaults to thread-local-random similar to Netty, secure-random produces FIPS compliant random numbers but
-        # could block looking for entropy (these are short integers so are easy to bruit-force, use thread-local-random
-        # unless you really require FIPS compliant random numbers).
-        id-generator-policy = thread-local-random
+        # The policy used to generate dns transaction ids. Options are `thread-local-random`,
+        # `enhanced-double-hash-random` or `secure-random`. Defaults to `enhanced-double-hash-random` which uses an
+        # enhanced double hashing algorithm optimized for minimizing collisions with a FIPS compliant initial seed.
+        # `thread-local-random` is similar to Netty and `secure-random` produces FIPS compliant random numbers every
+        # time but could block looking for entropy (these are short integers so are easy to brute-force, use
+        # `enhanced-double-hash-random` unless you really require FIPS compliant random numbers).
+        id-generator-policy = enhanced-double-hash-random
       }
     }
   }

actor/src/main/scala/org/apache/pekko/io/dns/DnsSettings.scala

@@ -28,6 +28,7 @@ import pekko.actor.ExtendedActorSystem
 import pekko.annotation.InternalApi
 import pekko.event.Logging
 import pekko.io.dns.CachePolicy.{ CachePolicy, Forever, Never, Ttl }
+import pekko.io.dns.IdGenerator.Policy
 import pekko.io.dns.internal.{ ResolvConf, ResolvConfParser }
 import pekko.util.Helpers
 import pekko.util.Helpers.Requiring
@@ -67,10 +68,17 @@ private[dns] final class DnsSettings(system: ExtendedActorSystem, c: Config) {
   val PositiveCachePolicy: CachePolicy = getTtl("positive-ttl")
   val NegativeCachePolicy: CachePolicy = getTtl("negative-ttl")
 
-  lazy val IdGeneratorPolicy: IdGenerator.Policy = IdGenerator
-    .Policy(c.getString("id-generator-policy"))
-    .getOrElse(throw new IllegalArgumentException("id-generator-policy must be 'thread-local-random' or " +
-      s"'secure-random' value was '${c.getString("id-generator-policy")}'"))
+  lazy val IdGeneratorPolicy: IdGenerator.Policy = {
+    c.getString("id-generator-policy") match {
+      case "thread-local-random"                               => Policy.ThreadLocalRandom
+      case "secure-random"                                     => Policy.SecureRandom
+      case s if s.isEmpty | s == "enhanced-double-hash-random" => Policy.EnhancedDoubleHashRandom
+      case _ =>
+        throw new IllegalArgumentException(
+          "Invalid value for id-generator-policy, id-generator-policy must be 'thread-local-random', 'secure-random' or" +
+          s"`enhanced-double-hash-random`")
+    }
+  }
 
   private def getTtl(path: String): CachePolicy =
     c.getString(path) match {

actor/src/main/scala/org/apache/pekko/io/dns/IdGenerator.scala

@@ -12,6 +12,7 @@ package org.apache.pekko.io.dns
 import org.apache.pekko.annotation.InternalApi
 
 import java.security.SecureRandom
+import java.util.Random
 import java.util.concurrent.ThreadLocalRandom
 
 /**
@@ -39,18 +40,13 @@ private[pekko] object IdGenerator {
   object Policy {
     case object ThreadLocalRandom extends Policy
     case object SecureRandom extends Policy
-    val Default: Policy = ThreadLocalRandom
-
-    def apply(name: String): Option[Policy] = name.toLowerCase match {
-      case "thread-local-random" => Some(ThreadLocalRandom)
-      case "secure-random"       => Some(SecureRandom)
-      case _                     => Some(ThreadLocalRandom)
-    }
+    case object EnhancedDoubleHashRandom extends Policy
   }
 
   def apply(policy: Policy): IdGenerator = policy match {
-    case Policy.ThreadLocalRandom => random(ThreadLocalRandom.current())
-    case Policy.SecureRandom      => random(new SecureRandom())
+    case Policy.ThreadLocalRandom        => random(ThreadLocalRandom.current())
+    case Policy.SecureRandom             => random(new SecureRandom())
+    case Policy.EnhancedDoubleHashRandom => new EnhancedDoubleHashGenerator(new SecureRandom())
   }
 
   def apply(): IdGenerator = random(ThreadLocalRandom.current())
@@ -60,4 +56,31 @@ private[pekko] object IdGenerator {
    */
   def random(rand: java.util.Random): IdGenerator =
     () => (rand.nextInt(UnsignedShortBound) + Short.MinValue).toShort
+
+  private[pekko] class EnhancedDoubleHashGenerator(seed: Random) extends IdGenerator {
+
+    /**
+     * An efficient thread safe generator of pseudo random shorts based on
+     * https://en.wikipedia.org/wiki/Double_hashing#Enhanced_double_hashing.
+     *
+     * Note that due to the usage of synchronized this method is optimized
+     * for the happy case (i.e. least contention) on multiple threads.
+     */
+    private var index = seed.nextLong
+    private var increment = seed.nextLong
+    private var count = 1L
+
+    override final def nextId(): Short = synchronized {
+      val result = (0xFFFFFFFF & index).asInstanceOf[Short]
+      index -= increment
+
+      // Incorporate the counter into the increment to create a
+      // tetrahedral number additional term.
+      increment -= {
+        count += 1
+        count - 1
+      }
+      result
+    }
+  }
 }

bench-jmh/src/main/scala/org/apache/pekko/io/dns/IdGeneratorBanchmark.scala

@@ -29,16 +29,34 @@ import java.security.SecureRandom
 @OutputTimeUnit(TimeUnit.NANOSECONDS)
 @Warmup(iterations = 3, time = 5, timeUnit = TimeUnit.SECONDS)
 @Measurement(iterations = 3, time = 5)
-@Threads(8)
 @Fork(1)
 @State(Scope.Benchmark)
 class IdGeneratorBanchmark {
   val threadLocalRandom = IdGenerator.random(ThreadLocalRandom.current())
   val secureRandom = IdGenerator.random(new SecureRandom())
+  val enhancedDoubleHash = new IdGenerator.EnhancedDoubleHashGenerator(new SecureRandom())
 
+  @Threads(1)
   @Benchmark
   def measureThreadLocalRandom(): Short = threadLocalRandom.nextId()
 
+  @Threads(1)
   @Benchmark
   def measureSecureRandom(): Short = secureRandom.nextId()
+
+  @Threads(1)
+  @Benchmark
+  def measureEnhancedDoubleHash(): Short = enhancedDoubleHash.nextId()
+
+  @Threads(2)
+  @Benchmark
+  def multipleThreadsMeasureThreadLocalRandom(): Short = threadLocalRandom.nextId()
+
+  @Threads(2)
+  @Benchmark
+  def multipleThreadsMeasureSecureRandom(): Short = secureRandom.nextId()
+
+  @Threads(2)
+  @Benchmark
+  def multipleThreadsMeasureEnhancedDoubleHash(): Short = enhancedDoubleHash.nextId()
 }