天下无处不乒乓
2009-06-24 12:46 by 老赵, 21222 visits在消息传递（Message Passing）领域，PingPong是最常见的测试之一。它的功能简单的有些无聊，一个Ping Actor和一个Pong Actor之间互相传递消息，你Ping过来我Pong过去。也正因为如此简单，PingPong的目标仅仅是测试纯粹的消息传递机制的效率。也正因为如此，各Actor模型往往都将其作为展示自己功能的第一个示例。老赵从互联网上收集了一些最为常见的，不同语言/平台下Actor模型实现PingPong的示例，可作“观赏”之用。
由于语言特性不同，有的Actor模型内置于语言之中的（如Erlang），其他大都由框架进行补充。有的Actor模型使用字符串作为消息，有的却通过语言功能而可以传递强类型的消息。简单的罗列各种实现，把这些实现从表面上进行简单的对比，也可以说颇有趣味。当然，这些实现虽然都使用了PingPong作为演示，但是其中细节还是略有不同的。例如有些会让Ping/Pong过程永远持续下去，而有些会让它们在进行一段时间过后就中止；有些会对命令进行识别，而有些对此并不在意。老赵认为这些区别无伤大雅，对此也未作任何修改。
本文会涉及以下Actor模型的实现（不少语言缺少语法着色，如果您有好用的HTML高亮方案请不吝指明）：
- Erlang (source)
- Scala (source)
- Haskell (source)
- Ruby (source)
- Python (source)
- Axum with Channel (source)
- Axum with Ordered Interaction Points (source)
- F# (source)
- F# with ActorLite (source)
在这些示例中，您会发现缺少了C#下的Actor——因为老赵打算下次单独开篇对此进行说明和介绍。:)
Erlang
-module(tut15).
-export([start/0, ping/2, pong/0]).
ping(0, Pong_PID) ->
Pong_PID ! finished,
io:format("ping finished~n", []);
ping(N, Pong_PID) ->
Pong_PID ! {ping, self()},
receive
pong ->
io:format("Ping received pong~n", [])
end,
ping(N - 1, Pong_PID).
pong() ->
receive
finished ->
io:format("Pong finished~n", []);
{ping, Ping_PID} ->
io:format("Pong received ping~n", []),
Ping_PID ! pong,
pong()
end.
start() ->
Pong_PID = spawn(tut15, pong, []),
spawn(tut15, ping, [3, Pong_PID]).
Scala
import scala.actors.Actor
import scala.actors.Actor._
class Ping(count: int, pong: Actor) extends Actor {
def act() {
var pingsLeft = count - 1
pong ! Ping
loop {
react {
case Pong =>
if (pingsLeft % 1000 == 0)
Console.println("Ping: pong")
if (pingsLeft > 0) {
pong ! Ping
pingsLeft -= 1
} else {
Console.println("Ping: stop")
pong ! Stop
exit()
}
}
}
}
}
class Pong extends Actor {
def act() {
var pongCount = 0
loop {
react {
case Ping =>
if (pongCount % 1000 == 0)
Console.println("Pong: ping "+pongCount)
sender ! Pong
pongCount = pongCount + 1
case Stop =>
Console.println("Pong: stop")
exit()
}
}
}
}
object pingpong extends Application {
val pong = new Pong
val ping = new Ping(100000, pong)
ping.start
pong.start
}
Ruby
require 'concurrent/actors'
include Concurrent::Actors
Message = Struct.new :ping, :pong
ping_thread = Actor.spawn do
loop do
Actor.receive do |f|
f.when Message do |m|
puts "PING"
sleep(1)
m.pong << Message.new(m.ping, m.pong)
end
end
end
end
pong_thread = Actor.spawn do
loop do
Actor.receive do |f|
f.when Message do |m|
puts "PONG"
sleep(1)
m.ping << Message.new(m.ping, m.pong)
end
end
end
end
ping_thread << Message.new(ping_thread, pong_thread)
while(true)
puts("WAITING...")
sleep(5)
end
Python
#
# pingpong_stackless.py
#
import stackless
ping_channel = stackless.channel()
pong_channel = stackless.channel()
def ping():
while ping_channel.receive(): #blocks here
print "PING"
pong_channel.send("from ping")
def pong():
while pong_channel.receive():
print "PONG"
ping_channel.send("from pong")
stackless.tasklet(ping)()
stackless.tasklet(pong)()
# we need to 'prime' the game by sending a start message
# if not, both tasklets will block
stackless.tasklet(ping_channel.send)('startup')
stackless.run()
Axum with Channel
using System; using System.Concurrency; using System.Collections.Generic; using Microsoft.Axum; using System.Concurrency.Messaging; public channel PingPongStatus { input int HowMany; output int Done; Start: { HowMany, Done -> End; } } public agent Ping : channel PingPongStatus { public Ping() { // How many are we supposed to do? var iters = receive(PrimaryChannel::HowMany); // Create pong and send how many to do var chan = Pong.CreateInNewDomain(); chan::HowMany <-- iters; // Send pings and receive pongs for (int i = 0; i < iters; i++) { chan::Ping <-- Signal.Value; receive(chan::Pong); Console.WriteLine("Ping received Pong"); } // How many did Pong process? int pongIters = receive(chan::Done); PrimaryChannel::Done <-- 0; } } public channel PingPong { input int HowMany; output int Done; input Signal Ping; output Signal Pong; } public agent Pong : channel PingPong { public Pong() { // Get how many we're supposed to do var iters = receive(PrimaryChannel::HowMany); int i = 0; // Receive our ping and send back our pong for (; i < iters; i++) { receive(PrimaryChannel::Ping); Console.WriteLine("Pong received Ping"); PrimaryChannel::Pong <-- Signal.Value; } PrimaryChannel::Done <-- i; } } public agent Program : channel Application { public Program() { // Wait for our command args var cmdArgs = receive(PrimaryChannel::CommandLine); // Set the iterations to be some number var iters = 3; // Create instance of ping and send msg var chan = Ping.CreateInNewDomain(); chan::HowMany <-- iters; // Receive how many done receive(chan::Done); PrimaryChannel::Done <-- Signal.Value; } }
F#
open System type message = Finished | Msg of int * MailboxProcessor<message> let ping iters (outbox : MailboxProcessor<message>) = MailboxProcessor.Start(fun inbox -> let rec loop n = async { if n > 0 then outbox.Post( Msg(n, inbox) ) let! msg = inbox.Receive() Console.WriteLine("ping received pong") return! loop(n-1) else outbox.Post(Finished) Console.WriteLine("ping finished") return ()} loop iters) let pong () = MailboxProcessor.Start(fun inbox -> let rec loop () = async { let! msg = inbox.Receive() match msg with | Finished -> Console.WriteLine("pong finished") return () | Msg(n, outbox) -> Console.WriteLine("pong received ping") outbox.Post(Msg(n, inbox) return! loop() } loop()) let ponger = pong() do (ping 10 ponger) |> ignore
F# with ActorLite
#light open System open ActorLite let (<=) (m:_ Actor) msg = m.Post msg type message = string * message Actor let pong = { new message Actor() with override self.Receive(message) = let msg, ping = message Console.WriteLine(msg) ping <= ("Pong", self) } let ping = { new message Actor() with override self.Receive(message) = let msg, pong = message Console.WriteLine(msg) pong <= ("Ping", self) } ping <= ("Start", pong) Console.ReadLine |> ignore
不错，够全的了