Commit 7a415741 authored by neverlord's avatar neverlord

maintenance

parent ac3711b6
...@@ -17,7 +17,7 @@ class ThreadedTestee(parent: Actor) extends Actor { ...@@ -17,7 +17,7 @@ class ThreadedTestee(parent: Actor) extends Actor {
case Spread(s) => case Spread(s) =>
if (s > 0) { if (s > 0) {
val next = Spread(s-1) val next = Spread(s-1)
(new ThreadedTestee(this)).start ! Spread(s-1) (new ThreadedTestee(this)).start ! next
(new ThreadedTestee(this)).start ! next (new ThreadedTestee(this)).start ! next
receive { receive {
case Result(v1) => case Result(v1) =>
...@@ -103,14 +103,16 @@ object ActorCreation { ...@@ -103,14 +103,16 @@ object ActorCreation {
} }
val n = args(1).toInt val n = args(1).toInt
if (args(0) == "threaded") { if (args(0) == "threaded") {
actor { val newMax = (1 << n) + 100
System.setProperty("actors.maxPoolSize", newMax.toString)
//actor {
(new ThreadedTestee(self)).start ! Spread(n) (new ThreadedTestee(self)).start ! Spread(n)
receive { receive {
case Result(v) => case Result(v) =>
if (v != (1 << n)) if (v != (1 << n))
Console.println("ERROR: expected " + (1 << n) + ", received " + v) Console.println("ERROR: expected " + (1 << n) + ", received " + v)
} }
} //}
} }
else if (args(0) == "threadless") { else if (args(0) == "threadless") {
actor { actor {
......
...@@ -10,32 +10,39 @@ case class Calc(value: Long) ...@@ -10,32 +10,39 @@ case class Calc(value: Long)
case class Factors(values: List[Long]) case class Factors(values: List[Long])
object global { object global {
final val taskN = ((Long) 86028157) * 329545133 final val taskN: Long = 86028157l * 329545133
final val factor1: Long = 86028157 final val factor1: Long = 86028157
final val factor2: Long = 329545133 final val factor2: Long = 329545133
final val numMessages = 1000 final val numMessages = 1000
val latch = new java.util.concurrent.CountDownLatch(1) val latch = new java.util.concurrent.CountDownLatch(1)
val checkFactors(f: List[Long]) { def checkFactors(f: List[Long]) {
assert(f.length == 2 && f(1) == factor1 && f(2) == factor2) assert(f.length == 2 && f(1) == factor1 && f(2) == factor2)
} }
} def factorize(arg: Long): List[Long] = {
object factorize {
def apply(arg: Long): List[Long] {
var n = arg var n = arg
if (n <= 3) if (n <= 3)
return List[Long](n) return List[Long](n)
var result = new List[Long] var result = new scala.collection.mutable.LinkedList[Long]
var d: Long = 2 var d: Long = 2
while (d < n) { while (d < n) {
if ((n % d) == 0) { if ((n % d) == 0) {
result append d result :+ d
n = n / d n = n / d
} }
else else
d = if (d == 2) 3 else d + 2 d = if (d == 2) 3 else d + 2
} }
result append d (result :+ d) toList
}
}
class ThreadedWorker extends Actor {
override def act() {
receive {
case Calc(value) =>
reply(global.factorize(value))
act()
}
} }
} }
...@@ -50,101 +57,69 @@ class ThreadedChainLink(next: Actor) extends Actor { ...@@ -50,101 +57,69 @@ class ThreadedChainLink(next: Actor) extends Actor {
} }
class ThreadedChainMaster extends Actor { class ThreadedChainMaster extends Actor {
@tailrec @tailrec final def newRing(a: Actor, i: Int): Actor = {
final def newRing(a: Actor, i: Int) { val next = (new ThreadedChainLink(a)).start
val next = new ThreadedChainLink(a)
if (i > 0) newRing(next, i-1) else next if (i > 0) newRing(next, i-1) else next
}
def checkFactors(f: List[Long]) {
} }
override def act() { override def act() {
val worker = actor { val worker = (new ThreadedWorker).start
receive {
case Calc(value) => reply factorize(value)
}
}
receive { receive {
case Init(rsize, iterations) => case Init(rsize, iterations) =>
var remainingFactors = iterations var remainingFactors = 0
for (_ <- 0 until iterations) { for (_ <- 0 until iterations) {
val next = newRing(rsize) val next = newRing(this, rsize)
remainingFactors += 1
worker ! Calc(global.taskN) worker ! Calc(global.taskN)
while (state != 0x2) var done = false
while (done == false)
receive { receive {
case Token(value) => case Token(value) =>
if (value > 0) next ! Token(value-1) if (value > 0) next ! Token(value-1) else done = true
case Factors(f) =>
global.checkFactors(f)
remainingFactors -= 1
} }
} }
while (remainingFactors > 0)
receive {
case Factors(f) =>
global.checkFactors(f)
remainingFactors -= 1
}
} }
} }
} }
class ThreadedReceiver(n: Long) extends Actor { object MixedCase {
override def act() {
var i: Long = 0
while (i < n)
receive {
case Msg => i += 1
}
}
}
class ThreadlessReceiver(n: Long) extends Actor {
var i: Long = 0
override def act() {
react {
case Msg =>
i += 1
if (i < n) act
}
}
}
class AkkaReceiver(n: Long) extends akka.actor.Actor {
var received: Long = 0
def receive = {
case Msg =>
received += 1
if (received == n) {
global.latch.countDown
self.exit
}
}
}
object MailboxPerformance {
def usage() { def usage() {
Console println "usage: (threaded|threadless|akka) (num_threads) (msgs_per_thread)" Console println "usage: (threaded|threadless|akka) (ring_size) (repetitions)"
} }
def main(args: Array[String]) = { def main(args: Array[String]) = {
if (args.size != 3) { if (args.size != 3) {
usage usage
throw new IllegalArgumentException("") throw new IllegalArgumentException("")
} }
val threads = args(1).toInt val ringSize = args(1).toInt
val msgs = args(2).toInt val repetitions = args(2).toInt
val impl = List("threaded", "threadless", "akka").indexOf(args(0)) val impl = List("threaded", "threadless", "akka").indexOf(args(0))
if (impl == -1) { if (impl == -1) {
usage usage
} }
else if (impl < 2) { else if (impl == 0) {
val rcvRef = if (impl == 0) (new ThreadedReceiver(threads*msgs)).start System.setProperty("actors.maxPoolSize", ((11 * ringSize) + 10).toString)
else (new ThreadlessReceiver(threads*msgs)).start for (_ <- 0 until 10) {
for (i <- 0 until threads) val a = (new ThreadedChainMaster()).start
(new java.lang.Thread { a ! Init(ringSize, repetitions)
override def run() { for (_ <- 0 until msgs) rcvRef ! Msg } }
}).start
} }
else { /*else {
val rcvRef = actorOf(new AkkaReceiver(threads*msgs)).start val rcvRef = actorOf(new AkkaReceiver(threads*msgs)).start
for (i <- 0 until threads) for (i <- 0 until threads)
(new java.lang.Thread { (new java.lang.Thread {
override def run() { for (_ <- 0 until msgs) rcvRef ! Msg } override def run() { for (_ <- 0 until msgs) rcvRef ! Msg }
}).start }).start
global.latch.await global.latch.await
} }*/
} }
} }
...@@ -45,12 +45,9 @@ using namespace cppa; ...@@ -45,12 +45,9 @@ using namespace cppa;
struct testee : fsm_actor<testee> struct testee : fsm_actor<testee>
{ {
uint32_t m_v1;
actor_ptr m_parent; actor_ptr m_parent;
behavior init_state; behavior init_state;
behavior wait4result1; testee(actor_ptr const& parent) : m_parent(parent)
behavior wait4result2;
testee(actor_ptr const& parent) : m_v1(0), m_parent(parent)
{ {
init_state = init_state =
( (
...@@ -61,31 +58,28 @@ struct testee : fsm_actor<testee> ...@@ -61,31 +58,28 @@ struct testee : fsm_actor<testee>
any_tuple msg = make_tuple(atom("spread"), x - 1); any_tuple msg = make_tuple(atom("spread"), x - 1);
spawn(new testee(this)) << msg; spawn(new testee(this)) << msg;
spawn(new testee(this)) << msg; spawn(new testee(this)) << msg;
become(&wait4result1); become
(
on<atom("result"), uint32_t>() >> [=](uint32_t v1)
{
become
(
on<atom("result"), uint32_t>() >> [=](uint32_t v2)
{
send(m_parent, atom("result"), v1 + v2);
become_void();
}
);
}
);
} }
else else
{ {
send(m_parent, atom("result"), (std::uint32_t) 1); send(m_parent, atom("result"), (std::uint32_t) 1);
quit(exit_reason::normal); become_void();
} }
} }
); );
wait4result1 =
(
on<atom("result"), uint32_t>() >> [=](uint32_t v1)
{
m_v1 = v1;
become(&wait4result2);
}
);
wait4result2 =
(
on<atom("result"), uint32_t>() >> [=](uint32_t v2)
{
send(m_parent, atom("result"), m_v1 + v2);
quit(exit_reason::normal);
}
);
} }
}; };
...@@ -129,17 +123,28 @@ void usage() ...@@ -129,17 +123,28 @@ void usage()
<< endl; << endl;
} }
template<typename T>
T rd(char const* cstr)
{
char* endptr = nullptr;
T result = static_cast<T>(strtol(cstr, &endptr, 10));
if (endptr == nullptr || *endptr != '\0')
{
std::string errstr;
errstr += "\"";
errstr += cstr;
errstr += "\" is not an integer";
usage();
throw std::invalid_argument(errstr);
}
return result;
}
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
if (argc == 3) if (argc == 3)
{ {
char* endptr = nullptr; int num = rd<int>(argv[2]);
int num = static_cast<int>(strtol(argv[2], &endptr, 10));
if (endptr == nullptr || *endptr != '\0')
{
cerr << "\"" << argv[2] << "\" is not an integer" << endl;
return 1;
}
if (strcmp(argv[1], "stacked") == 0) if (strcmp(argv[1], "stacked") == 0)
{ {
send(spawn(stacked_actor, self), atom("spread"), num); send(spawn(stacked_actor, self), atom("spread"), num);
......
#!/bin/bash #!/bin/bash
read -r cmd read -r cmd
export JAVA_OPTS="-Xmx5120M" export JAVA_OPTS="-Xmx4096M"
/usr/bin/time -p $cmd 2>&1 #| grep "^real" | grep -o -P "[0-9]*(\.[0-9]*)?" /usr/bin/time -p $cmd 2>&1 #| grep "^real" | grep -o -P "[0-9]*(\.[0-9]*)?"
...@@ -92,25 +92,29 @@ void usage() ...@@ -92,25 +92,29 @@ void usage()
<< endl; << endl;
} }
template<typename T>
T rd(char const* cstr)
{
char* endptr = nullptr;
T result = static_cast<T>(strtol(cstr, &endptr, 10));
if (endptr == nullptr || *endptr != '\0')
{
std::string errstr;
errstr += "\"";
errstr += cstr;
errstr += "\" is not an integer";
usage();
throw std::invalid_argument(errstr);
}
return result;
}
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
if (argc == 4) if (argc == 4)
{ {
char* endptr = nullptr; int64_t num_sender = rd<int64_t>(argv[2]);
int64_t num_sender = static_cast<int64_t>(strtol(argv[2], &endptr, 10)); int64_t num_msgs = rd<int64_t>(argv[3]);
if (endptr == nullptr || *endptr != '\0')
{
cerr << "\"" << argv[2] << "\" is not an integer" << endl;
usage();
return 1;
}
int64_t num_msgs = static_cast<int64_t>(strtol(argv[3], &endptr, 10));
if (endptr == nullptr || *endptr != '\0')
{
cerr << "\"" << argv[3] << "\" is not an integer" << endl;
usage();
return 1;
}
actor_ptr testee; actor_ptr testee;
if (strcmp(argv[1], "stacked") == 0) if (strcmp(argv[1], "stacked") == 0)
{ {
......
...@@ -277,30 +277,58 @@ void chain_master() ...@@ -277,30 +277,58 @@ void chain_master()
} }
template<typename F> template<typename F>
void run_test(F&& spawn_impl) void run_test(F&& spawn_impl, int ring_size, int repetitions)
{ {
std::vector<actor_ptr> masters; // of the universe std::vector<actor_ptr> masters; // of the universe
for (int i = 0; i < 10; ++i) for (int i = 0; i < 10; ++i)
{ {
masters.push_back(spawn_impl()); masters.push_back(spawn_impl());
send(masters.back(), atom("init"), 50, 20); send(masters.back(), atom("init"), ring_size, repetitions);
} }
await_all_others_done(); await_all_others_done();
} }
void usage()
{
cout << "usage: mailbox_performance "
"(stacked|event-based) (ring size) (repetitions)" << endl
<< endl;
}
template<typename T>
T rd(char const* cstr)
{
char* endptr = nullptr;
T result = static_cast<T>(strtol(cstr, &endptr, 10));
if (endptr == nullptr || *endptr != '\0')
{
std::string errstr;
errstr += "\"";
errstr += cstr;
errstr += "\" is not an integer";
usage();
throw std::invalid_argument(errstr);
}
return result;
}
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
announce<factors>(); announce<factors>();
if (argc == 2) if (argc == 4)
{ {
int ring_size = rd<int>(argv[2]);
int repetitions = rd<int>(argv[3]);
if (strcmp(argv[1], "event-based") == 0) if (strcmp(argv[1], "event-based") == 0)
{ {
run_test([]() { return spawn(new fsm_chain_master); }); run_test([]() { return spawn(new fsm_chain_master); },
ring_size, repetitions);
return 0; return 0;
} }
else if (strcmp(argv[1], "stacked") == 0) else if (strcmp(argv[1], "stacked") == 0)
{ {
run_test([]() { return spawn(chain_master); }); run_test([]() { return spawn(chain_master); },
ring_size, repetitions);
return 0; return 0;
} }
} }
......
#!/bin/bash #!/bin/bash
for i in *.scala; do for i in *.scala; do
echo "compile $i ..." echo "scalac -cp ../../akka-microkernel-1.2/lib/akka/akka-actor-1.2.jar \"$i\""
scalac -cp ../../akka-microkernel-1.2/lib/akka/akka-actor-1.2.jar "$i" scalac -cp ../../akka-microkernel-1.2/lib/akka/akka-actor-1.2.jar "$i"
done done
echo done echo done
...@@ -89,6 +89,8 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor ...@@ -89,6 +89,8 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor
protected: protected:
abstract_event_based_actor();
struct stack_element struct stack_element
{ {
......
...@@ -117,7 +117,7 @@ class abstract_scheduled_actor : public abstract_actor<local_actor> ...@@ -117,7 +117,7 @@ class abstract_scheduled_actor : public abstract_actor<local_actor>
static constexpr int blocked = 0x02; static constexpr int blocked = 0x02;
static constexpr int about_to_block = 0x04; static constexpr int about_to_block = 0x04;
abstract_scheduled_actor(); abstract_scheduled_actor(int state = done);
template<typename Scheduler> template<typename Scheduler>
abstract_scheduled_actor(void (*enqueue_fun)(Scheduler*, abstract_scheduled_actor(void (*enqueue_fun)(Scheduler*,
......
...@@ -64,7 +64,8 @@ class thread_pool_scheduler : public scheduler ...@@ -64,7 +64,8 @@ class thread_pool_scheduler : public scheduler
scheduled_actor_dummy m_dummy; scheduled_actor_dummy m_dummy;
thread m_supervisor; thread m_supervisor;
actor_ptr spawn_impl(abstract_scheduled_actor* what); actor_ptr spawn_impl(abstract_scheduled_actor* what,
bool push_to_queue = true);
static void worker_loop(worker*); static void worker_loop(worker*);
static void supervisor_loop(job_queue*, abstract_scheduled_actor*); static void supervisor_loop(job_queue*, abstract_scheduled_actor*);
......
...@@ -43,12 +43,15 @@ class event_based_actor : public event_based_actor_base<event_based_actor> ...@@ -43,12 +43,15 @@ class event_based_actor : public event_based_actor_base<event_based_actor>
typedef abstract_event_based_actor::stack_element stack_element; typedef abstract_event_based_actor::stack_element stack_element;
void clear();
// has_ownership == false // has_ownership == false
void do_become(behavior* bhvr); void do_become(behavior* bhvr);
void do_become(invoke_rules* bhvr, bool has_ownership); void do_become(invoke_rules* bhvr, bool has_ownership);
void do_become(timed_invoke_rules* bhvr, bool has_ownership); void do_become(timed_invoke_rules* bhvr, bool has_ownership);
public:
void become_void();
}; };
} // namespace cppa } // namespace cppa
......
...@@ -48,6 +48,7 @@ class stacked_event_based_actor : public event_based_actor_base<stacked_event_ba ...@@ -48,6 +48,7 @@ class stacked_event_based_actor : public event_based_actor_base<stacked_event_ba
protected: protected:
void unbecome(); void unbecome();
void become_void();
}; };
......
...@@ -34,6 +34,11 @@ ...@@ -34,6 +34,11 @@
namespace cppa { namespace cppa {
abstract_event_based_actor::abstract_event_based_actor()
: super(abstract_event_based_actor::blocked)
{
}
void abstract_event_based_actor::dequeue(invoke_rules&) void abstract_event_based_actor::dequeue(invoke_rules&)
{ {
quit(exit_reason::unallowed_function_call); quit(exit_reason::unallowed_function_call);
......
...@@ -38,9 +38,9 @@ namespace cppa { namespace detail { ...@@ -38,9 +38,9 @@ namespace cppa { namespace detail {
namespace { void dummy_enqueue(void*, abstract_scheduled_actor*) { } } namespace { void dummy_enqueue(void*, abstract_scheduled_actor*) { } }
abstract_scheduled_actor::abstract_scheduled_actor() abstract_scheduled_actor::abstract_scheduled_actor(int state)
: next(nullptr) : next(nullptr)
, m_state(abstract_scheduled_actor::done) , m_state(state)
, m_enqueue_to_scheduler(dummy_enqueue, static_cast<void*>(nullptr), this) , m_enqueue_to_scheduler(dummy_enqueue, static_cast<void*>(nullptr), this)
, m_has_pending_timeout_request(false) , m_has_pending_timeout_request(false)
, m_active_timeout_id(0) , m_active_timeout_id(0)
......
...@@ -122,11 +122,12 @@ size_t actor_registry::running() const ...@@ -122,11 +122,12 @@ size_t actor_registry::running() const
void actor_registry::dec_running() void actor_registry::dec_running()
{ {
size_t new_val = --m_running; size_t new_val = --m_running;
/*
if (new_val == std::numeric_limits<size_t>::max()) if (new_val == std::numeric_limits<size_t>::max())
{ {
throw std::underflow_error("actor_count::dec()"); throw std::underflow_error("actor_count::dec()");
} }
else if (new_val <= 1) else*/ if (new_val <= 1)
{ {
unique_lock<mutex> guard(m_running_mtx); unique_lock<mutex> guard(m_running_mtx);
m_running_cv.notify_all(); m_running_cv.notify_all();
......
...@@ -32,9 +32,9 @@ ...@@ -32,9 +32,9 @@
namespace cppa { namespace cppa {
void event_based_actor::clear() void event_based_actor::become_void()
{ {
if (!m_loop_stack.empty()) m_loop_stack.pop(); while (!m_loop_stack.empty()) m_loop_stack.pop();
} }
void event_based_actor::do_become(behavior* bhvr) void event_based_actor::do_become(behavior* bhvr)
...@@ -51,14 +51,14 @@ void event_based_actor::do_become(behavior* bhvr) ...@@ -51,14 +51,14 @@ void event_based_actor::do_become(behavior* bhvr)
void event_based_actor::do_become(invoke_rules* bhvr, bool has_ownership) void event_based_actor::do_become(invoke_rules* bhvr, bool has_ownership)
{ {
clear(); become_void();
reset_timeout(); reset_timeout();
m_loop_stack.push(stack_element(bhvr, has_ownership)); m_loop_stack.push(stack_element(bhvr, has_ownership));
} }
void event_based_actor::do_become(timed_invoke_rules* bhvr, bool has_ownership) void event_based_actor::do_become(timed_invoke_rules* bhvr, bool has_ownership)
{ {
clear(); become_void();
request_timeout(bhvr->timeout()); request_timeout(bhvr->timeout());
m_loop_stack.push(stack_element(bhvr, has_ownership)); m_loop_stack.push(stack_element(bhvr, has_ownership));
} }
......
...@@ -32,6 +32,11 @@ ...@@ -32,6 +32,11 @@
namespace cppa { namespace cppa {
void stacked_event_based_actor::become_void()
{
while (!m_loop_stack.empty()) m_loop_stack.pop();
}
void stacked_event_based_actor::unbecome() void stacked_event_based_actor::unbecome()
{ {
if (!m_loop_stack.empty()) m_loop_stack.pop(); if (!m_loop_stack.empty()) m_loop_stack.pop();
......
...@@ -112,7 +112,6 @@ struct thread_pool_scheduler::worker ...@@ -112,7 +112,6 @@ struct thread_pool_scheduler::worker
void exec_done() void exec_done()
{ {
if (!job->deref()) delete job; if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count(); dec_actor_count();
job = nullptr; job = nullptr;
} }
...@@ -245,20 +244,22 @@ void thread_pool_scheduler::schedule(abstract_scheduled_actor* what) ...@@ -245,20 +244,22 @@ void thread_pool_scheduler::schedule(abstract_scheduled_actor* what)
m_queue.push_back(what); m_queue.push_back(what);
} }
actor_ptr thread_pool_scheduler::spawn_impl(abstract_scheduled_actor* what) actor_ptr thread_pool_scheduler::spawn_impl(abstract_scheduled_actor* what,
bool push_to_queue)
{ {
inc_actor_count(); inc_actor_count();
CPPA_MEMORY_BARRIER(); CPPA_MEMORY_BARRIER();
intrusive_ptr<abstract_scheduled_actor> ctx(what); intrusive_ptr<abstract_scheduled_actor> ctx(what);
ctx->ref(); ctx->ref();
m_queue.push_back(ctx.get()); if (push_to_queue) m_queue.push_back(ctx.get());
return std::move(ctx); return std::move(ctx);
} }
actor_ptr thread_pool_scheduler::spawn(abstract_event_based_actor* what) actor_ptr thread_pool_scheduler::spawn(abstract_event_based_actor* what)
{ {
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this)); // do NOT push event-based actors to the queue on startup
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this), false);
} }
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* behavior, actor_ptr thread_pool_scheduler::spawn(scheduled_actor* behavior,
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment