Add distributed mutex test

JetBrains · olgalupuleac · Dec 1, 2020 · Dec 1, 2020 · Dec 7, 2020 · Dec 9, 2020
commit 277d3c6291f0319485b9a278a3bb5dd5e87370b8
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Environment.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Environment.kt
@@ -25,7 +25,9 @@ interface Environment<Message> {
      * @param i is the index
      * @return
      */
-    fun getAddress(cls: Class<*>, i: Int): Int
+    fun getAddress(cls: Class<out Node<Message>>, i: Int): Int
+
+    fun getNumberOfNodeType(cls: Class<out Node<Message>>) : Int
 
     /**
      * @param timer

diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Node.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/Node.kt
@@ -28,17 +28,17 @@ interface Node<Message> {
 
 
 abstract class BlockingReceiveNodeImp<Message> : Node<Message> {
-    private val messageQueue = LinkedBlockingQueue<Message>()
+    private val messageQueue = LinkedBlockingQueue<Pair<Message, Int>>()
 
     override fun onMessage(message: Message, sender : Int) {
-        messageQueue.add(message)
+        messageQueue.add(Pair(message, sender))
     }
 
-    fun receive(timeout : Long, timeUnit: TimeUnit = TimeUnit.MILLISECONDS) : Message? {
+    fun receive(timeout : Long, timeUnit: TimeUnit = TimeUnit.MILLISECONDS) : Pair<Message, Int>? {
         return messageQueue.poll(timeout, timeUnit)
     }
 
-    fun receive() : Message? {
-        return messageQueue.poll()
+    fun receive() : Pair<Message, Int> {
+        return messageQueue.peek()
     }
 }
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/stress/DistributedRunner.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/stress/DistributedRunner.kt
@@ -72,7 +72,7 @@ internal class SynchronousMessageQueue<Message> : MessageQueue<Message> {
     }
 }
 
-class FifoMessageQueue<Message>(numberOfNodes: Int) : MessageQueue<Message> {
+internal class FifoMessageQueue<Message>(numberOfNodes: Int) : MessageQueue<Message> {
     private val messageQueues = Array(numberOfNodes) {
         LinkedBlockingQueue<MessageWrapper<Message>>()
     }
@@ -93,7 +93,7 @@ class FifoMessageQueue<Message>(numberOfNodes: Int) : MessageQueue<Message> {
     }
 }
 
-class AsynchronousMessageQueue<Message>(private val numberOfNodes: Int) : MessageQueue<Message> {
+internal class AsynchronousMessageQueue<Message>(private val numberOfNodes: Int) : MessageQueue<Message> {
     private val messageQueues = Array(numberOfNodes) {
         LinkedBlockingQueue<MessageWrapper<Message>>()
     }
@@ -115,6 +115,7 @@ class AsynchronousMessageQueue<Message>(private val numberOfNodes: Int) : Messag
     }
 }
 
+internal class NodeFailureException(val nodeId: Int) : Exception()
 
 open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
                              val testCfg: DistributedCTestConfiguration<Message>,
@@ -142,7 +143,7 @@ open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
                 if (failedProcesses[message.receiver].get()) {
                     continue
                 }
-                //println("[${message.receiver}]: Received message $message from process ${message.sender}")
+                //println("[${message.receiver}]: Received message ${message.msg} from process ${message.sender}")
                 testInstances[message.receiver].onMessage(message.msg, message.sender)
             }
         }
@@ -163,9 +164,10 @@ open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
     override fun initialize() {
         super.initialize()
         check(scenario.parallelExecution.size == testCfg.threads) {
-            "Parrallel execution size is ${scenario.parallelExecution.size}"
+            "Parallel execution size is ${scenario.parallelExecution.size}"
         }
         testThreadExecutions = Array(testCfg.threads) { t ->
+           // println(scenario.parallelExecution[t].size)
             TestThreadExecutionGenerator.create(this, t, scenario.parallelExecution[t], null, false)
         }
         testThreadExecutions.forEach { it.allThreadExecutions = testThreadExecutions }
@@ -211,14 +213,14 @@ open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
         testThreadExecutions.map { executor.submit(it) }.forEachIndexed { i, future ->
             try {
                 future.get(20, TimeUnit.SECONDS)
-            } catch (e: TimeoutException) {
-                println("Timeout exception")
-                if (!failedProcesses[i].get()) {
-                    isRunning = false
-                    messageBroker.join()
-                    val threadDump = Thread.getAllStackTraces().filter { (t, _) -> t is FixedActiveThreadsExecutor.TestThread }
-                    return DeadlockInvocationResult(threadDump)
-                }
+            } catch(e : NodeFailureException) {
+
+            }
+            catch (e: TimeoutException) {
+                isRunning = false
+                messageBroker.join()
+                val threadDump = Thread.getAllStackTraces().filter { (t, _) -> t is FixedActiveThreadsExecutor.TestThread }
+                return DeadlockInvocationResult(threadDump)
             } catch (e: ExecutionException) {
                 return UnexpectedExceptionInvocationResult(e.cause!!)
             }
@@ -269,9 +271,6 @@ open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
     private inner class EnvironmentImpl(override val nodeId: Int,
                                         override val numberOfNodes:
                                         Int) : Environment<Message> {
-        override fun getAddress(cls: Class<*>, i: Int): Int {
-            TODO("Not yet implemented")
-        }
 
         override fun setTimer(timer: String, time: Int, timeUnit: TimeUnit) {
             TODO("Not yet implemented")
@@ -320,5 +319,13 @@ open class DistributedRunner<Message>(strategy: DistributedStrategy<Message>,
 
         override val events: List<Event>
             get() = TODO("Not yet implemented")
+
+        override fun getAddress(cls: Class<out Node<Message>>, i: Int): Int {
+            TODO("Not yet implemented")
+        }
+
+        override fun getNumberOfNodeType(cls: Class<out Node<Message>>): Int {
+            TODO("Not yet implemented")
+        }
     }
 }
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/stress/DistributedStrategy.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/distributed/stress/DistributedStrategy.kt
@@ -56,14 +56,13 @@ class DistributedStrategy<Message>(val testCfg: DistributedCTestConfiguration<Me
         try {
             // Run invocations
             for (invocation in 0 until invocations) {
-                println("INVOCATION $invocation")
+               println("INVOCATION $invocation")
                 val ir = runner.run()
                 when (ir) {
                     is CompletedInvocationResult -> {
                         if (!verifier.verifyResults(scenario, ir.results)) {
                             return IncorrectResultsFailure(scenario, ir.results)
                         }
-
                     }
                     else ->
                     {

diff --git a/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/kvstorage/KVStorageServer.kt b/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/kvstorage/KVStorageServer.kt
@@ -50,7 +50,7 @@ class KVStorageClient(private val environment: Environment<Command>) : BlockingR
             environment.send(command, serverAddr)
             val response = receive(10, TimeUnit.MILLISECONDS)
             if (response != null) {
-                commandResults[response.id] = response
+                commandResults[response.first.id] = response.first
             }
             if (commandResults.containsKey(command.id)) {
                 val res = commandResults[command.id]!!
@@ -113,4 +113,4 @@ class KVStorageServerTestClass {
                         .invocationsPerIteration(100)
                         .iterations(100))
     }
-}
+}
diff --git a/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/mutex/LamportMutex.kt b/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/mutex/LamportMutex.kt
@@ -0,0 +1,183 @@
+/*
+ * Lincheck
+ *
+ * Copyright (C) 2019 - 2020 JetBrains s.r.o.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Lesser Public License for more details.
+ *
+ * You should have received a copy of the GNU General Lesser Public
+ * License along with this program.  If not, see
+ * <http://www.gnu.org/licenses/lgpl-3.0.html>
+ */
+
+package org.jetbrains.kotlinx.lincheck.test.distributed.mutex
+
+import kotlinx.atomicfu.locks.withLock
+import org.jetbrains.kotlinx.lincheck.LinChecker
+import org.jetbrains.kotlinx.lincheck.LincheckAssertionError
+import org.jetbrains.kotlinx.lincheck.annotations.Operation
+import org.jetbrains.kotlinx.lincheck.annotations.Validate
+import org.jetbrains.kotlinx.lincheck.distributed.DistributedOptions
+import org.jetbrains.kotlinx.lincheck.distributed.Environment
+import org.jetbrains.kotlinx.lincheck.distributed.MessageOrder
+import org.jetbrains.kotlinx.lincheck.distributed.Node
+import org.junit.Test
+import java.lang.Integer.max
+import java.util.concurrent.locks.ReentrantLock
+
+sealed class MutexMessage(val msgTime: Int)
+
+class Req(msgTime : Int, val reqTime : Int) : MutexMessage(msgTime) {
+    override fun toString(): String {
+        return "REQ($msgTime, $reqTime)"
+    }
+}
+class Ok(msgTime : Int) : MutexMessage(msgTime){
+    override fun toString(): String {
+        return "OK($msgTime)"
+    }
+}
+class Rel(msgTime: Int) : MutexMessage(msgTime) {
+    override fun toString(): String {
+        return "REL($msgTime)"
+    }
+}
+class Lock(msgTime: Int) : MutexMessage(msgTime)
+class Unlock(msgTime: Int) : MutexMessage(msgTime)
+
+@Volatile
+internal var counter = 0
+
+class LamportMutex(private val env : Environment<MutexMessage>) : Node<MutexMessage> {
+    private val inf = Int.MAX_VALUE
+    private var clock = 0 // logical time
+    private var inCS = false // are we in critical section?
+    private val req = IntArray(env.numberOfNodes) { inf } // time of last REQ message
+    private val ok = IntArray(env.numberOfNodes) // time of last OK message
+    private val lock = ReentrantLock()
+    private val condition = lock.newCondition()
+
+    override fun onMessage(message: MutexMessage, sender : Int) {
+        lock.withLock {
+            val time = message.msgTime
+            clock = max(clock, time) + 1
+            when(message) {
+                is Req -> {
+                    req[sender] = message.reqTime
+                    env.send(Ok(++clock), sender)
+                }
+                is Ok -> {
+                    ok[sender] = time
+                }
+                is Rel -> {
+                    req[sender] = inf
+                }
+                else -> throw RuntimeException("Unexpected message type")
+            }
+            checkInCS()
+        }
+    }
+
+    @Validate
+    fun validate() {
+        counter = 0
+    }
+
+    private fun checkInCS() {
+        val myReqTime = req[env.nodeId]
+        if (myReqTime == inf || inCS) {
+            return
+        }
+        for (i in 0 until env.numberOfNodes) {
+            if (i == env.nodeId) {
+                continue
+            }
+            if (req[i] < myReqTime || req[i] == myReqTime && i < env.nodeId) {
+                return
+            }
+            if (ok[i] <= myReqTime) {
+                return
+            }
+        }
+        inCS = true
+        condition.signal()
+      //  env.sendLocal(Lock(clock))
+    }
+
+    @Operation
+    fun lock() : Int {
+       // println("[${env.nodeId}]: request lock")
+        lock.withLock {
+            check (req[env.nodeId] == inf) {
+                Thread.currentThread()
+            }
+            val myReqTime = ++clock
+            req[env.nodeId] = myReqTime
+            broadcast(Req(++clock, myReqTime))
+            if (env.numberOfNodes == 1) {
+                inCS = true
+            }
+            while(!inCS) {
+                condition.await()
+            }
+        }
+        val res = ++counter
+        unlock()
+        //println("[${env.nodeId}]: unlock")
+        return res
+    }
+
+    private fun unlock() {
+        lock.withLock {
+            if (!inCS) return
+            //env.sendLocal(Unlock(clock))
+            inCS = false
+            req[env.nodeId] = inf
+            broadcast(Rel(++clock))
+        }
+    }
+
+    private fun broadcast(msg : MutexMessage) {
+        for (i in 0 until env.numberOfNodes) {
+            if (i == env.nodeId) {
+                continue
+            }
+            env.send(msg, i)
+        }
+    }
+}
+
+class Counter {
+    var cnt = 0
+    fun lock() : Int {
+        return ++cnt
+    }
+}
+
+class LamportMutexTest {
+    @Test
+    fun testSimple() {
+        LinChecker.check(LamportMutex::class
+                .java, DistributedOptions<MutexMessage>().requireStateEquivalenceImplCheck
+        (false).sequentialSpecification(Counter::class.java).threads
+        (5).messageOrder(MessageOrder.SYNCHRONOUS)
+                .invocationsPerIteration(100).iterations(1000))
+    }
+
+    @Test(expected = LincheckAssertionError::class)
+    fun testNoFifo() {
+        LinChecker.check(LamportMutex::class
+                .java, DistributedOptions<MutexMessage>().requireStateEquivalenceImplCheck
+        (false).sequentialSpecification(Counter::class.java).threads
+        (3).messageOrder(MessageOrder.ASYNCHRONOUS)
+                .invocationsPerIteration(100).iterations(1000))
+    }
+}
diff --git a/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/mutex/MutexTest.kt b/src/jvm/test/org/jetbrains/kotlinx/lincheck/test/distributed/mutex/MutexTest.kt
@@ -0,0 +1,41 @@
+/*
+ * Lincheck
+ *
+ * Copyright (C) 2019 - 2020 JetBrains s.r.o.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Lesser Public License for more details.
+ *
+ * You should have received a copy of the GNU General Lesser Public
+ * License along with this program.  If not, see
+ * <http://www.gnu.org/licenses/lgpl-3.0.html>
+ */
+/*
+package org.jetbrains.kotlinx.lincheck.test.distributed.mutex
+
+import org.jetbrains.kotlinx.lincheck.LinChecker
+import org.jetbrains.kotlinx.lincheck.distributed.DistributedOptions
+import org.jetbrains.kotlinx.lincheck.distributed.MessageOrder
+
+class MutexTest {
+    fun testMutex() {
+        LinChecker.check(LamportMutex::class.java,
+                DistributedOptions()
+                        .threads(5)
+                        .requireStateEquivalenceImplCheck(false)
+                        .messageOrder(MessageOrder.FIFO)
+                        .useVectorClock(true)
+                        .maxMessageCount(14)
+                        .invocationsPerIteration(100)
+                        .iterations(100))
+    }
+}
+
+ */