Merge remote-tracking branch 'open-source/master' into thomas-merge

core/src/test/kotlin/net/corda/core/flows/FastThreadLocalTest.kt

@@ -0,0 +1,165 @@
+package net.corda.core.flows
+
+import co.paralleluniverse.fibers.Fiber
+import co.paralleluniverse.fibers.FiberExecutorScheduler
+import co.paralleluniverse.fibers.Suspendable
+import co.paralleluniverse.io.serialization.ByteArraySerializer
+import co.paralleluniverse.strands.SuspendableCallable
+import io.netty.util.concurrent.FastThreadLocal
+import io.netty.util.concurrent.FastThreadLocalThread
+import net.corda.core.internal.concurrent.OpenFuture
+import net.corda.core.internal.concurrent.openFuture
+import net.corda.core.internal.rootCause
+import net.corda.core.utilities.getOrThrow
+import org.assertj.core.api.Assertions.catchThrowable
+import org.hamcrest.Matchers.lessThanOrEqualTo
+import org.junit.After
+import org.junit.Assert.assertThat
+import org.junit.Test
+import java.util.*
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+class FastThreadLocalTest {
+    private inner class ExpensiveObj {
+        init {
+            expensiveObjCount.andIncrement
+        }
+    }
+
+    private val expensiveObjCount = AtomicInteger()
+    private lateinit var pool: ExecutorService
+    private lateinit var scheduler: FiberExecutorScheduler
+    private fun init(threadCount: Int, threadImpl: (Runnable) -> Thread) {
+        pool = Executors.newFixedThreadPool(threadCount, threadImpl)
+        scheduler = FiberExecutorScheduler(null, pool)
+    }
+
+    @After
+    fun poolShutdown() = try {
+        pool.shutdown()
+    } catch (e: UninitializedPropertyAccessException) {
+        // Do nothing.
+    }
+
+    @After
+    fun schedulerShutdown() = try {
+        scheduler.shutdown()
+    } catch (e: UninitializedPropertyAccessException) {
+        // Do nothing.
+    }
+
+    @Test
+    fun `ThreadLocal with plain old Thread is fiber-local`() {
+        init(3, ::Thread)
+        val threadLocal = object : ThreadLocal<ExpensiveObj>() {
+            override fun initialValue() = ExpensiveObj()
+        }
+        assertEquals(0, runFibers(100, threadLocal::get))
+        assertEquals(100, expensiveObjCount.get())
+    }
+
+    @Test
+    fun `ThreadLocal with FastThreadLocalThread is fiber-local`() {
+        init(3, ::FastThreadLocalThread)
+        val threadLocal = object : ThreadLocal<ExpensiveObj>() {
+            override fun initialValue() = ExpensiveObj()
+        }
+        assertEquals(0, runFibers(100, threadLocal::get))
+        assertEquals(100, expensiveObjCount.get())
+    }
+
+    @Test
+    fun `FastThreadLocal with plain old Thread is fiber-local`() {
+        init(3, ::Thread)
+        val threadLocal = object : FastThreadLocal<ExpensiveObj>() {
+            override fun initialValue() = ExpensiveObj()
+        }
+        assertEquals(0, runFibers(100, threadLocal::get))
+        assertEquals(100, expensiveObjCount.get())
+    }
+
+    @Test
+    fun `FastThreadLocal with FastThreadLocalThread is not fiber-local`() {
+        init(3, ::FastThreadLocalThread)
+        val threadLocal = object : FastThreadLocal<ExpensiveObj>() {
+            override fun initialValue() = ExpensiveObj()
+        }
+        runFibers(100, threadLocal::get) // Return value could be anything.
+        assertThat(expensiveObjCount.get(), lessThanOrEqualTo(3))
+    }
+
+    /** @return the number of times a different expensive object was obtained post-suspend. */
+    private fun runFibers(fiberCount: Int, threadLocalGet: () -> ExpensiveObj): Int {
+        val fibers = (0 until fiberCount).map { Fiber(scheduler, FiberTask(threadLocalGet)) }
+        val startedFibers = fibers.map { it.start() }
+        return startedFibers.map { it.get() }.count { it }
+    }
+
+    private class FiberTask(private val threadLocalGet: () -> ExpensiveObj) : SuspendableCallable<Boolean> {
+        @Suspendable
+        override fun run(): Boolean {
+            val first = threadLocalGet()
+            Fiber.sleep(1)
+            return threadLocalGet() != first
+        }
+    }
+
+    private class UnserializableObj {
+        @Suppress("unused")
+        private val fail: Nothing by lazy { throw UnsupportedOperationException("Nice try.") }
+    }
+
+    @Test
+    fun `ThreadLocal content is not serialized`() {
+        contentIsNotSerialized(object : ThreadLocal<UnserializableObj>() {
+            override fun initialValue() = UnserializableObj()
+        }::get)
+    }
+
+    @Test
+    fun `FastThreadLocal content is not serialized`() {
+        contentIsNotSerialized(object : FastThreadLocal<UnserializableObj>() {
+            override fun initialValue() = UnserializableObj()
+        }::get)
+    }
+
+    private fun contentIsNotSerialized(threadLocalGet: () -> UnserializableObj) {
+        init(1, ::FastThreadLocalThread)
+        // Use false like AbstractKryoSerializationScheme, the default of true doesn't work at all:
+        val serializer = Fiber.getFiberSerializer(false)
+        val returnValue = UUID.randomUUID()
+        val deserializedFiber = serializer.read(openFuture<ByteArray>().let {
+            Fiber(scheduler, FiberTask2(threadLocalGet, false, serializer, it, returnValue)).start()
+            it.getOrThrow()
+        }) as Fiber<*>
+        assertEquals(returnValue, Fiber.unparkDeserialized(deserializedFiber, scheduler).get())
+        assertEquals("Nice try.", openFuture<ByteArray>().let {
+            Fiber(scheduler, FiberTask2(threadLocalGet, true, serializer, it, returnValue)).start()
+            catchThrowable { it.getOrThrow() }
+        }.rootCause.message)
+    }
+
+    private class FiberTask2(
+            @Transient private val threadLocalGet: () -> UnserializableObj,
+            private val retainObj: Boolean,
+            @Transient private val serializer: ByteArraySerializer,
+            @Transient private val bytesFuture: OpenFuture<ByteArray>,
+            private val returnValue: UUID) : SuspendableCallable<UUID> {
+        @Suspendable
+        override fun run(): UUID {
+            var obj: UnserializableObj? = threadLocalGet()
+            assertNotNull(obj)
+            if (!retainObj) {
+                @Suppress("UNUSED_VALUE")
+                obj = null
+            }
+            // In retainObj false case, check this doesn't attempt to serialize fields of currentThread:
+            Fiber.parkAndSerialize { fiber, _ -> bytesFuture.capture { serializer.write(fiber) } }
+            return returnValue
+        }
+    }
+}

docs/source/api-flows.rst

@@ -682,3 +682,35 @@ We then update the progress tracker's current step as we progress through the fl
         :start-after: DOCSTART 18
         :end-before: DOCEND 18
         :dedent: 12
+
+
+Concurrency, Locking and Waiting
+--------------------------------
+This is an advanced topic.  Because Corda is designed to:
+
+* run many flows in parallel,
+* may persist flows to storage and resurrect those flows much later,
+* (in the future) migrate flows between JVMs,
+
+flows should avoid use of locks and typically not even attempt to interact with objects shared between flows (except
+``ServiceHub`` and other carefully crafted services such as Oracles.  See :doc:`oracles`).
+Locks will significantly reduce the scalability of the node, in the best case, and can cause the node to deadlock if they
+remain locked across flow context switch boundaries (such as sending and receiving
+from peers discussed above, and the sleep discussed below).
+
+If you need activities that are scheduled, you should investigate the use of ``SchedulableState``.
+However, we appreciate that Corda support for some more advanced patterns is still in the future, and if there is a need
+for brief pauses in flows then you should use ``FlowLogic.sleep`` in place of where you might have used ``Thread.sleep``.
+Flows should expressly not use ``Thread.sleep``, since this will prevent the node from processing other flows
+in the meantime, significantly impairing the performance of the node.
+Even ``FlowLogic.sleep`` is not to be used to create long running flows, since the Corda ethos is for short lived flows
+(otherwise upgrading nodes or CorDapps is much more complicated), or as a substitute to using the ``SchedulableState`` scheduler.
+
+Currently the ``finance`` package uses ``FlowLogic.sleep`` to make several attempts at coin selection, where necessary,
+when many states are soft locked and we wish to wait for those, or other new states in their place, to become unlocked.
+
+    .. literalinclude:: ../../finance/src/main/kotlin/net/corda/finance/contracts/asset/cash/selection/AbstractCashSelection.kt
+        :language: kotlin
+        :start-after: DOCSTART CASHSELECT 1
+        :end-before: DOCEND CASHSELECT 1
+        :dedent: 8

docs/source/corda-configuration-file.rst

@@ -1,64 +1,47 @@
 Node configuration
 ==================
 
+.. contents::
+
 File location
 -------------
+When starting a node, the ``corda.jar`` file defaults to reading the node's configuration from a ``node.conf`` file in
+the directory from which the command to launch Corda is executed. There are two command-line options to override this
+behaviour:
 
-The Corda all-in-one ``corda.jar`` file is generated by the ``gradle buildCordaJAR`` task and defaults to reading configuration
+* The ``--config-file`` command line option allows you to specify a configuration file with a different name, or at
-from a ``node.conf`` file in the present working directory. This behaviour can be overidden using the ``--config-file``
+  different file location. Paths are relative to the current working directory
-command line option to target configuration files with different names, or different file location (relative paths are
-relative to the current working directory). Also, the ``--base-directory`` command line option alters the Corda node
-workspace location and if specified a ``node.conf`` configuration file is then expected in the root of the workspace.
 
-The configuration file templates used for the ``gradle deployNodes`` task are to be found in the ``/config/dev`` folder.
+* The ``--base-directory`` command line option allows you to specify the node's workspace location. A ``node.conf``
-Also note that there is a basic set of defaults loaded from the built in resource file ``/node/src/main/resources/reference.conf``
+  configuration file is then expected in the root of this workspace
-of the ``:node`` gradle module. All properties in this can be overidden in the file configuration and for rarely changed
+
-properties this defaulting allows the property to be excluded from the configuration file.
+If you specify both command line arguments at the same time, the node will fail to start.
 
 Format
 ------
+The Corda configuration file uses the HOCON format which is superset of JSON. Please visit
+`<https://github.com/typesafehub/config/blob/master/HOCON.md>`_ for further details.
 
-The Corda configuration file uses the HOCON format which is superset of JSON. It has several features which makes it
+Defaults
-very useful as a configuration format. Please visit their `page <https://github.com/typesafehub/config/blob/master/HOCON.md>`_
-for further details.
-
-Examples
 --------
+A set of default configuration options are loaded from the built-in resource file ``/node/src/main/resources/reference.conf``.
+This file can be found in the ``:node`` gradle module of the `Corda repository <https://github.com/corda/corda>`_. Any
+options you do not specify in your own ``node.conf`` file will use these defaults.
 
-General node configuration file for hosting the IRSDemo services.
+Here are the contents of the ``reference.conf`` file:
 
-.. literalinclude:: example-code/src/main/resources/example-node.conf
+.. literalinclude:: ../../node/src/main/resources/reference.conf
     :language: javascript
 
-Simple Notary configuration file.
-
-.. parsed-literal::
-
-    myLegalName : "O=Notary Service,OU=corda,L=London,C=GB"
-    keyStorePassword : "cordacadevpass"
-    trustStorePassword : "trustpass"
-    p2pAddress : "localhost:12345"
-    rpcSettings = {
-        useSsl = false
-        standAloneBroker = false
-        address : "my-corda-node:10003"
-        adminAddress : "my-corda-node:10004"
-    }
-    webAddress : "localhost:12347"
-    notary : {
-        validating : false
-    }
-    devMode : true
-    compatibilityZoneURL : "https://cz.corda.net"
-
 Fields
 ------
+The available config fields are listed below. ``baseDirectory`` is available as a substitution value and contains the
+absolute path to the node's base directory.
 
-The available config fields are listed below. ``baseDirectory`` is available as a substitution value, containing the absolute
+:myLegalName: The legal identity of the node. This acts as a human-readable alias to the node's public key and can be used with
-path to the node's base directory.
+    the network map to look up the node's info. This is the name that is used in the node's certificates (either when requesting them
-
+    from the doorman, or when auto-generating them in dev mode). At runtime, Corda checks whether this name matches the
-:myLegalName: The legal identity of the node acts as a human readable alias to the node's public key and several demos use
+    name in the node's certificates.
-    this to lookup the NodeInfo.
 
 :keyStorePassword: The password to unlock the KeyStore file (``<workspace>/certificates/sslkeystore.jks``) containing the
     node certificate and private key.
@@ -222,4 +205,32 @@ path to the node's base directory.
             :port: Port the graphite instance is listening at.
             :prefix: Optional prefix string to identify metrics from this node, will default to a string made up
                     from Organisation Name and ip address.
-            :sampleIntervallSeconds: optional wait time between pushing metrics. This will default to 60 seconds.
+            :sampleIntervallSeconds: optional wait time between pushing metrics. This will default to 60 seconds.
+Examples
+--------
+
+General node configuration file for hosting the IRSDemo services:
+
+.. literalinclude:: example-code/src/main/resources/example-node.conf
+   :language: javascript
+
+Simple notary configuration file:
+
+.. parsed-literal::
+
+    myLegalName : "O=Notary Service,OU=corda,L=London,C=GB"
+    keyStorePassword : "cordacadevpass"
+    trustStorePassword : "trustpass"
+    p2pAddress : "localhost:12345"
+    rpcSettings = {
+        useSsl = false
+        standAloneBroker = false
+        address : "my-corda-node:10003"
+        adminAddress : "my-corda-node:10004"
+    }
+    webAddress : "localhost:12347"
+    notary : {
+        validating : false
+    }
+    devMode : true
+    compatibilityZoneURL : "https://cz.corda.net"

docs/source/hello-world-template.rst

@@ -67,7 +67,7 @@ To prevent build errors later on, we should delete the following files before we
 
 * Java: ``cordapp/src/main/java/com/template/TemplateClient.java``
 
-* Kotlin: ``cordapp/src/main/kotlin/com/template/TemplateClient.kt``
+* Kotlin: ``cordapp/src/main/kotlin/com/template/Client.kt``
 
 Progress so far
 ---------------

docs/source/node-database.rst

@@ -3,17 +3,12 @@ Node database
 
 Default in-memory database
 --------------------------
-
+By default, nodes store their data in an H2 database. You can connect directly to a running node's database to see its
-By default nodes store their data in an H2 database.
+stored states, transactions and attachments as follows:
-
-You can connect directly to a running node's database to see its stored states, transactions and attachments as
-follows:
 
 * Download the `h2 platform-independent zip <http://www.h2database.com/html/download.html>`_, unzip the zip, and
   navigate in a terminal window to the unzipped folder
-* Change directories to the bin folder:
+* Change directories to the bin folder: ``cd h2/bin``
-
-  ``cd h2/bin``
 
 * Run the following command to open the h2 web console in a web browser tab:
 
@@ -93,6 +88,17 @@ The property ``database.schema`` is optional. The value of ``database.schema`` i
 to preserve case-sensitivity (e.g. `AliceCorp` becomes `"AliceCorp"`, without quotes PostgresSQL would treat the value as `alicecorp`).
 
 Example node configuration for PostgreSQL:
+=======
+PostgreSQL
+----------
+Nodes also have untested support for PostgreSQL 9.6, using PostgreSQL JDBC Driver 42.1.4.
+
+.. warning:: This is an experimental community contribution, and is currently untested. We welcome pull requests to add
+   tests and additional support for this feature.
+
+Configuration
+~~~~~~~~~~~~~
+Here is an example node configuration for PostgreSQL:
 
 .. sourcecode:: groovy
 
@@ -107,3 +113,9 @@ Example node configuration for PostgreSQL:
         schema = [SCHEMA]
     }
     jarDirs = [PATH_TO_JDBC_DRIVER_DIR]
+
+Note that:
+
+* The ``database.schema`` property is optional
+* The value of ``database.schema`` is not wrapped in double quotes and Postgres always treats it as a lower-case value
+  (e.g. ``AliceCorp`` becomes ``alicecorp``)

docs/source/quickstart-index.rst

@@ -1,10 +1,7 @@
 Quickstart
 ==========
 
-.. toctree::
+* :doc:`Set up your machine for CorDapp development <getting-set-up>`
-   :maxdepth: 1
+* :doc:`Run the Example CorDapp <tutorial-cordapp>`
-
+* `View CorDapps in Corda Explore <http://explore.corda.zone/>`_
-   getting-set-up
+* `Download sample CorDapps <https://www.corda.net/samples/>`_
-   tutorial-cordapp
-   Other CorDapps <https://www.corda.net/samples/>
-   Utilities <https://www.corda.net/utilities/>

docs/source/shell.rst

@@ -186,7 +186,7 @@ Starting a flow
 
 We would start the ``CashIssue`` flow as follows:
 
-``flow start CashIssue amount: $1000, issueRef: 1234, recipient: "O=Bank A,L=London,C=GB", notary: "O=Notary Service,OU=corda,L=London,C=GB"``
+``flow start CashIssueFlow amount: $1000, issuerBankPartyRef: 1234, notary: "O=Controller, L=London, C=GB"``
 
 This breaks down as follows:
 

finance/src/main/kotlin/net/corda/finance/contracts/asset/cash/selection/AbstractCashSelection.kt

@@ -12,7 +12,10 @@ import net.corda.core.node.ServiceHub
 import net.corda.core.node.services.StatesNotAvailableException
 import net.corda.core.utilities.*
 import net.corda.finance.contracts.asset.Cash
-import java.sql.*
+import java.sql.Connection
+import java.sql.DatabaseMetaData
+import java.sql.ResultSet
+import java.sql.SQLException
 import java.util.*
 import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.locks.ReentrantLock
@@ -99,6 +102,7 @@ abstract class AbstractCashSelection(private val maxRetries : Int = 8, private v
                                         withIssuerRefs: Set<OpaqueBytes> = emptySet()): List<StateAndRef<Cash.State>> {
         val stateAndRefs = mutableListOf<StateAndRef<Cash.State>>()
 
+        // DOCSTART CASHSELECT 1
         for (retryCount in 1..maxRetries) {
             if (!attemptSpend(services, amount, lockId, notary, onlyFromIssuerParties, withIssuerRefs, stateAndRefs)) {
                 log.warn("Coin selection failed on attempt $retryCount")
@@ -114,6 +118,7 @@ abstract class AbstractCashSelection(private val maxRetries : Int = 8, private v
                 break
             }
         }
+        // DOCEND CASHSELECT 1
         return stateAndRefs
     }
 
@@ -169,4 +174,4 @@ abstract class AbstractCashSelection(private val maxRetries : Int = 8, private v
         }
         return false
     }
-}
+}

node-api/src/main/kotlin/net/corda/nodeapi/internal/ClassloaderUtils.kt

@@ -24,9 +24,9 @@ fun scanJarForContracts(cordappJarPath: String): List<ContractClassName> {
     val contracts = (scanResult.getNamesOfClassesImplementing(Contract::class.qualifiedName) ).distinct()
 
     // Only keep instantiable contracts
-    val classLoader = URLClassLoader(arrayOf(File(cordappJarPath).toURL()), currentClassLoader)
+    return URLClassLoader(arrayOf(File(cordappJarPath).toURL()), currentClassLoader).use {
-    val concreteContracts = contracts.map(classLoader::loadClass).filter { !it.isInterface && !Modifier.isAbstract(it.modifiers) }
+        contracts.map(it::loadClass).filter { !it.isInterface && !Modifier.isAbstract(it.modifiers) }
-    return concreteContracts.map { it.name }
+    }.map { it.name }
 }
 
 fun <T> withContractsInJar(jarInputStream: InputStream, withContracts: (List<ContractClassName>, InputStream) -> T): T {

node/src/main/kotlin/net/corda/node/internal/ServicesForResolutionImpl.kt

@@ -29,4 +29,4 @@ data class ServicesForResolutionImpl(
             it.value.map { StateAndRef(baseTx.outputs[it.index], it) }
         }.toSet()
     }
-}
+}

node/src/main/kotlin/net/corda/node/services/messaging/RPCServer.kt

@@ -210,7 +210,7 @@ class RPCServer(
         return thread(name = "rpc-server-sender", isDaemon = true) {
             var deduplicationSequenceNumber = 0L
             while (true) {
-                val job = sendJobQueue.poll()
+                val job = sendJobQueue.take()
                 when (job) {
                     is RpcSendJob.Send -> handleSendJob(deduplicationSequenceNumber++, job)
                     RpcSendJob.Stop -> return@thread

node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt

@@ -248,7 +248,7 @@ object BFTSMaRt {
         /** Generates a transaction signature over the specified transaction [txId]. */
         protected fun sign(txId: SecureHash): TransactionSignature {
             val signableData = SignableData(txId, SignatureMetadata(services.myInfo.platformVersion, Crypto.findSignatureScheme(notaryIdentityKey).schemeNumberID))
-            return services.keyManagementService.sign(signableData, notaryIdentityKey)
+            return services.database.transaction { services.keyManagementService.sign(signableData, notaryIdentityKey) }
         }
 
         // TODO:

node/src/main/kotlin/net/corda/node/utilities/AffinityExecutor.kt

@@ -1,6 +1,7 @@
 package net.corda.node.utilities
 
 import com.google.common.util.concurrent.SettableFuture
+import io.netty.util.concurrent.FastThreadLocalThread
 import java.util.*
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.Executor
@@ -55,8 +56,8 @@ interface AffinityExecutor : Executor {
         private val threads = Collections.synchronizedSet(HashSet<Thread>())
 
         init {
-            setThreadFactory(fun(runnable: Runnable): Thread {
+            setThreadFactory { runnable ->
-                val thread = object : Thread() {
+                val thread = object : FastThreadLocalThread() {
                     override fun run() {
                         try {
                             runnable.run()
@@ -68,8 +69,8 @@ interface AffinityExecutor : Executor {
                 thread.isDaemon = true
                 thread.name = threadName
                 threads += thread
-                return thread
+                thread
-            })
+            }
         }
 
         override val isOnThread: Boolean get() = Thread.currentThread() in threads

samples/irs-demo/src/system-test/kotlin/IRSDemoDockerTest.kt

@@ -72,15 +72,9 @@ class IRSDemoDockerTest {
 
         //Wait for deals to appear in a rows table
         val dealsList = driverWait.until<WebElement>({
-            makeScreenshot(driver, "second")
             it?.findElement(By.cssSelector("table#deal-list tbody tr"))
         })
 
         assertNotNull(dealsList)
     }
-
-    private fun makeScreenshot(driver: PhantomJSDriver, name: String) {
-        val screenshotAs = driver.getScreenshotAs(OutputType.FILE)
-        Files.copy(screenshotAs.toPath(), Paths.get("/Users", "maksymilianpawlak", "phantomjs", name + System.currentTimeMillis() + ".png"), StandardCopyOption.REPLACE_EXISTING)
-    }
 }

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/ProcessUtilities.kt

@@ -44,13 +44,12 @@ object ProcessUtilities {
             if (maximumHeapSize != null) add("-Xmx$maximumHeapSize")
             add("-XX:+UseG1GC")
             addAll(extraJvmArguments)
-            add("-cp")
-            add(classpath)
             add(className)
             addAll(arguments)
         }
         return ProcessBuilder(command).apply {
             inheritIO()
+            environment().put("CLASSPATH", classpath)
             if (workingDirectory != null) {
                 redirectError((workingDirectory / "$className.stderr.log").toFile())
                 redirectOutput((workingDirectory / "$className.stdout.log").toFile())

testing/test-utils/build.gradle

@@ -20,6 +20,7 @@ dependencies {
 
     // Unit testing helpers.
     compile "junit:junit:$junit_version"
+    compile 'org.hamcrest:hamcrest-library:1.3'
     compile "com.nhaarman:mockito-kotlin:1.1.0"
 
     // Guava: Google test library (collections test suite)