Merge pull request #338 from modelix/bugfix/ignite-nested-transactions

fix(model-server): writing to a branch using the v1 API failed

.gitignore

@@ -1,6 +1,7 @@
 .gradle/
 /build/
 /*/build/
+/*/ignite/
 .DS_Store
 .gradletasknamecache
 .idea/

commitlint.config.js

@@ -32,4 +32,7 @@ module.exports = {
     // No need to restrict the body line length. That only gives issues with URLs etc.
     "body-max-line-length": [0, 'always']
   },
+  ignores: [
+    (message) => message.includes('skip-lint')
+  ],
 };

model-server/build.gradle.kts

@@ -62,6 +62,7 @@ dependencies {
     testImplementation(libs.junit)
     testImplementation(libs.cucumber.java)
     testImplementation(libs.ktor.server.test.host)
+    testImplementation(libs.kotlin.coroutines.test)
     testImplementation(kotlin("test"))
     testImplementation(project(":modelql-untyped"))
 }

model-server/src/main/kotlin/org/modelix/model/server/Main.kt

@@ -56,11 +56,11 @@ import org.modelix.model.server.store.IStoreClient
 import org.modelix.model.server.store.IgniteStoreClient
 import org.modelix.model.server.store.InMemoryStoreClient
 import org.modelix.model.server.store.LocalModelClient
+import org.modelix.model.server.store.loadDump
+import org.modelix.model.server.store.writeDump
 import org.modelix.model.server.templates.PageWithMenuBar
 import org.slf4j.LoggerFactory
 import java.io.File
-import java.io.FileReader
-import java.io.FileWriter
 import java.io.IOException
 import java.nio.charset.StandardCharsets
 import java.time.Duration
@@ -111,18 +111,16 @@ object Main {
                 }
                 storeClient = InMemoryStoreClient()
                 if (cmdLineArgs.dumpInName != null) {
-                    val file = File(cmdLineArgs.dumpInName)
-                    val keys = storeClient.load(FileReader(file))
-                    println(
-                        "Values loaded from " + file.absolutePath + " (" + keys + ")",
-                    )
+                    val file = File(cmdLineArgs.dumpInName!!)
+                    val keys = storeClient.loadDump(file)
+                    println("Values loaded from " + file.absolutePath + " (" + keys + ")")
                 }
                 if (cmdLineArgs.dumpOutName != null) {
                     Runtime.getRuntime()
                         .addShutdownHook(
                             DumpOutThread(
                                 storeClient,
-                                cmdLineArgs.dumpOutName,
+                                cmdLineArgs.dumpOutName ?: "dump",
                             ),
                         )
                 }
@@ -243,24 +241,14 @@ object Main {
         }
     }
 
-    private class DumpOutThread internal constructor(inMemoryStoreClient: InMemoryStoreClient, dumpName: String?) :
+    private class DumpOutThread internal constructor(storeClient: IStoreClient, dumpName: String) :
         Thread(
             Runnable {
-                var fw: FileWriter? = null
                 try {
-                    fw = FileWriter(File(dumpName))
-                    inMemoryStoreClient.dump(fw!!)
+                    storeClient.writeDump(File(dumpName))
                     println("[Saved memory store into $dumpName]")
                 } catch (e: IOException) {
                     e.printStackTrace()
-                } finally {
-                    if (fw != null) {
-                        try {
-                            fw!!.close()
-                        } catch (e: IOException) {
-                            e.printStackTrace()
-                        }
-                    }
                 }
             },
         )

model-server/src/main/kotlin/org/modelix/model/server/store/IStoreClient.kt

@@ -19,12 +19,15 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
 import org.modelix.model.IKeyListener
+import java.io.File
+import java.io.IOException
 import kotlin.time.Duration.Companion.seconds
 
-interface IStoreClient {
+interface IStoreClient : AutoCloseable {
     operator fun get(key: String): String?
     fun getAll(keys: List<String>): List<String?>
     fun getAll(keys: Set<String>): Map<String, String?>
+    fun getAll(): Map<String, String?>
     fun put(key: String, value: String?, silent: Boolean = false)
     fun putAll(entries: Map<String, String?>, silent: Boolean = false)
     fun listen(key: String, listener: IKeyListener)
@@ -79,3 +82,30 @@ suspend fun pollEntry(storeClient: IStoreClient, key: String, lastKnownValue: St
     }
     return result
 }
+
+fun IStoreClient.loadDump(file: File): Int {
+    var n = 0
+    file.useLines { lines ->
+        val entries = lines.associate { line ->
+            val parts = line.split("#".toRegex(), limit = 2)
+            n++
+            parts[0] to parts[1]
+        }
+        putAll(entries, silent = true)
+    }
+    return n
+}
+
+@Synchronized
+@Throws(IOException::class)
+fun IStoreClient.writeDump(file: File) {
+    file.writer().use { writer ->
+        for ((key, value) in getAll()) {
+            if (value == null) continue
+            writer.append(key)
+            writer.append("#")
+            writer.append(value)
+            writer.append("\n")
+        }
+    }
+}

model-server/src/main/kotlin/org/modelix/model/server/store/IgniteStoreClient.kt

@@ -14,23 +14,28 @@
  */
 package org.modelix.model.server.store
 
-import com.google.common.collect.MultimapBuilder
+import mu.KotlinLogging
 import org.apache.ignite.Ignite
 import org.apache.ignite.IgniteCache
 import org.apache.ignite.Ignition
 import org.modelix.model.IKeyListener
+import org.modelix.model.persistent.HashUtil
 import java.io.File
 import java.io.FileReader
 import java.io.IOException
 import java.util.*
-import java.util.concurrent.Executors
 import java.util.stream.Collectors
 
-class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
-    private val ignite: Ignite
+private val LOG = KotlinLogging.logger { }
+
+class IgniteStoreClient(jdbcConfFile: File? = null, inmemory: Boolean = false) : IStoreClient, AutoCloseable {
+    private val ENTRY_CHANGED_TOPIC = "entryChanged"
+    private lateinit var ignite: Ignite
     private val cache: IgniteCache<String, String?>
-    private val timer = Executors.newScheduledThreadPool(1)
-    private val listeners = MultimapBuilder.hashKeys().hashSetValues().build<String, IKeyListener>()
+    private val changeNotifier = ChangeNotifier(this)
+    private val pendingChangeMessages = PendingChangeMessages {
+        ignite.message().send(ENTRY_CHANGED_TOPIC, it)
+    }
 
     /**
      * Istantiate an IgniteStoreClient
@@ -63,11 +68,18 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
                 )
             }
         }
-        ignite = Ignition.start(javaClass.getResource("ignite.xml"))
+        ignite = Ignition.start(javaClass.getResource(if (inmemory) "ignite-inmemory.xml" else "ignite.xml"))
         cache = ignite.getOrCreateCache("model")
         //        timer.scheduleAtFixedRate(() -> {
         //            System.out.println("stats: " + cache.metrics());
         //        }, 10, 10, TimeUnit.SECONDS);
+
+        ignite.message().localListen(ENTRY_CHANGED_TOPIC) { nodeId: UUID?, key: Any? ->
+            if (key is String) {
+                changeNotifier.notifyListeners(key)
+            }
+            true
+        }
     }
 
     override fun get(key: String): String? {
@@ -83,51 +95,38 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
         return cache.getAll(keys)
     }
 
+    override fun getAll(): Map<String, String?> {
+        return cache.associate { it.key to it.value }
+    }
+
     override fun put(key: String, value: String?, silent: Boolean) {
         putAll(Collections.singletonMap(key, value), silent)
     }
 
     override fun putAll(entries: Map<String, String?>, silent: Boolean) {
         val deletes = entries.filterValues { it == null }
         val puts = entries.filterValues { it != null }
-        if (deletes.isNotEmpty()) cache.removeAll(deletes.keys)
-        if (puts.isNotEmpty()) cache.putAll(puts)
-        if (!silent) {
-            for ((key, value) in entries) {
-                ignite.message().send(key, value ?: IKeyListener.NULL_VALUE)
+        runTransaction {
+            if (deletes.isNotEmpty()) cache.removeAll(deletes.keys)
+            if (puts.isNotEmpty()) cache.putAll(puts)
+            if (!silent) {
+                for (key in entries.keys) {
+                    if (HashUtil.isSha256(key)) continue
+                    pendingChangeMessages.entryChanged(key)
+                }
             }
         }
     }
 
     override fun listen(key: String, listener: IKeyListener) {
-        synchronized(listeners) {
-            val wasSubscribed = listeners.containsKey(key)
-            listeners.put(key, listener)
-            if (!wasSubscribed) {
-                ignite.message()
-                    .localListen(
-                        key,
-                    ) { nodeId: UUID?, value: Any? ->
-                        if (value is String) {
-                            synchronized(listeners) {
-                                for (l in listeners[key].toList()) {
-                                    try {
-                                        l.changed(key, if (value == IKeyListener.NULL_VALUE) null else value)
-                                    } catch (ex: Exception) {
-                                        println(ex.message)
-                                        ex.printStackTrace()
-                                    }
-                                }
-                            }
-                        }
-                        true
-                    }
-            }
-        }
+        // Entries where the key is the SHA hash over the value are not expected to change and listening is unnecessary.
+        require(!HashUtil.isSha256(key)) { "Listener for $key will never get notified." }
+
+        changeNotifier.addListener(key, listener)
     }
 
     override fun removeListener(key: String, listener: IKeyListener) {
-        synchronized(listeners) { listeners.remove(key, listener) }
+        changeNotifier.removeListener(key, listener)
     }
 
     override fun generateId(key: String): Long {
@@ -136,14 +135,83 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
 
     override fun <T> runTransaction(body: () -> T): T {
         val transactions = ignite.transactions()
-        transactions.txStart().use { tx ->
-            val result = body()
-            tx.commit()
-            return result
+        if (transactions.tx() == null) {
+            transactions.txStart().use { tx ->
+                val result = body()
+                tx.commit()
+                pendingChangeMessages.flushChangeMessages()
+                return result
+            }
+        } else {
+            // already in a transaction
+            return body()
         }
     }
 
     fun dispose() {
         ignite.close()
     }
+
+    override fun close() {
+        dispose()
+    }
+}
+
+class PendingChangeMessages(private val notifier: (String) -> Unit) {
+    private val pendingChangeMessages = Collections.synchronizedSet(HashSet<String>())
+
+    @Synchronized
+    fun flushChangeMessages() {
+        for (pendingChangeMessage in pendingChangeMessages) {
+            notifier(pendingChangeMessage)
+        }
+        pendingChangeMessages.clear()
+    }
+
+    @Synchronized
+    fun entryChanged(key: String) {
+        pendingChangeMessages += key
+    }
+}
+
+class ChangeNotifier(val store: IStoreClient) {
+    private val changeNotifiers = HashMap<String, EntryChangeNotifier>()
+
+    @Synchronized
+    fun notifyListeners(key: String) {
+        changeNotifiers[key]?.notifyIfChanged()
+    }
+
+    @Synchronized
+    fun addListener(key: String, listener: IKeyListener) {
+        changeNotifiers.getOrPut(key) { EntryChangeNotifier(key) }.listeners.add(listener)
+    }
+
+    @Synchronized
+    fun removeListener(key: String, listener: IKeyListener) {
+        val notifier = changeNotifiers[key] ?: return
+        notifier.listeners.remove(listener)
+        if (notifier.listeners.isEmpty()) {
+            changeNotifiers.remove(key)
+        }
+    }
+
+    private inner class EntryChangeNotifier(val key: String) {
+        val listeners = HashSet<IKeyListener>()
+        private var lastNotifiedValue: String? = null
+
+        fun notifyIfChanged() {
+            val value = store.get(key)
+            if (value == lastNotifiedValue) return
+            lastNotifiedValue = value
+
+            for (listener in listeners) {
+                try {
+                    listener.changed(key, value)
+                } catch (ex: Exception) {
+                    LOG.error("Exception in listener of $key", ex)
+                }
+            }
+        }
+    }
 }