Commit fcb7d04f authored by neverlord's avatar neverlord

scheduling bugfix

parent 017ddd5c
...@@ -13,6 +13,8 @@ test ...@@ -13,6 +13,8 @@ test
*.dump *.dump
*.beam *.beam
*.class *.class
benchmarks/mailbox_performance
benchmarks/actor_creation
.libs/ .libs/
.deps/ .deps/
libcppa.pc libcppa.pc
......
import scala.actors.Actor import scala.actors.Actor
import scala.actors.Actor._ import scala.actors.Actor._
import akka.actor.Actor.actorOf
case object GoAhead
case class Spread(value: Int)
case class Result(value: Int) case class Result(value: Int)
class ThreadedTestee(parent: Actor, n: Int) extends Actor { object global {
val latch = new java.util.concurrent.CountDownLatch(1)
}
class ThreadedTestee(parent: Actor) extends Actor {
def act() { def act() {
if (n > 0) { receive {
(new ThreadedTestee(this, n-1)).start case Spread(s) =>
(new ThreadedTestee(this, n-1)).start if (s > 0) {
val next = Spread(s-1)
(new ThreadedTestee(this)).start ! Spread(s-1)
(new ThreadedTestee(this)).start ! next
receive { receive {
case Result(v1) => case Result(v1) =>
receive { receive {
case Result(v2) => case Result(v2) =>
parent ! new Result(v1 + v2) parent ! Result(v1 + v2)
} }
} }
} }
else { else {
parent ! new Result(1) parent ! Result(1)
}
} }
} }
} }
class ThreadlessTestee(parent: Actor, n: Int) extends Actor { class ThreadlessTestee(parent: Actor) extends Actor {
def act() { def act() {
if (n > 0) { react {
(new ThreadlessTestee(this, n-1)).start case Spread(s) =>
(new ThreadlessTestee(this, n-1)).start if (s > 0) {
val next = Spread(s-1)
(new ThreadlessTestee(this)).start ! next
(new ThreadlessTestee(this)).start ! next
react { react {
case Result(v1) => case Result(v1) =>
react { react {
case Result(v2) => case Result(v2) =>
parent ! new Result(v1 + v2) parent ! Result(v1 + v2)
} }
} }
} }
else { else {
parent ! new Result(1) parent ! Result(1)
}
}
}
}
class AkkaTestee(parent: akka.actor.ActorRef) extends akka.actor.Actor {
def receive = {
case Spread(s) =>
if (s > 0) {
val msg = Spread(s-1)
actorOf(new AkkaTestee(self)).start ! msg
actorOf(new AkkaTestee(self)).start ! msg
}
else {
parent ! Result(1)
self.stop
}
case Result(v1) =>
become {
case Result(v2) =>
parent ! Result(v1 + v2)
self.exit
}
} }
}
class AkkaRootTestee(n: Int) extends akka.actor.Actor {
def receive = {
case GoAhead =>
actorOf(new AkkaTestee(self)).start ! Spread(n)
case Result(v) =>
if (v != (1 << n)) {
Console.println("Expected " + (1 << n) + ", received " + v)
System.exit(42)
}
global.latch.countDown
self.exit
} }
} }
object ActorCreation { object ActorCreation {
def usage() {
Console println "usage: (threaded|threadless|akka) POW\n creates 2^POW actors of given impl"
}
def main(args: Array[String]) = { def main(args: Array[String]) = {
if (args.size != 2) { if (args.size != 2) {
Console println "usage: (threaded|threadless|akka) POW\n creates 2^POW actors of given impl" usage
throw new IllegalArgumentException("")
} }
val n = args(1).toInt val n = args(1).toInt
if (args(0) == "threaded") { if (args(0) == "threaded") {
actor { actor {
(new ThreadedTestee(self, n-1)).start (new ThreadedTestee(self)).start ! Spread(n)
(new ThreadedTestee(self, n-1)).start
receive { receive {
case Result(v1) => case Result(v) =>
receive { if (v != (1 << n))
case Result(v2) => Console.println("ERROR: expected " + (1 << n) + ", received " + v)
if ((v1 + v2) != (1 << n)) {
throw new RuntimeException("Expected " + (1 << n) + ", received " + (v1 + v2))
} }
} }
} }
} else if (args(0) == "threadless") {
} else if (args(0) == "threadless") {
actor { actor {
(new ThreadlessTestee(self, n-1)).start (new ThreadlessTestee(self)).start ! Spread(n)
(new ThreadlessTestee(self, n-1)).start
react { react {
case Result(v1) => case Result(v) =>
react { if (v != (1 << n))
case Result(v2) => Console.println("ERROR: expected " + (1 << n) + ", received " + v)
if ((v1 + v2) != (1 << n)) {
throw new RuntimeException("Expected " + (1 << n) + ", received " + (v1 + v2))
}
} }
} }
} }
else if (args(0) == "akka") {
actorOf(new AkkaRootTestee(n)).start ! GoAhead
global.latch.await
} }
else usage
} }
} }
...@@ -3,11 +3,13 @@ ACLOCAL_AMFLAGS = -I ../m4 ...@@ -3,11 +3,13 @@ ACLOCAL_AMFLAGS = -I ../m4
AM_CXXFLAGS = -I../ --std=c++0x -pedantic -Wall -Wextra AM_CXXFLAGS = -I../ --std=c++0x -pedantic -Wall -Wextra
noinst_PROGRAMS = actor_creation noinst_PROGRAMS = actor_creation mailbox_performance
actor_creation_SOURCES = actor_creation.cpp actor_creation_SOURCES = actor_creation.cpp
mailbox_performance_SOURCES = mailbox_performance.cpp
EXAMPLES_LIBS = -L../.libs/ -lcppa $(BOOST_LDFLAGS) $(BOOST_THREAD_LIB) EXAMPLES_LIBS = -L../.libs/ -lcppa $(BOOST_LDFLAGS) $(BOOST_THREAD_LIB)
actor_creation_LDADD = $(EXAMPLES_LIBS) actor_creation_LDADD = $(EXAMPLES_LIBS)
mailbox_performance_LDADD = $(EXAMPLES_LIBS)
#! /bin/sh
# actor_creation - temporary wrapper script for .libs/actor_creation
# Generated by libtool (GNU libtool) 2.4.2
#
# The actor_creation program cannot be directly executed until all the libtool
# libraries that it depends on are installed.
#
# This wrapper script should never be moved out of the build directory.
# If it is, it will not operate correctly.
# Sed substitution that helps us do robust quoting. It backslashifies
# metacharacters that are still active within double-quoted strings.
sed_quote_subst='s/\([`"$\\]\)/\\\1/g'
# Be Bourne compatible
if test -n "${ZSH_VERSION+set}" && (emulate sh) >/dev/null 2>&1; then
emulate sh
NULLCMD=:
# Zsh 3.x and 4.x performs word splitting on ${1+"$@"}, which
# is contrary to our usage. Disable this feature.
alias -g '${1+"$@"}'='"$@"'
setopt NO_GLOB_SUBST
else
case `(set -o) 2>/dev/null` in *posix*) set -o posix;; esac
fi
BIN_SH=xpg4; export BIN_SH # for Tru64
DUALCASE=1; export DUALCASE # for MKS sh
# The HP-UX ksh and POSIX shell print the target directory to stdout
# if CDPATH is set.
(unset CDPATH) >/dev/null 2>&1 && unset CDPATH
relink_command=""
# This environment variable determines our operation mode.
if test "$libtool_install_magic" = "%%%MAGIC variable%%%"; then
# install mode needs the following variables:
generated_by_libtool_version='2.4.2'
notinst_deplibs=' /Users/neverlord/libcppa/.libs/libcppa.la'
else
# When we are sourced in execute mode, $file and $ECHO are already set.
if test "$libtool_execute_magic" != "%%%MAGIC variable%%%"; then
file="$0"
# A function that is used when there is no print builtin or printf.
func_fallback_echo ()
{
eval 'cat <<_LTECHO_EOF
$1
_LTECHO_EOF'
}
ECHO="printf %s\\n"
fi
# Very basic option parsing. These options are (a) specific to
# the libtool wrapper, (b) are identical between the wrapper
# /script/ and the wrapper /executable/ which is used only on
# windows platforms, and (c) all begin with the string --lt-
# (application programs are unlikely to have options which match
# this pattern).
#
# There are only two supported options: --lt-debug and
# --lt-dump-script. There is, deliberately, no --lt-help.
#
# The first argument to this parsing function should be the
# script's ../libtool value, followed by no.
lt_option_debug=
func_parse_lt_options ()
{
lt_script_arg0=$0
shift
for lt_opt
do
case "$lt_opt" in
--lt-debug) lt_option_debug=1 ;;
--lt-dump-script)
lt_dump_D=`$ECHO "X$lt_script_arg0" | /opt/local/bin/gsed -e 's/^X//' -e 's%/[^/]*$%%'`
test "X$lt_dump_D" = "X$lt_script_arg0" && lt_dump_D=.
lt_dump_F=`$ECHO "X$lt_script_arg0" | /opt/local/bin/gsed -e 's/^X//' -e 's%^.*/%%'`
cat "$lt_dump_D/$lt_dump_F"
exit 0
;;
--lt-*)
$ECHO "Unrecognized --lt- option: '$lt_opt'" 1>&2
exit 1
;;
esac
done
# Print the debug banner immediately:
if test -n "$lt_option_debug"; then
echo "actor_creation:actor_creation:${LINENO}: libtool wrapper (GNU libtool) 2.4.2" 1>&2
fi
}
# Used when --lt-debug. Prints its arguments to stdout
# (redirection is the responsibility of the caller)
func_lt_dump_args ()
{
lt_dump_args_N=1;
for lt_arg
do
$ECHO "actor_creation:actor_creation:${LINENO}: newargv[$lt_dump_args_N]: $lt_arg"
lt_dump_args_N=`expr $lt_dump_args_N + 1`
done
}
# Core function for launching the target application
func_exec_program_core ()
{
if test -n "$lt_option_debug"; then
$ECHO "actor_creation:actor_creation:${LINENO}: newargv[0]: $progdir/$program" 1>&2
func_lt_dump_args ${1+"$@"} 1>&2
fi
exec "$progdir/$program" ${1+"$@"}
$ECHO "$0: cannot exec $program $*" 1>&2
exit 1
}
# A function to encapsulate launching the target application
# Strips options in the --lt-* namespace from $@ and
# launches target application with the remaining arguments.
func_exec_program ()
{
case " $* " in
*\ --lt-*)
for lt_wr_arg
do
case $lt_wr_arg in
--lt-*) ;;
*) set x "$@" "$lt_wr_arg"; shift;;
esac
shift
done ;;
esac
func_exec_program_core ${1+"$@"}
}
# Parse options
func_parse_lt_options "$0" ${1+"$@"}
# Find the directory that this script lives in.
thisdir=`$ECHO "$file" | /opt/local/bin/gsed 's%/[^/]*$%%'`
test "x$thisdir" = "x$file" && thisdir=.
# Follow symbolic links until we get to the real thisdir.
file=`ls -ld "$file" | /opt/local/bin/gsed -n 's/.*-> //p'`
while test -n "$file"; do
destdir=`$ECHO "$file" | /opt/local/bin/gsed 's%/[^/]*$%%'`
# If there was a directory component, then change thisdir.
if test "x$destdir" != "x$file"; then
case "$destdir" in
[\\/]* | [A-Za-z]:[\\/]*) thisdir="$destdir" ;;
*) thisdir="$thisdir/$destdir" ;;
esac
fi
file=`$ECHO "$file" | /opt/local/bin/gsed 's%^.*/%%'`
file=`ls -ld "$thisdir/$file" | /opt/local/bin/gsed -n 's/.*-> //p'`
done
# Usually 'no', except on cygwin/mingw when embedded into
# the cwrapper.
WRAPPER_SCRIPT_BELONGS_IN_OBJDIR=no
if test "$WRAPPER_SCRIPT_BELONGS_IN_OBJDIR" = "yes"; then
# special case for '.'
if test "$thisdir" = "."; then
thisdir=`pwd`
fi
# remove .libs from thisdir
case "$thisdir" in
*[\\/].libs ) thisdir=`$ECHO "$thisdir" | /opt/local/bin/gsed 's%[\\/][^\\/]*$%%'` ;;
.libs ) thisdir=. ;;
esac
fi
# Try to get the absolute directory name.
absdir=`cd "$thisdir" && pwd`
test -n "$absdir" && thisdir="$absdir"
program='actor_creation'
progdir="$thisdir/.libs"
if test -f "$progdir/$program"; then
# Add our own library path to DYLD_LIBRARY_PATH
DYLD_LIBRARY_PATH="/Users/neverlord/libcppa/.libs:$DYLD_LIBRARY_PATH"
# Some systems cannot cope with colon-terminated DYLD_LIBRARY_PATH
# The second colon is a workaround for a bug in BeOS R4 sed
DYLD_LIBRARY_PATH=`$ECHO "$DYLD_LIBRARY_PATH" | /opt/local/bin/gsed 's/::*$//'`
export DYLD_LIBRARY_PATH
if test "$libtool_execute_magic" != "%%%MAGIC variable%%%"; then
# Run the actual program with our arguments.
func_exec_program ${1+"$@"}
fi
else
# The program doesn't exist.
$ECHO "$0: error: \`$progdir/$program' does not exist" 1>&2
$ECHO "This script is just a wrapper for $program." 1>&2
$ECHO "See the libtool documentation for more information." 1>&2
exit 1
fi
fi
...@@ -43,57 +43,72 @@ using std::uint32_t; ...@@ -43,57 +43,72 @@ using std::uint32_t;
using namespace cppa; using namespace cppa;
struct testee : event_based_actor struct testee : fsm_actor<testee>
{ {
uint32_t m_v1;
actor_ptr m_parent; actor_ptr m_parent;
int m_x; behavior init_state;
testee(actor_ptr const& parent, int x) : m_parent(parent), m_x(x) behavior wait4result1;
behavior wait4result2;
testee(actor_ptr const& parent) : m_v1(0), m_parent(parent)
{ {
} init_state =
void init()
{
if (m_x > 0)
{
spawn(new testee(this, m_x - 1));
spawn(new testee(this, m_x - 1));
become
( (
on<atom("result"),uint32_t>() >> [=](uint32_t value1) on<atom("spread"), int>() >> [=](int x)
{ {
become if (x > 0)
( {
on<atom("result"),uint32_t>() >> [=](uint32_t value2) any_tuple msg = make_tuple(atom("spread"), x - 1);
spawn(new testee(this)) << msg;
spawn(new testee(this)) << msg;
become(&wait4result1);
}
else
{ {
send(m_parent, atom("result"), value1 + value2); send(m_parent, atom("result"), (std::uint32_t) 1);
quit(exit_reason::normal); quit(exit_reason::normal);
} }
);
} }
); );
wait4result1 =
(
on<atom("result"), uint32_t>() >> [=](uint32_t v1)
{
m_v1 = v1;
become(&wait4result2);
} }
else );
wait4result2 =
(
on<atom("result"), uint32_t>() >> [=](uint32_t v2)
{ {
send(m_parent, atom("result"), (std::uint32_t) 1); send(m_parent, atom("result"), m_v1 + v2);
quit(exit_reason::normal);
} }
);
} }
}; };
void cr_stacked_actor(actor_ptr parent, int x) void stacked_actor(actor_ptr parent)
{ {
receive
(
on<atom("spread"), int>() >> [&](int x)
{
if (x > 0) if (x > 0)
{ {
spawn(cr_stacked_actor, self, x - 1); any_tuple msg = make_tuple(atom("spread"), x - 1);
spawn(cr_stacked_actor, self, x - 1); spawn(stacked_actor, self) << msg;
spawn(stacked_actor, self) << msg;
receive receive
( (
on<atom("result"),uint32_t>() >> [&](uint32_t value1) on<atom("result"), uint32_t>() >> [&](uint32_t v1)
{ {
receive receive
( (
on<atom("result"),uint32_t>() >> [&](uint32_t value2) on<atom("result"),uint32_t>() >> [&](uint32_t v2)
{ {
send(parent, atom("result"), value1 + value2); send(parent, atom("result"), v1 + v2);
quit(exit_reason::normal);
} }
); );
} }
...@@ -103,6 +118,8 @@ void cr_stacked_actor(actor_ptr parent, int x) ...@@ -103,6 +118,8 @@ void cr_stacked_actor(actor_ptr parent, int x)
{ {
send(parent, atom("result"), (std::uint32_t) 1); send(parent, atom("result"), (std::uint32_t) 1);
} }
}
);
} }
void usage() void usage()
...@@ -125,11 +142,11 @@ int main(int argc, char** argv) ...@@ -125,11 +142,11 @@ int main(int argc, char** argv)
} }
if (strcmp(argv[1], "stacked") == 0) if (strcmp(argv[1], "stacked") == 0)
{ {
spawn(cr_stacked_actor, self, num); send(spawn(stacked_actor, self), atom("spread"), num);
} }
else if (strcmp(argv[1], "event-based") == 0) else if (strcmp(argv[1], "event-based") == 0)
{ {
spawn(new testee(self, num)); send(spawn(new testee(self)), atom("spread"), num);
} }
else else
{ {
......
-module(actor_creation). -module(actor_creation).
-export([start/1, testee/2]). -export([start/1, testee/1]).
testee(Pid, 0) -> testee(Pid) ->
receive
{spread, 0} ->
Pid ! {result, 1}; Pid ! {result, 1};
{spread, X} ->
testee(Pid, N) -> spawn(actor_creation, testee, [self()]) ! {spread, X-1},
spawn(actor_creation, testee, [self(),N-1]), spawn(actor_creation, testee, [self()]) ! {spread, X-1},
spawn(actor_creation, testee, [self(),N-1]),
receive receive
{result, X1} -> {result, X1} ->
receive receive
{result, X2} -> {result, X2} ->
Pid ! {result, (X1+X2)} Pid ! {result, (X1+X2)}
end end
end
end. end.
start(X) -> start(X) ->
[H|_] = X, [H|_] = X,
N = list_to_integer(atom_to_list(H)), N = list_to_integer(atom_to_list(H)),
spawn(actor_creation, testee, [self(),N]), spawn(actor_creation, testee, [self()]) ! {spread, N},
spawn(actor_creation, testee, [self(),N]),
receive receive
{result, X1} -> {result, X} ->
receive
{result, X2} ->
if if
(X1+X2) == (2 bsl N) -> X == (2 bsl N) ->
X1+X2; X;
true -> true ->
error("unexpted result!") error("unexpted result!")
end end
end
end. end.
...@@ -245,3 +245,4 @@ src/self.cpp ...@@ -245,3 +245,4 @@ src/self.cpp
cppa/behavior.hpp cppa/behavior.hpp
src/receive.cpp src/receive.cpp
benchmarks/actor_creation.cpp benchmarks/actor_creation.cpp
benchmarks/mailbox_performance.cpp
...@@ -51,7 +51,7 @@ class either ...@@ -51,7 +51,7 @@ class either
bool m_is_left; bool m_is_left;
void check_flag(bool flag, char const* side) void check_flag(bool flag, char const* side) const
{ {
if (m_is_left != flag) if (m_is_left != flag)
{ {
...@@ -212,25 +212,25 @@ class either ...@@ -212,25 +212,25 @@ class either
Left& left() Left& left()
{ {
check_flag(true, "left"); //check_flag(true, "left");
return m_left; return m_left;
} }
Left const& left() const Left const& left() const
{ {
check_flag(true, "left"); //check_flag(true, "left");
return m_left; return m_left;
} }
Right& right() Right& right()
{ {
check_flag(false, "right"); //check_flag(false, "right");
return m_right; return m_right;
} }
Right const& right() const Right const& right() const
{ {
check_flag(false, "right"); //check_flag(false, "right");
return m_right; return m_right;
} }
......
...@@ -80,20 +80,20 @@ void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& nod ...@@ -80,20 +80,20 @@ void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& nod
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node) void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node)
{ {
if (m_loop_stack.top().is_left()) auto& bhvr = m_loop_stack.top();
if (bhvr.is_left())
{ {
handle_message(node, m_loop_stack.top().left()); handle_message(node, bhvr.left());
} }
else else
{ {
handle_message(node, m_loop_stack.top().right()); handle_message(node, bhvr.right());
} }
} }
void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback) void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
{ {
self.set(this); self.set(this);
//set_self(this);
auto done_cb = [&]() auto done_cb = [&]()
{ {
m_state.store(abstract_scheduled_actor::done); m_state.store(abstract_scheduled_actor::done);
...@@ -102,9 +102,9 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback) ...@@ -102,9 +102,9 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
callback->exec_done(); callback->exec_done();
}; };
bool actor_done = false;
std::unique_ptr<queue_node> node; std::unique_ptr<queue_node> node;
do for (;;)
//do
{ {
if (m_loop_stack.empty()) if (m_loop_stack.empty())
{ {
...@@ -149,20 +149,17 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback) ...@@ -149,20 +149,17 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
catch (actor_exited& what) catch (actor_exited& what)
{ {
cleanup(what.reason()); cleanup(what.reason());
actor_done = true; done_cb();
return;
} }
catch (...) catch (...)
{ {
cleanup(exit_reason::unhandled_exception); cleanup(exit_reason::unhandled_exception);
actor_done = true;
}
if (actor_done)
{
done_cb(); done_cb();
return; return;
} }
} }
while (callback->still_ready()); //while (callback->still_ready());
} }
void abstract_event_based_actor::on_exit() void abstract_event_based_actor::on_exit()
......
...@@ -86,25 +86,27 @@ struct thread_pool_scheduler::worker ...@@ -86,25 +86,27 @@ struct thread_pool_scheduler::worker
void operator()() void operator()()
{ {
typedef decltype(now()) time_type; //typedef decltype(now()) time_type;
// enqueue as idle worker // enqueue as idle worker
m_supervisor_queue->push_back(this); m_supervisor_queue->push_back(this);
util::fiber fself; util::fiber fself;
struct handler : abstract_scheduled_actor::resume_callback struct handler : abstract_scheduled_actor::resume_callback
{ {
time_type timeout;
bool reschedule;
abstract_scheduled_actor* job; abstract_scheduled_actor* job;
handler() : timeout(now()), reschedule(false), job(nullptr) //time_type timeout;
//bool reschedule;
handler() : job(nullptr)//, timeout(now()), reschedule(false)
{ {
} }
bool still_ready() bool still_ready()
{ {
/*
if (timeout >= now()) if (timeout >= now())
{ {
reschedule = true; reschedule = true;
return false; return false;
} }
*/
return true; return true;
} }
void exec_done() void exec_done()
...@@ -128,6 +130,7 @@ struct thread_pool_scheduler::worker ...@@ -128,6 +130,7 @@ struct thread_pool_scheduler::worker
if (m_done) return; if (m_done) return;
} }
h.job = const_cast<abstract_scheduled_actor*>(m_job); h.job = const_cast<abstract_scheduled_actor*>(m_job);
/*
// run actor up to 300ms // run actor up to 300ms
h.reschedule = false; h.reschedule = false;
h.timeout = now(); h.timeout = now();
...@@ -137,6 +140,8 @@ struct thread_pool_scheduler::worker ...@@ -137,6 +140,8 @@ struct thread_pool_scheduler::worker
{ {
m_job_queue->push_back(h.job); m_job_queue->push_back(h.job);
} }
*/
h.job->resume(&fself, &h);
m_job = nullptr; m_job = nullptr;
CPPA_MEMORY_BARRIER(); CPPA_MEMORY_BARRIER();
m_supervisor_queue->push_back(this); m_supervisor_queue->push_back(this);
...@@ -156,7 +161,9 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -156,7 +161,9 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
worker_queue wqueue; worker_queue wqueue;
std::vector<worker_ptr> workers; std::vector<worker_ptr> workers;
// init with at least two workers // init with at least two workers
size_t num_workers = std::max<size_t>(thread::hardware_concurrency(), 2); //size_t num_workers = std::max<size_t>(thread::hardware_concurrency(), 2);
// init with 2 threads per core but no less than 4
size_t num_workers = std::max<size_t>(thread::hardware_concurrency() * 2, 4);
auto new_worker = [&]() auto new_worker = [&]()
{ {
worker_ptr wptr(new worker(&wqueue, jqueue)); worker_ptr wptr(new worker(&wqueue, jqueue));
...@@ -180,10 +187,10 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -180,10 +187,10 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
else else
{ {
// fetch next idle worker (wait up to 500ms) // fetch next idle worker (wait up to 500ms)
worker* w = nullptr; //worker* w = nullptr;
auto timeout = now(); //auto timeout = now();
timeout += std::chrono::milliseconds(500); //timeout += std::chrono::milliseconds(500);
while (!w) /*while (!w)
{ {
w = wqueue.try_pop(timeout); w = wqueue.try_pop(timeout);
// all workers are blocked since 500ms, start a new one // all workers are blocked since 500ms, start a new one
...@@ -192,6 +199,8 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue, ...@@ -192,6 +199,8 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
new_worker(); new_worker();
} }
} }
*/
worker* w = wqueue.pop();
// lifetime scope of guard // lifetime scope of guard
{ {
guard_type guard(w->m_mtx); guard_type guard(w->m_mtx);
......
...@@ -28,6 +28,8 @@ ...@@ -28,6 +28,8 @@
\******************************************************************************/ \******************************************************************************/
#include <iostream>
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
#include "cppa/self.hpp" #include "cppa/self.hpp"
#include "cppa/detail/invokable.hpp" #include "cppa/detail/invokable.hpp"
...@@ -144,8 +146,9 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback) ...@@ -144,8 +146,9 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback)
} }
case yield_state::ready: case yield_state::ready:
{ {
if (callback->still_ready()) break; break;
else return; //if (callback->still_ready()) break;
//else return;
} }
case yield_state::blocked: case yield_state::blocked:
{ {
...@@ -154,8 +157,9 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback) ...@@ -154,8 +157,9 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback)
{ {
case abstract_scheduled_actor::ready: case abstract_scheduled_actor::ready:
{ {
if (callback->still_ready()) break; break;
else return; //if (callback->still_ready()) break;
//else return;
} }
case abstract_scheduled_actor::blocked: case abstract_scheduled_actor::blocked:
{ {
......
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