I have a tree in my application from which I need to periodically take a snapshot. I've been using locks so far, but after running into a few deadlock issues I decided to create a lock-free implementation. I'm using a persistent Map implementation together with an actor. My goals were:
- Consistent snapshot from the
Mapin a concurrent environment - Flattened tree structure
- Thread safety when writing
I have completed a preliminary implementation, but I'm not 100% sure that my implementation is idiomatic. I'm not that concerned with complexity, as writes won't be often, but I really need the consistent snapshot and the thread safety. I'd like some guidance on the
- overall design
- idiomatic use of coroutines / persistent data structures
- possible design oversights or thread safety issues
- anything that I missed (it happened in the past that I wrote something and it turned out that there is a well-known algorithm that does the same but better)
So without further ado, this is the implementation:
// primary interfaces
import kotlinx.collections.immutable.PersistentList
import kotlinx.collections.immutable.persistentListOf
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.math.max
interface Tree<D : Any> {
// this is the flattened tree
val descendants: Sequence<TreeNode<D>>
val children: Sequence<TreeNode<D>>
// children can only be created through a reference to a Tree or a TreeNode
fun createChild(data: D? = null): TreeNode<D>
companion object {
fun <D : Any> create(): Tree<D> {
return ConcurrentTree()
}
}
}
interface TreeNode<D : Any> {
val data: D?
val parent: TreeNode<D>?
val hasParent: Boolean
get() = parent != null
val children: Sequence<TreeNode<D>>
fun createChild(data: D? = null): TreeNode<D>
fun remove()
}
The TreeNode implementations look like this:
open class ConcurrentTreeNode<D : Any>(
override val parent: ConcurrentTreeNode<D>? = null,
override val data: D? = null,
private val tree: ConcurrentTree<D>
) : TreeNode<D> {
override val children: Sequence<TreeNode<D>>
get() = tree.children.filter { it.parent == this }
override fun createChild(data: D?): ConcurrentTreeNode<D> {
val parent = this
return ConcurrentTreeNode(
parent = this,
data = data,
tree = tree
).apply {
tree.addChildTo(parent, this)
}
}
override fun remove() {
tree.remove(this)
}
override fun toString(): String {
return "TreeNode(data=$data)"
}
}
private class RootNode<D : Any>(
parent: ConcurrentTreeNode<D>? = null,
data: D? = null,
tree: ConcurrentTree<D>
) : ConcurrentTreeNode<D>(
parent = parent,
data = data,
tree = tree
) {
override fun remove() {
throw RuntimeException("A root node can't be removed.")
}
}
And this is the Tree implementation:
class ConcurrentTree<D : Any> : Tree<D> {
sealed class Message {
data class AddChildTo<D : Any>(
val parent: TreeNode<D>,
val child: TreeNode<D>
) : Message()
data class DeleteChild<D : Any>(
val child: TreeNode<D>
) : Message()
}
private val treeScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
private val actor = treeScope.actor<Message> {
for (msg in channel) { // iterate over incoming messages
when (msg) {
is AddChildTo<*> -> {
val (parent, child) = msg
val lastChildIdx = backend.indexOfLast { it.parent == parent }
val parentIdx = backend.indexOf(parent)
val idx = max(lastChildIdx, parentIdx) + 1
backend = if (idx == backend.lastIndex) {
backend.add(child as TreeNode<D>)
} else {
backend.add(idx, child as TreeNode<D>)
}
}
is Message.DeleteChild<*> -> {
val parent = msg.child.parent
val child = msg.child
val toDelete = backend
.dropWhile { it !== child }
.takeWhile { it === child || it.parent !== parent }
backend = backend.removeAll(toDelete)
}
}
}
}
private val root = RootNode(
parent = null,
data = null,
tree = this
)
private var backend: PersistentList<TreeNode<D>> = persistentListOf(root)
override val descendants: Sequence<TreeNode<D>>
get() = backend.asSequence().drop(1)
override val children: Sequence<TreeNode<D>>
get() = root.children
override fun createChild(data: D?) = root.createChild(data)
fun addChildTo(parent: TreeNode<D>, child: TreeNode<D>) {
treeScope.launch {
actor.send(AddChildTo(parent, child))
}
}
fun remove(node: TreeNode<D>) {
treeScope.launch {
actor.send(Message.DeleteChild(node))
}
}
override fun toString(): String {
return "Tree(descendants=${descendants.joinToString()})"
}
}
Complexity of operations is O(n), but more importantly snapshot creation is O(1), as I'm using a PersistentList as a backend for the whole thing.
The operations (apart from the snapshot) are eventually consistent, but this doesn't matter in my case, I only want to see an operation either completed successfully or the previous state.
Can I do better? Is this idiomatic Kotlin code?
