Merge pull request #25 from RichardBradley/issue-19-windows-race-condition-file-per-thread

#19 Fix the Windows race condition in the measurement file by using one file per thread.

Author: sksamuel

src/main/scala/scoverage/IOUtils.scala

@@ -7,12 +7,12 @@ object IOUtils {
 
   // loads all the invoked statement ids
   def invoked(dir: File): Seq[Int] = {
-    dir.listFiles.map(file => {
+    dir.listFiles.flatMap { file =>
       val reader = new BufferedReader(new FileReader(file))
       val line = reader.readLine()
       reader.close()
       line.split(";").filterNot(_.isEmpty).map(_.toInt)
-    }).flatten.toSeq
+    }
   }
 
   def serialize(coverage: Coverage, file: File) {

src/main/scala/scoverage/Invoker.scala

@@ -8,17 +8,19 @@ object Invoker {
   /**
    * We record that the given id has been invoked by appending its id to the coverage
    * data file.
+   *
    * This will happen concurrently on as many threads as the application is using,
-   * but appending small amounts of data to a file is atomic on both POSIX and Windows
-   * if it is a single write of a small enough string.
+   * so we use one file per thread, named for the thread id.
    *
-   * @see http://stackoverflow.com/questions/1154446/is-file-append-atomic-in-unix
-   * @see http://stackoverflow.com/questions/3032482/is-appending-to-a-file-atomic-with-windows-ntfs
+   * This method is not thread-safe if the threads are in different JVMs, because
+   * the thread IDs may collide.
+   * You may not use `scoverage` on multiple processes in parallel without risking
+   * corruption of the measurement file.
    */
   def invoked(id: Int, path: String) = {
-    val dir = new File(path)
-    dir.mkdirs()
-    val file = new File(path + "/" + Thread.currentThread.getId)
+    // Each thread writes to a separate measurement file, to reduce contention
+    // and because file appends via FileWriter are not atomic on Windows.
+    val file = new File(path, Thread.currentThread.getId.toString)
     val writer = new FileWriter(file, true)
     writer.append(id.toString + ';')
     writer.close()

src/main/scala/scoverage/plugin.scala

@@ -76,7 +76,6 @@ class ScoverageComponent(val global: Global, options: ScoverageOptions)
     def safeSource(tree: Tree): Option[SourceFile] = if (tree.pos.isDefined) Some(tree.pos.source) else None
 
     def invokeCall(id: Int): Tree = {
-      val file = Env.measurementFile(options.dataDir).getAbsolutePath
       Apply(
         Select(
           Select(
@@ -90,7 +89,7 @@ class ScoverageComponent(val global: Global, options: ScoverageOptions)
             Constant(id)
           ),
           Literal(
-            Constant(file)
+            Constant(getMeasurementFilesPath())
           )
         )
       )
@@ -140,6 +139,20 @@ class ScoverageComponent(val global: Global, options: ScoverageOptions)
       }
     }
 
+    /**
+     * Get the configured dir in which the measurement files should be written
+     * (see [[Invoker.invoked()]])
+     *
+     * This path will be hardcoded into the instrumented class files.
+     *
+     * We create the dir now (rather than at client runtime), for efficiency.
+     */
+    def getMeasurementFilesPath(): String = {
+      val dir = Env.measurementFile(options.dataDir).getAbsoluteFile
+      dir.mkdirs()
+      dir.getPath
+    }
+
     def isIncluded(t: Tree): Boolean = {
       new CoverageFilter(options.excludedPackages).isIncluded(t.symbol.fullNameString)
     }

src/test/scala/scoverage/InvokerConcurrencyTest.scala

@@ -0,0 +1,51 @@
+package scoverage
+
+import org.scalatest.FunSuite
+import org.scalatest.BeforeAndAfter
+import java.io.File
+import scala.concurrent._
+import scala.concurrent.duration._
+import ExecutionContext.Implicits.global
+import scala.collection.breakOut
+
+/**
+ * Verify that [[Invoker.invoked()]] is thread-safe
+ */
+class InvokerConcurrencyTest extends FunSuite with BeforeAndAfter {
+
+  val measurementFile = new File("invoker-test.measurement")
+
+  before {
+    deleteMeasurementFiles()
+  }
+
+  test("calling Invoker.invoked on multiple threads does not corrupt the measurement file") {
+
+    val testIds: Set[Int] = (1 to 1000).toSet
+
+    // Create 1k "invoked" calls on the common thread pool, to stress test
+    // the method
+    val futures: List[Future[Unit]] = testIds.map { i: Int =>
+      future {
+        Invoker.invoked(i, measurementFile.toString)
+      }
+    }(breakOut)
+
+    futures.foreach(Await.result(_, 1.second))
+
+    // Now verify that the measurement file is not corrupted by loading it
+    val idsFromFile = IOUtils.invoked(measurementFile).toSet
+
+    idsFromFile === testIds
+  }
+
+  after {
+    deleteMeasurementFiles()
+  }
+
+  private def deleteMeasurementFiles(): Unit = {
+    measurementFile.getAbsoluteFile.getParentFile.listFiles()
+      .filter(_.getName.startsWith(measurementFile.getName))
+      .foreach(_.delete())
+  }
+}