concurrent-programming-in-scala / learning-examples Goto Github PK

View Code? Open in Web Editor NEW

214.0 214.0 106.0 1.19 MB

Code examples for the book Learning Concurrent Programming in Scala

Home Page: http://www.amazon.com/Learning-Concurrent-Programming-Aleksandar-Prokopec/dp/1783281413

License: MIT License

Scala 100.00%

learning-examples's People

Contributors

Stargazers

Watchers

Forkers

axel22 chemccandless davecromberge ilovejs sslavic sioncheng mironor chrisgdatasciencetraining sweaterr websdotcom r-wheeler qyoom yhyuan theonlyianwood meiliangwhut narayana-glassbeam annazhou vmarrazzo starblood ryblovav krasnyanskiy concerned3rdparty adeel-shahzad speedcom benjamintanweihao lannerate sunrenjie gatsbynewton vladmlch ssmylh bash-horatio mkunikow kindlychung ouyangshourui gongbp linearregression lorancechen javierfortea youngkwonjo maxin8899 amoszhou baohan1990 tywe ignaciomosca nagesantosh letonchanh studyindepth k1s djfwan renxiangnan awantik pascal19821003 leoneuwald jakeway zakiindrasukma oneils cnvp2003 tylerfowler pavel-ilyushko engineerkhan cesarchamal nileshkuchekar pobrienjhu hungnguyengoc tiagoooliveira farhadulislam tolsi dhirajhimani ijunjie evgenanufriev cmhteixeira vtiwari227 qiqipchy subhraj07 felal rodrigo-mendes deromed2000 yangfhit wkrea yohannesm bobo82529 duanshuaimin salvalcantara dineshdataengineer traintravel sanjeev1612 vincentk1991 vvsalpatel shipengcheng1230 greatmengqi vladoriginal yanghao sureshmaddela ivandyachenko josewil-creator ryazhapov liang3zy22 yaroslav73 nox213 elivais

learning-examples's Issues

deprecated code elements so soon in the book's lifecycle?

[info] Set current project to concurrency-examples
> set scalacOptions in ThisBuild ++= Seq("-deprecation")
[info] Defining {.}/*:scalacOptions
[info] The new value will be used by compile:scalacOptions
[info] Reapplying settings...
[info] Set current project to concurrency-examples (in build file:/repos/learning-examples/)
> compile
[info] Compiling 50 Scala sources to /repos/learning-examples/target/scala-2.11/classes...
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch4/Promises.scala:131: Adaptation of argument list by inserting () has been deprecated: this is unlikely to be what you want.
[warn]         signature: Promise.success(value: T): Promise.this.type
[warn]   given arguments: <none>
[warn]  after adaptation: Promise.success((): Unit)
[warn]       def run() = p success ()
[warn]                     ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch4/Promises.scala:176: Adaptation of argument list by inserting () has been deprecated: this is unlikely to be what you want.
[warn]         signature: Promise.trySuccess(value: T): Boolean
[warn]   given arguments: <none>
[warn]  after adaptation: Promise.trySuccess((): Unit)
[warn]   cancel trySuccess ()
[warn]          ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch4/Futures.scala:183: match may not be exhaustive.
[warn] It would fail on the following input: Failure(_)
[warn]   longestBuildLine onComplete {
[warn]                               ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch3/Collections.scala:27: trait BufferProxy in package mutable is deprecated: Proxying is deprecated due to lack of use and compiler-level support.
[warn]   val buffer = new mutable.BufferProxy[Int] with mutable.SynchronizedBuffer[Int] {
[warn]                            ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch3/Collections.scala:27: trait SynchronizedBuffer in package mutable is deprecated: Synchronization via traits is deprecated as it is inherently unreliable.  Consider java.util.concurrent.ConcurrentLinkedQueue as an alternative.
[warn]   val buffer = new mutable.BufferProxy[Int] with mutable.SynchronizedBuffer[Int] {
[warn]                                                          ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch3/Processes.scala:47: method lines_! in trait ProcessBuilder is deprecated: Use lineStream_! instead.
[warn]     val proclines = "ps -A".lines_!
[warn]                             ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch3/Processes.scala:67: method lines_! in trait ProcessBuilder is deprecated: Use lineStream_! instead.
[warn]   def files(pattern: String): Stream[String] = s"find /home/ -name $pattern".lines_!
[warn]                                                                              ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch3/Processes.scala:79: method lines_! in trait ProcessBuilder is deprecated: Use lineStream_! instead.
[warn]     val proclines = "ps -A" #| s"grep $pattern" lines_!;
[warn]                                                 ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch6/Composition.scala:78: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]   def randomQuote = Observable.create[String] { obs =>
[warn]                                ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch6/Observables.scala:57: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]   val vms = Observable.create[String] { obs =>
[warn]                        ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch6/Observables.scala:82: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]   val o = Observable.create[String] { obs =>
[warn]                      ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch6/Observables.scala:130: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]     Observable.create { observer =>
[warn]                ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch6/Observables.scala:168: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]     Observable.create { observer =>
[warn]                ^
[warn] /repos/learning-examples/src/main/scala/org/learningconcurrency/ch9/package.scala:54: method create in object Observable is deprecated: Use `apply[T](Subscriber[T] => Unit)` instead
[warn]     Observable.create { obs =>
[warn]                ^
[warn] there were 4 feature warning(s); re-run with -feature for details
[warn] 15 warnings found

> about
[info] This is sbt 0.13.7
[info] The current project is ... learning-examples 1.0
[info] The current project is built against Scala 2.11.1
[info] Available Plugins: sbt.plugins.IvyPlugin, sbt.plugins.JvmPlugin, sbt.plugins.CorePlugin, sbt.plugins.JUnitXmlReportPlugin, com.typesafe.sbteclipse.plugin.EclipsePlugin
[info] sbt, sbt plugins, and build definitions are using Scala 2.10.4

Or should it rather be compiled differently or with certain Scala versions for the sbt build definitions?

Sample solutions

Are there any sample solutions available to ensure one's own?

[error] sbt.librarymanagement.ResolveException: Error downloading org.ensime:sbt-ensime;sbtVersion=1.0;scalaVersion=2.12:1.11.0

/home/tom/.jdks/adopt-openjdk-1.8.0_282/bin/java -Djline.terminal=jline.UnsupportedTerminal -Dsbt.log.noformat=true -Dfile.encoding=UTF-8 -Didea.managed=true -Dfile.encoding=UTF-8 -jar /home/tom/.local/share/JetBrains/IntelliJIdea2020.3/Scala/launcher/sbt-launch.jar
[info] welcome to sbt 1.4.3 (AdoptOpenJDK Java 1.8.0_282)
[info] loading settings for project learning-examples-build from plugins.sbt ...
[info] loading project definition from /home/tom/code/learning-examples/project
[warn]
[warn] 	Note: Some unresolved dependencies have extra attributes.  Check that these dependencies exist with the requested attributes.
[warn] 		org.ensime:sbt-ensime:1.11.0 (sbtVersion=1.0, scalaVersion=2.12)
[warn]
[warn] 	Note: Unresolved dependencies path:
[error] sbt.librarymanagement.ResolveException: Error downloading org.ensime:sbt-ensime;sbtVersion=1.0;scalaVersion=2.12:1.11.0
[error]   Not found
[error]   Not found
[error]   not found: https://repo1.maven.org/maven2/org/ensime/sbt-ensime_2.12_1.0/1.11.0/sbt-ensime-1.11.0.pom
[error]   not found: /home/tom/.ivy2/local/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml
[error]   not found: https://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml
[error]   not found: https://repo.typesafe.com/typesafe/ivy-releases/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml
[error] 	at lmcoursier.CoursierDependencyResolution.unresolvedWarningOrThrow(CoursierDependencyResolution.scala:258)
[error] 	at lmcoursier.CoursierDependencyResolution.$anonfun$update$38(CoursierDependencyResolution.scala:227)
[error] 	at scala.util.Either$LeftProjection.map(Either.scala:573)
[error] 	at lmcoursier.CoursierDependencyResolution.update(CoursierDependencyResolution.scala:227)
[error] 	at sbt.librarymanagement.DependencyResolution.update(DependencyResolution.scala:60)
[error] 	at sbt.internal.LibraryManagement$.resolve$1(LibraryManagement.scala:53)
[error] 	at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$12(LibraryManagement.scala:103)
[error] 	at sbt.util.Tracked$.$anonfun$lastOutput$1(Tracked.scala:73)
[error] 	at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$20(LibraryManagement.scala:116)
[error] 	at scala.util.control.Exception$Catch.apply(Exception.scala:228)
[error] 	at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$11(LibraryManagement.scala:116)
[error] 	at sbt.internal.LibraryManagement$.$anonfun$cachedUpdate$11$adapted(LibraryManagement.scala:97)
[error] 	at sbt.util.Tracked$.$anonfun$inputChangedW$1(Tracked.scala:219)
[error] 	at sbt.internal.LibraryManagement$.cachedUpdate(LibraryManagement.scala:130)
[error] 	at sbt.Classpaths$.$anonfun$updateTask0$5(Defaults.scala:3484)
[error] 	at scala.Function1.$anonfun$compose$1(Function1.scala:49)
[error] 	at sbt.internal.util.$tilde$greater.$anonfun$$u2219$1(TypeFunctions.scala:62)
[error] 	at sbt.std.Transform$$anon$4.work(Transform.scala:68)
[error] 	at sbt.Execute.$anonfun$submit$2(Execute.scala:282)
[error] 	at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:23)
[error] 	at sbt.Execute.work(Execute.scala:291)
[error] 	at sbt.Execute.$anonfun$submit$1(Execute.scala:282)
[error] 	at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
[error] 	at sbt.CompletionService$$anon$2.call(CompletionService.scala:64)
[error] 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[error] 	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
[error] 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[error] 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
[error] 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
[error] 	at java.lang.Thread.run(Thread.java:748)
[error] (update) sbt.librarymanagement.ResolveException: Error downloading org.ensime:sbt-ensime;sbtVersion=1.0;scalaVersion=2.12:1.11.0
[error]   Not found
[error]   Not found
[error]   not found: https://repo1.maven.org/maven2/org/ensime/sbt-ensime_2.12_1.0/1.11.0/sbt-ensime-1.11.0.pom
[error]   not found: /home/tom/.ivy2/local/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml
[error]   not found: https://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml
[error]   not found: https://repo.typesafe.com/typesafe/ivy-releases/org.ensime/sbt-ensime/scala_2.12/sbt_1.0/1.11.0/ivys/ivy.xml

Unit test for Concurrent Code

I almost finished reading the book. Thanks @axel22 for writing and publishing the book. I really enjoy reading it. It was also very informative for me.

One of the questions that I left after reading the book: is it possible to run unit testing suite for concurrent code?

Q: `trySuccess` vs `success`

Within fileCreated (page 122), why use trySuccess instead of success (assumed -Ywarn-value-discard is off)?

Just before this example, you stated:

As long as we do not use the trySuccess, tryFailure, and tryComplete methods, and none of the success, failure, and complete methods ever throws an exception, we can use promises and retain determinism in our programs.

Can you elaborate why, e.g., trySuccess does not retain determinism?

Deterministic means that, given that no exception is thrown in the program, the result of the program will always be the same, regardless of the execution schedule of the parallel program.

So why does trySuccess violate here? Or, how does trySuccess depend on the execution schedule?

What does 'the last event' mean in Ex4 of Ch6 ?

I don't know what 'the last event' means in Exercise 4 of Chapter 6.
With this answer, 'the last event' means the event emitted just before completed:

val o = Subject[String]()
val s = o.toSignal

o.onNext("a")
println(s()) // null

o.onNext("b")
println(s()) // null

o.onCompleted()
println(s()) // b

But I guess that 'the last event' means the event emitted just before Signal.apply is called:

val o = Subject[String]()
val s = o.toSignal

o.onNext("a")
println(s()) // a

o.onNext("b")
println(s()) // b

o.onCompleted()
println(s()) // b

What does 'the last event' mean ?

First I wanna thank you for makings this book. Currently taking Parallel Computing at SUNY Oswego with Professor Doug Lea. I think there is a race condition in your CH3 FileSystem example. When the logger is created in the constructor it may access the messages queue before it is instantiated below, resulting in a NullPointerException. I just fixed the issue by pulling up the line instantiating the messages queue up before the logger is created.

Again, thank you for creating your book, it is helping me through Professor Lea's class immensely. Cheers.

Q: timeout & callback

On page 124 you stated

The future returned by timeout can be used to install a callback

Are the following mental extensions correct?

The future returned by timeout can be used to install a callback which will be called not before the timeout

The future returned by timeout can be used to install a callback which will be called after the timeout occurred

`AtomicBuffer#+=` needs to be final or private

Hi,

class AtomicBuffer[T] {

  private val buffer = new AtomicReference[List[T]](Nil)

  @tailrec def +=(x: T): Unit = {
    val xs = buffer.get
    val nxs = x :: xs
    if (!buffer.compareAndSet(xs, nxs))
      this += x // retry
    else ()
  }

}

I get the following compiler error:

could not optimize @tailrec annotated method +=: it is neither private nor final so can be overridden

Adding final works just fine.

Something wrong in ch9 about ParLongAccumulator

The source code:

  class ParLongAccumulator(z: Long)(op: (Long, Long) => Long) {
    private val par = Runtime.getRuntime.availableProcessors * 128
    private val values = new AtomicLongArray(par)
    @tailrec final def add(v: Long): Unit = {
      val id = Thread.currentThread.getId.toInt
      val i = math.abs(scala.util.hashing.byteswap32(id)) % par
      val ov = values.get(i)
      val nv = op(ov, v)
      if (!values.compareAndSet(i, ov, nv)) add(v)
    }
    def apply(): Long = {
      var total = 0L
      for (i <- 0 until values.length) total = op(total, values.get(i))
      total
    }
  }

missing z.maybe total=0L should be placed by total=z

Q: timeout and `or` combinator

On page 125 you demo cancellation future.

What is the difference between the following two code snippets:

object PromisesAndTimers extends App {

  implicit class FutureOps[T](val self: Future[T]) {
    def or(that: Future[T]): Future[T] = {
      val p = Promise[T]
      self onComplete { case x => println(x); p tryComplete x }
      that onComplete { case y => println(y); p tryComplete y }
      p.future
    }
  }

  { //[1]

    println("Start...")
    val f = timeout(1000).map(_ => "[1] timeout!") or Future {
      Thread.sleep(999)
      "[1] work completed!"
    }
    f foreach { case text => log(text) }

  }

  { //[2]

    println("Start...")

    // `f0` gets completed with `()` after 1sec
    val f0 = timeout(1000)

    // `f1` gets completed with "timeout" after
    // the mapping over the completed `f0`
    val f1 = f0.map(_ => "[2] timeout!")

    // `f2` gets completed with "work completed" after 999ms
    val f2 = Future {
      Thread.sleep(999)
      "[2] work completed!"
    }

    (f1 or f2) foreach { case text => log(text) }

  }

  Thread.sleep(2000)

}

I increased the last sleep to 2000 so that I can see that eventually all threads are completed --- added printlns to or combinator.

So my questions:

Are comments in [2] correct?

Yes.

I ran PromiseAndTimer several times directly after one another, and each time [1] prints “timeout” and [2] prints “work completed”. Why, in case of [1], is 999 not enough — although this is your original example?

My guess is that this is a combination of your particular processor type, OS and the JVM version. I get pretty non-deterministic results:

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Compiling 1 Scala source to
C:\cygwin\home\axel22\workspaces\scala\learning-examples\target\scala-2.11\classes...
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-5: [1] timeout!
Start...
ForkJoinPool-1-worker-7: [2] work completed!
[success] Total time: 5 s, completed May 21, 2015 10:51:26 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-15: [1] work completed!
Start...
ForkJoinPool-1-worker-7: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:31 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-9: [1] work completed!
Start...
ForkJoinPool-1-worker-15: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:36 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-7: [1] work completed!
Start...
ForkJoinPool-1-worker-15: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:41 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-5: [1] timeout!
Start...
ForkJoinPool-1-worker-15: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:46 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-7: [1] work completed!
Start...
ForkJoinPool-1-worker-15: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:50 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-5: [1] work completed!
Start...
ForkJoinPool-1-worker-7: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:51:54 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-15: [1] work completed!
Start...
ForkJoinPool-1-worker-5: [2] work completed!
[success] Total time: 4 s, completed May 21, 2015 10:51:59 PM

> runMain org.learningconcurrency.ch4.PromisesAndTimers
[info] Running org.learningconcurrency.ch4.PromisesAndTimers
ForkJoinPool-1-worker-15: [1] work completed!
Start...
ForkJoinPool-1-worker-7: [2] timeout!
[success] Total time: 4 s, completed May 21, 2015 10:52:03 PM

Code I used:

object PromisesAndTimers extends App {
  import java.util._
  import scala.concurrent._
  import ExecutionContext.Implicits.global
  import PromisesAndCustomOperations._

  private val timer = new Timer(true)

  def timeout(millis: Long): Future[Unit] = {
    val p = Promise[Unit]
    timer.schedule(new TimerTask {
      def run() = p.success(())
    }, millis)
    p.future
  }

  val f = timeout(1000).map(_ => "[1] timeout!") or Future {
    Thread.sleep(999)
    "[1] work completed!"
  }

  f foreach {
    case text => log(text)
  }

  Thread.sleep(2000)

  { //[2]

    println("Start...")

    // `f0` gets completed with `()` after 1sec
    val f0 = timeout(1000)

    // `f1` gets completed with "timeout" after
    // the mapping over the completed `f0`
    val f1 = f0.map(_ => "[2] timeout!")

    // `f2` gets completed with "work completed" after 999ms
    val f2 = Future {
      Thread.sleep(999)
      "[2] work completed!"
    }

    (f1 or f2) foreach { case text => log(text) }

  }

  Thread.sleep(2000)
}

Decreasing to 900 yields “work completed” in the foreach, though.

What is the difference between [1] and [2]?

There should be no difference as far as I can tell. This might be a particular artifact of how JIT optimizes code, or maybe the artifact of how OS schedules threads. The fact that I cannot reproduce it is in favour.

Just to be sure, I inspected the bytecode -- you can see below that in both cases, the sequence of operations is:

timeout
map
Future.apply
or

  public final void
delayedEndpoint$org$learningconcurrency$ch4$PromisesAndTimers$1();
    flags: ACC_PUBLIC, ACC_FINAL

    LineNumberTable:
      line 126: 0
      line 136: 12
      line 141: 69
      line 145: 91
      line 149: 97
      line 152: 105
      line 156: 113
      line 159: 133
      line 164: 153
      line 168: 182
    LocalVariableTable:
      Start  Length  Slot  Name   Signature
             0     189     0  this
Lorg/learningconcurrency/ch4/PromisesAndTimers$;
           113      69     1    f0   Lscala/concurrent/Future;
           133      49     2    f1   Lscala/concurrent/Future;
           153      29     3    f2   Lscala/concurrent/Future;
    Code:
      stack=5, locals=4, args_size=1
         0: aload_0
         1: new           #83                 // class java/util/Timer
         4: dup
         5: iconst_1
         6: invokespecial #102                // Method
java/util/Timer."<init>":(Z)V
         9: putfield      #63                 // Field
timer:Ljava/util/Timer;
        12: aload_0
        13: getstatic     #107                // Field
org/learningconcurrency/ch4/PromisesAndCustomOperations$.MODULE$:Lorg/learningconcurrency/ch4/PromisesAndCustomOperations$;
        16: aload_0
        17: ldc2_w        #108                // long 1000l
        20: invokevirtual #111                // Method
timeout:(J)Lscala/concurrent/Future;
        23: new           #113                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$9
        26: dup
        27: invokespecial #114                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$9."<init>":()V
        30: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
        33: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
        36: invokeinterface #129,  3          // InterfaceMethod
scala/concurrent/Future.map:(Lscala/Function1;Lscala/concurrent/ExecutionContext;)Lscala/concurrent/Future;
        41: invokevirtual #133                // Method
org/learningconcurrency/ch4/PromisesAndCustomOperations$.FutureOps:(Lscala/concurrent/Future;)Lorg/learningconcurrency/ch4/PromisesAndCustomOperations$FutureOps;
        44: getstatic     #138                // Field
scala/concurrent/Future$.MODULE$:Lscala/concurrent/Future$;
        47: new           #140                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$10
        50: dup
        51: invokespecial #141                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$10."<init>":()V
        54: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
        57: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
        60: invokevirtual #144                // Method
scala/concurrent/Future$.apply:(Lscala/Function0;Lscala/concurrent/ExecutionContext;)Lscala/concurrent/Future;
        63: invokevirtual #150                // Method
org/learningconcurrency/ch4/PromisesAndCustomOperations$FutureOps.or:(Lscala/concurrent/Future;)Lscala/concurrent/Future;
        66: putfield      #98                 // Field
f:Lscala/concurrent/Future;
        69: aload_0
        70: invokevirtual #152                // Method
f:()Lscala/concurrent/Future;
        73: new           #154                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$11
        76: dup
        77: invokespecial #155                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$11."<init>":()V
        80: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
        83: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
        86: invokeinterface #159,  3          // InterfaceMethod
scala/concurrent/Future.foreach:(Lscala/Function1;Lscala/concurrent/ExecutionContext;)V
        91: ldc2_w        #160                // long 2000l
        94: invokestatic  #166                // Method
java/lang/Thread.sleep:(J)V
        97: getstatic     #171                // Field
scala/Predef$.MODULE$:Lscala/Predef$;
       100: ldc           #173                // String Start...
       102: invokevirtual #177                // Method
scala/Predef$.println:(Ljava/lang/Object;)V
       105: aload_0
       106: ldc2_w        #108                // long 1000l
       109: invokevirtual #111                // Method
timeout:(J)Lscala/concurrent/Future;
       112: astore_1
       113: aload_1
       114: new           #179                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$12
       117: dup
       118: invokespecial #180                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$12."<init>":()V
       121: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
       124: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
       127: invokeinterface #129,  3          // InterfaceMethod
scala/concurrent/Future.map:(Lscala/Function1;Lscala/concurrent/ExecutionContext;)Lscala/concurrent/Future;
       132: astore_2
       133: getstatic     #138                // Field
scala/concurrent/Future$.MODULE$:Lscala/concurrent/Future$;
       136: new           #182                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$13
       139: dup
       140: invokespecial #183                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$13."<init>":()V
       143: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
       146: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
       149: invokevirtual #144                // Method
scala/concurrent/Future$.apply:(Lscala/Function0;Lscala/concurrent/ExecutionContext;)Lscala/concurrent/Future;
       152: astore_3
       153: getstatic     #107                // Field
org/learningconcurrency/ch4/PromisesAndCustomOperations$.MODULE$:Lorg/learningconcurrency/ch4/PromisesAndCustomOperations$;
       156: aload_2
       157: invokevirtual #133                // Method
org/learningconcurrency/ch4/PromisesAndCustomOperations$.FutureOps:(Lscala/concurrent/Future;)Lorg/learningconcurrency/ch4/PromisesAndCustomOperations$FutureOps;
       160: aload_3
       161: invokevirtual #150                // Method
org/learningconcurrency/ch4/PromisesAndCustomOperations$FutureOps.or:(Lscala/concurrent/Future;)Lscala/concurrent/Future;
       164: new           #185                // class
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$14
       167: dup
       168: invokespecial #186                // Method
org/learningconcurrency/ch4/PromisesAndTimers$$anonfun$14."<init>":()V
       171: getstatic     #119                // Field
scala/concurrent/ExecutionContext$Implicits$.MODULE$:Lscala/concurrent/ExecutionContext$Implicits$;
       174: invokevirtual #123                // Method
scala/concurrent/ExecutionContext$Implicits$.global:()Lscala/concurrent/ExecutionContextExecutor;
       177: invokeinterface #159,  3          // InterfaceMethod
scala/concurrent/Future.foreach:(Lscala/Function1;Lscala/concurrent/ExecutionContext;)V
       182: ldc2_w        #160                // long 2000l
       185: invokestatic  #166                // Method
java/lang/Thread.sleep:(J)V
       188: return

So, there should really be no difference. Try inverting the order of [1] and [2] to see if you get an opposite effect.

FYI, in case of [2] I separated the futures to make clear that the or combinator actually is operating on f1 and f2 and to make clear that actually three futures are created (f0, f1, and f2) --- neglecting the one created by the or combinator`.

Sure.

The difference might be more pronounced if the or operator were to take a by-name parameter. In this case f2 would be evaluated only after or is invoked in [1], but earlier in [2]. Which is, in this case, not a huge time difference, but might influence the execution schedule somewhat.

`success` in `timeout` needs a value

On page 123

def timeout(t: Long): Future[Unit] = {
  val p = Promise[Unit]
  timer.schedule(new TimerTask {
    def run() = {
      p success ()   // <<< p success (())
      timer.cancel()
    } 
  }, t)
  p.future 
}

It's only a warning, but still ;)

Possible error on the book? [both editions]

I might have found an error. Happy to submit errata @ http://packtpub.com/submit-errata, if I am right.

This concerns a snippet of code on the book, not on the exercises.

On chapter 2, section Volatile variables, page 53 (on paper print of 1st edition) you have the following snippet of code to illustrate volatile variables:

class Page(val txt: String, var position: Int)

object Volatile extends App {
  val pages = for (i <- 1 to 5) yield
    new Page("Na" * (100 - 20 * i) + " Batman!", -1)
  @volatile var found = false
  for (p <- pages) yield thread {
    var i = 0
    while (i < p.txt.length && !found)
      if (p.txt(i) == '!') {
        p.position = i
        found = true
      } else i += 1
  }
  while (!found) {}
  log(s"results: ${pages.map(_.position)}")
}

On the following page, you say:

For the purposes of this example, the main thread busy-waits until it reads found, which is
true. It then prints the positions. Note that a write to position occurs before the write to
found in the spawned threads, which in turn occurs before reading found in the main
thread. This means that the main thread always sees the write of the thread that set found,
and hence prints at least one position other than -1.

This makes total sense.

However, shouldn't the variable position on class Page also be volatile?
I am concerned about a scenario on which a spawned thread finds an exclamation point, but the main thread prints the position -1.
Is there anything forbidding the following scenario:

A spawned thread finds an exclamation point.
It updates the variable position only to the local caches, not to main memory because position is not volatile.
It updates variable found to main memory.
The main thread reads found to be true.
The main thread prints the results, but the page which was found to contain an exclamation point by the spawned thread still has the -1 on the position variable.

Ch8 Actors - CheckActors gives dead letter errors

Hello, I'm having trouble with the CheckActors example in the book. When I try to run the following code:

class CheckActor extends Actor {

  import akka.actor.{Identify, ActorIdentity}
  val log = Logging(context.system, this)

  override def receive = {
    case path: String =>
      log.info(s"checking path $path")
      log.info(context.self.toString)
      log.info(context.parent.toString)
      context.actorSelection(path) ! Identify(path)
    case ActorIdentity(path, Some(ref)) =>
      log.info(s"found actor $ref at $path")
    case ActorIdentity(path, None) =>
      log.info(s"count not find an actor at path")

  }

}

// In main:
    val checker:ActorRef = ourSystem.actorOf(Props[CheckActor], "checker")
    checker ! "../*"

I get the following error (when running in the sbt console):

[INFO] [08/03/2016 14:55:29.998] [OurExampleSystem-akka.actor.default-dispatcher-2] [akka://OurExampleSystem/user/checker] checking path ../*
[INFO] [08/03/2016 14:55:29.999] [OurExampleSystem-akka.actor.default-dispatcher-2] [akka://OurExampleSystem/user/checker] Actor[akka://OurExampleSystem/user/checker#419406266]
[INFO] [08/03/2016 14:55:29.999] [OurExampleSystem-akka.actor.default-dispatcher-2] [akka://OurExampleSystem/user/checker] Actor[akka://OurExampleSystem/user]
[INFO] [08/03/2016 14:55:30.006] [OurExampleSystem-akka.actor.default-dispatcher-3] [akka://OurExampleSystem/user/checker] Message [akka.actor.ActorIdentity] from Actor[akka://OurExampleSystem/deadLetters] to Actor[akka://OurExampleSystem/user/checker#419406266] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[success] Total time: 0 s, completed Aug 3, 2016 2:55:30 PM

I tried it with the code from the code files too, and get the same result (without the extra logging statements). Can you suggest what I might be doing wrong? Thanks.

Isn't ensime dead?

The ensime dependency causes some problems while trying to build, at least for me.

learning-examples/build.sbt

Line 8 in 5fccc9c

ensimeScalaVersion in ThisBuild := "2.11.1"

Deadlock at page 45: Why `t1` locks `a` and `t2` locks `b`?

Why can it happen that thread t1 locks a and t2 locks b?

b is sychronized within the a.sychronized block and if t1 has locked a, how can it be that t2 enters the a.synchronized block to lock b?

What am I missing here?

class Account(val name: String, var money: Int)

def send(a: Account, b: Account, n: Int) = a.synchronized {
  b.synchronized {
    a.money -= n
    b.money += n
  }
}

val a = new Account("Jack", 1000)
val b = new Account("Jill", 2000)
val t1 = thread { for (i<- 0 until 100) send(a, b, 1) }
val t2 = thread { for (i<- 0 until 100) send(b, a, 1) }
t1.join(); t2.join()
log(s"a = ${a.money}, b = ${b.money}")

CAS example

On page 69 you state:

The compare-and-set operation, sometimes called compare-and-swap (CAS), takes the expected previous value and the new value for the atomic variable and atomically replaces the current value with the new value only if the current value is equal to the expected value.

With respect to your code example, I guess "the expected previous value" is ov, and "the new value" is nv.

def compareAndSet(ov: Long, nv: Long): Boolean =
  this.synchronized {
    if (this.get == ov) false else {
      this.set(nv)
      true
    } 
  }

So, "if the current value is equal to the expected value" translates to if (this.get == ov). If equality holds the replacement is executed. But within the code snippet it is the other way around, no?

Learning Resource

Thanks for sharing your repo for learning concurrency in Scala. I'm wondering what book are you using for exercises?

Is the CAS implemention demo wrong?

in the page 113 of the book:
def compareAndSet(ov: Long, nv: Long): Boolean=
this.synchronized {
if (this.get == ov) false else {
this.set(nv)
true
}
}

Is the result of condition wrong ?
if (this.get == ov) this.set(nv) true else { false }

Example org.learningconcurrency.ch6.ObservablesSubscriptions runs infinitely

We'd expect subscription.unsubscribe() to causes fileMonitor.stop() to be invoked. But that is not the case. In my environment, the monitor runs infinitely. Any suggestions?

Misleading exercice solution ch2, ex4 ?

I believe the solution introduces a race condition when there are more than 1 producer threads.
While the exercise describes only 1 producer thread, I would say it is still a bit misleading for the reader (if I am correct of course.).
Do you agree?

val producer = thread {
    var x = 0
    while (x < 15) {
      if (syncVar.isEmpty) {
        syncVar.put(x)
        x = x + 1
      }

    }
  }

class SyncVar[T] {

    private var empty: Boolean = true

    private var x: T = null.asInstanceOf[T]


    def put(x: T): Unit = this.synchronized {
      if (!empty) throw new Exception("must be empty")
      else {
        empty = false
        this.x = x
      }
    }

    def isEmpty = synchronized {
      empty
    }

[not shown code]

  }

The call to methods isEmpty and put(x:T) is synchronized. However on the producer thread, they are called on separate lines. Is the following possible:

Some producer Thread-1 calls the method isEmpty, obtaining the monitor of the syncVar instance.
The isEmptymethod returns true.
The producer Thread-1 releases the monitor over the syncVar instance.
Some other producer Thread-2 calls the same method isEmpty on the same syncVar instance.
It also obtains the value of true.
Producer Thread-2 calls method put(x:T), obtaining the monitor, and effectively inserting a value.
Producer Thread-1 calls method put(x:T), obtains the monitor, but because the syncVar variable x is no longer empty, due to step 6, if throws an exception.

ArrayIndexOutOfBoundsException

When I try to run FTP Client, I get ArrayIndexOutOfBoundsException.

com.intellij.rt.execution.application.AppMain org.learningconcurrency.ch9.FTPClient
Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 0
at org.learningconcurrency.ch9.FTPClient$.main(FTPClient.scala:305)
at org.learningconcurrency.ch9.FTPClient.main(FTPClient.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)

Process finished with exit code 1