diff --git a/examples/src/actors/Actors.kt b/examples/src/actors/Actors.kt deleted file mode 100644 index dafc9869748..00000000000 --- a/examples/src/actors/Actors.kt +++ /dev/null @@ -1,186 +0,0 @@ -package actors - -import std.concurrent.* -import std.util.* - -import java.util.concurrent.* -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater -import java.util.Date - -/* -Message handling class which process only one message at any given moment - -Three main ways to use it -- post and forget -- post and await result -- post and assign callback to be called when message processed -*/ -abstract class Actor(protected val executor: Executor) { - // we can not do it private or protected - // otherwise updater defined in class object will not be able to access it - volatile var queue : FunctionalQueue = emptyQueue - - /* - Handle message and return result - This method guaranteed to be executed only one per object at any given time - */ - protected abstract fun onMessage(message: Any) : Any? - - /* - Post message to the actor. - The method returns immediately and the message itself will be processed later - */ - fun post(message: Any) { - while(true) { - val q = queue - if(q.empty) { - if(queueUpdater.compareAndSet(this, q, busyEmptyQueue)) { - executor.execute { process(message) } - break - } - } - else { - val newQueue = (if(q == busyEmptyQueue) emptyQueue else q) add message - if(queueUpdater.compareAndSet(this, q, newQueue)) { - break - } - } - } - } - - /* - Post message to the actor and schedule callback to be executed on given executor when message processed - */ - fun post(message: Any, executor: Executor = this.executor, callback: (Any?)->Unit) = - post(Callback(message, executor, callback)) - - /* - Send message to the actor and await result - */ - fun send(message: Any) : Any? { - val request = Request(message) - post(request) - request.await() - return request.result - } - - private fun nextMessage() { - while(true) { - val q = queue - if(q == busyEmptyQueue) { - if(queueUpdater.compareAndSet(this, busyEmptyQueue, emptyQueue)) { - break; - } - } - else { - val removed = q.removeFirst() - val newQueue = if(removed._2.empty) busyEmptyQueue else removed._2 - if(queueUpdater.compareAndSet(this, q, newQueue)) { - executor.execute { process(removed._1) } - break; - } - } - } - } - - private fun process(message: Any) { - when(message) { - is Request -> { - message.result = onMessage(message.message) - message.countDown() - } - is Callback -> { - val result = onMessage(message.message) - message.executor execute { - val callback = message.callback; - callback(result) - } - } - else -> onMessage(message) - } - - nextMessage() - } - - /* - Create actor on the same executor - */ - fun actor(handler: (Any)->Any?) = executor.actor(handler) - - class object { - val queueUpdater = AtomicReferenceFieldUpdater.newUpdater(typeinfo.javaClassForType,typeinfo>().javaClassForType, "queue").sure() - val emptyQueue = FunctionalQueue() - val busyEmptyQueue = FunctionalQueue() add "busy empty queue" - - class Request(val message: Any) : java.util.concurrent.CountDownLatch(1) { - var result: Any? = null - } - - class Callback(val message: Any, val executor: Executor, val callback: (Any?) -> Unit) - } -} - -fun Executor.actor(handler: (Any)->Any?) : Actor = object: Actor(this) { - override fun onMessage(message: Any) { - handler(message) - } -} - -fun singleThreadActor(handler: (Any)->Any?) : Actor = Executors.newSingleThreadExecutor().sure().actor(handler) - -class App() : Actor(Executors.newFixedThreadPool(10).sure()) { - private val logger = executor.actor { message -> - println("${Date()}:\t\t$message") - } - - private val actors = Array(100, { createChild(it) }); - - private var finishedChildren = actors.size - - override fun onMessage(message: Any) { - when(message) { - "start" -> { - logger post "app started" - for(a in actors) { - a post "start" - } - } - "child finished" -> { - if(--finishedChildren == 0) { - logger send "app finished" - (executor as ExecutorService).shutdown() - } - } - else -> { - logger post "unknown message $message" - } - } - } - - private fun createChild(index: Int) : Actor = actor { message -> - val next = (index + 1) % actors.size - when(message) { - "start" -> { - logger post "$index started" - actors[next] post #(index, 0) - } - is Tuple2 -> { - logger post "$index received $message" - val from = message._1 - val value = message._2 - if(next != from) { - actors[next] post #(from, value+1) - } - else { - logger post "$index finished" - this@App post "child finished" - } - } - else -> {} - } - } -} - -fun main(args: Array) { - App() post "start" -} diff --git a/examples/src/actors/actors.iml b/examples/src/actors/actors.iml new file mode 100644 index 00000000000..00069f36085 --- /dev/null +++ b/examples/src/actors/actors.iml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + diff --git a/examples/src/actors/actors.ipr b/examples/src/actors/actors.ipr new file mode 100644 index 00000000000..d7df3849c28 --- /dev/null +++ b/examples/src/actors/actors.ipr @@ -0,0 +1,173 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/examples/src/actors/actors.iws b/examples/src/actors/actors.iws new file mode 100644 index 00000000000..989317a4ff9 --- /dev/null +++ b/examples/src/actors/actors.iws @@ -0,0 +1,523 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + localhost + 5050 + + + + + + + 1328866126960 + 1328866126960 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + No facets are configured + + + + + + + + + + + + + + + 1.6 + + + + + + + + actors + + + + + + + + 1.6 + + + + + + + + KotlinRuntime + + + + + + + + + diff --git a/examples/src/actors/src/Actors.kt b/examples/src/actors/src/Actors.kt new file mode 100644 index 00000000000..af3510f486d --- /dev/null +++ b/examples/src/actors/src/Actors.kt @@ -0,0 +1,184 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import std.util.* + +import java.util.concurrent.* +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater +import java.util.Date +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.atomic.AtomicLong + +/* +Message handling class which process only one message at any given moment + +Three main ways to use it +- post and forget +- post and await result +- post and assign callback to be called when message processed +*/ +abstract class Actor(protected val executor: Executor, val fair: Boolean = false) { + private val queue = Ref() + + class Ref() : AtomicReference>(emptyQueue), Runnable { + override fun run() { + if (!fair) + runUnfair() + else + runFair() + } + + private fun runFair() { + while(true) { + val q = get() + if(q.identityEquals(busyEmptyQueue)) { + if(compareAndSet(busyEmptyQueue, emptyQueue)) { + break + } + } + else { + val removed = q.removeFirst() + val newQueue = if(removed._2.empty) busyEmptyQueue else removed._2 + if(compareAndSet(q, newQueue)) { + doProcess(removed._1) + executor.execute(this) + break + } + } + } + } + + private fun runUnfair() { + while (true) { + val q = get() + if(q.identityEquals(busyEmptyQueue)) { + if (compareAndSet(q, emptyQueue)) { + break + } + } + else { + if (compareAndSet(q, busyEmptyQueue)) { + var l = q.output + while (!l.empty) { + doProcess(l.head) + l = l.tail + } + l = q.input.reversed() + while (!l.empty) { + doProcess(l.head) + l = l.tail + } + } + } + } + } + + override fun toString(): String? = null + } + + /* + Handle message and return result + This method guaranteed to be executed only one per object at any given time + */ + protected abstract fun onMessage(message: Any) : Any? + + /* + Post message to the actor. + The method returns immediately and the message itself will be processed later + */ + fun post(message: Any) { + messagesSent.incrementAndGet() + while(true) { + val q = queue.get() + val newQueue = (if (q.identityEquals(busyEmptyQueue)) emptyQueue else q) add message + if(queue.compareAndSet(q, newQueue)) { + if (q.identityEquals(emptyQueue)) { + executor.execute(queue) + } + break + } + } + } + + fun postFirst(message: Any) { + messagesSent.incrementAndGet() + while(true) { + val q = queue.get() + val newQueue = (if (q.identityEquals(busyEmptyQueue)) emptyQueue else q) addFirst message + if(queue.compareAndSet(q, newQueue)) { + if (q.identityEquals(emptyQueue)) { + executor.execute(queue) + } + break + } + } + } + + /* + Post message to the actor and schedule callback to be executed on given executor when message processed + */ + fun post(message: Any, executor: Executor = this.executor, callback: (Any?)->Unit) = + post(Callback(message, executor, callback)) + + /* + Send message to the actor and await result + */ + fun send(message: Any) : Any? { + val request = Request(message) + post(request) + request.await() + return request.result + } + + private fun doProcess(message: Any) { + messagesProcessed.incrementAndGet() + when(message) { + is Request -> { + message.result = onMessage(message.message) + message.countDown() + } + is Callback -> { + val result = onMessage(message.message) + message.executor execute { + val callback = message.callback; + callback(result) + } + } + else -> onMessage(message) + } + } + + /* + Create actor on the same executor + */ + fun actor(handler: (Any)->Any?) = executor.actor(handler) + + class object { + val emptyQueue = FunctionalQueue() + val busyEmptyQueue = FunctionalQueue() add "busy empty queue" + + class Request(val message: Any) : java.util.concurrent.CountDownLatch(1) { + var result: Any? = null + } + + class Callback(val message: Any, val executor: Executor, val callback: (Any?) -> Unit) + + val messagesSent = AtomicLong() + val messagesProcessed = AtomicLong() + + val timer = fixedRateTimer(daemon=true, period=5000.lng) { + val sent = messagesSent.get() + val received = messagesProcessed.get() + println("Actors stat: Sent: $sent Processed: $received Pending: ${sent-received}") + } + } +} + +fun Executor.actor(handler: (Any)->Any?) : Actor = object: Actor(this) { + override fun onMessage(message: Any) { + handler(message) + } +} + +fun singleThreadActor(handler: (Any)->Any?) : Actor = Executors.newSingleThreadExecutor().sure().actor(handler) diff --git a/examples/src/actors/src/Main.kt b/examples/src/actors/src/Main.kt new file mode 100644 index 00000000000..8915a3b5fdf --- /dev/null +++ b/examples/src/actors/src/Main.kt @@ -0,0 +1,32 @@ +package org.jetbrains.kotlin.examples.actors + +import java.util.concurrent.Executors + +fun main(args: Array) { + val numberOfSymbols = 10000 + val numberOfShards = 10 + val numberOfClients = 100000 + val stocksPerClient = 4 + + val stockServer = StockServer(numberOfShards) + + for (i in 0..numberOfSymbols) + stockServer post RegisterStock(i.toString()) + + println("Stock server started") + + val executor = Executors.newFixedThreadPool(4*Runtime.getRuntime().sure().availableProcessors()).sure() + + for(i in 0..numberOfClients) { + val client = executor.actor{ message -> + // println(message) + } + + for (k in 1..stocksPerClient) { + val stock = (Math.random() * numberOfSymbols).int + stockServer post Subscribe(stock.toString(), client) + } + } + + println("clients connected") +} diff --git a/examples/src/actors/src/Messages.kt b/examples/src/actors/src/Messages.kt new file mode 100644 index 00000000000..5e815cce3a6 --- /dev/null +++ b/examples/src/actors/src/Messages.kt @@ -0,0 +1,23 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import java.util.concurrent.Executor +import java.util.LinkedList +import java.util.concurrent.Executors +import java.util.HashMap +import java.util.HashSet + +open class StockMessage(val symbol: String) { + class object { + val UpdateModel = "update" + val DoUpdateModel = "do update" + } +} + +class RegisterStock(symbol: String) : StockMessage(symbol) + +class Subscribe(symbol: String, val subscriber: Actor) : StockMessage(symbol) + +class Unsubscribe(symbol: String, val subscriber: Actor) : StockMessage(symbol) + +class PriceUpdate(symbol: String, val price: Double) : StockMessage(symbol) diff --git a/examples/src/actors/src/StatCalculator.kt b/examples/src/actors/src/StatCalculator.kt new file mode 100644 index 00000000000..d884c6ff1fd --- /dev/null +++ b/examples/src/actors/src/StatCalculator.kt @@ -0,0 +1,34 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import org.jetbrains.kotlin.examples.actors.Actor +import java.util.concurrent.Executors +import java.util.LinkedList + +class StatCalculator() : Actor(Executors.newSingleThreadExecutor().sure()) { + val list = LinkedList () + var average = 0.lng + var sum = 0.lng + var cnt = 0.lng + + val timer = fixedRateTimer(period=2000.lng, daemon=true) { + this@StatCalculator post "print" + } + + override fun onMessage(msg: Any) { + when(msg) { + is Long -> { + list.add(msg) + sum += msg + if (list.size() > 20) + sum -= list.removeFirst() + average = sum / list.size() + cnt++ + } + "print" -> { + println("Avr. period:[${average}ms] # updates:[$cnt]") + } + else -> {} + } + } +} diff --git a/examples/src/actors/src/Stock.kt b/examples/src/actors/src/Stock.kt new file mode 100644 index 00000000000..5d1d0d07c38 --- /dev/null +++ b/examples/src/actors/src/Stock.kt @@ -0,0 +1,22 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import java.util.concurrent.Executor +import java.util.LinkedList +import java.util.concurrent.Executors +import java.util.HashMap +import java.util.HashSet + +class Stock(val symbol: String) { + val subscribers = HashSet() + + var price = 10*Math.random() + + fun update() { + price += Math.random() - 0.5 + val priceUpdate = PriceUpdate(symbol, price) + for(s in subscribers) { + s post priceUpdate + } + } +} diff --git a/examples/src/actors/src/StockServer.kt b/examples/src/actors/src/StockServer.kt new file mode 100644 index 00000000000..4c8ef744412 --- /dev/null +++ b/examples/src/actors/src/StockServer.kt @@ -0,0 +1,38 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import java.util.concurrent.Executor +import java.util.LinkedList +import java.util.concurrent.Executors +import java.util.HashMap +import java.util.HashSet + +class StockServer(val numberOfShards : Int) : Actor(Executors.newFixedThreadPool(numberOfShards+1).sure()) { + private val stockToServer = HashMap () + + private val statCalculator = StatCalculator() + + private val shards = Array (numberOfShards, { (i: Int) -> StockServerShard(statCalculator, executor) }) + + private val timer = fixedRateTimer(period=1000.lng, daemon=true) { + for(s in shards) + s post StockMessage.UpdateModel + } + + override fun onMessage(message: Any): Unit { + when (message) { + is RegisterStock -> { + val shard = shards[stockToServer.size() % shards.size] + stockToServer.put(message.symbol, shard) + shard post message + } + is Subscribe -> { + stockToServer.get(message.symbol)?.post(message) + } + is Unsubscribe -> { + stockToServer.get(message.symbol)?.post(message) + } + else -> {} + } + } +} \ No newline at end of file diff --git a/examples/src/actors/src/StockServerShard.kt b/examples/src/actors/src/StockServerShard.kt new file mode 100644 index 00000000000..33d9a11358c --- /dev/null +++ b/examples/src/actors/src/StockServerShard.kt @@ -0,0 +1,44 @@ +package org.jetbrains.kotlin.examples.actors + +import std.concurrent.* +import java.util.concurrent.Executor +import java.util.LinkedList +import java.util.concurrent.Executors +import java.util.HashMap +import java.util.HashSet + +class StockServerShard(val statCalculator : StatCalculator, executor: Executor) : Actor(executor) { + private val stocks = HashMap(); + + private var previousTime = System.currentTimeMillis(); + + private var updateScheduled = false + + override fun onMessage(message: Any) { + when(message) { + StockMessage.DoUpdateModel -> { + val timeMillis = System.currentTimeMillis() + for(s in stocks.values()) { + s.update() + } + statCalculator post (timeMillis - previousTime) + previousTime = timeMillis + updateScheduled = false + } + StockMessage.UpdateModel -> { + if (!updateScheduled) { + this post StockMessage.DoUpdateModel + updateScheduled = true + } + } + is RegisterStock -> + stocks.put(message.symbol, Stock(message.symbol)) + is Subscribe -> + stocks.get(message.symbol)?.subscribers?.add(message.subscriber) + is Unsubscribe -> + stocks.get(message.symbol)?.subscribers?.add(message.subscriber) + else -> {} + } + } +} + diff --git a/stdlib/ktSrc/concurrent/FunctionalList.kt b/stdlib/ktSrc/concurrent/FunctionalList.kt index b285fedabc2..d3b4f2cd1ff 100644 --- a/stdlib/ktSrc/concurrent/FunctionalList.kt +++ b/stdlib/ktSrc/concurrent/FunctionalList.kt @@ -1,11 +1,11 @@ package std.concurrent -abstract class FunctionalList(val size: Int) { - abstract val head: T - abstract val tail: FunctionalList +abstract class FunctionalList(public val size: Int) { + public abstract val head: T + public abstract val tail: FunctionalList val empty : Boolean - get() = size == 0 + get() = size == 0 fun add(element: T) : FunctionalList = FunctionalList.Standard(element, this)