diff options
author | Aleksandar Prokopec <axel22@gmail.com> | 2012-02-01 19:54:50 +0100 |
---|---|---|
committer | Aleksandar Prokopec <axel22@gmail.com> | 2012-02-01 19:54:50 +0100 |
commit | 5fe2d8b109abf3ff3e2d82dd4f248200846795c3 (patch) | |
tree | b50c45368759198fdcd0521016138a3fd7019322 | |
parent | 8aa87f15e3887dbeb1a39bfea002b56cf68c445a (diff) | |
download | scala-5fe2d8b109abf3ff3e2d82dd4f248200846795c3.tar.gz scala-5fe2d8b109abf3ff3e2d82dd4f248200846795c3.tar.bz2 scala-5fe2d8b109abf3ff3e2d82dd4f248200846795c3.zip |
Add the Ctrie concurrent map implementation.
Ctrie is a scalable concurrent map implementation that supports
constant time lock-free lazy snapshots.
Due to the well-known private volatile field problem, atomic
reference updaters cannot be used efficiently in Scala yet.
For this reason, 4 java files had to be included as well.
None of these pollute the namespace, as most of the classes
are private.
Unit tests and a scalacheck check is also included.
-rw-r--r-- | src/library/scala/collection/mutable/BasicNode.java | 20 | ||||
-rw-r--r-- | src/library/scala/collection/mutable/Ctrie.scala | 906 | ||||
-rw-r--r-- | src/library/scala/collection/mutable/Gen.java | 18 | ||||
-rw-r--r-- | src/library/scala/collection/mutable/INodeBase.java | 35 | ||||
-rw-r--r-- | src/library/scala/collection/mutable/MainNode.java | 36 | ||||
-rw-r--r-- | test/files/run/ctries/DumbHash.scala | 14 | ||||
-rw-r--r-- | test/files/run/ctries/Wrap.scala | 9 | ||||
-rw-r--r-- | test/files/run/ctries/concmap.scala | 169 | ||||
-rw-r--r-- | test/files/run/ctries/iterator.scala | 279 | ||||
-rw-r--r-- | test/files/run/ctries/lnode.scala | 58 | ||||
-rw-r--r-- | test/files/run/ctries/main.scala | 45 | ||||
-rw-r--r-- | test/files/run/ctries/snapshot.scala | 267 | ||||
-rw-r--r-- | test/files/scalacheck/Ctrie.scala | 199 |
13 files changed, 2055 insertions, 0 deletions
diff --git a/src/library/scala/collection/mutable/BasicNode.java b/src/library/scala/collection/mutable/BasicNode.java new file mode 100644 index 0000000000..b934aed24f --- /dev/null +++ b/src/library/scala/collection/mutable/BasicNode.java @@ -0,0 +1,20 @@ +/* __ *\ +** ________ ___ / / ___ Scala API ** +** / __/ __// _ | / / / _ | (c) 2003-2012, LAMP/EPFL ** +** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** +** /____/\___/_/ |_/____/_/ | | ** +** |/ ** +\* */ + +package scala.collection.mutable; + + + + + + +abstract class BasicNode { + + public abstract String string(int lev); + +}
\ No newline at end of file diff --git a/src/library/scala/collection/mutable/Ctrie.scala b/src/library/scala/collection/mutable/Ctrie.scala new file mode 100644 index 0000000000..d02e0ce178 --- /dev/null +++ b/src/library/scala/collection/mutable/Ctrie.scala @@ -0,0 +1,906 @@ +/* __ *\ +** ________ ___ / / ___ Scala API ** +** / __/ __// _ | / / / _ | (c) 2003-2012, LAMP/EPFL ** +** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** +** /____/\___/_/ |_/____/_/ | | ** +** |/ ** +\* */ + +package scala.collection.mutable + + + +import java.util.concurrent.atomic._ +import collection.immutable.{ ListMap => ImmutableListMap } +import annotation.tailrec +import annotation.switch + + + +private[mutable] final class INode[K, V](bn: MainNode[K, V], g: Gen) extends INodeBase[K, V](g) { + import INodeBase._ + + WRITE(bn) + + def this(g: Gen) = this(null, g) + + @inline final def WRITE(nval: MainNode[K, V]) = INodeBase.updater.set(this, nval) + + @inline final def CAS(old: MainNode[K, V], n: MainNode[K, V]) = INodeBase.updater.compareAndSet(this, old, n) + + @inline final def GCAS_READ(ct: Ctrie[K, V]): MainNode[K, V] = { + val m = /*READ*/mainnode + val prevval = /*READ*/m.prev + if (prevval eq null) m + else GCAS_Complete(m, ct) + } + + @tailrec private def GCAS_Complete(m: MainNode[K, V], ct: Ctrie[K, V]): MainNode[K, V] = if (m eq null) null else { + // complete the GCAS + val prev = /*READ*/m.prev + val ctr = ct.RDCSS_READ_ROOT(true) + + prev match { + case null => + m + case fn: FailedNode[_, _] => // try to commit to previous value + if (CAS(m, fn.prev)) fn.prev + else GCAS_Complete(/*READ*/mainnode, ct) + case vn: MainNode[_, _] => + // Assume that you've read the root from the generation G. + // Assume that the snapshot algorithm is correct. + // ==> you can only reach nodes in generations <= G. + // ==> `gen` is <= G. + // We know that `ctr.gen` is >= G. + // ==> if `ctr.gen` = `gen` then they are both equal to G. + // ==> otherwise, we know that either `ctr.gen` > G, `gen` < G, + // or both + if ((ctr.gen eq gen) && ct.nonReadOnly) { + // try to commit + if (m.CAS_PREV(prev, null)) m + else GCAS_Complete(m, ct) + } else { + // try to abort + m.CAS_PREV(prev, new FailedNode(prev)) + GCAS_Complete(/*READ*/mainnode, ct) + } + } + } + + @inline final def GCAS(old: MainNode[K, V], n: MainNode[K, V], ct: Ctrie[K, V]): Boolean = { + n.WRITE_PREV(old) + if (CAS(old, n)) { + GCAS_Complete(n, ct) + /*READ*/n.prev eq null + } else false + } + + @inline private def inode(cn: MainNode[K, V]) = { + val nin = new INode[K, V](gen) + nin.WRITE(cn) + nin + } + + @inline final def copyToGen(ngen: Gen, ct: Ctrie[K, V]) = { + val nin = new INode[K, V](ngen) + val main = GCAS_READ(ct) + nin.WRITE(main) + nin + } + + /** Inserts a key value pair, overwriting the old pair if the keys match. + * + * @return true if successful, false otherwise + */ + @tailrec final def rec_insert(k: K, v: V, hc: Int, lev: Int, parent: INode[K, V], startgen: Gen, ct: Ctrie[K, V]): Boolean = { + val m = GCAS_READ(ct) // use -Yinline! + + m match { + case cn: CNode[K, V] => // 1) a multiway node + val idx = (hc >>> lev) & 0x1f + val flag = 1 << idx + val bmp = cn.bitmap + val mask = flag - 1 + val pos = Integer.bitCount(bmp & mask) + if ((bmp & flag) != 0) { + // 1a) insert below + cn.array(pos) match { + case in: INode[K, V] => + if (startgen eq in.gen) in.rec_insert(k, v, hc, lev + 5, this, startgen, ct) + else { + if (GCAS(cn, cn.renewed(startgen, ct), ct)) rec_insert(k, v, hc, lev, parent, startgen, ct) + else false + } + case sn: SNode[K, V] => + if (sn.hc == hc && sn.k == k) GCAS(cn, cn.updatedAt(pos, new SNode(k, v, hc), gen), ct) + else { + val rn = if (cn.gen eq gen) cn else cn.renewed(gen, ct) + val nn = rn.updatedAt(pos, inode(CNode.dual(sn, sn.hc, new SNode(k, v, hc), hc, lev + 5, gen)), gen) + GCAS(cn, nn, ct) + } + } + } else { + val rn = if (cn.gen eq gen) cn else cn.renewed(gen, ct) + val ncnode = rn.insertedAt(pos, flag, new SNode(k, v, hc), gen) + GCAS(cn, ncnode, ct) + } + case tn: TNode[K, V] => + clean(parent, ct, lev - 5) + false + case ln: LNode[K, V] => // 3) an l-node + val nn = ln.inserted(k, v) + GCAS(ln, nn, ct) + } + } + + /** Inserts a new key value pair, given that a specific condition is met. + * + * @param cond null - don't care if the key was there; KEY_ABSENT - key wasn't there; KEY_PRESENT - key was there; other value `v` - key must be bound to `v` + * @return null if unsuccessful, Option[V] otherwise (indicating previous value bound to the key) + */ + @tailrec final def rec_insertif(k: K, v: V, hc: Int, cond: AnyRef, lev: Int, parent: INode[K, V], startgen: Gen, ct: Ctrie[K, V]): Option[V] = { + val m = GCAS_READ(ct) // use -Yinline! + + m match { + case cn: CNode[K, V] => // 1) a multiway node + val idx = (hc >>> lev) & 0x1f + val flag = 1 << idx + val bmp = cn.bitmap + val mask = flag - 1 + val pos = Integer.bitCount(bmp & mask) + if ((bmp & flag) != 0) { + // 1a) insert below + cn.array(pos) match { + case in: INode[K, V] => + if (startgen eq in.gen) in.rec_insertif(k, v, hc, cond, lev + 5, this, startgen, ct) + else { + if (GCAS(cn, cn.renewed(startgen, ct), ct)) rec_insertif(k, v, hc, cond, lev, parent, startgen, ct) + else null + } + case sn: SNode[K, V] => cond match { + case null => + if (sn.hc == hc && sn.k == k) { + if (GCAS(cn, cn.updatedAt(pos, new SNode(k, v, hc), gen), ct)) Some(sn.v) else null + } else { + val rn = if (cn.gen eq gen) cn else cn.renewed(gen, ct) + val nn = rn.updatedAt(pos, inode(CNode.dual(sn, sn.hc, new SNode(k, v, hc), hc, lev + 5, gen)), gen) + if (GCAS(cn, nn, ct)) None + else null + } + case INode.KEY_ABSENT => + if (sn.hc == hc && sn.k == k) Some(sn.v) + else { + val rn = if (cn.gen eq gen) cn else cn.renewed(gen, ct) + val nn = rn.updatedAt(pos, inode(CNode.dual(sn, sn.hc, new SNode(k, v, hc), hc, lev + 5, gen)), gen) + if (GCAS(cn, nn, ct)) None + else null + } + case INode.KEY_PRESENT => + if (sn.hc == hc && sn.k == k) { + if (GCAS(cn, cn.updatedAt(pos, new SNode(k, v, hc), gen), ct)) Some(sn.v) else null + } else None + case otherv: V => + if (sn.hc == hc && sn.k == k && sn.v == otherv) { + if (GCAS(cn, cn.updatedAt(pos, new SNode(k, v, hc), gen), ct)) Some(sn.v) else null + } else None + } + } + } else cond match { + case null | INode.KEY_ABSENT => + val rn = if (cn.gen eq gen) cn else cn.renewed(gen, ct) + val ncnode = rn.insertedAt(pos, flag, new SNode(k, v, hc), gen) + if (GCAS(cn, ncnode, ct)) None else null + case INode.KEY_PRESENT => None + case otherv: V => None + } + case sn: TNode[K, V] => + clean(parent, ct, lev - 5) + null + case ln: LNode[K, V] => // 3) an l-node + @inline def insertln() = { + val nn = ln.inserted(k, v) + GCAS(ln, nn, ct) + } + cond match { + case null => + val optv = ln.get(k) + if (insertln()) optv else null + case INode.KEY_ABSENT => + ln.get(k) match { + case None => if (insertln()) None else null + case optv => optv + } + case INode.KEY_PRESENT => + ln.get(k) match { + case Some(v0) => if (insertln()) Some(v0) else null + case None => None + } + case otherv: V => + ln.get(k) match { + case Some(v0) if v0 == otherv => if (insertln()) Some(otherv) else null + case _ => None + } + } + } + } + + /** Looks up the value associated with the key. + * + * @return null if no value has been found, RESTART if the operation wasn't successful, or any other value otherwise + */ + @tailrec final def rec_lookup(k: K, hc: Int, lev: Int, parent: INode[K, V], startgen: Gen, ct: Ctrie[K, V]): AnyRef = { + val m = GCAS_READ(ct) // use -Yinline! + + m match { + case cn: CNode[K, V] => // 1) a multinode + val idx = (hc >>> lev) & 0x1f + val flag = 1 << idx + val bmp = cn.bitmap + if ((bmp & flag) == 0) null // 1a) bitmap shows no binding + else { // 1b) bitmap contains a value - descend + val pos = if (bmp == 0xffffffff) idx else Integer.bitCount(bmp & (flag - 1)) + val sub = cn.array(pos) + sub match { + case in: INode[K, V] => + if (ct.isReadOnly || (startgen eq in.gen)) in.rec_lookup(k, hc, lev + 5, this, startgen, ct) + else { + if (GCAS(cn, cn.renewed(startgen, ct), ct)) rec_lookup(k, hc, lev, parent, startgen, ct) + else return RESTART // used to be throw RestartException + } + case sn: SNode[K, V] => // 2) singleton node + if (sn.hc == hc && sn.k == k) sn.v.asInstanceOf[AnyRef] + else null + } + } + case tn: TNode[K, V] => // 3) non-live node + def cleanReadOnly(tn: TNode[K, V]) = if (ct.nonReadOnly) { + clean(parent, ct, lev - 5) + RESTART // used to be throw RestartException + } else { + if (tn.hc == hc && tn.k == k) tn.v.asInstanceOf[AnyRef] + else null + } + cleanReadOnly(tn) + case ln: LNode[K, V] => // 5) an l-node + ln.get(k).asInstanceOf[Option[AnyRef]].orNull + } + } + + /** Removes the key associated with the given value. + * + * @param v if null, will remove the key irregardless of the value; otherwise removes only if binding contains that exact key and value + * @return null if not successful, an Option[V] indicating the previous value otherwise + */ + final def rec_remove(k: K, v: V, hc: Int, lev: Int, parent: INode[K, V], startgen: Gen, ct: Ctrie[K, V]): Option[V] = { + val m = GCAS_READ(ct) // use -Yinline! + + m match { + case cn: CNode[K, V] => + val idx = (hc >>> lev) & 0x1f + val bmp = cn.bitmap + val flag = 1 << idx + if ((bmp & flag) == 0) None + else { + val pos = Integer.bitCount(bmp & (flag - 1)) + val sub = cn.array(pos) + val res = sub match { + case in: INode[K, V] => + if (startgen eq in.gen) in.rec_remove(k, v, hc, lev + 5, this, startgen, ct) + else { + if (GCAS(cn, cn.renewed(startgen, ct), ct)) rec_remove(k, v, hc, lev, parent, startgen, ct) + else null + } + case sn: SNode[K, V] => + if (sn.hc == hc && sn.k == k && (v == null || sn.v == v)) { + val ncn = cn.removedAt(pos, flag, gen).toContracted(lev) + if (GCAS(cn, ncn, ct)) Some(sn.v) else null + } else None + } + + if (res == None || (res eq null)) res + else { + @tailrec def cleanParent(nonlive: AnyRef) { + val pm = parent.GCAS_READ(ct) + pm match { + case cn: CNode[K, V] => + val idx = (hc >>> (lev - 5)) & 0x1f + val bmp = cn.bitmap + val flag = 1 << idx + if ((bmp & flag) == 0) {} // somebody already removed this i-node, we're done + else { + val pos = Integer.bitCount(bmp & (flag - 1)) + val sub = cn.array(pos) + if (sub eq this) nonlive match { + case tn: TNode[K, V] => + val ncn = cn.updatedAt(pos, tn.copyUntombed, gen).toContracted(lev - 5) + if (!parent.GCAS(cn, ncn, ct)) + if (ct.RDCSS_READ_ROOT().gen == startgen) cleanParent(nonlive) + } + } + case _ => // parent is no longer a cnode, we're done + } + } + + if (parent ne null) { // never tomb at root + val n = GCAS_READ(ct) + if (n.isInstanceOf[TNode[_, _]]) + cleanParent(n) + } + + res + } + } + case tn: TNode[K, V] => + clean(parent, ct, lev - 5) + null + case ln: LNode[K, V] => + if (v == null) { + val optv = ln.get(k) + val nn = ln.removed(k) + if (GCAS(ln, nn, ct)) optv else null + } else ln.get(k) match { + case optv @ Some(v0) if v0 == v => + val nn = ln.removed(k) + if (GCAS(ln, nn, ct)) optv else null + case _ => None + } + } + } + + private def clean(nd: INode[K, V], ct: Ctrie[K, V], lev: Int) { + val m = nd.GCAS_READ(ct) + m match { + case cn: CNode[K, V] => nd.GCAS(cn, cn.toCompressed(ct, lev, gen), ct) + case _ => + } + } + + final def isNullInode(ct: Ctrie[K, V]) = GCAS_READ(ct) eq null + + /* this is a quiescent method! */ + def string(lev: Int) = "%sINode -> %s".format(" " * lev, mainnode match { + case null => "<null>" + case tn: TNode[_, _] => "TNode(%s, %s, %d, !)".format(tn.k, tn.v, tn.hc) + case cn: CNode[_, _] => cn.string(lev) + case ln: LNode[_, _] => ln.string(lev) + case x => "<elem: %s>".format(x) + }) + +} + + +private[mutable] object INode { + val KEY_PRESENT = new AnyRef + val KEY_ABSENT = new AnyRef + + def newRootNode[K, V] = { + val gen = new Gen + val cn = new CNode[K, V](0, new Array(0), gen) + new INode[K, V](cn, gen) + } +} + + +private[mutable] final class FailedNode[K, V](p: MainNode[K, V]) extends MainNode[K, V] { + WRITE_PREV(p) + + def string(lev: Int) = throw new UnsupportedOperationException + + override def toString = "FailedNode(%s)".format(p) +} + + +private[mutable] trait KVNode[K, V] { + def kvPair: (K, V) +} + + +private[mutable] final class SNode[K, V](final val k: K, final val v: V, final val hc: Int) +extends BasicNode with KVNode[K, V] { + final def copy = new SNode(k, v, hc) + final def copyTombed = new TNode(k, v, hc) + final def copyUntombed = new SNode(k, v, hc) + final def kvPair = (k, v) + final def string(lev: Int) = (" " * lev) + "SNode(%s, %s, %x)".format(k, v, hc) +} + + +private[mutable] final class TNode[K, V](final val k: K, final val v: V, final val hc: Int) +extends MainNode[K, V] with KVNode[K, V] { + final def copy = new TNode(k, v, hc) + final def copyTombed = new TNode(k, v, hc) + final def copyUntombed = new SNode(k, v, hc) + final def kvPair = (k, v) + final def string(lev: Int) = (" " * lev) + "TNode(%s, %s, %x, !)".format(k, v, hc) +} + + +private[mutable] final class LNode[K, V](final val listmap: ImmutableListMap[K, V]) +extends MainNode[K, V] { + def this(k: K, v: V) = this(ImmutableListMap(k -> v)) + def this(k1: K, v1: V, k2: K, v2: V) = this(ImmutableListMap(k1 -> v1, k2 -> v2)) + def inserted(k: K, v: V) = new LNode(listmap + ((k, v))) + def removed(k: K): MainNode[K, V] = { + val updmap = listmap - k + 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 + } + } + def get(k: K) = listmap.get(k) + def string(lev: Int) = (" " * lev) + "LNode(%s)".format(listmap.mkString(", ")) +} + + +private[mutable] final class CNode[K, V](final val bitmap: Int, final val array: Array[BasicNode], final val gen: Gen) +extends MainNode[K, V] { + + final def updatedAt(pos: Int, nn: BasicNode, gen: Gen) = { + val len = array.length + val narr = new Array[BasicNode](len) + Array.copy(array, 0, narr, 0, len) + narr(pos) = nn + new CNode[K, V](bitmap, narr, gen) + } + + final def removedAt(pos: Int, flag: Int, gen: Gen) = { + val arr = array + val len = arr.length + val narr = new Array[BasicNode](len - 1) + Array.copy(arr, 0, narr, 0, pos) + Array.copy(arr, pos + 1, narr, pos, len - pos - 1) + new CNode[K, V](bitmap ^ flag, narr, gen) + } + + final def insertedAt(pos: Int, flag: Int, nn: BasicNode, gen: Gen) = { + val len = array.length + val bmp = bitmap + val narr = new Array[BasicNode](len + 1) + Array.copy(array, 0, narr, 0, pos) + narr(pos) = nn + Array.copy(array, pos, narr, pos + 1, len - pos) + new CNode[K, V](bmp | flag, narr, gen) + } + + /** Returns a copy of this cnode such that all the i-nodes below it are copied + * to the specified generation `ngen`. + */ + final def renewed(ngen: Gen, ct: Ctrie[K, V]) = { + var i = 0 + val arr = array + val len = arr.length + val narr = new Array[BasicNode](len) + while (i < len) { + arr(i) match { + case in: INode[K, V] => narr(i) = in.copyToGen(ngen, ct) + case bn: BasicNode => narr(i) = bn + } + i += 1 + } + new CNode[K, V](bitmap, narr, ngen) + } + + private def resurrect(inode: INode[K, V], inodemain: AnyRef): BasicNode = inodemain match { + case tn: TNode[_, _] => tn.copyUntombed + case _ => inode + } + + final def toContracted(lev: Int): MainNode[K, V] = if (array.length == 1 && lev > 0) array(0) match { + case sn: SNode[K, V] => sn.copyTombed + case _ => this + } else this + + // - if the branching factor is 1 for this CNode, and the child + // is a tombed SNode, returns its tombed version + // - otherwise, if there is at least one non-null node below, + // returns the version of this node with at least some null-inodes + // removed (those existing when the op began) + // - if there are only null-i-nodes below, returns null + final def toCompressed(ct: Ctrie[K, V], lev: Int, gen: Gen) = { + var bmp = bitmap + var i = 0 + val arr = array + val tmparray = new Array[BasicNode](arr.length) + while (i < arr.length) { // construct new bitmap + val sub = arr(i) + sub match { + case in: INode[K, V] => + val inodemain = in.GCAS_READ(ct) + assert(inodemain ne null) + tmparray(i) = resurrect(in, inodemain) + case sn: SNode[K, V] => + tmparray(i) = sn + } + i += 1 + } + + new CNode[K, V](bmp, tmparray, gen).toContracted(lev) + } + + private[mutable] def string(lev: Int): String = "CNode %x\n%s".format(bitmap, array.map(_.string(lev + 1)).mkString("\n")) + + /* quiescently consistent - don't call concurrently to anything involving a GCAS!! */ + protected def collectElems: Seq[(K, V)] = array flatMap { + case sn: SNode[K, V] => Some(sn.kvPair) + case in: INode[K, V] => in.mainnode match { + case tn: TNode[K, V] => Some(tn.kvPair) + case ln: LNode[K, V] => ln.listmap.toList + case cn: CNode[K, V] => cn.collectElems + } + } + + protected def collectLocalElems: Seq[String] = array flatMap { + case sn: SNode[K, V] => Some(sn.kvPair._2.toString) + case in: INode[K, V] => Some(in.toString.drop(14) + "(" + in.gen + ")") + } + + override def toString = { + val elems = collectLocalElems + "CNode(sz: %d; %s)".format(elems.size, elems.sorted.mkString(", ")) + } +} + + +private[mutable] object CNode { + + def dual[K, V](x: SNode[K, V], xhc: Int, y: SNode[K, V], yhc: Int, lev: Int, gen: Gen): MainNode[K, V] = if (lev < 35) { + val xidx = (xhc >>> lev) & 0x1f + val yidx = (yhc >>> lev) & 0x1f + val bmp = (1 << xidx) | (1 << yidx) + if (xidx == yidx) { + val subinode = new INode[K, V](gen)//(Ctrie.inodeupdater) + subinode.mainnode = dual(x, xhc, y, yhc, lev + 5, gen) + new CNode(bmp, Array(subinode), gen) + } else { + if (xidx < yidx) new CNode(bmp, Array(x, y), gen) + else new CNode(bmp, Array(y, x), gen) + } + } else { + new LNode(x.k, x.v, y.k, y.v) + } + +} + + +private[mutable] case class RDCSS_Descriptor[K, V](old: INode[K, V], expectedmain: MainNode[K, V], nv: INode[K, V]) { + @volatile var committed = false +} + + +class Ctrie[K, V] private (r: AnyRef, rtupd: AtomicReferenceFieldUpdater[Ctrie[K, V], AnyRef]) +extends ConcurrentMap[K, V] +{ + private val rootupdater = rtupd + @volatile var root = r + + def this() = this( + INode.newRootNode, + AtomicReferenceFieldUpdater.newUpdater(classOf[Ctrie[K, V]], classOf[AnyRef], "root") + ) + + /* internal methods */ + + @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] = { + val r = /*READ*/root + r match { + case in: INode[K, V] => in + case desc: RDCSS_Descriptor[K, V] => RDCSS_Complete(abort) + } + } + + @tailrec private def RDCSS_Complete(abort: Boolean): INode[K, V] = { + val v = /*READ*/root + v match { + case in: INode[K, V] => in + case desc: RDCSS_Descriptor[K, V] => + val RDCSS_Descriptor(ov, exp, nv) = desc + if (abort) { + if (CAS_ROOT(desc, ov)) ov + else RDCSS_Complete(abort) + } else { + val oldmain = ov.GCAS_READ(this) + if (oldmain eq exp) { + if (CAS_ROOT(desc, nv)) { + desc.committed = true + nv + } else RDCSS_Complete(abort) + } else { + if (CAS_ROOT(desc, ov)) ov + else RDCSS_Complete(abort) + } + } + } + } + + private def RDCSS_ROOT(ov: INode[K, V], expectedmain: MainNode[K, V], nv: INode[K, V]): Boolean = { + val desc = RDCSS_Descriptor(ov, expectedmain, nv) + if (CAS_ROOT(ov, desc)) { + RDCSS_Complete(false) + /*READ*/desc.committed + } 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) + } + + @tailrec private def insertifhc(k: K, hc: Int, v: V, cond: AnyRef): Option[V] = { + val r = RDCSS_READ_ROOT() + + val ret = r.rec_insertif(k, v, hc, cond, 0, null, r.gen, this) + if (ret eq null) insertifhc(k, hc, v, cond) + else ret + } + + @tailrec private def lookuphc(k: K, hc: Int): AnyRef = { + val r = RDCSS_READ_ROOT() + val res = r.rec_lookup(k, hc, 0, null, r.gen, this) + if (res eq INodeBase.RESTART) lookuphc(k, hc) + else res + } + + /* + //@tailrec + private def lookuphc(k: K, hc: Int): AnyRef = { + val r = RDCSS_READ_ROOT() + try { + r.rec_lookup(k, hc, 0, null, r.gen, this) + } catch { + case RestartException => + lookuphc(k, hc) + } + } + */ + + @tailrec private def removehc(k: K, v: V, hc: Int): Option[V] = { + val r = RDCSS_READ_ROOT() + val res = r.rec_remove(k, v, hc, 0, null, r.gen, this) + if (res ne null) res + else removehc(k, v, hc) + } + + def string = RDCSS_READ_ROOT().string(0) + + /* public methods */ + + @inline final def isReadOnly = rootupdater eq null + + @inline final def nonReadOnly = rootupdater ne null + + @tailrec final def snapshot(): Ctrie[K, V] = { + val r = RDCSS_READ_ROOT() + val expmain = r.GCAS_READ(this) + if (RDCSS_ROOT(r, expmain, r.copyToGen(new Gen, this))) new Ctrie(r.copyToGen(new Gen, this), rootupdater) + else snapshot() + } + + @tailrec final def readOnlySnapshot(): collection.Map[K, V] = { + val r = RDCSS_READ_ROOT() + val expmain = r.GCAS_READ(this) + if (RDCSS_ROOT(r, expmain, r.copyToGen(new Gen, this))) new Ctrie(r, null) + else readOnlySnapshot() + } + + @tailrec final override def clear() { + val r = RDCSS_READ_ROOT() + if (!RDCSS_ROOT(r, r.GCAS_READ(this), INode.newRootNode[K, V])) clear() + } + + final def lookup(k: K): V = { + val hc = computeHash(k) + lookuphc(k, hc).asInstanceOf[V] + } + + final override def apply(k: K): V = { + val hc = computeHash(k) + val res = lookuphc(k, hc) + if (res eq null) throw new NoSuchElementException + else res.asInstanceOf[V] + } + + final def get(k: K): Option[V] = { + val hc = computeHash(k) + Option(lookuphc(k, hc)).asInstanceOf[Option[V]] + } + + override def put(key: K, value: V): Option[V] = { + val hc = computeHash(key) + insertifhc(key, hc, value, null) + } + + final override def update(k: K, v: V) { + val hc = computeHash(k) + inserthc(k, hc, v) + } + + final def +=(kv: (K, V)) = { + update(kv._1, kv._2) + this + } + + final override def remove(k: K): Option[V] = { + val hc = computeHash(k) + removehc(k, null.asInstanceOf[V], hc) + } + + final def -=(k: K) = { + remove(k) + this + } + + def putIfAbsent(k: K, v: V): Option[V] = { + val hc = computeHash(k) + insertifhc(k, hc, v, INode.KEY_ABSENT) + } + + def remove(k: K, v: V): Boolean = { + val hc = computeHash(k) + removehc(k, v, hc).nonEmpty + } + + def replace(k: K, oldvalue: V, newvalue: V): Boolean = { + val hc = computeHash(k) + insertifhc(k, hc, newvalue, oldvalue.asInstanceOf[AnyRef]).nonEmpty + } + + def replace(k: K, v: V): Option[V] = { + val hc = computeHash(k) + insertifhc(k, hc, v, INode.KEY_PRESENT) + } + + def iterator: Iterator[(K, V)] = + if (nonReadOnly) readOnlySnapshot().iterator + else new CtrieIterator(this) + +} + + +object Ctrie { + val inodeupdater = AtomicReferenceFieldUpdater.newUpdater(classOf[INodeBase[_, _]], classOf[AnyRef], "mainnode") +} + + +private[mutable] class CtrieIterator[K, V](ct: Ctrie[K, V], mustInit: Boolean = true) extends Iterator[(K, V)] { + var stack = new Array[Array[BasicNode]](7) + var stackpos = new Array[Int](7) + var depth = -1 + var subiter: Iterator[(K, V)] = null + var current: KVNode[K, V] = null + + if (mustInit) initialize() + + def hasNext = (current ne null) || (subiter ne null) + + def next() = if (hasNext) { + var r: (K, V) = null + if (subiter ne null) { + r = subiter.next() + checkSubiter() + } else { + r = current.kvPair + advance() + } + r + } else Iterator.empty.next() + + private def readin(in: INode[K, V]) = in.GCAS_READ(ct) match { + case cn: CNode[K, V] => + depth += 1 + stack(depth) = cn.array + stackpos(depth) = -1 + advance() + case tn: TNode[K, V] => + current = tn + case ln: LNode[K, V] => + subiter = ln.listmap.iterator + checkSubiter() + case null => + current = null + } + + @inline private def checkSubiter() = if (!subiter.hasNext) { + subiter = null + advance() + } + + @inline private def initialize() { + assert(ct.isReadOnly) + + val r = ct.RDCSS_READ_ROOT() + readin(r) + } + + def advance(): Unit = if (depth >= 0) { + val npos = stackpos(depth) + 1 + if (npos < stack(depth).length) { + stackpos(depth) = npos + stack(depth)(npos) match { + case sn: SNode[K, V] => + current = sn + case in: INode[K, V] => + readin(in) + } + } else { + depth -= 1 + advance() + } + } else current = null + + /** Returns a sequence of iterators over subsets of this iterator. + * It's used to ease the implementation of splitters for a parallel version of the Ctrie. + */ + protected def subdivide: Seq[Iterator[(K, V)]] = if (subiter ne null) { + // the case where an LNode is being iterated + val it = subiter + subiter = null + advance() + Seq(it, this) + } else if (depth == -1) Seq(this) else { + var d = 0 + while (d <= depth) { + val rem = stack(d).length - 1 - stackpos(d) + if (rem > 0) { + val (arr1, arr2) = stack(d).drop(stackpos(d) + 1).splitAt(rem / 2) + stack(d) = arr1 + stackpos(d) = -1 + val it = new CtrieIterator[K, V](ct, false) + it.stack(0) = arr2 + it.stackpos(0) = -1 + it.depth = 0 + it.advance() // <-- fix it + return Seq(this, it) + } + d += 1 + } + Seq(this) + } + + private def print { + println("ctrie iterator") + println(stackpos.mkString(",")) + println("depth: " + depth) + println("curr.: " + current) + println(stack.mkString("\n")) + } + +} + + +private[mutable] object RestartException extends util.control.ControlThrowable + + +private[mutable] object Debug { + import collection._ + + lazy val logbuffer = new java.util.concurrent.ConcurrentLinkedQueue[AnyRef] + + def log(s: AnyRef) = logbuffer.add(s) + + def flush() { + for (s <- JavaConversions.asScalaIterator(logbuffer.iterator())) Console.out.println(s.toString) + logbuffer.clear() + } + + def clear() { + logbuffer.clear() + } + +} + + + + + + + + + + diff --git a/src/library/scala/collection/mutable/Gen.java b/src/library/scala/collection/mutable/Gen.java new file mode 100644 index 0000000000..0c9a30d198 --- /dev/null +++ b/src/library/scala/collection/mutable/Gen.java @@ -0,0 +1,18 @@ +/* __ *\ +** ________ ___ / / ___ Scala API ** +** / __/ __// _ | / / / _ | (c) 2003-2012, LAMP/EPFL ** +** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** +** /____/\___/_/ |_/____/_/ | | ** +** |/ ** +\* */ + +package scala.collection.mutable; + + + + + + +final class Gen { +} + diff --git a/src/library/scala/collection/mutable/INodeBase.java b/src/library/scala/collection/mutable/INodeBase.java new file mode 100644 index 0000000000..487b5cfc28 --- /dev/null +++ b/src/library/scala/collection/mutable/INodeBase.java @@ -0,0 +1,35 @@ +/* __ *\ +** ________ ___ / / ___ Scala API ** +** / __/ __// _ | / / / _ | (c) 2003-2012, LAMP/EPFL ** +** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** +** /____/\___/_/ |_/____/_/ | | ** +** |/ ** +\* */ + +package scala.collection.mutable; + + + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + + + +abstract class INodeBase<K, V> extends BasicNode { + + public static final AtomicReferenceFieldUpdater<INodeBase, MainNode> updater = AtomicReferenceFieldUpdater.newUpdater(INodeBase.class, MainNode.class, "mainnode"); + + public static final Object RESTART = new Object(); + + public volatile MainNode<K, V> mainnode = null; + + public final Gen gen; + + public INodeBase(Gen generation) { + gen = generation; + } + + public BasicNode prev() { + return null; + } + +}
\ No newline at end of file diff --git a/src/library/scala/collection/mutable/MainNode.java b/src/library/scala/collection/mutable/MainNode.java new file mode 100644 index 0000000000..09bc858edc --- /dev/null +++ b/src/library/scala/collection/mutable/MainNode.java @@ -0,0 +1,36 @@ +/* __ *\ +** ________ ___ / / ___ Scala API ** +** / __/ __// _ | / / / _ | (c) 2003-2012, LAMP/EPFL ** +** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** +** /____/\___/_/ |_/____/_/ | | ** +** |/ ** +\* */ + +package scala.collection.mutable; + + + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + + + +abstract class MainNode<K, V> extends BasicNode { + + public static final AtomicReferenceFieldUpdater<MainNode, MainNode> updater = AtomicReferenceFieldUpdater.newUpdater(MainNode.class, MainNode.class, "prev"); + + public volatile MainNode<K, V> prev = null; + + public boolean CAS_PREV(MainNode<K, V> oldval, MainNode<K, V> nval) { + return updater.compareAndSet(this, oldval, nval); + } + + public void WRITE_PREV(MainNode<K, V> nval) { + updater.set(this, nval); + } + + // do we need this? unclear in the javadocs... + public MainNode<K, V> READ_PREV() { + return updater.get(this); + } + +}
\ No newline at end of file diff --git a/test/files/run/ctries/DumbHash.scala b/test/files/run/ctries/DumbHash.scala new file mode 100644 index 0000000000..8ef325b67c --- /dev/null +++ b/test/files/run/ctries/DumbHash.scala @@ -0,0 +1,14 @@ + + + + + + +class DumbHash(val i: Int) { + override def equals(other: Any) = other match { + case that: DumbHash => that.i == this.i + case _ => false + } + override def hashCode = i % 5 + override def toString = "DH(%s)".format(i) +} diff --git a/test/files/run/ctries/Wrap.scala b/test/files/run/ctries/Wrap.scala new file mode 100644 index 0000000000..7b645c1612 --- /dev/null +++ b/test/files/run/ctries/Wrap.scala @@ -0,0 +1,9 @@ + + + + + + +case class Wrap(i: Int) { + override def hashCode = i * 0x9e3775cd +} diff --git a/test/files/run/ctries/concmap.scala b/test/files/run/ctries/concmap.scala new file mode 100644 index 0000000000..85a305ce5b --- /dev/null +++ b/test/files/run/ctries/concmap.scala @@ -0,0 +1,169 @@ + + + +import collection.mutable.Ctrie + + +object ConcurrentMapSpec extends Spec { + + val initsz = 500 + val secondsz = 750 + + def test() { + "support put" in { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until initsz) assert(ct.put(new Wrap(i), i) == None) + for (i <- 0 until initsz) assert(ct.put(new Wrap(i), -i) == Some(i)) + } + + "support put if absent" in { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until initsz) ct.update(new Wrap(i), i) + for (i <- 0 until initsz) assert(ct.putIfAbsent(new Wrap(i), -i) == Some(i)) + for (i <- 0 until initsz) assert(ct.putIfAbsent(new Wrap(i), -i) == Some(i)) + for (i <- initsz until secondsz) assert(ct.putIfAbsent(new Wrap(i), -i) == None) + for (i <- initsz until secondsz) assert(ct.putIfAbsent(new Wrap(i), i) == Some(-i)) + } + + "support remove if mapped to a specific value" in { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until initsz) ct.update(new Wrap(i), i) + for (i <- 0 until initsz) assert(ct.remove(new Wrap(i), -i - 1) == false) + for (i <- 0 until initsz) assert(ct.remove(new Wrap(i), i) == true) + for (i <- 0 until initsz) assert(ct.remove(new Wrap(i), i) == false) + } + + "support replace if mapped to a specific value" in { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until initsz) ct.update(new Wrap(i), i) + for (i <- 0 until initsz) assert(ct.replace(new Wrap(i), -i - 1, -i - 2) == false) + for (i <- 0 until initsz) assert(ct.replace(new Wrap(i), i, -i - 2) == true) + for (i <- 0 until initsz) assert(ct.replace(new Wrap(i), i, -i - 2) == false) + for (i <- initsz until secondsz) assert(ct.replace(new Wrap(i), i, 0) == false) + } + + "support replace if present" in { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until initsz) ct.update(new Wrap(i), i) + for (i <- 0 until initsz) assert(ct.replace(new Wrap(i), -i) == Some(i)) + for (i <- 0 until initsz) assert(ct.replace(new Wrap(i), i) == Some(-i)) + for (i <- initsz until secondsz) assert(ct.replace(new Wrap(i), i) == None) + } + + def assertEqual(a: Any, b: Any) = { + if (a != b) println(a, b) + assert(a == b) + } + + "support replace if mapped to a specific value, using several threads" in { + val ct = new Ctrie[Wrap, Int] + val sz = 55000 + for (i <- 0 until sz) ct.update(new Wrap(i), i) + + class Updater(index: Int, offs: Int) extends Thread { + override def run() { + var repeats = 0 + for (i <- 0 until sz) { + val j = (offs + i) % sz + var k = Int.MaxValue + do { + if (k != Int.MaxValue) repeats += 1 + k = ct.lookup(new Wrap(j)) + } while (!ct.replace(new Wrap(j), k, -k)) + } + //println("Thread %d repeats: %d".format(index, repeats)) + } + } + + val threads = for (i <- 0 until 16) yield new Updater(i, sz / 32 * i) + threads.foreach(_.start()) + threads.foreach(_.join()) + + for (i <- 0 until sz) assertEqual(ct(new Wrap(i)), i) + + val threads2 = for (i <- 0 until 15) yield new Updater(i, sz / 32 * i) + threads2.foreach(_.start()) + threads2.foreach(_.join()) + + for (i <- 0 until sz) assertEqual(ct(new Wrap(i)), -i) + } + + "support put if absent, several threads" in { + val ct = new Ctrie[Wrap, Int] + val sz = 110000 + + class Updater(offs: Int) extends Thread { + override def run() { + for (i <- 0 until sz) { + val j = (offs + i) % sz + ct.putIfAbsent(new Wrap(j), j) + assert(ct.lookup(new Wrap(j)) == j) + } + } + } + + val threads = for (i <- 0 until 16) yield new Updater(sz / 32 * i) + threads.foreach(_.start()) + threads.foreach(_.join()) + + for (i <- 0 until sz) assert(ct(new Wrap(i)) == i) + } + + "support remove if mapped to a specific value, several threads" in { + val ct = new Ctrie[Wrap, Int] + val sz = 55000 + for (i <- 0 until sz) ct.update(new Wrap(i), i) + + class Remover(offs: Int) extends Thread { + override def run() { + for (i <- 0 until sz) { + val j = (offs + i) % sz + ct.remove(new Wrap(j), j) + assert(ct.get(new Wrap(j)) == None) + } + } + } + + val threads = for (i <- 0 until 16) yield new Remover(sz / 32 * i) + threads.foreach(_.start()) + threads.foreach(_.join()) + + for (i <- 0 until sz) assert(ct.get(new Wrap(i)) == None) + } + + "have all or none of the elements depending on the oddity" in { + val ct = new Ctrie[Wrap, Int] + val sz = 65000 + for (i <- 0 until sz) ct(new Wrap(i)) = i + + class Modifier(index: Int, offs: Int) extends Thread { + override def run() { + for (j <- 0 until sz) { + val i = (offs + j) % sz + var success = false + do { + if (ct.contains(new Wrap(i))) { + success = ct.remove(new Wrap(i)) != None + } else { + success = ct.putIfAbsent(new Wrap(i), i) == None + } + } while (!success) + } + } + } + + def modify(n: Int) = { + val threads = for (i <- 0 until n) yield new Modifier(i, sz / n * i) + threads.foreach(_.start()) + threads.foreach(_.join()) + } + + modify(16) + for (i <- 0 until sz) assertEqual(ct.get(new Wrap(i)), Some(i)) + modify(15) + for (i <- 0 until sz) assertEqual(ct.get(new Wrap(i)), None) + } + + } + +} diff --git a/test/files/run/ctries/iterator.scala b/test/files/run/ctries/iterator.scala new file mode 100644 index 0000000000..1cef4f66ea --- /dev/null +++ b/test/files/run/ctries/iterator.scala @@ -0,0 +1,279 @@ + + + + +import collection._ +import collection.mutable.Ctrie + + + +object IteratorSpec extends Spec { + + def test() { + "work for an empty trie" in { + val ct = new Ctrie + val it = ct.iterator + + it.hasNext shouldEqual (false) + evaluating { it.next() }.shouldProduce [NoSuchElementException] + } + + def nonEmptyIteratorCheck(sz: Int) { + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct.put(new Wrap(i), i) + + val it = ct.iterator + val tracker = mutable.Map[Wrap, Int]() + for (i <- 0 until sz) { + assert(it.hasNext == true) + tracker += it.next + } + + it.hasNext shouldEqual (false) + evaluating { it.next() }.shouldProduce [NoSuchElementException] + tracker.size shouldEqual (sz) + tracker shouldEqual (ct) + } + + "work for a 1 element trie" in { + nonEmptyIteratorCheck(1) + } + + "work for a 2 element trie" in { + nonEmptyIteratorCheck(2) + } + + "work for a 3 element trie" in { + nonEmptyIteratorCheck(3) + } + + "work for a 5 element trie" in { + nonEmptyIteratorCheck(5) + } + + "work for a 10 element trie" in { + nonEmptyIteratorCheck(10) + } + + "work for a 20 element trie" in { + nonEmptyIteratorCheck(20) + } + + "work for a 50 element trie" in { + nonEmptyIteratorCheck(50) + } + + "work for a 100 element trie" in { + nonEmptyIteratorCheck(100) + } + + "work for a 1k element trie" in { + nonEmptyIteratorCheck(1000) + } + + "work for a 5k element trie" in { + nonEmptyIteratorCheck(5000) + } + + "work for a 75k element trie" in { + nonEmptyIteratorCheck(75000) + } + + "work for a 250k element trie" in { + nonEmptyIteratorCheck(500000) + } + + def nonEmptyCollideCheck(sz: Int) { + val ct = new Ctrie[DumbHash, Int] + for (i <- 0 until sz) ct.put(new DumbHash(i), i) + + val it = ct.iterator + val tracker = mutable.Map[DumbHash, Int]() + for (i <- 0 until sz) { + assert(it.hasNext == true) + tracker += it.next + } + + it.hasNext shouldEqual (false) + evaluating { it.next() }.shouldProduce [NoSuchElementException] + tracker.size shouldEqual (sz) + tracker shouldEqual (ct) + } + + "work for colliding hashcodes, 2 element trie" in { + nonEmptyCollideCheck(2) + } + + "work for colliding hashcodes, 3 element trie" in { + nonEmptyCollideCheck(3) + } + + "work for colliding hashcodes, 5 element trie" in { + nonEmptyCollideCheck(5) + } + + "work for colliding hashcodes, 10 element trie" in { + nonEmptyCollideCheck(10) + } + + "work for colliding hashcodes, 100 element trie" in { + nonEmptyCollideCheck(100) + } + + "work for colliding hashcodes, 500 element trie" in { + nonEmptyCollideCheck(500) + } + + "work for colliding hashcodes, 5k element trie" in { + nonEmptyCollideCheck(5000) + } + + def assertEqual(a: Map[Wrap, Int], b: Map[Wrap, Int]) { + if (a != b) { + println(a.size + " vs " + b.size) + // println(a) + // println(b) + // println(a.toSeq.sortBy((x: (Wrap, Int)) => x._1.i)) + // println(b.toSeq.sortBy((x: (Wrap, Int)) => x._1.i)) + } + assert(a == b) + } + + "be consistent when taken with concurrent modifications" in { + val sz = 25000 + val W = 25 + val S = 10 + val checks = 5 + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct.put(new Wrap(i), i) + + class Modifier extends Thread { + override def run() { + for (i <- 0 until sz) ct.putIfAbsent(new Wrap(i), i) match { + case Some(_) => ct.remove(new Wrap(i)) + case None => + } + } + } + + def consistentIteration(ct: Ctrie[Wrap, Int], checks: Int) { + class Iter extends Thread { + override def run() { + val snap = ct.readOnlySnapshot() + val initial = mutable.Map[Wrap, Int]() + for (kv <- snap) initial += kv + + for (i <- 0 until checks) { + assertEqual(snap.iterator.toMap, initial) + } + } + } + + val iter = new Iter + iter.start() + iter.join() + } + + val threads = for (_ <- 0 until W) yield new Modifier + threads.foreach(_.start()) + for (_ <- 0 until S) consistentIteration(ct, checks) + threads.foreach(_.join()) + } + + "be consistent with a concurrent removal with a well defined order" in { + val sz = 150000 + val sgroupsize = 40 + val sgroupnum = 20 + val removerslowdown = 50 + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct.put(new Wrap(i), i) + + class Remover extends Thread { + override def run() { + for (i <- 0 until sz) { + assert(ct.remove(new Wrap(i)) == Some(i)) + for (i <- 0 until removerslowdown) ct.get(new Wrap(i)) // slow down, mate + } + //println("done removing") + } + } + + def consistentIteration(it: Iterator[(Wrap, Int)]) = { + class Iter extends Thread { + override def run() { + val elems = it.toSeq + if (elems.nonEmpty) { + val minelem = elems.minBy((x: (Wrap, Int)) => x._1.i)._1.i + assert(elems.forall(_._1.i >= minelem)) + } + } + } + new Iter + } + + val remover = new Remover + remover.start() + for (_ <- 0 until sgroupnum) { + val iters = for (_ <- 0 until sgroupsize) yield consistentIteration(ct.iterator) + iters.foreach(_.start()) + iters.foreach(_.join()) + } + //println("done with iterators") + remover.join() + } + + "be consistent with a concurrent insertion with a well defined order" in { + val sz = 150000 + val sgroupsize = 30 + val sgroupnum = 30 + val inserterslowdown = 50 + val ct = new Ctrie[Wrap, Int] + + class Inserter extends Thread { + override def run() { + for (i <- 0 until sz) { + assert(ct.put(new Wrap(i), i) == None) + for (i <- 0 until inserterslowdown) ct.get(new Wrap(i)) // slow down, mate + } + //println("done inserting") + } + } + + def consistentIteration(it: Iterator[(Wrap, Int)]) = { + class Iter extends Thread { + override def run() { + val elems = it.toSeq + if (elems.nonEmpty) { + val maxelem = elems.maxBy((x: (Wrap, Int)) => x._1.i)._1.i + assert(elems.forall(_._1.i <= maxelem)) + } + } + } + new Iter + } + + val inserter = new Inserter + inserter.start() + for (_ <- 0 until sgroupnum) { + val iters = for (_ <- 0 until sgroupsize) yield consistentIteration(ct.iterator) + iters.foreach(_.start()) + iters.foreach(_.join()) + } + //println("done with iterators") + inserter.join() + } + + "work on a yet unevaluated snapshot" in { + val sz = 50000 + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct.update(new Wrap(i), i) + + val snap = ct.snapshot() + val it = snap.iterator + + while (it.hasNext) it.next() + } + + } + +} diff --git a/test/files/run/ctries/lnode.scala b/test/files/run/ctries/lnode.scala new file mode 100644 index 0000000000..28da4cc62f --- /dev/null +++ b/test/files/run/ctries/lnode.scala @@ -0,0 +1,58 @@ + + + +import collection.mutable.Ctrie + + +object LNodeSpec extends Spec { + + val initsz = 1500 + val secondsz = 1750 + + def test() { + "accept elements with the same hash codes" in { + val ct = new Ctrie[DumbHash, Int] + for (i <- 0 until initsz) ct.update(new DumbHash(i), i) + } + + "lookup 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.get(new DumbHash(i)) == Some(i)) + for (i <- initsz until secondsz) assert(ct.get(new DumbHash(i)) == None) + } + + "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) assert(ct.get(new DumbHash(i)) == None) + } + + "put elements with the same hash codes if absent" in { + val ct = new Ctrie[DumbHash, Int] + for (i <- 0 until initsz) ct.put(new DumbHash(i), i) + for (i <- 0 until initsz) assert(ct.lookup(new DumbHash(i)) == i) + for (i <- 0 until initsz) assert(ct.putIfAbsent(new DumbHash(i), i) == Some(i)) + for (i <- initsz until secondsz) assert(ct.putIfAbsent(new DumbHash(i), i) == None) + for (i <- initsz until secondsz) assert(ct.lookup(new DumbHash(i)) == i) + } + + "replace elements with the same hash codes" in { + val ct = new Ctrie[DumbHash, Int] + for (i <- 0 until initsz) assert(ct.put(new DumbHash(i), i) == None) + for (i <- 0 until initsz) assert(ct.lookup(new DumbHash(i)) == i) + for (i <- 0 until initsz) assert(ct.replace(new DumbHash(i), -i) == Some(i)) + for (i <- 0 until initsz) assert(ct.lookup(new DumbHash(i)) == -i) + for (i <- 0 until initsz) assert(ct.replace(new DumbHash(i), -i, i) == true) + } + + "remove elements with the same hash codes if mapped to a specific value" in { + val ct = new Ctrie[DumbHash, Int] + for (i <- 0 until initsz) assert(ct.put(new DumbHash(i), i) == None) + for (i <- 0 until initsz) assert(ct.remove(new DumbHash(i), i) == true) + } + + } + +} diff --git a/test/files/run/ctries/main.scala b/test/files/run/ctries/main.scala new file mode 100644 index 0000000000..8db7fcef54 --- /dev/null +++ b/test/files/run/ctries/main.scala @@ -0,0 +1,45 @@ + + + + + + + +object Test { + + def main(args: Array[String]) { + ConcurrentMapSpec.test() + IteratorSpec.test() + LNodeSpec.test() + SnapshotSpec.test() + } + +} + + +trait Spec { + + implicit def str2ops(s: String) = new { + def in[U](body: =>U) { + // just execute body + body + } + } + + implicit def any2ops(a: Any) = new { + def shouldEqual(other: Any) = assert(a == other) + } + + def evaluating[U](body: =>U) = new { + def shouldProduce[T <: Throwable: ClassManifest]() = { + var produced = false + try body + catch { + case e => if (e.getClass == implicitly[ClassManifest[T]].erasure) produced = true + } finally { + assert(produced, "Did not produce exception of type: " + implicitly[ClassManifest[T]]) + } + } + } + +} diff --git a/test/files/run/ctries/snapshot.scala b/test/files/run/ctries/snapshot.scala new file mode 100644 index 0000000000..69073d3f06 --- /dev/null +++ b/test/files/run/ctries/snapshot.scala @@ -0,0 +1,267 @@ + + + + +import collection._ +import collection.mutable.Ctrie + + + +object SnapshotSpec extends Spec { + + def test() { + "support snapshots" in { + val ctn = new Ctrie + ctn.snapshot() + ctn.readOnlySnapshot() + + val ct = new Ctrie[Int, Int] + for (i <- 0 until 100) ct.put(i, i) + ct.snapshot() + ct.readOnlySnapshot() + } + + "empty 2 quiescent snapshots in isolation" in { + val sz = 4000 + + class Worker(trie: Ctrie[Wrap, Int]) extends Thread { + override def run() { + for (i <- 0 until sz) { + assert(trie.remove(new Wrap(i)) == Some(i)) + for (j <- 0 until sz) + if (j <= i) assert(trie.get(new Wrap(j)) == None) + else assert(trie.get(new Wrap(j)) == Some(j)) + } + } + } + + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct.put(new Wrap(i), i) + val snapt = ct.snapshot() + + val original = new Worker(ct) + val snapshot = new Worker(snapt) + original.start() + snapshot.start() + original.join() + snapshot.join() + + for (i <- 0 until sz) { + assert(ct.get(new Wrap(i)) == None) + assert(snapt.get(new Wrap(i)) == None) + } + } + + def consistentReadOnly(name: String, readonly: Map[Wrap, Int], sz: Int, N: Int) { + @volatile var e: Exception = null + + // reads possible entries once and stores them + // then reads all these N more times to check if the + // state stayed the same + class Reader(trie: Map[Wrap, Int]) extends Thread { + setName("Reader " + name) + + override def run() = + try check() + catch { + case ex: Exception => e = ex + } + + def check() { + val initial = mutable.Map[Wrap, Int]() + for (i <- 0 until sz) trie.get(new Wrap(i)) match { + case Some(i) => initial.put(new Wrap(i), i) + case None => // do nothing + } + + for (k <- 0 until N) { + for (i <- 0 until sz) { + val tres = trie.get(new Wrap(i)) + val ires = initial.get(new Wrap(i)) + if (tres != ires) println(i, "initially: " + ires, "traversal %d: %s".format(k, tres)) + assert(tres == ires) + } + } + } + } + + val reader = new Reader(readonly) + reader.start() + reader.join() + + if (e ne null) { + e.printStackTrace() + throw e + } + } + + // traverses the trie `rep` times and modifies each entry + class Modifier(trie: Ctrie[Wrap, Int], index: Int, rep: Int, sz: Int) extends Thread { + setName("Modifier %d".format(index)) + + override def run() { + for (k <- 0 until rep) { + for (i <- 0 until sz) trie.putIfAbsent(new Wrap(i), i) match { + case Some(_) => trie.remove(new Wrap(i)) + case None => // do nothing + } + } + } + } + + // removes all the elements from the trie + class Remover(trie: Ctrie[Wrap, Int], index: Int, totremovers: Int, sz: Int) extends Thread { + setName("Remover %d".format(index)) + + override def run() { + for (i <- 0 until sz) trie.remove(new Wrap((i + sz / totremovers * index) % sz)) + } + } + + "have a consistent quiescent read-only snapshot" in { + val sz = 10000 + val N = 100 + val W = 10 + + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(new Wrap(i)) = i + val readonly = ct.readOnlySnapshot() + val threads = for (i <- 0 until W) yield new Modifier(ct, i, N, sz) + + threads.foreach(_.start()) + consistentReadOnly("qm", readonly, sz, N) + threads.foreach(_.join()) + } + + // now, we check non-quiescent snapshots, as these permit situations + // where a thread is caught in the middle of the update when a snapshot is taken + + "have a consistent non-quiescent read-only snapshot, concurrent with removes only" in { + val sz = 1250 + val W = 100 + val S = 5000 + + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(new Wrap(i)) = i + val threads = for (i <- 0 until W) yield new Remover(ct, i, W, sz) + + threads.foreach(_.start()) + for (i <- 0 until S) consistentReadOnly("non-qr", ct.readOnlySnapshot(), sz, 5) + threads.foreach(_.join()) + } + + "have a consistent non-quiescent read-only snapshot, concurrent with modifications" in { + val sz = 1000 + val N = 7000 + val W = 10 + val S = 7000 + + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(new Wrap(i)) = i + val threads = for (i <- 0 until W) yield new Modifier(ct, i, N, sz) + + threads.foreach(_.start()) + for (i <- 0 until S) consistentReadOnly("non-qm", ct.readOnlySnapshot(), sz, 5) + threads.foreach(_.join()) + } + + def consistentNonReadOnly(name: String, trie: Ctrie[Wrap, Int], sz: Int, N: Int) { + @volatile var e: Exception = null + + // reads possible entries once and stores them + // then reads all these N more times to check if the + // state stayed the same + class Worker extends Thread { + setName("Worker " + name) + + override def run() = + try check() + catch { + case ex: Exception => e = ex + } + + def check() { + val initial = mutable.Map[Wrap, Int]() + for (i <- 0 until sz) trie.get(new Wrap(i)) match { + case Some(i) => initial.put(new Wrap(i), i) + case None => // do nothing + } + + for (k <- 0 until N) { + // modify + for ((key, value) <- initial) { + val oldv = if (k % 2 == 0) value else -value + val newv = -oldv + trie.replace(key, oldv, newv) + } + + // check + for (i <- 0 until sz) if (initial.contains(new Wrap(i))) { + val expected = if (k % 2 == 0) -i else i + //println(trie.get(new Wrap(i))) + assert(trie.get(new Wrap(i)) == Some(expected)) + } else { + assert(trie.get(new Wrap(i)) == None) + } + } + } + } + + val worker = new Worker + worker.start() + worker.join() + + if (e ne null) { + e.printStackTrace() + throw e + } + } + + "have a consistent non-quiescent snapshot, concurrent with modifications" in { + val sz = 9000 + val N = 1000 + val W = 10 + val S = 400 + + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(new Wrap(i)) = i + val threads = for (i <- 0 until W) yield new Modifier(ct, i, N, sz) + + threads.foreach(_.start()) + for (i <- 0 until S) { + consistentReadOnly("non-qm", ct.snapshot(), sz, 5) + consistentNonReadOnly("non-qsnap", ct.snapshot(), sz, 5) + } + threads.foreach(_.join()) + } + + "work when many concurrent snapshots are taken, concurrent with modifications" in { + val sz = 12000 + val W = 10 + val S = 10 + val modifytimes = 1200 + val snaptimes = 600 + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(new Wrap(i)) = i + + class Snapshooter extends Thread { + setName("Snapshooter") + override def run() { + for (k <- 0 until snaptimes) { + val snap = ct.snapshot() + for (i <- 0 until sz) snap.remove(new Wrap(i)) + for (i <- 0 until sz) assert(!snap.contains(new Wrap(i))) + } + } + } + + val mods = for (i <- 0 until W) yield new Modifier(ct, i, modifytimes, sz) + val shooters = for (i <- 0 until S) yield new Snapshooter + val threads = mods ++ shooters + threads.foreach(_.start()) + threads.foreach(_.join()) + } + + } + +} diff --git a/test/files/scalacheck/Ctrie.scala b/test/files/scalacheck/Ctrie.scala new file mode 100644 index 0000000000..2950937278 --- /dev/null +++ b/test/files/scalacheck/Ctrie.scala @@ -0,0 +1,199 @@ + + + +import org.scalacheck._ +import Prop._ +import org.scalacheck.Gen._ +import collection._ +import collection.mutable.Ctrie + + + +case class Wrap(i: Int) { + override def hashCode = i // * 0x9e3775cd +} + + +/** A check mainly oriented towards checking snapshot correctness. + */ +object Test extends Properties("Ctrie") { + + /* generators */ + + val sizes = choose(0, 200000) + + val threadCounts = choose(2, 16) + + val threadCountsAndSizes = for { + p <- threadCounts + sz <- sizes + } yield (p, sz); + + + /* helpers */ + + def inParallel[T](totalThreads: Int)(body: Int => T): Seq[T] = { + val threads = for (idx <- 0 until totalThreads) yield new Thread { + setName("ParThread-" + idx) + private var res: T = _ + override def run() { + res = body(idx) + } + def result = { + this.join() + res + } + } + + threads foreach (_.start()) + threads map (_.result) + } + + def spawn[T](body: =>T): { def get: T } = { + val t = new Thread { + setName("SpawnThread") + private var res: T = _ + override def run() { + res = body + } + def result = res + } + t.start() + new { + def get: T = { + t.join() + t.result + } + } + } + + def elementRange(threadIdx: Int, totalThreads: Int, totalElems: Int): Range = { + val sz = totalElems + val idx = threadIdx + val p = totalThreads + val start = (sz / p) * idx + math.min(idx, sz % p) + val elems = (sz / p) + (if (idx < sz % p) 1 else 0) + val end = start + elems + (start until end) + } + + def hasGrown[K, V](last: Map[K, V], current: Map[K, V]) = { + (last.size <= current.size) && { + last forall { + case (k, v) => current.get(k) == Some(v) + } + } + } + + object err { + var buffer = new StringBuilder + def println(a: AnyRef) = buffer.append(a.toString).append("\n") + def clear() = buffer.clear() + def flush() = { + Console.out.println(buffer) + clear() + } + } + + + /* properties */ + + property("concurrent growing snapshots") = forAll(threadCounts, sizes) { + (numThreads, numElems) => + val p = 3 //numThreads + val sz = 102 //numElems + val ct = new Ctrie[Wrap, Int] + + // checker + val checker = spawn { + def check(last: Map[Wrap, Int], iterationsLeft: Int): Boolean = { + val current = ct.readOnlySnapshot() + if (!hasGrown(last, current)) false + else if (current.size >= sz) true + else if (iterationsLeft < 0) false + else check(current, iterationsLeft - 1) + } + check(ct.readOnlySnapshot(), 500) + } + + // fillers + inParallel(p) { + idx => + elementRange(idx, p, sz) foreach (i => ct.update(Wrap(i), i)) + } + + // wait for checker to finish + val growing = true//checker.get + + val ok = growing && ((0 until sz) forall { + case i => ct.get(Wrap(i)) == Some(i) + }) + + ok + } + + property("update") = forAll(sizes) { + (n: Int) => + val ct = new Ctrie[Int, Int] + for (i <- 0 until n) ct(i) = i + (0 until n) forall { + case i => ct(i) == i + } + } + + property("concurrent update") = forAll(threadCountsAndSizes) { + case (p, sz) => + val ct = new Ctrie[Wrap, Int] + + inParallel(p) { + idx => + for (i <- elementRange(idx, p, sz)) ct(Wrap(i)) = i + } + + (0 until sz) forall { + case i => ct(Wrap(i)) == i + } + } + + + property("concurrent remove") = forAll(threadCounts, sizes) { + (p, sz) => + val ct = new Ctrie[Wrap, Int] + for (i <- 0 until sz) ct(Wrap(i)) = i + + inParallel(p) { + idx => + for (i <- elementRange(idx, p, sz)) ct.remove(Wrap(i)) + } + + (0 until sz) forall { + case i => ct.get(Wrap(i)) == None + } + } + + + property("concurrent putIfAbsent") = forAll(threadCounts, sizes) { + (p, sz) => + val ct = new Ctrie[Wrap, Int] + + val results = inParallel(p) { + idx => + elementRange(idx, p, sz) find (i => ct.putIfAbsent(Wrap(i), i) != None) + } + + (results forall (_ == None)) && ((0 until sz) forall { + case i => ct.get(Wrap(i)) == Some(i) + }) + } + +} + + + + + + + + + + |