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
04c11648
Commit
04c11648
authored
May 31, 2018
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'topic/latency'
parents
7521d55e
05f70d8d
Changes
1
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
52 additions
and
12 deletions
+52
-12
libcaf_core/caf/policy/work_stealing.hpp
libcaf_core/caf/policy/work_stealing.hpp
+52
-12
No files found.
libcaf_core/caf/policy/work_stealing.hpp
View file @
04c11648
...
...
@@ -23,6 +23,8 @@
#include <thread>
#include <random>
#include <cstddef>
#include <mutex>
#include <condition_variable>
#include "caf/resumable.hpp"
#include "caf/actor_system_config.hpp"
...
...
@@ -53,6 +55,13 @@ public:
usec
sleep_duration
;
};
// what is needed to implement the waiting strategy.
struct
wait_strategy
{
std
::
mutex
lock
;
std
::
condition_variable
cv
;
bool
sleeping
{
false
};
};
// The coordinator has only a counter for round-robin enqueue to its workers.
struct
coordinator_data
{
inline
explicit
coordinator_data
(
scheduler
::
abstract_coordinator
*
)
...
...
@@ -90,6 +99,7 @@ public:
std
::
default_random_engine
rengine
;
std
::
uniform_int_distribution
<
size_t
>
uniform
;
poll_strategy
strategies
[
3
];
wait_strategy
waitdata
;
};
// Goes on a raid in quest for a shiny new job.
...
...
@@ -117,6 +127,14 @@ public:
template
<
class
Worker
>
void
external_enqueue
(
Worker
*
self
,
resumable
*
job
)
{
d
(
self
).
queue
.
append
(
job
);
auto
&
lock
=
d
(
self
).
waitdata
.
lock
;
auto
&
cv
=
d
(
self
).
waitdata
.
cv
;
{
// guard scope
std
::
unique_lock
<
std
::
mutex
>
guard
(
lock
);
// check if the worker is sleeping
if
(
d
(
self
).
waitdata
.
sleeping
&&
!
d
(
self
).
queue
.
empty
()
)
cv
.
notify_one
();
}
}
template
<
class
Worker
>
...
...
@@ -136,30 +154,52 @@ public:
// 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
// dequeue attempts
auto
&
strategies
=
d
(
self
).
strategies
;
resumable
*
job
=
nullptr
;
for
(
auto
&
strat
:
strategies
)
{
for
(
size_t
i
=
0
;
i
<
strat
.
attempts
;
i
+=
strat
.
step_size
)
{
for
(
int
k
=
0
;
k
<
2
;
++
k
)
{
// iterate over the first two strategies
for
(
size_t
i
=
0
;
i
<
strat
egies
[
k
].
attempts
;
i
+=
strategies
[
k
]
.
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
)
{
if
((
i
%
strat
egies
[
k
]
.
steal_interval
)
==
0
)
{
job
=
try_steal
(
self
);
if
(
job
)
return
job
;
}
if
(
strat
.
sleep_duration
.
count
()
>
0
)
std
::
this_thread
::
sleep_for
(
strat
.
sleep_duration
);
if
(
strat
egies
[
k
]
.
sleep_duration
.
count
()
>
0
)
std
::
this_thread
::
sleep_for
(
strat
egies
[
k
]
.
sleep_duration
);
}
}
// unreachable, because the last strategy loops
// until a job has been dequeued
return
nullptr
;
// we assume pretty much nothing is going on so we can relax polling
// and falling to sleep on a condition variable whose timeout is the one
// of the relaxed polling strategy
auto
&
relaxed
=
strategies
[
2
];
auto
&
sleeping
=
d
(
self
).
waitdata
.
sleeping
;
auto
&
lock
=
d
(
self
).
waitdata
.
lock
;
auto
&
cv
=
d
(
self
).
waitdata
.
cv
;
bool
notimeout
=
true
;
size_t
i
=
1
;
do
{
{
// guard scope
std
::
unique_lock
<
std
::
mutex
>
guard
(
lock
);
sleeping
=
true
;
if
(
!
cv
.
wait_for
(
guard
,
relaxed
.
sleep_duration
,
[
&
]
{
return
!
d
(
self
).
queue
.
empty
();
}))
notimeout
=
false
;
sleeping
=
false
;
}
if
(
notimeout
)
{
job
=
d
(
self
).
queue
.
take_head
();
}
else
{
notimeout
=
true
;
if
((
i
%
relaxed
.
steal_interval
)
==
0
)
job
=
try_steal
(
self
);
}
++
i
;
}
while
(
job
==
nullptr
);
return
job
;
}
template
<
class
Worker
,
class
UnaryFunction
>
...
...
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