JetBrains · olgalupuleac · Dec 1, 2020 · Dec 1, 2020 · Dec 7, 2020 · Dec 9, 2020
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -1,4 +1,4 @@
-import org.gradle.jvm.tasks.Jar
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
 // atomicfu
 buildscript {
@@ -15,12 +15,16 @@ plugins {
     id("maven-publish")
     id("maven")
     id("kotlinx.team.infra") version "0.2.0-dev-55"
+    kotlin("plugin.serialization") version "1.5.31"
 }
 
 repositories {
     mavenCentral()
     jcenter()
 }
+/*dependencies {
+    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.3.0")
+}*/
 
 kotlin {
     jvm {
@@ -43,6 +47,7 @@ kotlin {
             val kotlinxCoroutinesVersion: String by project
             val asmVersion: String by project
             val reflectionsVersion: String by project
+            val apacheCommonsVersion: String by project
             dependencies {
                 api("org.jetbrains.kotlin:kotlin-stdlib:$kotlinVersion")
                 api("org.jetbrains.kotlin:kotlin-stdlib-common:$kotlinVersion")
@@ -51,6 +56,7 @@ kotlin {
                 api("org.ow2.asm:asm-commons:$asmVersion")
                 api("org.ow2.asm:asm-util:$asmVersion")
                 api("org.reflections:reflections:$reflectionsVersion")
+                api("org.apache.commons:commons-math3:$apacheCommonsVersion")
             }
         }
 
@@ -88,17 +94,20 @@ tasks {
 
     withType<Test> {
         maxParallelForks = 1
-        jvmArgs("--add-opens", "java.base/jdk.internal.misc=ALL-UNNAMED",
-                "--add-exports", "java.base/jdk.internal.util=ALL-UNNAMED")
+        jvmArgs(
+            "--add-opens", "java.base/jdk.internal.misc=ALL-UNNAMED",
+            "--add-exports", "java.base/jdk.internal.util=ALL-UNNAMED"
+        )
     }
 
     withType<Jar> {
         manifest {
             val inceptionYear: String by project
             val lastCopyrightYear: String by project
-            attributes("Copyright" to
-                "Copyright (C) 2015 - 2019 Devexperts, LLC\n                                " +
-                "Copyright (C) $inceptionYear - $lastCopyrightYear JetBrains, s.r.o."
+            attributes(
+                "Copyright" to
+                        "Copyright (C) 2015 - 2019 Devexperts, LLC\n                                " +
+                        "Copyright (C) $inceptionYear - $lastCopyrightYear JetBrains, s.r.o."
             )
         }
     }

diff --git a/docs/distributed/overview.md b/docs/distributed/overview.md
@@ -0,0 +1,73 @@
+# Lincheck for distributed algorithms.
+
+This Lincheck extension provides an ability to test the prototypes of distributed algorithms implemented using the
+internal Lincheck API.
+
+## Algorithm implementation
+
+To be able to test the distributed algorithm, you have to implement it using the provided API. Each node in the system
+is represented by the instance of the class implemented by the user. The node class should implement `Node` interface
+and take `NodeEnvironment` as only constructor parameter. The `Node` and `NodeEnvironment` are parameterized by generic
+type `Message`.
+
+### Operations
+
+### Interface `Node`
+
+```kotlin
+interface Node<Message> {
+    fun onMessage(message: Message, sender: Int)
+    ...
+}
+```
+
+While implementing `Node` interface user should implement `onMessage` which will be called when `message` from node
+with `id` sender arrives (more information about nodes' ids see below). This method defines how to react on message.
+
+There are also over methods which can be optionally overridden (by default they do nothing):
+
+1. `onStart` is called before running operations (FIXME)
+2. `onScenarioFinished` is called after all operations from scenario for this node are completed. If the node doesn't
+   have any operations to execute, the method is not called.
+3. `recover` is called when the node restarts after crash (see below).
+4. `onNodeUnavailable(x)` is called when the node `x` crashes or becomes unreachable because of a network partition. It
+   can be used as an alternative to the failure detector (
+   FIXME switch off only event logging).
+5. `stateRepresentation` returns the string representation of the node's state. Could be used for debugging purposes (
+   see Events).
+
+### NodeEnvironment
+
+The `NodeEnvironment` is passed to each node as a constructor parameter and is used for sending messages between nodes
+and other specific purposes. 
+
+To get information about nodes in the system, use:
+
+* `id: Int` returns id of this node. The nodes are numbered from 0 until the total number of nodes. Ids are passed
+  to `Node::onMessage` and `NodeEnvironment::send` and helps to distinguish nodes from each other.
+* `nodes: Int` returns the total number in the system.
+* `getIds<NodeType>(): List<Int>` returns the ids of nodes for the specified type, when there are multiple types of
+  nodes in the system.
+
+```kotlin
+fun operation() {
+    ...
+    val server = env.getIds<Server>[0]
+    env.send(msg, server)
+}
+```
+
+To send messages between nodes:
+
+* `send(message: Message, receiver: Int)` sends `message` to the node with id `receiver`.
+* `broadcast(message: Message, skipItself: Boolean = true)` 
+* `broadcastToGroup<NodeType>(message: Message, skipItself: Boolean = true)`
+
+
+
+## Example
+
+Let's consider the simple example of distributed key-value storage with one server and multiple clients. Client has two
+types of operations `fun get(key: Int): Int?` and `put(key: Int, value: Int): Int?`. The operations are executed on the
+server.
+
diff --git a/gradle.properties b/gradle.properties
@@ -5,11 +5,12 @@ version=2.15-SNAPSHOT
 inceptionYear=2019
 lastCopyrightYear=2020
 
-kotlinVersion=1.4.0
+kotlinVersion=1.6.10
 kotlinxCoroutinesVersion=1.4.0
-asmVersion=9.0
+asmVersion=9.2
 atomicfuVersion=0.14.4
 reflectionsVersion=0.9.12
+apacheCommonsVersion=3.3
 
 junitVersion=4.13.1
 jctoolsVersion=2.1.0

diff --git a/settings.gradle.kts b/settings.gradle.kts
@@ -6,7 +6,7 @@ pluginManagement {
     }
 
     repositories {
-        maven(url = "https://dl.bintray.com/kotlin/kotlinx")
+        maven(url = "https://maven.pkg.jetbrains.space/kotlin/p/kotlinx/maven")
         mavenCentral()
         jcenter()
         gradlePluginPortal()

diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/ExecutionClassLoader.java b/src/jvm/main/org/jetbrains/kotlinx/lincheck/ExecutionClassLoader.java
@@ -22,6 +22,7 @@
  * #L%
  */
 
+import org.jetbrains.kotlinx.lincheck.runner.TestNodeExecution;
 import org.jetbrains.kotlinx.lincheck.runner.TestThreadExecution;
 
 /**
@@ -32,4 +33,8 @@ public class ExecutionClassLoader extends ClassLoader {
     public Class<? extends TestThreadExecution> defineClass(String className, byte[] bytecode) {
         return (Class<? extends TestThreadExecution>) super.defineClass(className, bytecode, 0, bytecode.length);
     }
+
+    public Class<? extends TestNodeExecution> defineNodeClass(String className, byte[] bytecode) {
+        return (Class<? extends TestNodeExecution>) super.defineClass(className, bytecode, 0, bytecode.length);
+    }
 }
diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/LinChecker.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/LinChecker.kt
@@ -21,16 +21,19 @@
  */
 package org.jetbrains.kotlinx.lincheck
 
-import org.jetbrains.kotlinx.lincheck.annotations.*
-import org.jetbrains.kotlinx.lincheck.execution.*
-import org.jetbrains.kotlinx.lincheck.strategy.*
-import org.jetbrains.kotlinx.lincheck.verifier.*
-import kotlin.reflect.*
+import org.jetbrains.kotlinx.lincheck.annotations.LogLevel
+import org.jetbrains.kotlinx.lincheck.distributed.DistributedCTestConfiguration
+import org.jetbrains.kotlinx.lincheck.distributed.DistributedVerifier
+import org.jetbrains.kotlinx.lincheck.distributed.random.ProbabilisticModel
+import org.jetbrains.kotlinx.lincheck.execution.ExecutionScenario
+import org.jetbrains.kotlinx.lincheck.strategy.LincheckFailure
+import org.jetbrains.kotlinx.lincheck.verifier.Verifier
+import kotlin.reflect.KClass
 
 /**
  * This class runs concurrent tests.
  */
-class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
+class LinChecker(private val testClass: Class<*>, options: Options<*, *>?) {
     private val testStructure = CTestStructure.getFromTestClass(testClass)
     private val testConfigurations: List<CTestConfiguration>
     private val reporter: Reporter
@@ -39,7 +42,7 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
         val logLevel = options?.logLevel ?: testClass.getAnnotation(LogLevel::class.java)?.value ?: DEFAULT_LOG_LEVEL
         reporter = Reporter(logLevel)
         testConfigurations = if (options != null) listOf(options.createTestConfigurations(testClass))
-                             else createFromTestClassAnnotations(testClass)
+        else createFromTestClassAnnotations(testClass)
     }
 
     /**
@@ -79,7 +82,7 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
             val failure = scenario.run(this, verifier)
             if (failure != null) {
                 val minimizedFailedIteration = if (!minimizeFailedScenario) failure
-                                               else failure.minimize(this)
+                else failure.minimize(this)
                 reporter.logFailedIteration(minimizedFailedIteration)
                 return minimizedFailedIteration
             }
@@ -99,6 +102,10 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
         while (true) {
             minimizedFailure = minimizedFailure.scenario.tryMinimize(testCfg) ?: break
         }
+        // For distributed algorithms number of nodes and number of crashes and partitions can be minimized as well.
+        if (testCfg is DistributedCTestConfiguration<*>) {
+            return minimizedFailure.minimizeDistributedFailure(testCfg)
+        }
         return minimizedFailure
     }
 
@@ -121,7 +128,11 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
         return null
     }
 
-    private fun ExecutionScenario.tryMinimize(threadId: Int, position: Int, testCfg: CTestConfiguration): LincheckFailure? {
+    private fun ExecutionScenario.tryMinimize(
+        threadId: Int,
+        position: Int,
+        testCfg: CTestConfiguration
+    ): LincheckFailure? {
         val newScenario = this.copy()
         val actors = newScenario[threadId] as MutableList<Actor>
         actors.removeAt(position)
@@ -135,6 +146,53 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
         } else null
     }
 
+    private fun LincheckFailure.minimizeDistributedFailure(
+        distributedTestCfg: DistributedCTestConfiguration<*>
+    ): LincheckFailure {
+        var minimizedFailure = this
+        var testCfg = distributedTestCfg
+        var minimizedOnIteration: Boolean
+        // While possible, tries to decrease number of nodes for each type.
+        do {
+            minimizedOnIteration = false
+            for (nextCfg in testCfg.nextConfigurations(minimizedFailure.scenario)) {
+                val failure =
+                    minimizedFailure.scenario.run(nextCfg, nextCfg.createVerifier(checkStateEquivalence = false))
+                if (failure != null) {
+                    testCfg = nextCfg
+                    minimizedFailure = failure
+                    minimizedOnIteration = true
+                    break
+                }
+            }
+        } while (minimizedOnIteration)
+        // Minimizes number of crashes
+        while (minimizedFailure.crashes > 0) {
+            ProbabilisticModel.crashedNodesExpectation.set(minimizedFailure.crashes - 1)
+            val failure =
+                minimizedFailure.scenario.run(testCfg, testCfg.createVerifier(checkStateEquivalence = false))
+            if (failure != null && failure.crashes < minimizedFailure.crashes) {
+                minimizedFailure = failure
+            } else {
+                break
+            }
+        }
+        // Minimizes number of partitions.
+        while (minimizedFailure.partitions > 0) {
+            ProbabilisticModel.networkPartitionExpectation.set(minimizedFailure.partitions - 1)
+            val failure =
+                minimizedFailure.scenario.run(testCfg, testCfg.createVerifier(checkStateEquivalence = false))
+            if (failure != null && failure.partitions < minimizedFailure.partitions) {
+                minimizedFailure = failure
+            } else {
+                break
+            }
+        }
+        ProbabilisticModel.crashedNodesExpectation.remove()
+        ProbabilisticModel.networkPartitionExpectation.remove()
+        return minimizedFailure
+    }
+
     private fun ExecutionScenario.run(testCfg: CTestConfiguration, verifier: Verifier): LincheckFailure? =
         testCfg.createStrategy(
             testClass = testClass,
@@ -168,27 +226,36 @@ class LinChecker (private val testClass: Class<*>, options: Options<*, *>?) {
         }
     }
 
-    private val ExecutionScenario.hasSuspendableActorsInInitPart get() =
-        initExecution.stream().anyMatch(Actor::isSuspendable)
-    private val ExecutionScenario.hasPostPartAndSuspendableActors get() =
-        (parallelExecution.stream().anyMatch { actors -> actors.stream().anyMatch { it.isSuspendable } } && postExecution.size > 0)
-    private val ExecutionScenario.isParallelPartEmpty get() =
-        parallelExecution.map { it.size }.sum() == 0
+    private val ExecutionScenario.hasSuspendableActorsInInitPart
+        get() =
+            initExecution.stream().anyMatch(Actor::isSuspendable)
+    private val ExecutionScenario.hasPostPartAndSuspendableActors
+        get() =
+            (parallelExecution.stream()
+                .anyMatch { actors -> actors.stream().anyMatch { it.isSuspendable } } && postExecution.size > 0)
+    private val ExecutionScenario.isParallelPartEmpty
+        get() =
+            parallelExecution.map { it.size }.sum() == 0
 
 
     private fun CTestConfiguration.createVerifier(checkStateEquivalence: Boolean) =
-        verifierClass.getConstructor(Class::class.java).newInstance(sequentialSpecification).also {
-            if (!checkStateEquivalence) return@also
-            val stateEquivalenceCorrect = it.checkStateEquivalenceImplementation()
-            if (!stateEquivalenceCorrect) {
-                if (requireStateEquivalenceImplCheck) {
-                    val errorMessage = StringBuilder().appendStateEquivalenceViolationMessage(sequentialSpecification).toString()
-                    error(errorMessage)
-                } else {
-                    reporter.logStateEquivalenceViolation(sequentialSpecification)
-                }
-            }
-        }
+        //TODO: better way
+        if (!DistributedVerifier::class.java.isAssignableFrom(verifierClass))
+            verifierClass.getConstructor(Class::class.java)
+                .newInstance(sequentialSpecification).also {
+                    if (!checkStateEquivalence) return@also
+                    val stateEquivalenceCorrect = it.checkStateEquivalenceImplementation()
+                    if (!stateEquivalenceCorrect) {
+                        if (requireStateEquivalenceImplCheck) {
+                            val errorMessage =
+                                StringBuilder().appendStateEquivalenceViolationMessage(sequentialSpecification)
+                                    .toString()
+                            error(errorMessage)
+                        } else {
+                            reporter.logStateEquivalenceViolation(sequentialSpecification)
+                        }
+                    }
+                } else verifierClass.getConstructor().newInstance()
 
     private fun CTestConfiguration.createExecutionGenerator() =
         generatorClass.getConstructor(

diff --git a/src/jvm/main/org/jetbrains/kotlinx/lincheck/Options.kt b/src/jvm/main/org/jetbrains/kotlinx/lincheck/Options.kt
@@ -31,7 +31,7 @@ import org.jetbrains.kotlinx.lincheck.verifier.*
 abstract class Options<OPT : Options<OPT, CTEST>, CTEST : CTestConfiguration> {
     internal var logLevel = DEFAULT_LOG_LEVEL
     protected var iterations = CTestConfiguration.DEFAULT_ITERATIONS
-    protected var threads = CTestConfiguration.DEFAULT_THREADS
+    protected open var threads = CTestConfiguration.DEFAULT_THREADS
     protected var actorsPerThread = CTestConfiguration.DEFAULT_ACTORS_PER_THREAD
     protected var actorsBefore = CTestConfiguration.DEFAULT_ACTORS_BEFORE
     protected var actorsAfter = CTestConfiguration.DEFAULT_ACTORS_AFTER