Make 'Message' a generic parameter in distributed algorithms

JetBrains · olgalupuleac · Dec 1, 2020 · Dec 1, 2020 · Dec 7, 2020 · Dec 9, 2020
commit 3aa7f0cad26d5d91c77cda2fafcd5841581fa8af
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/DistributedCTestConfiguration.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/DistributedCTestConfiguration.kt
@@ -31,7 +31,7 @@ import org.jetbrains.kotlinx.lincheck.verifier.Verifier
 import java.lang.reflect.Method
 
 
-class DistributedCTestConfiguration(testClass: Class<*>, iterations: Int,
+class DistributedCTestConfiguration<Message>(testClass: Class<*>, iterations: Int,
                                     threads: Int, actorsPerThread: Int,
                                     generatorClass: Class<out ExecutionGenerator>,
                                     verifierClass: Class<out Verifier>,
@@ -57,6 +57,6 @@ class DistributedCTestConfiguration(testClass: Class<*>, iterations: Int,
     }
 
     override fun createStrategy(testClass: Class<*>, scenario: ExecutionScenario, validationFunctions: List<Method>, stateRepresentationMethod: Method?, verifier: Verifier): Strategy {
-        return DistributedStrategy(this, testClass, scenario, validationFunctions, stateRepresentationMethod, verifier)
+        return DistributedStrategy<Message>(this, testClass, scenario, validationFunctions, stateRepresentationMethod, verifier)
     }
 }
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/DistributedOptions.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/DistributedOptions.kt
@@ -9,8 +9,8 @@ enum class MessageOrder {
     ASYNCHRONOUS
 }
 
-class DistributedOptions : Options<DistributedOptions,
-        DistributedCTestConfiguration>() {
+class DistributedOptions<Message> : Options<DistributedOptions<Message>,
+        DistributedCTestConfiguration<Message>>() {
     var networkReliability: Double = 1.0
     var messageOrder: MessageOrder = MessageOrder.SYNCHRONOUS
     var maxNumberOfFailedNodes: Int = 0
@@ -20,30 +20,31 @@ class DistributedOptions : Options<DistributedOptions,
     var maxMessageCount : Int = Int.MAX_VALUE
     var maxMessagePerProcess : Int = Int.MAX_VALUE
     var duplicationRate : Int = 1
-    var testClasses = HashMap<Class<out Node>, Int>()
+    var testClasses = HashMap<Class<out Node<Message>>, Int>()
+    var useVectorClock = false
 
-    fun delay(delay : Int) : DistributedOptions {
+    fun delay(delay : Int) : DistributedOptions<Message> {
         this.maxDelay = delay
         return this
     }
 
 
-    fun networkReliability(networkReliability: Double): DistributedOptions {
+    fun networkReliability(networkReliability: Double): DistributedOptions<Message> {
         this.networkReliability = networkReliability
         return this
     }
 
-    fun testClass(cls : Class<out Node>, numberOfInstances : Int): DistributedOptions {
+    fun testClass(cls : Class<out Node<Message>>, numberOfInstances : Int): DistributedOptions<Message> {
         this.testClasses[cls] = numberOfInstances
         return this
     }
 
-    fun messageOrder(messageOrder: MessageOrder): DistributedOptions {
+    fun messageOrder(messageOrder: MessageOrder): DistributedOptions<Message> {
         this.messageOrder = messageOrder
         return this
     }
 
-    fun maxNumberOfFailedNodes(maxNumOfFailedNodes: Int): DistributedOptions {
+    fun maxNumberOfFailedNodes(maxNumOfFailedNodes: Int): DistributedOptions<Message> {
         if (maxNumOfFailedNodes > threads) {
             throw IllegalArgumentException("Maximum number of failed nodes " +
                     "is more than total number of nodes")
@@ -52,39 +53,43 @@ class DistributedOptions : Options<DistributedOptions,
         return this
     }
 
-    fun supportRecovery(supportRecovery: Boolean): DistributedOptions {
+    fun supportRecovery(supportRecovery: Boolean): DistributedOptions<Message> {
         this.supportRecovery = supportRecovery
         return this
     }
 
-    fun invocationsPerIteration(invocations: Int): DistributedOptions {
+    fun invocationsPerIteration(invocations: Int): DistributedOptions<Message> {
         this.invocationsPerIteration = invocations
         return this
     }
 
-    fun maxMessageCount(count : Int) : DistributedOptions {
+    fun maxMessageCount(count : Int) : DistributedOptions<Message> {
         this.maxMessageCount = count
         return this
     }
 
-    fun maxMessageCountPerProcess(count : Int) : DistributedOptions {
+    fun maxMessageCountPerProcess(count : Int) : DistributedOptions<Message> {
         this.maxMessagePerProcess = count
         return this
     }
 
-    fun duplicationRate(rate : Int) : DistributedOptions {
+    fun duplicationRate(rate : Int) : DistributedOptions<Message> {
         this.duplicationRate = rate
         return this
     }
 
-    override fun createTestConfigurations(testClass: Class<*>): DistributedCTestConfiguration {
+    fun useVectorClock(useClock : Boolean) : DistributedOptions<Message> {
+        this.useVectorClock = useClock
+        return this
+    }
+
+    override fun createTestConfigurations(testClass: Class<*>): DistributedCTestConfiguration<Message> {
         return DistributedCTestConfiguration(testClass, iterations, threads,
                 actorsPerThread, executionGenerator,
                 verifier, invocationsPerIteration, networkReliability,
                 messageOrder, maxNumberOfFailedNodes, supportRecovery,
                 maxDelay, maxMessageCount, maxMessagePerProcess, duplicationRate,
                 requireStateEquivalenceImplementationCheck, false,
                 chooseSequentialSpecification(sequentialSpecification, testClass), timeoutMs)
-
     }
 }
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Environment.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Environment.kt
@@ -5,7 +5,7 @@ import java.util.concurrent.TimeUnit
 /**
  * Environment interface for communication with other processes.
  */
-interface Environment {
+interface Environment<Message> {
     /**
      * Identifier of this node (from 0 to [numberOfNodes]).
      */
@@ -25,30 +25,28 @@ interface Environment {
      * @param i is the index
      * @return
      */
-    fun getAddress(cls : Class<*>, i : Int) : Int
+    fun getAddress(cls: Class<*>, i: Int): Int
 
     /**
      * @param timer
      * @param time
      * @param timeUnit
      */
-    fun setTimer(timer : String, time : Int, timeUnit : TimeUnit = TimeUnit.MILLISECONDS)
+    fun setTimer(timer: String, time: Int, timeUnit: TimeUnit = TimeUnit.MILLISECONDS)
 
     /**
      * @param timer
      */
-    fun cancelTimer(timer : String)
+    fun cancelTimer(timer: String)
 
     /**
      * Sends the specified [message] to the process [destId] (from 0 to [numberOfNodes]).
      */
-    fun send(message: Message, receiver : Int)
+    fun send(message: Message, receiver: Int)
 
     /**
      * Sends the specified [message] to all processes (from 0 to
      * [numberOfNodes]).
-     *
-     * @param message is a message to be sent
      */
     fun broadcast(message: Message) {
         for (i in 0 until numberOfNodes) {
@@ -57,37 +55,23 @@ interface Environment {
     }
 
     /**
-     *
-     */
-    fun sendLocal(message : Message)
-
-    /**
-     *
-     */
-    fun checkLocalMessages(atMostOnce : Boolean = false, atLeastOnce : Boolean = false, preserveOrder : Boolean = false)
-
-    /**
-     *
-     */
-    val messages : List<Message>
-
-    /**
-     *
+     * Sends local message. Can be used to point out that some action is completed
+     * (e.g. message is delivered to a user, or lock is acquired). After the end of the execution all local messages are available and can be checked by the user.
      */
-    val processes : List<ProcessExecution>
+    fun sendLocal(message: Message)
 
-    /**
-     *
-     */
-    val processExecution : ProcessExecution?
+    val events: List<Event>
 }
 
-/**
- *
- */
-data class ProcessExecution(
-        val id : Int,
-        val isAlive : Boolean,
-        val sentMessages : List<Message>,
-        val receivedMessages : List<Message>,
-        val localMessages : List<Message>)
+sealed class Event
+class MessageSentEvent<Message>(val message: Message, val sender: Int, val receiver: Int, val id: Int) : Event()
+class MessageReceivedEvent<Message>(val message: Message, val sender: Int, val receiver: Int, val id: Int) : Event()
+class LocalMessageSentEvent<Message>(val message: Message, val sender: Int, val id: Int) : Event()
+class ProcessFailureEvent(val processId: Int) : Event()
+class ProcessRecoveryEvent(val processId: Int) : Event()
+class TimerEvent(val processId: Int, val timer: String) : Event()
+
+fun <Message> Environment<Message>.correctProcesses() = (0 until numberOfNodes).subtract(events.filterIsInstance<ProcessFailureEvent>().map { it.processId })
+fun <Message> Environment<Message>.sentMessages(processId: Int) = events.filterIsInstance<MessageSentEvent<Message>>().filter { it.sender == processId }
+fun <Message> Environment<Message>.receivedMessage(processId: Int) = events.filterIsInstance<MessageReceivedEvent<Message>>().filter { it.receiver == processId }
+fun <Message> Environment<Message>.localMessage(processId: Int) = events.filterIsInstance<LocalMessageSentEvent<Message>>()
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Message.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Message.kt
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Node.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Node.kt
@@ -6,12 +6,12 @@ import java.util.concurrent.TimeUnit
 /**
  * Interface for a single node in a distributed algorithm.
  */
-interface Node {
+interface Node<Message> {
     /**
      * Called when a new message arrives.
      * @param message is a message from another node
      */
-    fun onMessage(message : Message)
+    fun onMessage(message : Message, sender : Int)
 
     /**
      * Called if the [timer] expires. The timer can be set using the environment
@@ -27,10 +27,10 @@ interface Node {
 }
 
 
-abstract class BlockingReceiveNodeImp : Node {
+abstract class BlockingReceiveNodeImp<Message> : Node<Message> {
     private val messageQueue = LinkedBlockingQueue<Message>()
 
-    override fun onMessage(message: Message) {
+    override fun onMessage(message: Message, sender : Int) {
         messageQueue.add(message)
     }