Incrementally update NestedArrays by GraphChanges

README.md

@@ -163,6 +163,20 @@ networking.firewall.allowedTCPPorts = [ 12345 ];
 
 The images are automatically published to docker.woost.space once a day from `master`.
 
+# nodejs Stack traces in tests
+
+TODO: Why is this not working?
+
+```bash
+npm install source-map-support
+```
+
+then run tests using
+
+```
+SOURCEMAPS=true start nsbt
+```
+
 # Deployment
 Requirements:
 * docker

build.sbt

@@ -160,6 +160,9 @@ lazy val webSettings = Seq(
   scalaJSStage in Test := FastOptStage,
   scalaJSLinkerConfig in (Compile, fastOptJS) ~= { _.withSourceMap(withSourceMaps) },
   scalaJSLinkerConfig in (Compile, fullOptJS) ~= { _.withSourceMap(withSourceMaps) },
+  scalaJSLinkerConfig in (Test, fastOptJS) ~= { _.withSourceMap(withSourceMaps) },
+  scalaJSLinkerConfig in (Test, fullOptJS) ~= { _.withSourceMap(withSourceMaps) },
+
   npmDevDependencies in Compile ++= Deps.npm.webpackDependencies,
   version in webpack := Deps.webpackVersion,
   version in startWebpackDevServer := Deps.webpackDevServerVersion,

webApp/src/main/scala/wust/webApp/state/GlobalState.scala

@@ -2,6 +2,7 @@ package wust.webApp.state
 
 import scala.util.Try
 import com.github.ghik.silencer.silent
+import graphstate.GraphState
 import acyclic.file
 import monix.reactive.Observable
 import monix.reactive.subjects.PublishSubject
@@ -125,8 +126,7 @@ object GlobalState {
           val g = graph.addNodes(
             // these nodes are obviously not in the graph for an assumed user, since the user is not persisted yet.
             // if we start with an assumed user and just create new channels we will never get a graph from the backend.
-            user().toNode ::
-              Nil
+            user().toNode :: Nil
           )
           scribe.debug("  graph with added usernode: " + g)
           g
@@ -136,6 +136,13 @@ object GlobalState {
     }
   }
 
+  val graphState = Var(new GraphState(Graph.empty))
+  eventProcessor.graphEvents.foreach { 
+    case LocalGraphUpdateEvent.NewChanges(changes) => graphState.now.update(changes)
+    case LocalGraphUpdateEvent.NewGraph(graph) => graphState() = new GraphState(graph)
+  }
+
+
   val viewConfig: Rx[ViewConfig] = {
 
     var lastViewConfig: ViewConfig = ViewConfig(View.Empty, Page.empty)

webApp/src/main/scala/wust/webApp/state/graphstate/EdgeState.scala

@@ -0,0 +1,78 @@
+package wust.webApp.state.graphstate
+
+import acyclic.file
+import rx._
+import flatland._
+import wust.ids._
+import wust.util.algorithm._
+import wust.util.collection._
+import wust.util.macros.InlineList
+import wust.graph._
+import wust.util.time.time
+
+import scala.collection.{ breakOut, immutable, mutable }
+
+object EdgeState {
+  @inline def edgeKey(edge: Edge): (NodeId, NodeId) = edge.sourceId -> edge.targetId
+
+  def apply(nodeState: NodeState, edges: Array[Edge]): EdgeState = {
+    val edgeState = new EdgeState(nodeState)
+    edgeState.update(GraphChanges(addEdges = edges))
+    edgeState
+  }
+}
+
+final class EdgeState(nodeState: NodeState) {
+
+  val edgesNow: mutable.ArrayBuffer[Edge] = mutable.ArrayBuffer.empty
+  val idToIdxHashMap: mutable.HashMap[(NodeId, NodeId), Int] = mutable.HashMap.empty
+  val edgesRx = new LazyReactiveCollection[Edge](getCurrent = idx => edgesNow(idx))
+  var edgesIdxNow: InterleavedArrayInt = InterleavedArrayInt.empty
+
+  @inline def idToIdxFold[T](endPoints: (NodeId, NodeId))(default: => T)(f: Int => T): T = {
+    idToIdxHashMap.get(endPoints) match {
+      case Some(idx) => f(idx)
+      case None      => default
+    }
+  }
+  @inline def idToIdxForeach[U](endPoints: (NodeId, NodeId))(f: Int => U): Unit = idToIdxFold(endPoints)(())(f(_))
+  @inline def idToIdxMap[T](endPoints: (NodeId, NodeId))(f: Int => T): Option[T] = idToIdxFold(endPoints)(Option.empty[T])(idx => Some(f(idx)))
+  @inline def idToIdxOrThrow(endPoints: (NodeId, NodeId)): Int = idToIdxHashMap(endPoints)
+  def idToIdx(endPoints: (NodeId, NodeId)): Option[Int] = idToIdxFold[Option[Int]](endPoints)(None)(Some(_))
+
+  def update(changes: GraphChanges): Unit = {
+    // register new and updated edges
+
+    edgesNow.sizeHint(edgesNow.length + changes.addEdges.length)
+    edgesRx.sizeHint(edgesRx.length + changes.addEdges.length)
+    idToIdxHashMap.sizeHint(idToIdxHashMap.size + changes.addEdges.length)
+
+    val addEdgeIdxBuilder = InterleavedArrayInt.builder
+    changes.addEdges.foreachElement { edge =>
+      val key = EdgeState.edgeKey(edge)
+
+      idToIdxFold(key){
+        val newIdx = edgesNow.length
+        edgesNow += edge
+        edgesRx.grow()
+        idToIdxHashMap(key) = newIdx
+        nodeState.idToIdxForeach(edge.sourceId){ sourceIdx =>
+          nodeState.idToIdxForeach(edge.targetId){ targetIdx =>
+            addEdgeIdxBuilder.add(sourceIdx, targetIdx)
+          }
+        }
+      }{ idx =>
+        edgesNow(idx) = edge
+        edgesRx(idx)() = edge
+      }
+    }
+
+    assert(edgesNow.length == idToIdxHashMap.size)
+
+    time("graphstate:edgestate:updateInterleaved") {
+    edgesIdxNow = new InterleavedArrayInt(edgesIdxNow.interleaved ++ addEdgeIdxBuilder.result().interleaved)
+    }
+    assert(edgesRx.length == edgesNow.length)
+    assert(edgesIdxNow.elementCount == edgesNow.length)
+  }
+}

webApp/src/main/scala/wust/webApp/state/graphstate/GraphState.scala

@@ -0,0 +1,49 @@
+package wust.webApp.state.graphstate
+
+import acyclic.file
+import rx._
+import flatland._
+import wust.ids._
+import wust.util.algorithm._
+import wust.util.collection._
+import wust.util.macros.InlineList
+import wust.graph._
+import wust.util.time.time
+
+import scala.collection.{ breakOut, immutable, mutable }
+
+//TODO: idempotence
+//TODO: commutativity (e.g. store edges without loaded nodes until the nodes appear)
+//TODO: be immune to inconsistency
+
+final class GraphState(initialGraph: Graph) {
+  import initialGraph.{ nodes, edges }
+  val nodeState = new NodeState
+  val edgeState = new EdgeState(nodeState)
+
+  val children = new LayerState(edgeState, edgeDistributors.ifMyEdgeChild)
+  val read = new LayerState(edgeState, edgeDistributors.ifMyEdgeRead)
+
+  update(GraphChanges(addNodes = initialGraph.nodes, addEdges = initialGraph.edges))
+
+  def update(changes: GraphChanges) = {
+    time("graphstate") {
+      val layerChanges = time("graphstate:nodestate") {nodeState.update(changes)}
+      time("graphstate:edgestate") {edgeState.update(changes)}
+
+      children.update(layerChanges)
+      read.update(layerChanges)
+    }
+  }
+}
+
+object edgeDistributors {
+  def ifMyEdgeChild(code: (NodeId, NodeId) => Unit): Edge => Unit = {
+    case edge: Edge.Child => code(edge.parentId, edge.childId)
+    case _                =>
+  }
+  def ifMyEdgeRead(code: (NodeId, NodeId) => Unit): Edge => Unit = {
+    case edge: Edge.Read => code(edge.nodeId, edge.userId)
+    case _                =>
+  }
+}

webApp/src/main/scala/wust/webApp/state/graphstate/LayerState.scala

@@ -0,0 +1,133 @@
+package wust.webApp.state.graphstate
+
+import scala.reflect.ClassTag
+import scala.scalajs.js.JSConverters._
+import acyclic.file
+import rx._
+import flatland._
+import wust.ids._
+import wust.util.algorithm._
+import wust.util.collection._
+import wust.util.macros.InlineList
+import wust.graph._
+import wust.util.time.time
+import scala.scalajs.js
+
+import scala.collection.{ breakOut, immutable, mutable }
+import scala.scalajs.js.WrappedArray
+
+final class InterleavedJSArrayIntBuilder {
+  @inline private def extractHi(l: Long): Int = (l >> 32).toInt
+  @inline private def extractLo(l: Long): Int = l.toInt
+  @inline private def combineToLong(a: Int, b: Int): Long = ((a.toLong) << 32) | (b & 0xffffffffL)
+
+  val self = new js.Array[Long]
+
+  @inline def add(a: Int, b: Int): Unit = {
+    self += combineToLong(a, b)
+  }
+
+  @inline def result() = self
+}
+
+final class LayerState(val edgeState: EdgeState, ifMyEdge: ((NodeId, NodeId) => Unit) => (Edge => Unit)) {
+  import edgeState.edgesIdxNow
+
+  var edgeLookupNow: NestedArrayIntValues = NestedArrayInt.empty
+  var edgeRevLookupNow: NestedArrayIntValues = NestedArrayInt.empty
+  val edgeLookupRx = new LazyReactiveCollection[Array[Int]](getCurrent = idx => edgeLookupNow(idx).toArray)
+  val edgeRevLookupRx = new LazyReactiveCollection[Array[Int]](getCurrent = idx => edgeRevLookupNow(idx).toArray)
+
+  var lookupNow: NestedArrayIntMapped = edgeLookupNow.viewMapInt(edgesIdxNow.right)
+  var revLookupNow: NestedArrayIntMapped = edgeRevLookupNow.viewMapInt(edgesIdxNow.left)
+  def lookupRx(idx: Int)(implicit ctx: Ctx.Owner) = edgeLookupRx(idx).map(_.map(edgesIdxNow.right))
+  def revLookupRx(idx: Int)(implicit ctx: Ctx.Owner) = edgeRevLookupRx(idx).map(_.map(edgesIdxNow.left))
+
+  // @inline def ifMyEdge(code: (NodeId, NodeId) => Unit): PartialFunction[Edge, Unit]
+  @inline def update(changes: LayerChanges): Unit = {
+    // time("graphstate:update") {
+    val affectedSourceNodes = new mutable.ArrayBuilder.ofInt
+    val affectedTargetNodes = new mutable.ArrayBuilder.ofInt
+    val addElemBuilder = new InterleavedJSArrayIntBuilder
+    val addRevElemBuilder = new InterleavedJSArrayIntBuilder
+    time("graphstate:update:addEdges") {
+      changes.addEdges.foreachElement {
+        ifMyEdge { (sourceId, targetId) =>
+          edgeState.idToIdxForeach(sourceId -> targetId) { edgeIdx =>
+            val sourceIdx = edgesIdxNow.left(edgeIdx)
+            val targetIdx = edgesIdxNow.right(edgeIdx)
+            addElemBuilder.add(sourceIdx, edgeIdx)
+            addRevElemBuilder.add(targetIdx, edgeIdx)
+            affectedSourceNodes += sourceIdx
+            affectedTargetNodes += targetIdx
+          }
+        }
+      }
+    }
+    val addElem = time("graphstate:update:addElem") {
+      val interleavedLong = addElemBuilder.result()
+      // use native js Array.sort with compare function for long (faster than scala.util.Sorting.quickSort or java.util.Arrays.sort)
+      interleavedLong.sort(_ compare _)
+      new InterleavedArrayInt(interleavedLong.toArray) // TODO: is there a way to avoid copying? Either convert js.Array to scala array or call js.Array.sort on scala Array?
+    }
+    val addRevElem = time("graphstate:update:addRevElem") {
+      val interleavedLong = addRevElemBuilder.result()
+      // use native js Array.sort with compare function for long (faster than scala.util.Sorting.quickSort or java.util.Arrays.sort)
+      interleavedLong.sort(_ compare _)
+      new InterleavedArrayInt(interleavedLong.toArray) // TODO: is there a way to avoid copying? Either convert js.Array to scala array or call js.Array.sort on scala Array?
+    }
+
+    val delElemBuilder = new mutable.ArrayBuilder.ofRef[(Int, Int)]
+    val delRevElemBuilder = new mutable.ArrayBuilder.ofRef[(Int, Int)]
+    time("graphstate:update:delEdges") {
+      changes.delEdges.foreach {
+        ifMyEdge { (sourceId, targetId) =>
+          edgeState.idToIdxForeach(sourceId -> targetId) { edgeIdx =>
+            val sourceIdx = edgesIdxNow.left(edgeIdx)
+            val targetIdx = edgesIdxNow.right(edgeIdx)
+            delElemBuilder += sourceIdx -> lookupNow.indexOf(sourceIdx)(edgeIdx) // Remove the first occurence of the sourceId/targetId combination
+            delRevElemBuilder += targetIdx -> lookupNow.indexOf(targetIdx)(edgeIdx) // Remove the first occurence of the sourceId/targetId combination
+            affectedSourceNodes += sourceIdx
+            affectedTargetNodes += targetIdx
+          }
+        }
+      }
+    }
+    val delElem = time("graphstate:update:delElem") { InterleavedArrayInt(delElemBuilder.result().sortBy(_._1)) }
+    val delRevElem = time("graphstate:update:delRevElem") { InterleavedArrayInt(delRevElemBuilder.result().sortBy(_._1)) }
+
+    time("graphstate:update:change") {
+      // NestedArray.changed() parameters:
+      // addIdx: Int, // how many nodes are added
+      // addElem: InterleavedArrayInt // Array[idx -> elem]
+      // delElem: InterleavedArrayInt // Array[idx -> position]
+      if (scala.scalajs.LinkingInfo.developmentMode) {
+        edgeLookupNow = edgeLookupNow.changedWithAssertions(changes.addIdx, addElem, delElem)
+        edgeRevLookupNow = edgeRevLookupNow.changedWithAssertions(changes.addIdx, addRevElem, delRevElem)
+      } else {
+        edgeLookupNow = edgeLookupNow.changed(changes.addIdx, addElem, delElem)
+        edgeRevLookupNow = edgeRevLookupNow.changed(changes.addIdx, addRevElem, delRevElem)
+      }
+    }
+
+    time("graphstate:update:update-rx") {
+      lookupNow = edgeLookupNow.viewMapInt(edgesIdxNow.right)
+      revLookupNow = edgeRevLookupNow.viewMapInt(edgesIdxNow.left)
+
+      edgeLookupRx.sizeHint(edgeLookupRx.length + changes.addIdx)
+      edgeRevLookupRx.sizeHint(edgeRevLookupRx.length + changes.addIdx)
+      loop(changes.addIdx) { _ =>
+        edgeLookupRx.grow()
+        edgeRevLookupRx.grow()
+      }
+
+      affectedSourceNodes.result().foreachElement { sourceNodeIdx =>
+        edgeLookupRx.refresh(sourceNodeIdx)
+      }
+      affectedTargetNodes.result().foreachElement { targetNodeIdx =>
+        edgeRevLookupRx.refresh(targetNodeIdx)
+      }
+    }
+    // }
+  }
+}

webApp/src/main/scala/wust/webApp/state/graphstate/LazyReactiveCollection.scala

@@ -0,0 +1,44 @@
+package wust.webApp.state.graphstate
+
+import scala.reflect.ClassTag
+import scala.scalajs.js.JSConverters._
+import acyclic.file
+import rx._
+import flatland._
+import wust.ids._
+import wust.util.algorithm._
+import wust.util.collection._
+import wust.util.macros.InlineList
+import wust.graph._
+import wust.util.time.time
+import scala.scalajs.js
+
+import scala.collection.{ breakOut, immutable, mutable }
+import scala.scalajs.js.WrappedArray
+
+@inline final class LazyReactiveCollection[T](getCurrent: Int => T) {
+  val self: mutable.ArrayBuffer[Var[T]] = mutable.ArrayBuffer.empty
+
+  @inline def length = self.length
+
+  @inline def sizeHint(n:Int) = self.sizeHint(n)
+
+  @inline def grow(): Unit = { self += null }
+
+  @inline def refresh(idx: Int): Unit = {
+    if (self(idx) != null) {
+      self(idx)() = getCurrent(idx)
+    }
+  }
+
+  @inline def apply(idx: Int): Var[T] = {
+    if (self(idx) == null) {
+      val value = Var(getCurrent(idx))
+      self(idx) = value
+      value
+    } else {
+      self(idx)
+    }
+  }
+}
+