Add flow-less summary storage

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/ifds/Producer.kt

@@ -0,0 +1,152 @@
+/*
+ *  Copyright 2022 UnitTestBot contributors (utbot.org)
+ * <p>
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ * <p>
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.usvm.dataflow.ifds
+
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicReference
+
+interface Producer<T> {
+    fun produce(event: T)
+    fun subscribe(consumer: Consumer<T>)
+}
+
+fun interface Consumer<in T> {
+    fun consume(event: T)
+}
+
+class SyncProducer<T> : Producer<T> {
+    private val consumers: MutableList<Consumer<T>> = mutableListOf()
+    private val events: MutableList<T> = mutableListOf()
+
+    @Synchronized
+    override fun produce(event: T) {
+        for (consumer in consumers) {
+            consumer.consume(event)
+        }
+        events.add(event)
+    }
+
+    @Synchronized
+    override fun subscribe(consumer: Consumer<T>) {
+        for (event in events) {
+            consumer.consume(event)
+        }
+        consumers.add(consumer)
+    }
+}
+
+sealed interface ConsList<out T> : Iterable<T>
+
+data object Nil : ConsList<Nothing> {
+    override fun iterator(): Iterator<Nothing> = object : Iterator<Nothing> {
+        override fun hasNext(): Boolean = false
+        override fun next(): Nothing = throw NoSuchElementException()
+    }
+}
+
+data class Cons<out T>(
+    val value: T,
+    val tail: ConsList<T>,
+) : ConsList<T> {
+    override fun iterator(): Iterator<T> = Iter(this)
+
+    private class Iter<T>(private var list: ConsList<T>) : Iterator<T> {
+        override fun hasNext(): Boolean = list !is Nil
+
+        override fun next(): T = when (val list = list) {
+            is Nil -> throw NoSuchElementException()
+            is Cons -> {
+                val value = list.value
+                this.list = list.tail
+                value
+            }
+        }
+    }
+}
+
+class NonBlockingQueue<T> {
+    data class Node<T>(
+        val value: T,
+        @Volatile var next: Node<T>? = null,
+    )
+
+    var head: Node<T>? = null
+        private set
+    val tail: AtomicReference<Node<T>> = AtomicReference(head)
+    val size: AtomicInteger = AtomicInteger(0)
+
+    fun add(element: T) {
+        val node = Node(element)
+        var currentTail: Node<T>?
+        while (true) {
+            currentTail = tail.get()
+            if (tail.compareAndSet(currentTail, node)) break
+        }
+        if (currentTail != null) {
+            currentTail.next = node
+        } else {
+            head = node
+        }
+        size.incrementAndGet()
+    }
+}
+
+class ConcurrentProducer<T> : Producer<T> {
+    private var consumers: AtomicReference<ConsList<Consumer<T>>> = AtomicReference(Nil)
+    private val events: NonBlockingQueue<T> = NonBlockingQueue()
+
+    override fun produce(event: T) {
+        var currentConsumers: ConsList<Consumer<T>>
+        while (true) {
+            currentConsumers = consumers.get() ?: continue
+            if (consumers.compareAndSet(currentConsumers, null)) break
+        }
+
+        events.add(event)
+
+        try {
+            for (consumer in currentConsumers) {
+                consumer.consume(event)
+            }
+        } finally {
+            check(consumers.compareAndSet(null, currentConsumers))
+        }
+    }
+
+    override fun subscribe(consumer: Consumer<T>) {
+        var last: NonBlockingQueue.Node<T>? = null
+        while (true) {
+            val start = if (last != null) last.next else events.head
+            var current = start
+            while (current != null) {
+                last = current
+                consumer.consume(current.value)
+                current = current.next
+            }
+
+            val currentConsumers = consumers.get() ?: continue
+            if (!consumers.compareAndSet(currentConsumers, null)) continue
+            if (events.tail.get() === last) {
+                val newConsumers = Cons(consumer, currentConsumers)
+                check(consumers.compareAndSet(null, newConsumers))
+                break
+            } else {
+                check(consumers.compareAndSet(null, currentConsumers))
+            }
+        }
+    }
+}

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/ifds/Runner.kt

@@ -114,6 +114,7 @@ class UniRunner<Fact, Event, Method, Statement>(
 
             // Add edge to worklist:
             workList.trySend(edge).getOrThrow()
+            manager.handleControlEvent(queueIsNotEmpty)
 
             return true
         }
@@ -126,7 +127,6 @@ class UniRunner<Fact, Event, Method, Statement>(
             val edge = workList.tryReceive().getOrElse {
                 manager.handleControlEvent(queueIsEmpty)
                 val edge = workList.receive()
-                manager.handleControlEvent(queueIsNotEmpty)
                 edge
             }
             tabulationAlgorithmStep(edge, this@coroutineScope)

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/ifds/Summary.kt

@@ -16,7 +16,6 @@
 
 package org.usvm.dataflow.ifds
 
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import org.jacodb.api.common.CommonMethod
@@ -40,6 +39,7 @@ interface SummaryEdge<out Fact, out Statement : CommonInst> : Summary<CommonMeth
 }
 
 interface Vulnerability<out Fact, out Statement : CommonInst> : Summary<CommonMethod> {
+
     val message: String
     val sink: Vertex<Fact, Statement>
 
@@ -50,60 +50,79 @@ interface Vulnerability<out Fact, out Statement : CommonInst> : Summary<CommonMe
 /**
  * Contains summaries for many methods and allows to update them and subscribe for them.
  */
-interface SummaryStorage<T : Summary<*>> {
+class SummaryStorageWithFlows<T : Summary<*>> {
+    private val summaries = ConcurrentHashMap<CommonMethod, MutableSet<T>>()
+    private val outFlows = ConcurrentHashMap<CommonMethod, MutableSharedFlow<T>>()
 
     /**
-     * A list of all methods for which summaries are not empty.
+     * @return a list with all methods for which there are some summaries.
      */
     val knownMethods: List<CommonMethod>
+        get() = summaries.keys.toList()
+
+    private fun getFlow(method: CommonMethod): MutableSharedFlow<T> {
+        return outFlows.computeIfAbsent(method) {
+            MutableSharedFlow(replay = Int.MAX_VALUE)
+        }
+    }
 
     /**
-     * Adds [summary] the summaries storage of its method.
+     * Adds a new [fact] to the storage.
      */
-    fun add(summary: T)
+    fun add(fact: T) {
+        val isNew = summaries.computeIfAbsent(fact.method) { ConcurrentHashMap.newKeySet() }.add(fact)
+        if (isNew) {
+            val flow = getFlow(fact.method)
+            check(flow.tryEmit(fact))
+        }
+    }
 
     /**
      * @return a flow with all facts summarized for the given [method].
      * Already received facts, along with the facts that will be sent to this storage later,
      * will be emitted to the returned flow.
      */
-    fun getFacts(method: CommonMethod): Flow<T>
+    fun getFacts(method: CommonMethod): SharedFlow<T> {
+        return getFlow(method)
+    }
 
     /**
      * @return a list will all facts summarized for the given [method] so far.
      */
-    fun getCurrentFacts(method: CommonMethod): List<T>
+    fun getCurrentFacts(method: CommonMethod): List<T> {
+        return getFacts(method).replayCache
+    }
 }
 
-class SummaryStorageImpl<T : Summary<*>> : SummaryStorage<T> {
-
+class SummaryStorageWithProducers<T : Summary<*>>(
+    private val isConcurrent: Boolean = true,
+) {
     private val summaries = ConcurrentHashMap<CommonMethod, MutableSet<T>>()
-    private val outFlows = ConcurrentHashMap<CommonMethod, MutableSharedFlow<T>>()
-
-    override val knownMethods: List<CommonMethod>
-        get() = summaries.keys.toList()
-
-    private fun getFlow(method: CommonMethod): MutableSharedFlow<T> {
-        return outFlows.computeIfAbsent(method) {
-            MutableSharedFlow(replay = Int.MAX_VALUE)
+    private val producers = ConcurrentHashMap<CommonMethod, Producer<T>>()
+
+    val knownMethods: Collection<CommonMethod>
+        get() = summaries.keys
+
+    private fun getProducer(method: CommonMethod): Producer<T> {
+        return producers.computeIfAbsent(method) {
+            if (isConcurrent) {
+                ConcurrentProducer()
+            } else {
+                SyncProducer()
+            }
         }
     }
 
-    override fun add(summary: T) {
-        val isNew = summaries.computeIfAbsent(summary.method) {
-            ConcurrentHashMap.newKeySet()
-        }.add(summary)
+    fun add(fact: T) {
+        val isNew = summaries.computeIfAbsent(fact.method) { ConcurrentHashMap.newKeySet() }.add(fact)
         if (isNew) {
-            val flow = getFlow(summary.method)
-            check(flow.tryEmit(summary))
+            val producer = getProducer(fact.method)
+            producer.produce(fact)
         }
     }
 
-    override fun getFacts(method: CommonMethod): SharedFlow<T> {
-        return getFlow(method)
-    }
-
-    override fun getCurrentFacts(method: CommonMethod): List<T> {
-        return getFacts(method).replayCache
+    fun subscribe(method: CommonMethod, handler: (T) -> Unit) {
+        val producer = getProducer(method)
+        producer.subscribe(handler)
     }
 }

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/taint/TaintAnalyzers.kt

@@ -27,20 +27,22 @@ import org.usvm.dataflow.config.FactAwareConditionEvaluator
 import org.usvm.dataflow.ifds.Analyzer
 import org.usvm.dataflow.ifds.Edge
 import org.usvm.dataflow.ifds.Reason
+import org.usvm.dataflow.ifds.UnitResolver
 import org.usvm.dataflow.util.Traits
 
 private val logger = object : KLogging() {}.logger
 
 context(Traits<Method, Statement>)
 class TaintAnalyzer<Method, Statement>(
     private val graph: ApplicationGraph<Method, Statement>,
+    private val unitResolver: UnitResolver<Method>,
     private val getConfigForMethod: (Method) -> List<TaintConfigurationItem>?,
 ) : Analyzer<TaintDomainFact, TaintEvent<Statement>, Method, Statement>
     where Method : CommonMethod,
           Statement : CommonInst {
 
     override val flowFunctions: ForwardTaintFlowFunctions<Method, Statement> by lazy {
-        ForwardTaintFlowFunctions(graph, getConfigForMethod)
+        ForwardTaintFlowFunctions(graph, unitResolver, getConfigForMethod)
     }
 
     private fun isExitPoint(statement: Statement): Boolean {

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/taint/TaintBidiRunner.kt

@@ -20,6 +20,9 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
+import org.jacodb.api.common.CommonMethod
+import org.jacodb.api.common.analysis.ApplicationGraph
+import org.jacodb.api.common.cfg.CommonInst
 import org.usvm.dataflow.ifds.ControlEvent
 import org.usvm.dataflow.ifds.Edge
 import org.usvm.dataflow.ifds.IfdsResult
@@ -28,9 +31,6 @@ import org.usvm.dataflow.ifds.QueueEmptinessChanged
 import org.usvm.dataflow.ifds.Reason
 import org.usvm.dataflow.ifds.UnitResolver
 import org.usvm.dataflow.ifds.UnitType
-import org.jacodb.api.common.CommonMethod
-import org.jacodb.api.common.analysis.ApplicationGraph
-import org.jacodb.api.common.cfg.CommonInst
 
 class TaintBidiRunner<Method, Statement>(
     val manager: TaintManager<Method, Statement>,

usvm-dataflow/src/main/kotlin/org/usvm/dataflow/taint/TaintFlowFunctions.kt

@@ -45,6 +45,8 @@ import org.usvm.dataflow.config.TaintActionEvaluator
 import org.usvm.dataflow.ifds.ElementAccessor
 import org.usvm.dataflow.ifds.FlowFunction
 import org.usvm.dataflow.ifds.FlowFunctions
+import org.usvm.dataflow.ifds.UnitResolver
+import org.usvm.dataflow.ifds.UnknownUnit
 import org.usvm.dataflow.ifds.isOnHeap
 import org.usvm.dataflow.ifds.isStatic
 import org.usvm.dataflow.ifds.minus
@@ -57,6 +59,7 @@ private val logger = mu.KotlinLogging.logger {}
 context(Traits<Method, Statement>)
 class ForwardTaintFlowFunctions<Method, Statement>(
     private val graph: ApplicationGraph<Method, Statement>,
+    private val unitResolver: UnitResolver<Method>,
     val getConfigForMethod: (Method) -> List<TaintConfigurationItem>?,
 ) : FlowFunctions<TaintDomainFact, Method, Statement>
     where Method : CommonMethod,
@@ -321,7 +324,7 @@ class ForwardTaintFlowFunctions<Method, Statement>(
         //    to remove any marks from 'instance' and arguments.
         //   Currently, "analyzability" of the callee depends on the fact that the callee
         //    is "accessible" through the JcApplicationGraph::callees().
-        if (callee in graph.callees(callStatement)) {
+        if (callee in graph.callees(callStatement) && unitResolver.resolve(callee) != UnknownUnit) {
 
             if (fact.variable.isStatic) {
                 return@FlowFunction emptyList()