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
c6160a26
Commit
c6160a26
authored
Mar 07, 2016
by
Lingxi-Li
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Cleanup scheduling-related code
parent
11d1732a
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
100 additions
and
100 deletions
+100
-100
libcaf_core/caf/policy/unprofiled.hpp
libcaf_core/caf/policy/unprofiled.hpp
+78
-0
libcaf_core/caf/policy/work_sharing.hpp
libcaf_core/caf/policy/work_sharing.hpp
+14
-54
libcaf_core/caf/policy/work_stealing.hpp
libcaf_core/caf/policy/work_stealing.hpp
+4
-31
libcaf_core/caf/scheduler/abstract_coordinator.hpp
libcaf_core/caf/scheduler/abstract_coordinator.hpp
+0
-2
libcaf_core/caf/scheduler/coordinator.hpp
libcaf_core/caf/scheduler/coordinator.hpp
+1
-1
libcaf_core/caf/scheduler/worker.hpp
libcaf_core/caf/scheduler/worker.hpp
+3
-12
No files found.
libcaf_core/caf/policy/unprofiled.hpp
0 → 100644
View file @
c6160a26
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_UNPROFILED_HPP
#define CAF_POLICY_UNPROFILED_HPP
namespace
caf
{
namespace
policy
{
/// This class is intended to be used as a base class for actual polices.
/// It provides a default empty implementation for the customization points.
/// By deriving from it, actual policy classes only need to implement/override
/// the customization points they need. This class also serves as a place to
/// factor common utilities for implementing actual policies.
class
unprofiled
{
public:
virtual
~
unprofiled
()
=
default
;
/// Policy-specific data fields for the coordinator.
struct
coordinator_data
{
};
/// Policy-specific data fields for the worker.
struct
worker_data
{
};
/// Performs cleanup action before a shutdown takes place.
template
<
class
Worker
>
void
before_shutdown
(
Worker
*
)
{
// nop
}
/// Called immediately before resuming an actor.
template
<
class
Worker
>
void
before_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
/// Called whenever an actor has been resumed. This function can
/// prepare some fields before the next resume operation takes place
/// or perform cleanup actions between to actor runs.
template
<
class
Worker
>
void
after_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
/// Called whenever an actor has completed a job.
template
<
class
Worker
>
void
after_completion
(
Worker
*
,
resumable
*
)
{
// nop
}
protected:
// Convenience function to access the data field.
template
<
class
WorkerOrCoordinator
>
static
auto
d
(
WorkerOrCoordinator
*
self
)
->
decltype
(
self
->
data
())
{
return
self
->
data
();
}
};
}
// namespace policy
}
// namespace caf
#endif // CAF_POLICY_UNPROFILED_HPP
libcaf_core/caf/policy/work_sharing.hpp
View file @
c6160a26
...
@@ -26,37 +26,23 @@
...
@@ -26,37 +26,23 @@
#include <condition_variable>
#include <condition_variable>
#include "caf/resumable.hpp"
#include "caf/resumable.hpp"
#include "caf/policy/unprofiled.hpp"
namespace
caf
{
namespace
caf
{
namespace
policy
{
namespace
policy
{
/// @extends scheduler_policy
/// @extends scheduler_policy
class
work_sharing
{
class
work_sharing
:
public
unprofiled
{
public:
public:
// A thead-safe queue implementation.
// A th
r
ead-safe queue implementation.
using
queue_type
=
std
::
list
<
resumable
*>
;
using
queue_type
=
std
::
list
<
resumable
*>
;
struct
coordinator_data
{
struct
coordinator_data
{
queue_type
queue
;
queue_type
queue
;
std
::
mutex
lock
;
std
::
mutex
lock
;
std
::
condition_variable
cv
;
std
::
condition_variable
cv
;
inline
coordinator_data
()
{
// nop
}
};
};
struct
worker_data
{
inline
worker_data
()
{
// nop
}
};
// Convenience function to access the data field.
template
<
class
WorkerOrCoordinator
>
static
auto
d
(
WorkerOrCoordinator
*
self
)
->
decltype
(
self
->
data
())
{
return
self
->
data
();
}
template
<
class
Coordinator
>
template
<
class
Coordinator
>
void
enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
void
enqueue
(
Coordinator
*
self
,
resumable
*
job
)
{
queue_type
l
;
queue_type
l
;
...
@@ -78,12 +64,7 @@ public:
...
@@ -78,12 +64,7 @@ public:
template
<
class
Worker
>
template
<
class
Worker
>
void
internal_enqueue
(
Worker
*
self
,
resumable
*
job
)
{
void
internal_enqueue
(
Worker
*
self
,
resumable
*
job
)
{
auto
&
parent_data
=
d
(
self
->
parent
());
enqueue
(
self
->
parent
(),
job
);
queue_type
l
;
l
.
push_back
(
job
);
std
::
unique_lock
<
std
::
mutex
>
guard
(
parent_data
.
lock
);
parent_data
.
queue
.
splice
(
parent_data
.
queue
.
begin
(),
l
);
parent_data
.
cv
.
notify_one
();
}
}
template
<
class
Worker
>
template
<
class
Worker
>
...
@@ -97,54 +78,33 @@ public:
...
@@ -97,54 +78,33 @@ public:
resumable
*
dequeue
(
Worker
*
self
)
{
resumable
*
dequeue
(
Worker
*
self
)
{
auto
&
parent_data
=
d
(
self
->
parent
());
auto
&
parent_data
=
d
(
self
->
parent
());
std
::
unique_lock
<
std
::
mutex
>
guard
(
parent_data
.
lock
);
std
::
unique_lock
<
std
::
mutex
>
guard
(
parent_data
.
lock
);
while
(
parent_data
.
queue
.
empty
())
{
parent_data
.
cv
.
wait
(
guard
,
[
&
]
{
return
!
parent_data
.
queue
.
empty
();
});
parent_data
.
cv
.
wait
(
guard
);
}
resumable
*
job
=
parent_data
.
queue
.
front
();
resumable
*
job
=
parent_data
.
queue
.
front
();
parent_data
.
queue
.
pop_front
();
parent_data
.
queue
.
pop_front
();
return
job
;
return
job
;
}
}
template
<
class
Worker
>
template
<
class
Worker
,
class
UnaryFunction
>
void
before_shutdown
(
Worker
*
)
{
void
foreach_resumable
(
Worker
*
,
UnaryFunction
)
{
// nop
}
template
<
class
Worker
>
void
before_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
template
<
class
Worker
>
void
after_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
template
<
class
Worker
>
void
after_completion
(
Worker
*
,
resumable
*
)
{
// nop
// nop
}
}
template
<
class
Worker
,
class
UnaryFunction
>
template
<
class
Coordinator
,
class
UnaryFunction
>
void
foreach_resumable
(
Worker
*
self
,
UnaryFunction
f
)
{
void
foreach_central_resumable
(
Coordinator
*
self
,
UnaryFunction
f
)
{
auto
&
queue
=
d
(
self
).
queue
;
auto
next
=
[
&
]()
->
resumable
*
{
auto
next
=
[
&
]()
->
resumable
*
{
if
(
d
(
self
->
parent
()).
queue
.
empty
())
{
if
(
queue
.
empty
())
{
return
nullptr
;
return
nullptr
;
}
}
auto
front
=
d
(
self
->
parent
()).
queue
.
front
();
auto
front
=
queue
.
front
();
d
(
self
->
parent
()).
queue
.
pop_front
();
queue
.
pop_front
();
return
front
;
return
front
;
};
};
std
::
unique_lock
<
std
::
mutex
>
guard
(
d
(
self
->
parent
()
).
lock
);
std
::
unique_lock
<
std
::
mutex
>
guard
(
d
(
self
).
lock
);
for
(
auto
job
=
next
();
job
!=
nullptr
;
job
=
next
())
{
for
(
auto
job
=
next
();
job
!=
nullptr
;
job
=
next
())
{
f
(
job
);
f
(
job
);
}
}
}
}
template
<
class
Coordinator
,
class
UnaryFunction
>
void
foreach_central_resumable
(
Coordinator
*
,
UnaryFunction
)
{
// nop
}
};
};
}
// namespace policy
}
// namespace policy
...
...
libcaf_core/caf/policy/work_stealing.hpp
View file @
c6160a26
...
@@ -27,6 +27,7 @@
...
@@ -27,6 +27,7 @@
#include <cstddef>
#include <cstddef>
#include "caf/resumable.hpp"
#include "caf/resumable.hpp"
#include "caf/policy/unprofiled.hpp"
#include "caf/detail/double_ended_queue.hpp"
#include "caf/detail/double_ended_queue.hpp"
...
@@ -49,9 +50,9 @@ namespace policy {
...
@@ -49,9 +50,9 @@ namespace policy {
/// [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639
/// [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639
///
///
/// @extends scheduler_policy
/// @extends scheduler_policy
class
work_stealing
{
class
work_stealing
:
public
unprofiled
{
public:
public:
// A thead-safe queue implementation.
// A th
r
ead-safe queue implementation.
using
queue_type
=
detail
::
double_ended_queue
<
resumable
>
;
using
queue_type
=
detail
::
double_ended_queue
<
resumable
>
;
// 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.
...
@@ -67,22 +68,14 @@ public:
...
@@ -67,22 +68,14 @@ public:
// This queue is exposed to other workers that may attempt to steal jobs
// 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.
// from it and the central scheduling unit can push new jobs to the queue.
queue_type
queue
;
queue_type
queue
;
// needed by our engine
std
::
random_device
rdevice
;
// needed to generate pseudo random numbers
// needed to generate pseudo random numbers
std
::
default_random_engine
rengine
;
std
::
default_random_engine
rengine
;
// initialize random engine
// initialize random engine
inline
worker_data
()
:
r
device
(),
rengine
(
rdevice
())
{
inline
worker_data
()
:
r
engine
(
std
::
random_device
{}
())
{
// nop
// nop
}
}
};
};
// Convenience function to access the data field.
template
<
class
WorkerOrCoordinator
>
static
auto
d
(
WorkerOrCoordinator
*
self
)
->
decltype
(
self
->
data
())
{
return
self
->
data
();
}
// 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
)
{
...
@@ -167,26 +160,6 @@ public:
...
@@ -167,26 +160,6 @@ public:
return
nullptr
;
return
nullptr
;
}
}
template
<
class
Worker
>
void
before_shutdown
(
Worker
*
)
{
// nop
}
template
<
class
Worker
>
void
before_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
template
<
class
Worker
>
void
after_resume
(
Worker
*
,
resumable
*
)
{
// nop
}
template
<
class
Worker
>
void
after_completion
(
Worker
*
,
resumable
*
)
{
// nop
}
template
<
class
Worker
,
class
UnaryFunction
>
template
<
class
Worker
,
class
UnaryFunction
>
void
foreach_resumable
(
Worker
*
self
,
UnaryFunction
f
)
{
void
foreach_resumable
(
Worker
*
self
,
UnaryFunction
f
)
{
auto
next
=
[
&
]
{
return
d
(
self
).
queue
.
take_head
();
};
auto
next
=
[
&
]
{
return
d
(
self
).
queue
.
take_head
();
};
...
...
libcaf_core/caf/scheduler/abstract_coordinator.hpp
View file @
c6160a26
...
@@ -82,8 +82,6 @@ public:
...
@@ -82,8 +82,6 @@ public:
void
*
subtype_ptr
()
override
;
void
*
subtype_ptr
()
override
;
protected:
protected:
abstract_coordinator
();
void
stop_actors
();
void
stop_actors
();
// ID of the worker receiving the next enqueue
// ID of the worker receiving the next enqueue
...
...
libcaf_core/caf/scheduler/coordinator.hpp
View file @
c6160a26
...
@@ -59,7 +59,7 @@ protected:
...
@@ -59,7 +59,7 @@ protected:
void
start
()
override
{
void
start
()
override
{
// initialize workers vector
// initialize workers vector
auto
num
=
num_workers
();
auto
num
=
num_workers
();
//workers_.resiz
e(num);
workers_
.
reserv
e
(
num
);
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
for
(
size_t
i
=
0
;
i
<
num
;
++
i
)
workers_
.
emplace_back
(
new
worker_type
(
i
,
this
,
max_throughput_
));
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
...
...
libcaf_core/caf/scheduler/worker.hpp
View file @
c6160a26
...
@@ -91,13 +91,6 @@ public:
...
@@ -91,13 +91,6 @@ public:
return
this_thread_
;
return
this_thread_
;
}
}
void
detach_all
()
{
CAF_LOG_TRACE
(
""
);
policy_
.
foreach_resumable
(
this
,
[](
resumable
*
job
)
{
intrusive_ptr_release
(
job
);
});
}
actor_id
id_of
(
resumable
*
ptr
)
{
actor_id
id_of
(
resumable
*
ptr
)
{
abstract_actor
*
dptr
=
ptr
?
dynamic_cast
<
abstract_actor
*>
(
ptr
)
abstract_actor
*
dptr
=
ptr
?
dynamic_cast
<
abstract_actor
*>
(
ptr
)
:
nullptr
;
:
nullptr
;
...
@@ -124,25 +117,23 @@ private:
...
@@ -124,25 +117,23 @@ private:
CAF_LOG_DEBUG
(
"resume actor:"
<<
CAF_ARG
(
id_of
(
job
)));
CAF_LOG_DEBUG
(
"resume actor:"
<<
CAF_ARG
(
id_of
(
job
)));
CAF_PUSH_AID_FROM_PTR
(
dynamic_cast
<
abstract_actor
*>
(
job
));
CAF_PUSH_AID_FROM_PTR
(
dynamic_cast
<
abstract_actor
*>
(
job
));
policy_
.
before_resume
(
this
,
job
);
policy_
.
before_resume
(
this
,
job
);
switch
(
job
->
resume
(
this
,
max_throughput_
))
{
auto
res
=
job
->
resume
(
this
,
max_throughput_
);
policy_
.
after_resume
(
this
,
job
);
switch
(
res
)
{
case
resumable
:
:
resume_later
:
{
case
resumable
:
:
resume_later
:
{
policy_
.
after_resume
(
this
,
job
);
policy_
.
resume_job_later
(
this
,
job
);
policy_
.
resume_job_later
(
this
,
job
);
break
;
break
;
}
}
case
resumable
:
:
done
:
{
case
resumable
:
:
done
:
{
policy_
.
after_resume
(
this
,
job
);
policy_
.
after_completion
(
this
,
job
);
policy_
.
after_completion
(
this
,
job
);
intrusive_ptr_release
(
job
);
intrusive_ptr_release
(
job
);
break
;
break
;
}
}
case
resumable
:
:
awaiting_message
:
{
case
resumable
:
:
awaiting_message
:
{
// resumable will be enqueued again later
// resumable will be enqueued again later
policy_
.
after_resume
(
this
,
job
);
break
;
break
;
}
}
case
resumable
:
:
shutdown_execution_unit
:
{
case
resumable
:
:
shutdown_execution_unit
:
{
policy_
.
after_resume
(
this
,
job
);
policy_
.
after_completion
(
this
,
job
);
policy_
.
after_completion
(
this
,
job
);
policy_
.
before_shutdown
(
this
);
policy_
.
before_shutdown
(
this
);
return
;
return
;
...
...
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