Acting Concurrent

The Akka actor model for Java and Scala

By Dennis M. Sosnoski, ScalableScala.com / Sosnoski Software Associates Ltd

CompletableFuture blocking

    // task definitions
    private static CompletableFuture<Integer> task1(int input) {
        return TimedEventSupport.delayedSuccess(1, input + 1);
    }
    private static CompletableFuture<Integer> task2(int input) {
        return TimedEventSupport.delayedSuccess(2, input + 2);
    }
    private static CompletableFuture<Integer> task3(int input) {
        return TimedEventSupport.delayedSuccess(3, input + 3);
    }
    private static CompletableFuture<Integer> task4(int input) {
        return TimedEventSupport.delayedSuccess(1, input + 4);
    }
    
    /**
     * Run events with blocking waits.
     * 
     * @return future for result (already complete)
     */
    private static CompletableFuture<Integer> runBlocking() {
        Integer i1 = task1(1).join();
        CompletableFuture<Integer> future2 = task2(i1);
        CompletableFuture<Integer> future3 = task3(i1);
        Integer result = task4(future2.join() + future3.join()).join();
        return CompletableFuture.completedFuture(result);
    }

CompletableFuture non-blocking

    /** Run events with composition. */
    private static CompletableFuture<Integer> runNonblocking() {
        return task1(1).thenCompose(i1 -> ((CompletableFuture<Integer>)task2(i1)
            .thenCombine(task3(i1), (i2,i3) -> i2+i3)))
            .thenCompose(i4 -> task4(i4));
    }
    
    /** Run task2 and task3 and combine the results. This is just a refactoring of {@link
     * #runNonblocking()} to make it easier to understand the code. */
    private static CompletableFuture<Integer> runTask2and3(Integer i1) {
        CompletableFuture<Integer> task2 = task2(i1);
        CompletableFuture<Integer> task3 = task3(i1);
        BiFunction<Integer, Integer, Integer> sum = (a, b) -> a + b;
        return task2.thenCombine(task3, sum);
    }
    
    /** Run events with composition. This is just a refactoring of {@link #runNonblocking()} to make
     * it easier to understand the code. */
    private static CompletableFuture<Integer> runNonblockingAlt() {
        CompletableFuture<Integer> task1 = task1(1);
        CompletableFuture<Integer> comp123 = task1.thenCompose(EventComposition::runTask2and3);
        return comp123.thenCompose(EventComposition::task4);
    }

Streams

Automatically partitition work to be done across threads

    private final List<ChunkDistanceChecker> chunkCheckers;
    
    /**
     * Find best match to word executing chunk checkers in parallel. Each chunk checker finds the best
     * match for the set of words it knows. The individual results are combined here to find the 
     * overall best result.
     */
    public DistancePair bestMatch(String target) {
        return chunkCheckers.parallelStream()
            .map(checker -> checker.bestDistance(target))
            .reduce(DistancePair.worstMatch(), (a, b) -> DistancePair.best(a, b));
    }

Great for the special case of parallel operations

Promise / Future blocking

  // task definitions
  def task1(input: Int) = TimedEvent.delayedSuccess(1, input + 1)
  def task2(input: Int) = TimedEvent.delayedSuccess(2, input + 2)
  def task3(input: Int) = TimedEvent.delayedSuccess(3, input + 3)
  def task4(input: Int) = TimedEvent.delayedSuccess(1, input + 4)
  
  /** Run tasks with blocking waits. */
  def runBlocking() = {
    val v1 = Await.result(task1(1), Duration.Inf)
    val future2 = task2(v1)
    val future3 = task3(v1)
    val v2 = Await.result(future2, Duration.Inf)
    val v3 = Await.result(future3, Duration.Inf)
    val v4 = Await.result(task4(v2 + v3), Duration.Inf)
    val result = Promise[Int]
    result.success(v4)
    result.future
  }

Promise / Future non-blocking

Can use standard collections (monadic) operators such as forEach

  /** Run tasks with flatMap. */
  def runFlatMap() = {
    task1(1) flatMap {v1 =>
      val a = task2(v1)
      val b = task3(v1)
      a flatMap { v2 =>
        b flatMap { v3 => task4(v2 + v3) }}
    }
  }

for composition for more complex cases

async macro

Scala macros perform compile-time transformations of code

async macro converts what looks like sequential code to asynchronous:

  /** Run tasks with async macro. */
  def runAsync(): Future[Int] = {
    async {
      val v1 = await(task1(1))
      val a = task2(v1)
      val b = task3(v1)
      await(task4(await(a) + await(b)))
    }
  }

Similar to C# and F# feature, but cleaner and more flexible

Some limitations, but a very powerful abstraction

async details

async has several effects:

• Declares block to be asynchronous
• Executes block asynchronously (by default)
• Returns result as a future
• Allows non-blocking await to be used inside block (suspends until future completed)

Implemented with a generated state machine class

• Attaches to onComplete handling for each await in turn
• Completion passes on failure or moves to next await until done

Limitations a result of implementation:

• Can't use await inside nested closure, class definition
• Can't use await inside nested try / catch block
• Debugging can be confusing (as with other callback approaches)

Scala example setup

Creates an actor of type Root and sends it a Start message

• ! is an alias for the tell method to send a message with no response
• ? is an alias for the ask method to send a message with Future response

timeout is an implicit argument to some of the ask calls on the next slide

  case class Start()
  case class Generate()
  case class Number(n: Int)
  case class Report()
  case class Sum(n: Int)
  
  implicit val timeout = Timeout(2000)
  
  val system = ActorSystem("actor-demo")
  system.actorOf(Props[Root]) ! Start
  Thread sleep 2000
  system shutdown

Scala example actors

  class Root extends Actor {
    def receive = {
      case Start => {
        val summer = context.actorOf(Props[Summer])
        context.actorOf(Props[Generator]).tell(Generate, summer)
        context.actorOf(Props[Generator]).tell(Generate, summer)
        val sum = summer ? Report
        println("Sum of values is " + Await.result(sum, Duration.Inf))
      }
    }
  }

  class Generator extends Actor {
    var lastOut = 0
    def receive = {
      case Generate => {
        lastOut = lastOut + 1
        sender ! Number(lastOut)
      }
    }
  }
  
  class Summer extends Actor {
    var total = 0
    def receive = {
      case Number(n) => total = total + n
      case Report => sender ! Sum(total)
    }
  }

Java example messages and main

    public static final Object START = new Object();
    public static final Object STOP = new Object();
    public static final Object GENERATE = new Object();
    public static final Object REPORT = new Object();

    public static class Number {
        public final int value;
        public Number(int val) {
            value = val;
        }
    }

    public static class Sum {
        public final int value;
        public Sum(int val) {
            value = val;
        }
    }

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("actor-demo-java");
        ActorRef root = system.actorOf(Props.create(Root.class));
        root.tell(START, ActorRef.noSender());
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) { /* ignored */
        }
        root.tell(STOP, ActorRef.noSender());
    }

Java example actors part 1

    public static class Root extends UntypedActor {
        private final ActorRef summer = getContext().actorOf(Props.create(Summer.class));
        @Override
        public void onReceive(Object msg) throws Exception {
            if (msg == START) {
                getContext().actorOf(Props.create(Generator.class)).tell(GENERATE, summer);
                getContext().actorOf(Props.create(Generator.class)).tell(GENERATE, summer);
            } else if (msg == STOP) {
                Future<Object> future = ask(summer, REPORT, 1000);
                Object result = Await.result(future, Duration.create(1, TimeUnit.SECONDS));
                System.out.println("Sum of values is " + ((Sum)result).value);
            } else {
                unhandled(msg);
            }
        }
    }

    public static class Generator extends UntypedActor {
        private int lastOut;
        @Override
        public void onReceive(Object msg) throws Exception {
            if (msg == GENERATE) {
                getSender().tell(new Number(++lastOut), self());
            } else {
                unhandled(msg);
            }
        }
    }

Java example actors part 2

    public static class Summer extends UntypedActor {
        private int total;

        @Override
        public void onReceive(Object msg) throws Exception {
            if (msg instanceof Number) {
                total += ((Number) msg).value;
            } else if (msg == REPORT) {
                getSender().tell(new Sum(total), self());
            } else {
                unhandled(msg);
            }
        }

    }

Akka UDP server

UDP socket server example:

• Bind to listen to a local socket
• Echo any received datagrams back to sender

class UdpEchoService(address: InetSocketAddress) extends Actor {
  import context.system
  IO(Udp) ! Udp.Bind(self, address)

  def receive = {
    case Udp.Bound(_) =>
      context.become(ready(sender))
  }

  def ready(socket: ActorRef): Receive = {
    case Udp.Received(data, remote) =>
      socket ! Udp.Send(data, remote)
    case Udp.Unbind => socket ! Udp.Unbind
    case Udp.Unbound => context.stop(self)
  }
}

Akka UDP client

UDP socket client example:

• "Bind" to a local socket
• Send string messages as UDP to remote address
• Print received UDP messages to console

class UdpEchoClient(local: InetSocketAddress, remote: InetSocketAddress) extends Actor {
  import context.system
  IO(Udp) ! Udp.Bind(self, local)

  def receive = {
    case Udp.Bound(_) =>
      context.become(ready(sender))
  }

  def ready(socket: ActorRef): Receive = {
    case msg: String =>
      socket ! Udp.Send(ByteString(msg, "UTF-8"), remote)
    case Udp.Received(data, _) =>
      println(s"Client received ${data.decodeString("UTF-8")}")
    case Udp.Unbind => socket ! Udp.Unbind
    case Udp.Unbound => context.stop(self)
  }
}