Pollak D. Beginning Scala
Подождите немного. Документ загружается.
CHAPTER 6 ■ ACTORS AND CONCURRENCY
157
this.start
}
object TestAcct {
def doTest() = {
val dpp = new Acct
dpp ! Update("Savings", v => (v getOrElse 0) + 1000)
dpp ! Update("Checking", v => (v getOrElse 0) + 100)
val archer = new Acct
archer ! Update("Savings", v => (v getOrElse 0) + 2000)
archer ! Update("Checking", v => (v getOrElse 0) + 50)
println("Initial balances:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
var xid = 1
def transfer(who: Actor, from: String, to: String, amount: Int): Boolean = {
xid += 1
who ! BeginXAction(xid)
who !? (500, (xid, GetInfo)) match {
case Some(Info(bal)) =>
if (bal.getOrElse(from, 0) > amount) {
who ! (xid, Update(from, v => (v getOrElse 0) - amount))
who ! (xid, Update(to, v => (v getOrElse 0) + amount))
who ! (xid, CommitXAction)
true
} else {
who ! (xid, RollbackXAction)
false
}
case _ => who ! (xid, RollbackXAction)
false
}
}
transfer(dpp, "Savings", "Checking", 700)
println("xfer 1 dpp: "+(dpp !? GetInfo))
transfer(dpp, "Savings", "Checking", 700)
println("xfer 2 dpp: "+(dpp !? GetInfo))
19897ch06.fm Page 157 Monday, April 20, 2009 11:10 AM
158
CHAPTER 6
■ ACTORS AND CONCURRENCY
def transfer2(src: Actor, sact: String, dest: Actor,
dact: String, amount: Int): Boolean = {
xid += 1
src ! BeginXAction(xid)
dest ! BeginXAction(xid)
(src !? (500, (xid, GetInfo)), dest !? (500, (xid, GetInfo))) match {
case (Some(Info(sbal)), Some(Info(dbal))) =>
dest ! (xid, Update(dact, v => (v getOrElse 0) + amount))
if (sbal.getOrElse(sact, 0) > amount) {
src ! (xid, Update(sact, v => (v getOrElse 0) - amount))
src ! (xid, CommitXAction)
dest ! (xid, CommitXAction)
true
} else {
src ! (xid, RollbackXAction)
dest ! (xid, RollbackXAction)
false
}
case _ =>
src ! (xid, RollbackXAction)
dest ! (xid, RollbackXAction)
false
}
}
transfer2(dpp, "Checking", archer, "Checking", 700)
println("XFer 700 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
transfer2(dpp, "Checking", archer, "Checking", 700)
println("Again, XFer 700 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
transfer2(dpp, "Checking", archer, "Checking", 10)
println("XFer 10 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
}
}
19897ch06.fm Page 158 Monday, April 20, 2009 11:10 AM
CHAPTER 6 ■ ACTORS AND CONCURRENCY
159
Let’s step through the code. First some imports:
import scala.actors.Actor
import Actor._
import scala.actors.TIMEOUT
The TIMEOUT message will be useful for staying in the transaction boundary for a certain
period of time. Next, let’s define some messages:
case object GetInfo
case class Info(i: Map[String, Int])
case class SetInfo(n: String, v: Int)
case class Update(n: String, f: Option[Int] => Int)
case class BeginXAction(id: Int)
case object CommitXAction
case object RollbackXAction
The XAction messages allow us to define the transactional boundaries. Next, let’s define
an
Acct, account, Actor:
class Acct extends Actor {
private var info: Map[String, Int] = Map()
private var service = (normal, false)
The service variable defines how we’re going to service incoming requests. By default,
requests will be handled with the
normal handler, but if we’re in a transaction, the handler
can be changed. Next, let’s define the
normal handler:
private def normal: PartialFunction[Any, Unit] = {
case GetInfo => reply(Info(info))
case SetInfo(n, v) => info += n -> v
case Update(n, f) => info += n -> f(info.get(n))
case BeginXAction(id) => begin(id)
}
This looks just like our previous example, Listing 6-4, except this time, we have the
BeginXAction message. This message calls the begin method with the transaction ID.
private def begin(xActionId: Int) {
We now capture the current state of the Actor—both the info and the mechanism for
servicing incoming messages:
val oldInfo = info // capture
val oldService = service
19897ch06.fm Page 159 Monday, April 20, 2009 11:10 AM
160
CHAPTER 6
■ ACTORS AND CONCURRENCY
Next, we build up a partial function that will service incoming requests. If there’s a
TIMEOUT, we roll back the transaction by replacing the current state with the state at the
beginning of the transaction:
val tmp: PartialFunction[Any, Unit] = {
case TIMEOUT => // Rollback
info = oldInfo
service = oldService
If we get a Pair that contains the current transaction ID and the RollbackXAction
message, we roll back the transaction:
case (n, RollbackXAction) if n == xActionId =>
info = oldInfo
service = oldService
If we get a Pair that contains the current transaction ID and the CommitXAction message,
we commit the transaction by reverting to the old servicing mechanism and leaving the
data the way it was mutated by the current transaction:
case (n, CommitXAction) if n == xActionId => // Commit
service = oldService
If we get any other message and it has the correct transaction ID and can be handled by
the normal message handler, we pass it on to the normal message handler:
case (n, v) if n == xActionId &&
normal.isDefinedAt(v) => normal(v)
}
Finally, we set the servicing mechanism to the new partial function:
service = (tmp, true)
}
Next, we define our act loop. If we are outside of the transaction boundary, we process
the message the normal way with
react. However, if we’re inside a transaction, we call
reactWithin(500). This will process messages the same way as react, except if no message
matching the pattern is received in 500 milliseconds, the Actor is sent a
TIMEOUT message.
This allows us to terminate our transaction automatically if there’s no activity on the
transaction in a 500 millisecond period. Any messages received outside the transaction
will be left in the Actor’s mailbox and will be processed after the transaction is complete.
def act = loop {
service match {
case (pf, false) => react(pf)
case (pf, true) => reactWithin(500)(pf)
}
}
19897ch06.fm Page 160 Monday, April 20, 2009 11:10 AM
CHAPTER 6 ■ ACTORS AND CONCURRENCY
161
And always remember to start the Actor.
this.start
}
Testing the Transactional Actor
We’ve defined a transactional Actor. Let’s see how it works. Let’s create accounts for two
people,
dpp and archer, and put funds into the checking and savings accounts.
val dpp = new Acct
dpp ! Update("Savings", v => (v getOrElse 0) + 1000)
dpp ! Update("Checking", v => (v getOrElse 0) + 100)
val archer = new Acct
archer ! Update("Savings", v => (v getOrElse 0) + 2000)
archer ! Update("Checking", v => (v getOrElse 0) + 50)
println("Initial balances:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
Let’s see what’s printed on the console:
Initial balances:
dpp: Info(Map(Savings -> 1000, Checking -> 100))
archer: Info(Map(Savings -> 2000, Checking -> 50))
Let’s define a variable that keeps track of our transaction IDs:
var xid = 1
Next, let’s define a method that will transactionally transfer money between accounts
for a single Actor as long as there are sufficient funds:
def transfer(who: Actor, from: String, to: String, amount: Int): Unit = {
xid += 1
who ! BeginXAction(xid)
We begin the transaction and get the current balances:
who !? (500, (xid, GetInfo)) match {
case Some(Info(bal)) =>
19897ch06.fm Page 161 Monday, April 20, 2009 11:10 AM
162
CHAPTER 6
■ ACTORS AND CONCURRENCY
If the account has sufficient balances, then we debit one account and credit the other
account and commit the transaction.
if (bal.getOrElse(from, 0) > amount) {
who ! (xid, Update(from, v => (v getOrElse 0) - amount))
who ! (xid, Update(to, v => (v getOrElse 0) + amount))
who ! (xid, CommitXAction)
} else {
If we don’t get the balance within 500 milliseconds or there’s not sufficient funds, roll
back the transaction.
who ! (xid, RollbackXAction)
}
case _ => who ! (xid, RollbackXAction)
}
}
Let’s test out the transactional transfer and see whether it works by trying to transfer
$700 from savings to checking twice. The first transaction should succeed, but the second
should fail.
transfer(dpp, "Savings", "Checking", 700)
println("xfer 1 dpp: "+(dpp !? GetInfo))
transfer(dpp, "Savings", "Checking", 700)
println("xfer 2 dpp: "+(dpp !? GetInfo))
The console says:
xfer 1 dpp: Info(Map(Savings -> 300, Checking -> 800))
xfer 2 dpp: Info(Map(Savings -> 300, Checking -> 800))
Next, let’s define a method that transfers money between two accounts on two sepa-
rate Actors. Once again, there must be sufficient funds for the transfer, or the transaction
will be rolled back.
def transfer2(src: Actor, sact: String, dest: Actor,
dact: String, amount: Int): Unit = {
xid += 1
Begin the transaction for both the Actors.
src ! BeginXAction(xid)
dest ! BeginXAction(xid)
19897ch06.fm Page 162 Monday, April 20, 2009 11:10 AM
CHAPTER 6 ■ ACTORS AND CONCURRENCY
163
And get the current balance for both the Actors. Note that the second balance is not
used, but the result indicates that we’re in the transaction for both the Actors.
(src !? (500, (xid, GetInfo)), dest !? (500, (xid, GetInfo))) match {
case (Some(Info(sbal)), Some(Info(dbal))) =>
We’re going to create the destination account before debiting the source account. This
will demonstrate that rolling back the transaction works correctly.
dest ! (xid, Update(dact, v => (v getOrElse 0) + amount))
if (sbal.getOrElse(sact, 0) > amount) {
src ! (xid, Update(sact, v => (v getOrElse 0) - amount))
src ! (xid, CommitXAction)
dest ! (xid, CommitXAction)
} else {
src ! (xid, RollbackXAction)
dest ! (xid, RollbackXAction)
}
case _ =>
src ! (xid, RollbackXAction)
dest ! (xid, RollbackXAction)
}
}
We’ve defined the ability to transactionally transfer money between accounts on two
different Actors. Let’s see whether it works. Let’s transfer $700 from
dpp to archer:
transfer2(dpp, "Checking", archer, "Checking", 700)
println("XFer 700 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
What does the console say?
XFer 700 dpp -> archer:
dpp: Info(Map(Savings -> 300, Checking -> 100))
archer: Info(Map(Savings -> 2000, Checking -> 750))
Yes, it works correctly. What happens if we try the transfer again?
transfer2(dpp, "Checking", archer, "Checking", 700)
println("Again, XFer 700 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
19897ch06.fm Page 163 Monday, April 20, 2009 11:10 AM
164
CHAPTER 6
■ ACTORS AND CONCURRENCY
The console confirms that the transfer did not take place:
Again, XFer 700 dpp -> archer:
dpp: Info(Map(Savings -> 300, Checking -> 100))
archer: Info(Map(Savings -> 2000, Checking -> 750))
Now, let’s try transferring $10:
transfer2(dpp, "Checking", archer, "Checking", 10)
println("XFer 10 dpp -> archer:")
println("dpp: "+(dpp !? GetInfo))
println("archer: "+(archer !? GetInfo))
And the console confirms that the transfer was successful:
XFer 10 dpp -> archer:
dpp: Info(Map(Savings -> 300, Checking -> 90))
archer: Info(Map(Savings -> 2000, Checking -> 760))
In practice, Actors are very flexible. They can be modeled like coarse-grained objects,
yet they have built-in concurrency support. The message handling, including the mailbox
and flexibility of
react, allows you to dynamically control which messages are handled by
the Actor given the current state of the Actor. Combined with Scala’s built-in support for
immutable collections, it’s very easy to build transactional support for Actors. Finally,
Actor messaging is syntactically lightweight enough to encourage Actor use while at the
same time offering the cue to the developer that the cost of passing the message is higher
than a method invocation. In the next subsection, we’ll see how to compose an Actor out
of a series of traits.
Composing Actors
So far in this chapter, we’ve built Actors as monoliths. A single file contains all the Actor’s
message handling. However, as Scala is a hybrid OOP/FP language, we’re going to combine
Scala’s OOP and FP sides to create Actors by composing traits. Scala’s traits are like Java’s
interfaces, but traits can contain methods as well as define them. We are going to build
two generic traits and compose them together into a specific Actor to provide chat server
functionality that we’ve built in a monolithic fashion earlier in the chapter.
19897ch06.fm Page 164 Monday, April 20, 2009 11:10 AM
CHAPTER 6 ■ ACTORS AND CONCURRENCY
165
How many listener implementations have you written in your life? Lots, probably.
Actors and the listener pattern work well together. Why not build a generic listener and
then compose it with a trait that contains business logic into a single Actor? This fore-
shadows some of the exciting stuff we’re going to do in the next chapter, but let’s get to
writing the code (see Listing 6-6).
Listing 6-6. Composed Actors
import scala.actors.Actor
import Actor._
trait Buildable {
def handler: PartialFunction[Any, Unit] = Map.empty
}
case class Add(who: Actor)
case class Remove(who: Actor)
trait ListenerMgt extends Buildable {
private var listeners: List[Actor] = Nil
override def handler = super.handler orElse {
case Add(who) =>
listeners = who :: listeners
who ! updateMessage
case Remove(who) => listeners -= who
}
protected def updateListeners() {
val m = updateMessage
listeners.foreach(a => a ! m)
}
protected def updateMessage: Any
}
case object GetInfo
trait GetMgt extends Buildable {
override def handler = super.handler orElse {
case GetInfo => reply(updateMessage)
}
19897ch06.fm Page 165 Monday, April 20, 2009 11:10 AM
166
CHAPTER 6
■ ACTORS AND CONCURRENCY
protected def updateMessage: Any
}
case class Messages(msgs: List[String])
object Chat extends Actor with ListenerMgt with GetMgt {
private var msgs: List[String] = Nil
def act = loop {
react(handler orElse {
case s: String => msgs ::= s
updateListeners()
})
}
protected def updateMessage = Messages(msgs)
this.start
}
class Listen extends Actor {
def act = loop {
react {
case Messages(m) => println("Got "+m)
}
}
this.start
}
That’s the whole listing. Let’s take it piece by piece. First, we import our Actors:
import scala.actors.Actor
import Actor._
We define a generic trait called Buildable that has a single handler method. The handler
method has the signature that we need to pass to the
react method in an Actor, and the
implementation is to return a partial function that will match nothing. This is analogous
to a blank method in a class that you expect to be overloaded. Your subclass can imple-
ment functionality, and when they call super, there is no additional work done.
trait Buildable {
def handler: PartialFunction[Any, Unit] = Map.empty
}
19897ch06.fm Page 166 Monday, April 20, 2009 11:10 AM