ENT-6947 Intern common types to reduce heap footprint (#7239)

ENT-6947: Implement interning for SecureHash, CordaX500Name, PublicKey, AsbtractParty and SignatureAttachmentConstraint, including automatic detection of internable types off companion objects in AMQP & Kyro deserialization.  In some cases, add new factory methods to companion objects, and make main code base use them.

Performance tested in performance cluster with no negative impact visible (so default concurrency setting seems okay).

Testing suggests 5-6x memory saving for tokens in TokensSDK in memory selector.  Should see approx. 1 million tokens per GB or better (1.5 million for the tokens we tested with).

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/InterningSerializer.kt

@@ -0,0 +1,24 @@
+package net.corda.serialization.internal.amqp
+
+import net.corda.core.internal.utilities.PrivateInterner
+import net.corda.core.serialization.SerializationContext
+import org.apache.qpid.proton.codec.Data
+import java.lang.reflect.Type
+
+class InterningSerializer(private val delegate: ObjectSerializer, private val interner: PrivateInterner<Any>) : ObjectSerializer by delegate {
+    companion object {
+        fun maybeWrapForInterning(candidate: ObjectSerializer): ObjectSerializer {
+            val clazz = candidate.type as? Class<*>
+            val interner: PrivateInterner<Any>? = PrivateInterner.findFor(clazz)
+            return if (interner != null) InterningSerializer(candidate, interner) else candidate
+        }
+    }
+
+    override fun writeObject(obj: Any, data: Data, type: Type, output: SerializationOutput, context: SerializationContext, debugIndent: Int) {
+        delegate.writeObject(obj, data, type, output, context, debugIndent)
+    }
+
+    override fun readObject(obj: Any, schemas: SerializationSchemas, input: DeserializationInput, context: SerializationContext): Any {
+        return interner.intern(delegate.readObject(obj, schemas, input, context))
+    }
+}

serialization/src/main/kotlin/net/corda/serialization/internal/amqp/ObjectSerializer.kt

@@ -71,7 +71,7 @@ interface ObjectSerializer : AMQPSerializer<Any> {
         private fun makeForComposable(typeInformation: LocalTypeInformation.Composable,
                                       typeNotation: CompositeType,
                                       typeDescriptor: Symbol,
-                                      factory: LocalSerializerFactory): ComposableObjectSerializer {
+                                      factory: LocalSerializerFactory): ObjectSerializer {
             val propertySerializers = makePropertySerializers(typeInformation.properties, factory)
             val reader = ComposableObjectReader(
                     typeInformation.typeIdentifier,
@@ -83,13 +83,13 @@ interface ObjectSerializer : AMQPSerializer<Any> {
                     typeInformation.interfaces,
                     propertySerializers)
 
-            return ComposableObjectSerializer(
+            return InterningSerializer.maybeWrapForInterning(ComposableObjectSerializer(
                     typeInformation.observedType,
                     typeDescriptor,
                     propertySerializers,
                     typeNotation.fields,
                     reader,
-                    writer)
+                    writer))
         }
 
         private fun makePropertySerializers(properties: Map<PropertyName, LocalPropertyInformation>,

serialization/src/test/kotlin/net/corda/serialization/internal/amqp/RoundTripTests.kt

@@ -1,6 +1,7 @@
 package net.corda.serialization.internal.amqp
 
 import net.corda.core.contracts.ContractState
+import net.corda.core.contracts.SignatureAttachmentConstraint
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TransactionState
@@ -20,6 +21,8 @@ import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.Test
 import java.math.BigInteger
 import kotlin.test.assertEquals
+import kotlin.test.assertNotSame
+import kotlin.test.assertSame
 
 class RoundTripTests {
 
@@ -145,6 +148,42 @@ class RoundTripTests {
 	fun canSerializeClassesWithUntypedProperties() {
         val data = MembershipState<Any>(mapOf("foo" to "bar"))
         val party = Party(
+                CordaX500Name.interner.intern(CordaX500Name(organisation = "Test Corp", locality = "Madrid", country = "ES")),
+                entropyToKeyPair(BigInteger.valueOf(83)).public)
+        val transactionState = TransactionState(
+                data,
+                "foo",
+                party
+        )
+        val ref = StateRef(SecureHash.zeroHash, 0)
+        val instance = OnMembershipChanged(StateAndRef(
+                transactionState,
+                ref
+        ))
+
+        val factory = testDefaultFactoryNoEvolution().apply { register(PublicKeySerializer) }
+        val bytes = SerializationOutput(factory).serialize(instance)
+        val deserialized = DeserializationInput(factory).deserialize(bytes)
+        assertEquals(mapOf("foo" to "bar"), deserialized.changedMembership.state.data.metadata)
+        assertNotSame(instance.changedMembership.state.notary, deserialized.changedMembership.state.notary)
+        assertSame(instance.changedMembership.state.notary.name, deserialized.changedMembership.state.notary.name)
+        assertSame(instance.changedMembership.state.notary.owningKey, deserialized.changedMembership.state.notary.owningKey)
+    }
+
+    @Test(timeout = 300_000)
+    fun sigConstraintsInterned() {
+        val instance = SignatureAttachmentConstraint.create(entropyToKeyPair(BigInteger.valueOf(83)).public)
+
+        val factory = testDefaultFactoryNoEvolution().apply { register(PublicKeySerializer) }
+        val bytes = SerializationOutput(factory).serialize(instance)
+        val deserialized = DeserializationInput(factory).deserialize(bytes)
+        assertSame(instance, deserialized)
+    }
+
+    @Test(timeout = 300_000)
+    fun canSerializeClassesWithUntypedPropertiesWithInternedParty() {
+        val data = MembershipState<Any>(mapOf("foo" to "bar"))
+        val party = Party.create(
                 CordaX500Name(organisation = "Test Corp", locality = "Madrid", country = "ES"),
                 entropyToKeyPair(BigInteger.valueOf(83)).public)
         val transactionState = TransactionState(
@@ -162,6 +201,7 @@ class RoundTripTests {
         val bytes = SerializationOutput(factory).serialize(instance)
         val deserialized = DeserializationInput(factory).deserialize(bytes)
         assertEquals(mapOf("foo" to "bar"), deserialized.changedMembership.state.data.metadata)
+        assertSame(instance.changedMembership.state.notary, deserialized.changedMembership.state.notary)
     }
 
     interface I2<T> {
@@ -170,8 +210,8 @@ class RoundTripTests {
 
     data class C<A, B : A>(override val t: B) : I2<B>
 
-    @Test(timeout=300_000)
-	fun recursiveTypeVariableResolution() {
+    @Test(timeout = 300_000)
+    fun recursiveTypeVariableResolution() {
         val factory = testDefaultFactoryNoEvolution()
         val instance = C<Collection<String>, List<String>>(emptyList())