summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/library/scala/collection/mutable/Ctrie.scala103
-rw-r--r--test/files/jvm/serialization.check4
-rw-r--r--test/files/jvm/serialization.scala7
-rw-r--r--test/files/run/ctries/lnode.scala5
4 files changed, 106 insertions, 13 deletions
diff --git a/src/library/scala/collection/mutable/Ctrie.scala b/src/library/scala/collection/mutable/Ctrie.scala
index d02e0ce178..84cceb44eb 100644
--- a/src/library/scala/collection/mutable/Ctrie.scala
+++ b/src/library/scala/collection/mutable/Ctrie.scala
@@ -6,12 +6,14 @@
** |/ **
\* */
-package scala.collection.mutable
+package scala.collection
+package mutable
import java.util.concurrent.atomic._
import collection.immutable.{ ListMap => ImmutableListMap }
+import generic._
import annotation.tailrec
import annotation.switch
@@ -425,7 +427,7 @@ extends MainNode[K, V] {
if (updmap.size > 1) new LNode(updmap)
else {
val (k, v) = updmap.iterator.next
- new TNode(k, v, k.hashCode) // create it tombed so that it gets compressed on subsequent accesses
+ new TNode(k, v, Ctrie.computeHash(k)) // create it tombed so that it gets compressed on subsequent accesses
}
}
def get(k: K) = listmap.get(k)
@@ -568,10 +570,26 @@ private[mutable] case class RDCSS_Descriptor[K, V](old: INode[K, V], expectedmai
}
-class Ctrie[K, V] private (r: AnyRef, rtupd: AtomicReferenceFieldUpdater[Ctrie[K, V], AnyRef])
+/** A concurrent hash-trie or Ctrie is a concurrent thread-safe lock-free
+ * implementation of a hash array mapped trie. It is used to implement the
+ * concurrent map abstraction. It has particularly scalable concurrent insert
+ * and remove operations and is memory-efficient. It supports O(1), atomic,
+ * lock-free snapshots which are used to implement linearizable lock-free size,
+ * iterator and clear operations. The cost of evaluating the (lazy) snapshot is
+ * distributed across subsequent updates, thus making snapshot evaluation horizontally scalable.
+ *
+ * @author Aleksandar Prokopec
+ * @since 2.10
+ */
+@SerialVersionUID(0L - 6402774413839597105L)
+final class Ctrie[K, V] private (r: AnyRef, rtupd: AtomicReferenceFieldUpdater[Ctrie[K, V], AnyRef])
extends ConcurrentMap[K, V]
+ with MapLike[K, V, Ctrie[K, V]]
+ with Serializable
{
- private val rootupdater = rtupd
+ import Ctrie.computeHash
+
+ private var rootupdater = rtupd
@volatile var root = r
def this() = this(
@@ -581,6 +599,31 @@ extends ConcurrentMap[K, V]
/* internal methods */
+ private def writeObject(out: java.io.ObjectOutputStream) {
+ val it = iterator
+ while (it.hasNext) {
+ val (k, v) = it.next()
+ out.writeObject(k)
+ out.writeObject(v)
+ }
+ out.writeObject(CtrieSerializationEnd)
+ }
+
+ private def readObject(in: java.io.ObjectInputStream) {
+ root = INode.newRootNode
+ rootupdater = AtomicReferenceFieldUpdater.newUpdater(classOf[Ctrie[K, V]], classOf[AnyRef], "root")
+
+ var obj: AnyRef = null
+ do {
+ obj = in.readObject()
+ if (obj != CtrieSerializationEnd) {
+ val k = obj.asInstanceOf[K]
+ val v = in.readObject().asInstanceOf[V]
+ update(k, v)
+ }
+ } while (obj != CtrieSerializationEnd)
+ }
+
@inline final def CAS_ROOT(ov: AnyRef, nv: AnyRef) = rootupdater.compareAndSet(this, ov, nv)
@inline final def RDCSS_READ_ROOT(abort: Boolean = false): INode[K, V] = {
@@ -623,10 +666,6 @@ extends ConcurrentMap[K, V]
} else false
}
- @inline private def computeHash(k: K): Int = {
- k.hashCode
- }
-
@tailrec private def inserthc(k: K, hc: Int, v: V) {
val r = RDCSS_READ_ROOT()
if (!r.rec_insert(k, v, hc, 0, null, r.gen, this)) inserthc(k, hc, v)
@@ -647,7 +686,7 @@ extends ConcurrentMap[K, V]
else res
}
- /*
+ /* slower:
//@tailrec
private def lookuphc(k: K, hc: Int): AnyRef = {
val r = RDCSS_READ_ROOT()
@@ -671,10 +710,21 @@ extends ConcurrentMap[K, V]
/* public methods */
+ override def empty: Ctrie[K, V] = new Ctrie[K, V]
+
@inline final def isReadOnly = rootupdater eq null
@inline final def nonReadOnly = rootupdater ne null
+ /** Returns a snapshot of this Ctrie.
+ * This operation is lock-free and linearizable.
+ *
+ * The snapshot is lazily updated - the first time some branch
+ * in the snapshot or this Ctrie are accessed, they are rewritten.
+ * This means that the work of rebuilding both the snapshot and this
+ * Ctrie is distributed across all the threads doing updates or accesses
+ * subsequent to the snapshot creation.
+ */
@tailrec final def snapshot(): Ctrie[K, V] = {
val r = RDCSS_READ_ROOT()
val expmain = r.GCAS_READ(this)
@@ -682,6 +732,18 @@ extends ConcurrentMap[K, V]
else snapshot()
}
+ /** Returns a read-only snapshot of this Ctrie.
+ * This operation is lock-free and linearizable.
+ *
+ * The snapshot is lazily updated - the first time some branch
+ * of this Ctrie are accessed, it is rewritten. The work of creating
+ * the snapshot is thus distributed across subsequent updates
+ * and accesses on this Ctrie by all threads.
+ * Note that the snapshot itself is never rewritten unlike when calling
+ * the `snapshot` method, but the obtained snapshot cannot be modified.
+ *
+ * This method is used by other methods such as `size` and `iterator`.
+ */
@tailrec final def readOnlySnapshot(): collection.Map[K, V] = {
val r = RDCSS_READ_ROOT()
val expmain = r.GCAS_READ(this)
@@ -760,11 +822,25 @@ extends ConcurrentMap[K, V]
if (nonReadOnly) readOnlySnapshot().iterator
else new CtrieIterator(this)
+ override def stringPrefix = "Ctrie"
+
}
-object Ctrie {
- val inodeupdater = AtomicReferenceFieldUpdater.newUpdater(classOf[INodeBase[_, _]], classOf[AnyRef], "mainnode")
+object Ctrie extends MutableMapFactory[Ctrie] {
+ val inodeupdater = AtomicReferenceFieldUpdater.newUpdater(classOf[INodeBase[_, _]], classOf[MainNode[_, _]], "mainnode")
+
+ implicit def canBuildFrom[K, V]: CanBuildFrom[Coll, (K, V), Ctrie[K, V]] = new MapCanBuildFrom[K, V]
+
+ def empty[K, V]: Ctrie[K, V] = new Ctrie[K, V]
+
+ @inline final def computeHash[K](k: K): Int = {
+ var hcode = k.hashCode
+ hcode = hcode * 0x9e3775cd
+ hcode = java.lang.Integer.reverseBytes(hcode)
+ hcode * 0x9e3775cd
+ }
+
}
@@ -877,6 +953,11 @@ private[mutable] class CtrieIterator[K, V](ct: Ctrie[K, V], mustInit: Boolean =
private[mutable] object RestartException extends util.control.ControlThrowable
+/** Only used for ctrie serialization. */
+@SerialVersionUID(0L - 7237891413820527142L)
+private[mutable] case object CtrieSerializationEnd
+
+
private[mutable] object Debug {
import collection._
diff --git a/test/files/jvm/serialization.check b/test/files/jvm/serialization.check
index f58f763a76..cdfc100e0d 100644
--- a/test/files/jvm/serialization.check
+++ b/test/files/jvm/serialization.check
@@ -192,6 +192,10 @@ x = TreeSet(1, 2, 3)
y = TreeSet(1, 2, 3)
x equals y: true, y equals x: true
+x = Ctrie(1 -> one, 2 -> two, 3 -> three)
+y = Ctrie(1 -> one, 2 -> two, 3 -> three)
+x equals y: true, y equals x: true
+
x = xml:src="hello"
y = xml:src="hello"
x equals y: true, y equals x: true
diff --git a/test/files/jvm/serialization.scala b/test/files/jvm/serialization.scala
index 73bed2d46b..4e1ff368ab 100644
--- a/test/files/jvm/serialization.scala
+++ b/test/files/jvm/serialization.scala
@@ -286,7 +286,7 @@ object Test3_mutable {
import scala.collection.mutable.{
ArrayBuffer, ArrayBuilder, ArraySeq, ArrayStack, BitSet, DoubleLinkedList,
HashMap, HashSet, History, LinkedList, ListBuffer, Publisher, Queue,
- Stack, StringBuilder, WrappedArray, TreeSet}
+ Stack, StringBuilder, WrappedArray, TreeSet, Ctrie}
// in alphabetic order
try {
@@ -385,6 +385,11 @@ object Test3_mutable {
val ts1 = TreeSet[Int]() ++= Array(1, 2, 3)
val _ts1: TreeSet[Int] = read(write(ts1))
check(ts1, _ts1)
+
+ // Ctrie
+ val ct1 = Ctrie[Int, String]() ++= Array(1 -> "one", 2 -> "two", 3 -> "three")
+ val _ct1: Ctrie[Int, String] = read(write(ct1))
+ check(ct1, _ct1)
}
catch {
case e: Exception =>
diff --git a/test/files/run/ctries/lnode.scala b/test/files/run/ctries/lnode.scala
index 28da4cc62f..88cbeed1f6 100644
--- a/test/files/run/ctries/lnode.scala
+++ b/test/files/run/ctries/lnode.scala
@@ -25,7 +25,10 @@ object LNodeSpec extends Spec {
"remove elements with the same hash codes" in {
val ct = new Ctrie[DumbHash, Int]
for (i <- 0 until initsz) ct.update(new DumbHash(i), i)
- for (i <- 0 until initsz) assert(ct.remove(new DumbHash(i)) == Some(i))
+ for (i <- 0 until initsz) {
+ val remelem = ct.remove(new DumbHash(i))
+ assert(remelem == Some(i), "removing " + i + " yields " + remelem)
+ }
for (i <- 0 until initsz) assert(ct.get(new DumbHash(i)) == None)
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-30 01:51:16 +0000