Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
916326a4
Commit
916326a4
authored
Apr 28, 2017
by
Sebastian Woelke
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Implement NUMA-aware work stealing scheduler
parent
4c713f45
Changes
14
Hide whitespace changes
Inline
Side-by-side
Showing
14 changed files
with
491 additions
and
61 deletions
+491
-61
CMakeLists.txt
CMakeLists.txt
+11
-10
cmake/build_config.hpp.in
cmake/build_config.hpp.in
+1
-1
examples/CMakeLists.txt
examples/CMakeLists.txt
+1
-0
examples/hello_world.cpp
examples/hello_world.cpp
+3
-4
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+1
-0
libcaf_core/caf/policy/numa_aware_work_stealing.hpp
libcaf_core/caf/policy/numa_aware_work_stealing.hpp
+323
-0
libcaf_core/caf/policy/scheduler_policy.hpp
libcaf_core/caf/policy/scheduler_policy.hpp
+13
-0
libcaf_core/caf/policy/work_sharing.hpp
libcaf_core/caf/policy/work_sharing.hpp
+21
-2
libcaf_core/caf/policy/work_stealing.hpp
libcaf_core/caf/policy/work_stealing.hpp
+44
-17
libcaf_core/caf/scheduler/coordinator.hpp
libcaf_core/caf/scheduler/coordinator.hpp
+11
-15
libcaf_core/caf/scheduler/worker.hpp
libcaf_core/caf/scheduler/worker.hpp
+2
-1
libcaf_core/src/actor_system.cpp
libcaf_core/src/actor_system.cpp
+19
-9
libcaf_core/src/actor_system_config.cpp
libcaf_core/src/actor_system_config.cpp
+2
-2
libcaf_core/src/numa_aware_work_stealing.cpp
libcaf_core/src/numa_aware_work_stealing.cpp
+39
-0
No files found.
CMakeLists.txt
View file @
916326a4
...
@@ -277,6 +277,15 @@ endif(CAF_ENABLE_ADDRESS_SANITIZER)
...
@@ -277,6 +277,15 @@ endif(CAF_ENABLE_ADDRESS_SANITIZER)
if
(
NOT APPLE AND NOT WIN32
)
if
(
NOT APPLE AND NOT WIN32
)
set
(
EXTRA_FLAGS
"
${
EXTRA_FLAGS
}
-pthread"
)
set
(
EXTRA_FLAGS
"
${
EXTRA_FLAGS
}
-pthread"
)
endif
()
endif
()
# load hwloc library if not told otherwise
if
(
NOT CAF_NO_HWLOC
)
find_package
(
Hwloc
)
if
(
Hwloc_FOUND
)
set
(
LD_FLAGS
${
LD_FLAGS
}
${
Hwloc_LIBRARIES
}
)
else
()
set
(
CAF_NO_HWLOC yes
)
endif
()
endif
()
# -fPIC generates warnings on MinGW and Cygwin plus extra setup steps needed on MinGW
# -fPIC generates warnings on MinGW and Cygwin plus extra setup steps needed on MinGW
if
(
MINGW
)
if
(
MINGW
)
add_definitions
(
-D_WIN32_WINNT=0x0600
)
add_definitions
(
-D_WIN32_WINNT=0x0600
)
...
@@ -345,6 +354,7 @@ macro(to_int_value name)
...
@@ -345,6 +354,7 @@ macro(to_int_value name)
endif
()
endif
()
endmacro
()
endmacro
()
to_int_value
(
CAF_NO_HWLOC
)
to_int_value
(
CAF_LOG_LEVEL
)
to_int_value
(
CAF_LOG_LEVEL
)
to_int_value
(
CAF_NO_EXCEPTIONS
)
to_int_value
(
CAF_NO_EXCEPTIONS
)
to_int_value
(
CAF_NO_MEM_MANAGEMENT
)
to_int_value
(
CAF_NO_MEM_MANAGEMENT
)
...
@@ -496,6 +506,7 @@ endmacro()
...
@@ -496,6 +506,7 @@ endmacro()
# build core and I/O library
# build core and I/O library
add_caf_lib
(
core
)
add_caf_lib
(
core
)
set
(
CAF_CORE_LIBRARY libcaf_core_shared
)
add_optional_caf_lib
(
io
)
add_optional_caf_lib
(
io
)
# build opencl library if not told otherwise and OpenCL package was found
# build opencl library if not told otherwise and OpenCL package was found
...
@@ -509,16 +520,6 @@ if(NOT CAF_NO_OPENCL)
...
@@ -509,16 +520,6 @@ if(NOT CAF_NO_OPENCL)
endif
()
endif
()
endif
()
endif
()
# load hwloc library if not told otherwise
if
(
NOT CAF_NO_HWLOC
)
find_package
(
Hwloc
)
if
(
Hwloc_FOUND
)
add_optional_caf_lib
(
hwloc
)
else
()
set
(
CAF_NO_HWLOC yes
)
endif
()
endif
()
# build Python binding if not being told otherwise
# build Python binding if not being told otherwise
if
(
NOT CAF_NO_PYTHON AND EXISTS
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/libcaf_python/CMakeLists.txt"
)
if
(
NOT CAF_NO_PYTHON AND EXISTS
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/libcaf_python/CMakeLists.txt"
)
add_subdirectory
(
libcaf_python
)
add_subdirectory
(
libcaf_python
)
...
...
cmake/build_config.hpp.in
View file @
916326a4
...
@@ -42,7 +42,7 @@
...
@@ -42,7 +42,7 @@
#define CAF_NO_EXCEPTIONS
#define CAF_NO_EXCEPTIONS
#endif
#endif
#if @CAF_NO_HWLOC@ != -1
#if @CAF_NO_HWLOC
_INT
@ != -1
#define CAF_NO_HWLOC
#define CAF_NO_HWLOC
#endif
#endif
...
...
examples/CMakeLists.txt
View file @
916326a4
...
@@ -25,6 +25,7 @@ endmacro()
...
@@ -25,6 +25,7 @@ endmacro()
# introductionary applications
# introductionary applications
add
(
. aout
)
add
(
. aout
)
add
(
. hello_world
)
add
(
. hello_world
)
add
(
. fibonacci
)
# basic message passing primitives
# basic message passing primitives
add
(
message_passing cell
)
add
(
message_passing cell
)
...
...
examples/hello_world.cpp
View file @
916326a4
...
@@ -33,13 +33,12 @@ void hello_world(event_based_actor* self, const actor& buddy) {
...
@@ -33,13 +33,12 @@ void hello_world(event_based_actor* self, const actor& buddy) {
);
);
}
}
int
main
()
{
void
caf_main
(
actor_system
&
system
)
{
// our CAF environment
actor_system_config
cfg
;
actor_system
system
{
cfg
};
// create a new actor that calls 'mirror()'
// create a new actor that calls 'mirror()'
auto
mirror_actor
=
system
.
spawn
(
mirror
);
auto
mirror_actor
=
system
.
spawn
(
mirror
);
// create another actor that calls 'hello_world(mirror_actor)';
// create another actor that calls 'hello_world(mirror_actor)';
system
.
spawn
(
hello_world
,
mirror_actor
);
system
.
spawn
(
hello_world
,
mirror_actor
);
// system will wait until both actors are destroyed before leaving main
// system will wait until both actors are destroyed before leaving main
}
}
CAF_MAIN
()
libcaf_core/CMakeLists.txt
View file @
916326a4
...
@@ -70,6 +70,7 @@ set (LIBCAF_CORE_SRCS
...
@@ -70,6 +70,7 @@ set (LIBCAF_CORE_SRCS
src/message_view.cpp
src/message_view.cpp
src/monitorable_actor.cpp
src/monitorable_actor.cpp
src/node_id.cpp
src/node_id.cpp
src/numa_aware_work_stealing.cpp
src/outgoing_stream_multiplexer.cpp
src/outgoing_stream_multiplexer.cpp
src/parse_ini.cpp
src/parse_ini.cpp
src/private_thread.cpp
src/private_thread.cpp
...
...
libcaf_core/caf/policy/numa_aware_work_stealing.hpp
0 → 100644
View file @
916326a4
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_NUMA_AWARE_WORK_STEALING_HPP
#define CAF_POLICY_NUMA_AWARE_WORK_STEALING_HPP
#include <deque>
#include <chrono>
#include <thread>
#include <random>
#include <cstddef>
#include <hwloc.h>
#include "caf/policy/work_stealing.hpp"
namespace
caf
{
namespace
policy
{
#define CALL_CAF_CRITICAL(predicate, msg) \
if (predicate) \
CAF_CRITICAL(msg)
/// Implements scheduling of actors via a numa aware work stealing.
/// @extends scheduler_policy
class
numa_aware_work_stealing
:
public
work_stealing
{
public:
~
numa_aware_work_stealing
();
struct
hwloc_bitmap_topo_free
{
void
operator
()(
hwloc_topology_t
p
)
{
hwloc_topology_destroy
(
p
);
}
};
using
topo_ptr
=
std
::
unique_ptr
<
hwloc_topology
,
hwloc_bitmap_topo_free
>
;
template
<
class
Worker
>
struct
worker_deleter
{
worker_deleter
(
topo_ptr
&
t
)
:
topo
(
t
)
{
}
void
operator
()(
void
*
p
)
{
hwloc_free
(
topo
.
get
(),
p
,
sizeof
(
Worker
));
}
topo_ptr
&
topo
;
};
struct
hwloc_bitmap_free_wrapper
{
void
operator
()(
hwloc_bitmap_t
p
)
{
hwloc_bitmap_free
(
p
);
}
};
using
hwloc_bitmap_wrapper
=
std
::
unique_ptr
<
hwloc_bitmap_s
,
hwloc_bitmap_free_wrapper
>
;
static
hwloc_bitmap_wrapper
hwloc_bitmap_make_wrapper
()
{
return
hwloc_bitmap_wrapper
(
hwloc_bitmap_alloc
());
}
using
pu_id_t
=
int
;
using
node_id_t
=
int
;
using
pu_set_t
=
hwloc_bitmap_wrapper
;
using
node_set_t
=
hwloc_bitmap_wrapper
;
template
<
class
Worker
>
struct
coordinator_data
{
inline
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
{
int
res
;
hwloc_topology_t
raw_topo
;
res
=
hwloc_topology_init
(
&
raw_topo
);
CALL_CAF_CRITICAL
(
res
==
-
1
,
"hwloc_topology_init() failed"
);
topo
.
reset
(
raw_topo
);
res
=
hwloc_topology_load
(
topo
.
get
());
CALL_CAF_CRITICAL
(
res
==
-
1
,
"hwloc_topology_load() failed"
);
next_worker
=
0
;
}
topo_ptr
topo
;
std
::
vector
<
std
::
unique_ptr
<
Worker
,
worker_deleter
<
Worker
>>>
workers
;
std
::
map
<
pu_id_t
,
Worker
*>
worker_id_map
;
// used by central enqueue to balance new jobs between workers with round
// robin strategy
std
::
atomic
<
size_t
>
next_worker
;
};
template
<
class
Worker
>
struct
worker_data
{
using
worker_matrix_t
=
std
::
vector
<
std
::
vector
<
Worker
*>>
;
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
p
)
:
rengine
(
std
::
random_device
{}())
,
strategies
(
get_poll_strategies
(
p
))
{
// nop
}
worker_matrix_t
init_worker_matrix
(
Worker
*
self
,
const
pu_set_t
&
current_pu_id_set
)
{
auto
&
cdata
=
d
(
self
->
parent
());
auto
&
topo
=
cdata
.
topo
;
auto
current_node_set
=
hwloc_bitmap_make_wrapper
();
hwloc_cpuset_to_nodeset
(
topo
.
get
(),
current_pu_id_set
.
get
(),
current_node_set
.
get
());
CALL_CAF_CRITICAL
(
hwloc_bitmap_iszero
(
current_node_set
.
get
()),
"Current NUMA node_set is unknown"
);
auto
current_node_id
=
hwloc_bitmap_first
(
current_node_set
.
get
());
std
::
map
<
float
,
pu_set_t
>
dist_map
;
worker_matrix_t
result_matrix
;
// Distance matrix of NUMA nodes.
// It is possible to request the distance matrix on PU level,
// which would be a better match for our usecase
// but on all tested hardware it returned a nullptr, maybe future
// work?
auto
distance_matrix
=
hwloc_get_whole_distance_matrix_by_type
(
topo
.
get
(),
HWLOC_OBJ_NUMANODE
);
// if NUMA distance matrix is not available it is assumed that all PUs
// have the same distance
if
(
!
distance_matrix
||
!
distance_matrix
->
latency
)
{
auto
allowed_const_pus
=
hwloc_topology_get_allowed_cpuset
(
topo
.
get
());
hwloc_bitmap_wrapper
allowed_pus
;
allowed_pus
.
reset
(
hwloc_bitmap_dup
(
allowed_const_pus
));
dist_map
.
insert
(
std
::
make_pair
(
1.0
,
std
::
move
(
allowed_pus
)));
}
else
{
auto
num_of_dist_objs
=
distance_matrix
->
nbobjs
;
// relvant line for the current NUMA node in distance matrix
float
*
dist_pointer
=
&
distance_matrix
->
latency
[
num_of_dist_objs
*
static_cast
<
unsigned
int
>
(
current_node_id
)];
// iterate over all NUMA nodes and classify them in distance levels
// regarding to the current NUMA node
for
(
node_id_t
x
=
0
;
static_cast
<
unsigned
int
>
(
x
)
<
num_of_dist_objs
;
++
x
)
{
node_set_t
tmp_node_set
=
hwloc_bitmap_make_wrapper
();
hwloc_bitmap_set
(
tmp_node_set
.
get
(),
static_cast
<
unsigned
int
>
(
x
));
auto
tmp_pu_set
=
hwloc_bitmap_make_wrapper
();
hwloc_cpuset_from_nodeset
(
topo
.
get
(),
tmp_pu_set
.
get
(),
tmp_node_set
.
get
());
// you cannot steal from yourself
if
(
x
==
current_node_id
)
{
hwloc_bitmap_andnot
(
tmp_pu_set
.
get
(),
tmp_pu_set
.
get
(),
current_pu_id_set
.
get
());
}
auto
dist_it
=
dist_map
.
find
(
dist_pointer
[
x
]);
if
(
dist_it
==
dist_map
.
end
())
// create a new distane level
dist_map
.
insert
(
std
::
make_pair
(
dist_pointer
[
x
],
std
::
move
(
tmp_pu_set
)));
else
// add PUs to an available distance level
hwloc_bitmap_or
(
dist_it
->
second
.
get
(),
dist_it
->
second
.
get
(),
tmp_pu_set
.
get
());
}
}
// return PU matrix sorted by its distance
result_matrix
.
reserve
(
dist_map
.
size
());
for
(
auto
&
pu_set_it
:
dist_map
)
{
std
::
vector
<
Worker
*>
current_lvl
;
auto
pu_set
=
pu_set_it
.
second
.
get
();
for
(
pu_id_t
pu_id
=
hwloc_bitmap_first
(
pu_set
);
pu_id
!=
-
1
;
pu_id
=
hwloc_bitmap_next
(
pu_set
,
pu_id
))
{
auto
worker_id_it
=
cdata
.
worker_id_map
.
find
(
pu_id
);
// if worker id is not found less worker than available PUs
// have been started
if
(
worker_id_it
!=
cdata
.
worker_id_map
.
end
())
current_lvl
.
emplace_back
(
worker_id_it
->
second
);
}
// current_lvl can be empty if all pus of NUMA node are deactivated
if
(
!
current_lvl
.
empty
())
{
result_matrix
.
emplace_back
(
std
::
move
(
current_lvl
));
}
}
//accumulate scheduler_lvls - each lvl contains all lower lvls
auto
last_lvl_it
=
result_matrix
.
begin
();
for
(
auto
current_lvl_it
=
result_matrix
.
begin
();
current_lvl_it
!=
result_matrix
.
end
();
++
current_lvl_it
)
{
if
(
current_lvl_it
!=
result_matrix
.
begin
())
{
std
::
copy
(
last_lvl_it
->
begin
(),
last_lvl_it
->
end
(),
std
::
back_inserter
(
*
current_lvl_it
))
;
++
last_lvl_it
;
}
}
return
result_matrix
;
}
// This queue is exposed to other workers that may attempt to steal jobs
// from it and the central scheduling unit can push new jobs to the queue.
queue_type
queue
;
worker_matrix_t
worker_matrix
;
std
::
default_random_engine
rengine
;
std
::
uniform_int_distribution
<
size_t
>
uniform
;
std
::
vector
<
poll_strategy
>
strategies
;
};
/// Create x workers.
template
<
class
Coordinator
,
class
Worker
>
void
create_workers
(
Coordinator
*
self
,
size_t
num_workers
,
size_t
throughput
)
{
auto
&
cdata
=
d
(
self
);
auto
&
topo
=
cdata
.
topo
;
auto
allowed_pus
=
hwloc_topology_get_allowed_cpuset
(
topo
.
get
());
size_t
num_allowed_pus
=
static_cast
<
size_t
>
(
hwloc_bitmap_weight
(
allowed_pus
));
CALL_CAF_CRITICAL
(
num_allowed_pus
<
num_workers
,
"less PUs than worker"
);
cdata
.
workers
.
reserve
(
num_allowed_pus
);
auto
pu_set
=
hwloc_bitmap_make_wrapper
();
auto
node_set
=
hwloc_bitmap_make_wrapper
();
auto
pu_id
=
hwloc_bitmap_first
(
allowed_pus
);
size_t
worker_count
=
0
;
while
(
pu_id
!=
-
1
&&
worker_count
<
num_workers
)
{
hwloc_bitmap_only
(
pu_set
.
get
(),
static_cast
<
unsigned
int
>
(
pu_id
));
hwloc_cpuset_to_nodeset
(
topo
.
get
(),
pu_set
.
get
(),
node_set
.
get
());
auto
ptr
=
hwloc_alloc_membind_nodeset
(
topo
.
get
(),
sizeof
(
Worker
),
node_set
.
get
(),
HWLOC_MEMBIND_BIND
,
HWLOC_MEMBIND_THREAD
);
std
::
unique_ptr
<
Worker
,
worker_deleter
<
Worker
>>
worker
(
new
(
ptr
)
Worker
(
static_cast
<
unsigned
int
>
(
pu_id
),
self
,
throughput
),
worker_deleter
<
Worker
>
(
topo
));
cdata
.
worker_id_map
.
insert
(
std
::
make_pair
(
pu_id
,
worker
.
get
()));
cdata
.
workers
.
emplace_back
(
std
::
move
(
worker
));
pu_id
=
hwloc_bitmap_next
(
allowed_pus
,
pu_id
);
worker_count
++
;
}
}
/// Initalize worker thread.
template
<
class
Worker
>
void
init_worker_thread
(
Worker
*
self
)
{
auto
&
wdata
=
d
(
self
);
auto
&
cdata
=
d
(
self
->
parent
());
auto
pu_set
=
hwloc_bitmap_make_wrapper
();
hwloc_bitmap_set
(
pu_set
.
get
(),
static_cast
<
unsigned
int
>
(
self
->
id
()));
auto
res
=
hwloc_set_cpubind
(
cdata
.
topo
.
get
(),
pu_set
.
get
(),
HWLOC_CPUBIND_THREAD
|
HWLOC_CPUBIND_NOMEMBIND
);
CALL_CAF_CRITICAL
(
res
==
-
1
,
"hwloc_set_cpubind() failed"
);
wdata
.
worker_matrix
=
wdata
.
init_worker_matrix
(
self
,
pu_set
);
}
template
<
class
Worker
>
resumable
*
try_steal
(
Worker
*
self
,
size_t
&
scheduler_lvl_idx
,
size_t
&
steal_cnt
)
{
//auto p = self->parent();
auto
&
wdata
=
d
(
self
);
auto
&
cdata
=
d
(
self
->
parent
());
size_t
num_workers
=
cdata
.
workers
.
size
();
if
(
num_workers
<
2
)
{
// you can't steal from yourself, can you?
return
nullptr
;
}
auto
&
wmatrix
=
wdata
.
worker_matrix
;
auto
&
scheduler_lvl
=
wmatrix
[
scheduler_lvl_idx
];
auto
res
=
scheduler_lvl
[
wdata
.
uniform
(
wdata
.
rengine
)
%
scheduler_lvl
.
size
()]
->
data
()
.
queue
.
take_tail
();
++
steal_cnt
;
if
(
steal_cnt
>=
scheduler_lvl
.
size
())
{
steal_cnt
=
0
;
++
scheduler_lvl_idx
;
if
(
scheduler_lvl_idx
>=
wmatrix
.
size
())
{
scheduler_lvl_idx
=
wmatrix
.
size
()
-
1
;
}
}
return
res
;
}
template
<
class
Worker
>
resumable
*
dequeue
(
Worker
*
self
)
{
// we wait for new jobs by polling our external queue: first, we
// assume an active work load on the machine and perform aggresive
// polling, then we relax our polling a bit and wait 50 us between
// dequeue attempts, finally we assume pretty much nothing is going
// on and poll every 10 ms; this strategy strives to minimize the
// downside of "busy waiting", which still performs much better than a
// "signalizing" implementation based on mutexes and conition variables
size_t
scheduler_lvl_idx
=
0
;
size_t
steal_cnt
=
0
;
auto
&
strategies
=
d
(
self
).
strategies
;
resumable
*
job
=
nullptr
;
for
(
auto
&
strat
:
strategies
)
{
for
(
size_t
i
=
0
;
i
<
strat
.
attempts
;
i
+=
strat
.
step_size
)
{
job
=
d
(
self
).
queue
.
take_head
();
if
(
job
)
return
job
;
// try to steal every X poll attempts
if
((
i
%
strat
.
steal_interval
)
==
0
)
{
job
=
try_steal
(
self
,
scheduler_lvl_idx
,
steal_cnt
);
if
(
job
)
return
job
;
}
if
(
strat
.
sleep_duration
.
count
()
>
0
)
std
::
this_thread
::
sleep_for
(
strat
.
sleep_duration
);
}
}
// unreachable, because the last strategy loops
// until a job has been dequeued
return
nullptr
;
}
private:
// -- debug stuff --
friend
std
::
ostream
&
operator
<<
(
std
::
ostream
&
s
,
const
hwloc_bitmap_wrapper
&
w
);
};
}
// namespace policy
}
// namespace caf
#endif // CAF_POLICY_NUMA_AWARE_WORK_STEALING_HPP
libcaf_core/caf/policy/scheduler_policy.hpp
View file @
916326a4
...
@@ -31,15 +31,28 @@ namespace policy {
...
@@ -31,15 +31,28 @@ namespace policy {
class
scheduler_policy
{
class
scheduler_policy
{
public:
public:
/// Policy-specific data fields for the coordinator.
/// Policy-specific data fields for the coordinator.
template
<
class
Worker
>
struct
coordinator_data
{
struct
coordinator_data
{
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
);
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
);
/// Stores the pointer of all workes handeld by the scheduler
std
::
vector
<
std
::
unique_ptr
<
Worker
>>
workers
;
};
};
/// Policy-specific data fields for the worker.
/// Policy-specific data fields for the worker.
template
<
class
Worker
>
struct
worker_data
{
struct
worker_data
{
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
);
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
);
};
};
/// Create x workers.
template
<
class
Coordinator
,
class
Worker
>
void
create_workers
(
Coordinator
*
self
,
size_t
num_workers
,
size_t
throughput
);
/// Initalize worker thread.
template
<
class
Worker
>
void
init_worker_thread
(
Worker
*
self
);
/// Enqueues a new job to coordinator.
/// Enqueues a new job to coordinator.
template
<
class
Coordinator
>
template
<
class
Coordinator
>
void
central_enqueue
(
Coordinator
*
self
,
resumable
*
job
);
void
central_enqueue
(
Coordinator
*
self
,
resumable
*
job
);
...
...
libcaf_core/caf/policy/work_sharing.hpp
View file @
916326a4
...
@@ -40,22 +40,41 @@ public:
...
@@ -40,22 +40,41 @@ public:
~
work_sharing
()
override
;
~
work_sharing
()
override
;
template
<
class
Worker
>
struct
coordinator_data
{
struct
coordinator_data
{
inline
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
{
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
{
// nop
// nop
}
}
queue_type
queue
;
queue_type
queue
;
std
::
mutex
lock
;
std
::
mutex
lock
;
std
::
condition_variable
cv
;
std
::
condition_variable
cv
;
std
::
vector
<
std
::
unique_ptr
<
Worker
>>
workers
;
};
};
template
<
class
Worker
>
struct
worker_data
{
struct
worker_data
{
inline
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
)
{
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
)
{
// nop
// nop
}
}
};
};
// Create x workers.
template
<
class
Coordinator
,
class
Worker
>
void
create_workers
(
Coordinator
*
self
,
size_t
num_workers
,
size_t
throughput
)
{
for
(
size_t
i
=
0
;
i
<
num_workers
;
++
i
)
{
std
::
unique_ptr
<
Worker
>
worker
(
new
Worker
(
i
,
self
,
throughput
));
d
(
self
).
workers
.
emplace_back
(
std
::
move
(
worker
));
}
}
/// Initalize worker thread.
template
<
class
Worker
>
void
init_worker_thread
(
Worker
*
)
{
// nop
}
template
<
class
Coordinator
>
template
<
class
Coordinator
>
void
enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
void
enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
queue_type
l
;
queue_type
l
;
...
...
libcaf_core/caf/policy/work_stealing.hpp
View file @
916326a4
...
@@ -56,32 +56,42 @@ public:
...
@@ -56,32 +56,42 @@ public:
};
};
// The coordinator has only a counter for round-robin enqueue to its workers.
// The coordinator has only a counter for round-robin enqueue to its workers.
template
<
class
Worker
>
struct
coordinator_data
{
struct
coordinator_data
{
inline
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
:
next_worker
(
0
)
{
:
next_worker
(
0
)
{
// nop
// nop
}
}
std
::
atomic
<
size_t
>
next_worker
;
std
::
atomic
<
size_t
>
next_worker
;
std
::
vector
<
std
::
unique_ptr
<
Worker
>>
workers
;
};
};
static
std
::
vector
<
poll_strategy
>
get_poll_strategies
(
scheduler
::
abstract_coordinator
*
p
)
{
auto
&
x
=
p
->
system
().
config
();
std
::
vector
<
poll_strategy
>
res
{
{
x
.
work_stealing_aggressive_poll_attempts
,
1
,
x
.
work_stealing_aggressive_steal_interval
,
usec
{
0
}},
{
x
.
work_stealing_moderate_poll_attempts
,
1
,
x
.
work_stealing_moderate_steal_interval
,
usec
{
x
.
work_stealing_moderate_sleep_duration_us
}},
{
1
,
0
,
x
.
work_stealing_relaxed_steal_interval
,
usec
{
p
->
system
().
config
().
work_stealing_relaxed_sleep_duration_us
}}
};
return
res
;
}
// Holds job job queue of a worker and a random number generator.
// Holds job job queue of a worker and a random number generator.
template
<
class
Worker
>
struct
worker_data
{
struct
worker_data
{
inline
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
p
)
explicit
worker_data
(
scheduler
::
abstract_coordinator
*
p
)
:
rengine
(
std
::
random_device
{}())
,
:
rengine
(
std
::
random_device
{}())
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
// `uniform` will not be used anyway
uniform
(
0
,
p
->
num_workers
()
-
2
),
,
uniform
(
0
,
p
->
num_workers
()
-
2
)
strategies
{
,
strategies
(
get_poll_strategies
(
p
))
{
{
p
->
system
().
config
().
work_stealing_aggressive_poll_attempts
,
1
,
p
->
system
().
config
().
work_stealing_aggressive_steal_interval
,
usec
{
0
}},
{
p
->
system
().
config
().
work_stealing_moderate_poll_attempts
,
1
,
p
->
system
().
config
().
work_stealing_moderate_steal_interval
,
usec
{
p
->
system
().
config
().
work_stealing_moderate_sleep_duration_us
}},
{
1
,
0
,
p
->
system
().
config
().
work_stealing_relaxed_steal_interval
,
usec
{
p
->
system
().
config
().
work_stealing_relaxed_sleep_duration_us
}}
}
{
// nop
// nop
}
}
...
@@ -91,9 +101,25 @@ public:
...
@@ -91,9 +101,25 @@ public:
// needed to generate pseudo random numbers
// needed to generate pseudo random numbers
std
::
default_random_engine
rengine
;
std
::
default_random_engine
rengine
;
std
::
uniform_int_distribution
<
size_t
>
uniform
;
std
::
uniform_int_distribution
<
size_t
>
uniform
;
poll_strategy
strategies
[
3
]
;
std
::
vector
<
poll_strategy
>
strategies
;
};
};
// Create x workers.
template
<
class
Coordinator
,
class
Worker
>
void
create_workers
(
Coordinator
*
self
,
size_t
num_workers
,
size_t
throughput
)
{
for
(
size_t
i
=
0
;
i
<
num_workers
;
++
i
)
{
std
::
unique_ptr
<
Worker
>
worker
(
new
Worker
(
i
,
self
,
throughput
));
d
(
self
).
workers
.
emplace_back
(
std
::
move
(
worker
));
}
}
/// Initalize worker thread.
template
<
class
Worker
>
void
init_worker_thread
(
Worker
*
)
{
// nop
}
// Goes on a raid in quest for a shiny new job.
// Goes on a raid in quest for a shiny new job.
template
<
class
Worker
>
template
<
class
Worker
>
resumable
*
try_steal
(
Worker
*
self
)
{
resumable
*
try_steal
(
Worker
*
self
)
{
...
@@ -104,15 +130,16 @@ public:
...
@@ -104,15 +130,16 @@ public:
}
}
// roll the dice to pick a victim other than ourselves
// roll the dice to pick a victim other than ourselves
auto
victim
=
d
(
self
).
uniform
(
d
(
self
).
rengine
);
auto
victim
=
d
(
self
).
uniform
(
d
(
self
).
rengine
);
// in this case the worker id is similar to the worker index
if
(
victim
==
self
->
id
())
if
(
victim
==
self
->
id
())
victim
=
p
->
num_workers
()
-
1
;
victim
=
p
->
num_workers
()
-
1
;
// steal oldest element from the victim's queue
// steal oldest element from the victim's queue
return
d
(
p
->
worker_by_id
(
victim
)).
queue
.
take_tail
();
return
d
(
p
->
worker_by_id
x
(
victim
)).
queue
.
take_tail
();
}
}
template
<
class
Coordinator
>
template
<
class
Coordinator
>
void
central_enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
void
central_enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
auto
w
=
self
->
worker_by_id
(
d
(
self
).
next_worker
++
%
self
->
num_workers
());
auto
w
=
self
->
worker_by_id
x
(
d
(
self
).
next_worker
++
%
self
->
num_workers
());
w
->
external_enqueue
(
job
);
w
->
external_enqueue
(
job
);
}
}
...
...
libcaf_core/caf/scheduler/coordinator.hpp
View file @
916326a4
...
@@ -39,16 +39,17 @@ class coordinator : public abstract_coordinator {
...
@@ -39,16 +39,17 @@ class coordinator : public abstract_coordinator {
public:
public:
using
super
=
abstract_coordinator
;
using
super
=
abstract_coordinator
;
using
policy_data
=
typename
Policy
::
coordinator_data
;
using
worker_type
=
worker
<
Policy
>
;
using
policy_data
=
typename
Policy
::
template
coordinator_data
<
worker_type
>;
coordinator
(
actor_system
&
sys
)
:
super
(
sys
),
data_
(
this
)
{
coordinator
(
actor_system
&
sys
)
:
super
(
sys
),
data_
(
this
)
{
// nop
// nop
}
}
using
worker_type
=
worker
<
Policy
>
;
/// return a worker by its vector index
/// (this is not necessarily the id of the worker)
worker_type
*
worker_by_id
(
size_t
x
)
{
worker_type
*
worker_by_id
x
(
size_t
x
)
{
return
workers_
[
x
].
get
();
return
data_
.
workers
[
x
].
get
();
}
}
policy_data
&
data
()
{
policy_data
&
data
()
{
...
@@ -62,12 +63,9 @@ public:
...
@@ -62,12 +63,9 @@ public:
protected:
protected:
void
start
()
override
{
void
start
()
override
{
// initialize workers vector
// initialize workers vector
auto
num
=
num_workers
();
policy_
.
template
create_workers
<
coordinator
<
Policy
>,
worker_type
>
(
this
,
num_workers
(),
max_throughput_
);
workers_
.
reserve
(
num
);
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
workers_
.
emplace_back
(
new
worker_type
(
i
,
this
,
max_throughput_
));
// start all workers now that all workers have been initialized
// start all workers now that all workers have been initialized
for
(
auto
&
w
:
workers_
)
for
(
auto
&
w
:
data_
.
workers
)
w
->
start
();
w
->
start
();
// run remaining startup code
// run remaining startup code
super
::
start
();
super
::
start
();
...
@@ -103,7 +101,7 @@ protected:
...
@@ -103,7 +101,7 @@ protected:
std
::
set
<
worker_type
*>
alive_workers
;
std
::
set
<
worker_type
*>
alive_workers
;
auto
num
=
num_workers
();
auto
num
=
num_workers
();
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
{
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
{
alive_workers
.
insert
(
worker_by_id
(
i
));
alive_workers
.
insert
(
worker_by_id
x
(
i
));
sh
.
ref
();
// make sure reference count is high enough
sh
.
ref
();
// make sure reference count is high enough
}
}
while
(
!
alive_workers
.
empty
())
{
while
(
!
alive_workers
.
empty
())
{
...
@@ -120,12 +118,12 @@ protected:
...
@@ -120,12 +118,12 @@ protected:
// shutdown utility actors
// shutdown utility actors
stop_actors
();
stop_actors
();
// wait until all workers are done
// wait until all workers are done
for
(
auto
&
w
:
workers_
)
{
for
(
auto
&
w
:
data_
.
workers
)
{
w
->
get_thread
().
join
();
w
->
get_thread
().
join
();
}
}
// run cleanup code for each resumable
// run cleanup code for each resumable
auto
f
=
&
abstract_coordinator
::
cleanup_and_release
;
auto
f
=
&
abstract_coordinator
::
cleanup_and_release
;
for
(
auto
&
w
:
workers_
)
for
(
auto
&
w
:
data_
.
workers
)
policy_
.
foreach_resumable
(
w
.
get
(),
f
);
policy_
.
foreach_resumable
(
w
.
get
(),
f
);
policy_
.
foreach_central_resumable
(
this
,
f
);
policy_
.
foreach_central_resumable
(
this
,
f
);
}
}
...
@@ -135,8 +133,6 @@ protected:
...
@@ -135,8 +133,6 @@ protected:
}
}
private:
private:
// usually of size std::thread::hardware_concurrency()
std
::
vector
<
std
::
unique_ptr
<
worker_type
>>
workers_
;
// policy-specific data
// policy-specific data
policy_data
data_
;
policy_data
data_
;
// instance of our policy object
// instance of our policy object
...
...
libcaf_core/caf/scheduler/worker.hpp
View file @
916326a4
...
@@ -40,7 +40,7 @@ class worker : public execution_unit {
...
@@ -40,7 +40,7 @@ class worker : public execution_unit {
public:
public:
using
job_ptr
=
resumable
*
;
using
job_ptr
=
resumable
*
;
using
coordinator_ptr
=
coordinator
<
Policy
>*
;
using
coordinator_ptr
=
coordinator
<
Policy
>*
;
using
policy_data
=
typename
Policy
::
worker_data
;
using
policy_data
=
typename
Policy
::
template
worker_data
<
worker
<
Policy
>
>
;
worker
(
size_t
worker_id
,
coordinator_ptr
worker_parent
,
size_t
throughput
)
worker
(
size_t
worker_id
,
coordinator_ptr
worker_parent
,
size_t
throughput
)
:
execution_unit
(
&
worker_parent
->
system
()),
:
execution_unit
(
&
worker_parent
->
system
()),
...
@@ -106,6 +106,7 @@ public:
...
@@ -106,6 +106,7 @@ public:
private:
private:
void
run
()
{
void
run
()
{
CAF_SET_LOGGER_SYS
(
&
system
());
CAF_SET_LOGGER_SYS
(
&
system
());
policy_
.
init_worker_thread
(
this
);
// scheduling loop
// scheduling loop
for
(;;)
{
for
(;;)
{
auto
job
=
policy_
.
dequeue
(
this
);
auto
job
=
policy_
.
dequeue
(
this
);
...
...
libcaf_core/src/actor_system.cpp
View file @
916326a4
...
@@ -29,6 +29,7 @@
...
@@ -29,6 +29,7 @@
#include "caf/policy/work_sharing.hpp"
#include "caf/policy/work_sharing.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/policy/numa_aware_work_stealing.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp"
...
@@ -223,6 +224,7 @@ actor_system::actor_system(actor_system_config& cfg)
...
@@ -223,6 +224,7 @@ actor_system::actor_system(actor_system_config& cfg)
using
test
=
scheduler
::
test_coordinator
;
using
test
=
scheduler
::
test_coordinator
;
using
share
=
scheduler
::
coordinator
<
policy
::
work_sharing
>
;
using
share
=
scheduler
::
coordinator
<
policy
::
work_sharing
>
;
using
steal
=
scheduler
::
coordinator
<
policy
::
work_stealing
>
;
using
steal
=
scheduler
::
coordinator
<
policy
::
work_stealing
>
;
using
numa_steal
=
scheduler
::
coordinator
<
policy
::
numa_aware_work_stealing
>
;
using
profiled_share
=
scheduler
::
profiled_coordinator
<
policy
::
profiled
<
policy
::
work_sharing
>>
;
using
profiled_share
=
scheduler
::
profiled_coordinator
<
policy
::
profiled
<
policy
::
work_sharing
>>
;
using
profiled_steal
=
scheduler
::
profiled_coordinator
<
policy
::
profiled
<
policy
::
work_stealing
>>
;
using
profiled_steal
=
scheduler
::
profiled_coordinator
<
policy
::
profiled
<
policy
::
work_stealing
>>
;
// set scheduler only if not explicitly loaded by user
// set scheduler only if not explicitly loaded by user
...
@@ -231,29 +233,37 @@ actor_system::actor_system(actor_system_config& cfg)
...
@@ -231,29 +233,37 @@ actor_system::actor_system(actor_system_config& cfg)
stealing
=
0x0001
,
stealing
=
0x0001
,
sharing
=
0x0002
,
sharing
=
0x0002
,
testing
=
0x0003
,
testing
=
0x0003
,
numa_stealing
=
0x0004
,
profiled
=
0x0100
,
profiled
=
0x0100
,
profiled_stealing
=
0x0101
,
profiled_stealing
=
0x0101
,
profiled_sharing
=
0x0102
profiled_sharing
=
0x0102
};
};
sched_conf
sc
=
stealing
;
sched_conf
sc
=
numa_stealing
;
if
(
cfg
.
scheduler_policy
==
atom
(
"sharing"
))
if
(
cfg
.
scheduler_policy
==
atom
(
"stealing"
))
sc
=
stealing
;
else
if
(
cfg
.
scheduler_policy
==
atom
(
"sharing"
))
sc
=
sharing
;
sc
=
sharing
;
else
if
(
cfg
.
scheduler_policy
==
atom
(
"testing"
))
else
if
(
cfg
.
scheduler_policy
==
atom
(
"testing"
))
sc
=
testing
;
sc
=
testing
;
else
if
(
cfg
.
scheduler_policy
!=
atom
(
"stealing"
))
else
if
(
cfg
.
scheduler_policy
!=
atom
(
"numa-steal"
))
std
::
cerr
<<
"[WARNING] "
<<
deep_to_string
(
cfg
.
scheduler_policy
)
std
::
cerr
<<
" is an unrecognized scheduler pollicy, "
<<
"[WARNING] "
<<
deep_to_string
(
cfg
.
scheduler_policy
)
"falling back to 'stealing' (i.e. work-stealing)"
<<
" is an unrecognized scheduler pollicy, "
<<
std
::
endl
;
"falling back to 'numa-steal' (i.e. numa aware work-stealing)"
<<
std
::
endl
;
if
(
cfg
.
scheduler_enable_profiling
)
if
(
cfg
.
scheduler_enable_profiling
)
sc
=
static_cast
<
sched_conf
>
(
sc
|
profiled
);
sc
=
static_cast
<
sched_conf
>
(
sc
|
profiled
);
switch
(
sc
)
{
switch
(
sc
)
{
default:
// any invalid configuration falls back to work stealing
default:
// any invalid configuration falls back to
numa
work stealing
sched
.
reset
(
new
steal
(
*
this
));
sched
.
reset
(
new
numa_
steal
(
*
this
));
break
;
break
;
case
sharing
:
case
sharing
:
sched
.
reset
(
new
share
(
*
this
));
sched
.
reset
(
new
share
(
*
this
));
break
;
break
;
case
stealing
:
sched
.
reset
(
new
steal
(
*
this
));
break
;
case
profiled_stealing
:
case
profiled_stealing
:
sched
.
reset
(
new
profiled_steal
(
*
this
));
sched
.
reset
(
new
profiled_steal
(
*
this
));
break
;
break
;
...
...
libcaf_core/src/actor_system_config.cpp
View file @
916326a4
...
@@ -110,7 +110,7 @@ actor_system_config::actor_system_config()
...
@@ -110,7 +110,7 @@ actor_system_config::actor_system_config()
add_message_type_impl
<
std
::
vector
<
atom_value
>>
(
"std::vector<@atom>"
);
add_message_type_impl
<
std
::
vector
<
atom_value
>>
(
"std::vector<@atom>"
);
add_message_type_impl
<
std
::
vector
<
message
>>
(
"std::vector<@message>"
);
add_message_type_impl
<
std
::
vector
<
message
>>
(
"std::vector<@message>"
);
// (1) hard-coded defaults
// (1) hard-coded defaults
scheduler_policy
=
atom
(
"
stealing
"
);
scheduler_policy
=
atom
(
"
numa-steal
"
);
scheduler_max_threads
=
std
::
max
(
std
::
thread
::
hardware_concurrency
(),
scheduler_max_threads
=
std
::
max
(
std
::
thread
::
hardware_concurrency
(),
unsigned
{
4
});
unsigned
{
4
});
scheduler_max_throughput
=
std
::
numeric_limits
<
size_t
>::
max
();
scheduler_max_throughput
=
std
::
numeric_limits
<
size_t
>::
max
();
...
@@ -337,7 +337,7 @@ actor_system_config& actor_system_config::parse(message& args,
...
@@ -337,7 +337,7 @@ actor_system_config& actor_system_config::parse(message& args,
atom
(
"asio"
)
atom
(
"asio"
)
# endif
# endif
},
middleman_network_backend
,
"middleman.network-backend"
);
},
middleman_network_backend
,
"middleman.network-backend"
);
verify_atom_opt
({
atom
(
"stealing"
),
atom
(
"sharing"
)},
verify_atom_opt
({
atom
(
"stealing"
),
atom
(
"sharing"
)
,
atom
(
"numa-steal"
)
},
scheduler_policy
,
"scheduler.policy "
);
scheduler_policy
,
"scheduler.policy "
);
if
(
res
.
opts
.
count
(
"caf#dump-config"
)
!=
0u
)
{
if
(
res
.
opts
.
count
(
"caf#dump-config"
)
!=
0u
)
{
cli_helptext_printed
=
true
;
cli_helptext_printed
=
true
;
...
...
libcaf_core/src/numa_aware_work_stealing.cpp
0 → 100644
View file @
916326a4
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/numa_aware_work_stealing.hpp"
namespace
caf
{
namespace
policy
{
numa_aware_work_stealing
::~
numa_aware_work_stealing
()
{
// nop
}
std
::
ostream
&
operator
<<
(
std
::
ostream
&
s
,
const
numa_aware_work_stealing
::
hwloc_bitmap_wrapper
&
w
)
{
char
*
tmp
=
nullptr
;
hwloc_bitmap_asprintf
(
&
tmp
,
w
.
get
());
s
<<
std
::
string
(
tmp
);
free
(
tmp
);
return
s
;
}
}
// namespace policy
}
// namespace caf
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment