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
4f804d2a
Commit
4f804d2a
authored
Jul 30, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Move worker_hub to the core library
parent
af6de514
Changes
13
Hide whitespace changes
Inline
Side-by-side
Showing
13 changed files
with
294 additions
and
94 deletions
+294
-94
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+2
-0
libcaf_core/caf/detail/abstract_worker.hpp
libcaf_core/caf/detail/abstract_worker.hpp
+56
-0
libcaf_core/caf/detail/abstract_worker_hub.hpp
libcaf_core/caf/detail/abstract_worker_hub.hpp
+21
-26
libcaf_core/caf/detail/worker.hpp
libcaf_core/caf/detail/worker.hpp
+62
-0
libcaf_core/caf/detail/worker_hub.hpp
libcaf_core/caf/detail/worker_hub.hpp
+64
-0
libcaf_core/caf/fwd.hpp
libcaf_core/caf/fwd.hpp
+2
-0
libcaf_core/src/abstract_worker.cpp
libcaf_core/src/abstract_worker.cpp
+49
-0
libcaf_core/src/abstract_worker_hub.cpp
libcaf_core/src/abstract_worker_hub.cpp
+19
-20
libcaf_io/CMakeLists.txt
libcaf_io/CMakeLists.txt
+0
-1
libcaf_io/caf/io/basp/instance.hpp
libcaf_io/caf/io/basp/instance.hpp
+4
-3
libcaf_io/caf/io/basp/worker.hpp
libcaf_io/caf/io/basp/worker.hpp
+12
-23
libcaf_io/src/worker.cpp
libcaf_io/src/worker.cpp
+2
-19
libcaf_io/test/worker.cpp
libcaf_io/test/worker.cpp
+1
-2
No files found.
libcaf_core/CMakeLists.txt
View file @
4f804d2a
...
@@ -11,6 +11,8 @@ set(LIBCAF_CORE_SRCS
...
@@ -11,6 +11,8 @@ set(LIBCAF_CORE_SRCS
src/abstract_composable_behavior.cpp
src/abstract_composable_behavior.cpp
src/abstract_coordinator.cpp
src/abstract_coordinator.cpp
src/abstract_group.cpp
src/abstract_group.cpp
src/abstract_worker.cpp
src/abstract_worker_hub.cpp
src/actor.cpp
src/actor.cpp
src/actor_addr.cpp
src/actor_addr.cpp
src/actor_clock.cpp
src/actor_clock.cpp
...
...
libcaf_core/caf/detail/abstract_worker.hpp
0 → 100644
View file @
4f804d2a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
#include "caf/resumable.hpp"
namespace
caf
{
namespace
detail
{
class
abstract_worker
:
public
ref_counted
,
public
resumable
{
public:
// -- friends ----------------------------------------------------------------
friend
abstract_worker_hub
;
// -- constructors, destructors, and assignment operators --------------------
abstract_worker
();
~
abstract_worker
()
override
;
// -- implementation of resumable --------------------------------------------
subtype_t
subtype
()
const
override
;
void
intrusive_ptr_add_ref_impl
()
override
;
void
intrusive_ptr_release_impl
()
override
;
private:
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
std
::
atomic
<
abstract_worker
*>
next_
;
};
}
// namespace detail
}
// namespace caf
libcaf_
io/caf/io/basp/
worker_hub.hpp
→
libcaf_
core/caf/detail/abstract_
worker_hub.hpp
View file @
4f804d2a
...
@@ -23,54 +23,50 @@
...
@@ -23,54 +23,50 @@
#include <mutex>
#include <mutex>
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
namespace
caf
{
namespace
caf
{
namespace
io
{
namespace
detail
{
namespace
basp
{
/// A central place where
BASP
workers return to after finishing a task. A hub
/// A central place where workers return to after finishing a task. A hub
/// supports any number of workers that call `push`, but only a single master
/// supports any number of workers that call `push`, but only a single master
/// that calls `pop`. The hub takes ownership of all workers. Workers register
/// that calls `pop`. The hub takes ownership of all workers. Workers register
/// at the hub during construction and get destroyed when the hub gets
/// at the hub during construction and get destroyed when the hub gets
/// destroyed.
/// destroyed.
class
worker_hub
{
class
abstract_
worker_hub
{
public:
public:
// --
member types ---------------------------------------
--------------------
// --
constructors, destructors, and assignment operators
--------------------
using
pointer
=
worker
*
;
abstract_worker_hub
()
;
// -- constructors, destructors, and assignment operators --------------------
virtual
~
abstract_worker_hub
();
worker_hub
();
// -- synchronization --------------------------------------------------------
~
worker_hub
();
/// Waits until all workers are back at the hub.
void
await_workers
();
// -- properties -------------------------------------------------------------
protected:
// -- worker management ------------------------------------------------------
///
Creates a new worker and adds it
to the hub.
///
Adds a new worker
to the hub.
void
add_new_worker
(
message_queue
&
,
proxy_registry
&
);
void
push_new
(
abstract_worker
*
ptr
);
///
Add
a worker to the hub.
///
Returns
a worker to the hub.
void
push
(
pointer
ptr
);
void
push
_returning
(
abstract_worker
*
ptr
);
///
Get
a worker from the hub.
///
Tries to retrieve
a worker from the hub.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
/// hub is currently empty.
pointer
pop
();
abstract_worker
*
pop_impl
();
/// Check which worker would `pop` currently return.
/// Check
s
which worker would `pop` currently return.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
/// hub is currently empty.
pointer
peek
();
abstract_worker
*
peek_impl
();
/// Waits until all workers are back at the hub.
void
await_workers
();
private:
// -- member variables -------------------------------------------------------
// -- member variables -------------------------------------------------------
std
::
atomic
<
pointer
>
head_
;
std
::
atomic
<
abstract_worker
*
>
head_
;
std
::
atomic
<
size_t
>
running_
;
std
::
atomic
<
size_t
>
running_
;
...
@@ -79,6 +75,5 @@ private:
...
@@ -79,6 +75,5 @@ private:
std
::
condition_variable
cv_
;
std
::
condition_variable
cv_
;
};
};
}
// namespace basp
}
// namespace detail
}
// namespace io
}
// namespace caf
}
// namespace caf
libcaf_core/caf/detail/worker.hpp
0 → 100644
View file @
4f804d2a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
#include "caf/resumable.hpp"
namespace
caf
{
namespace
detail
{
class
worker
:
public
ref_counted
,
public
resumable
{
public:
// -- friends ----------------------------------------------------------------
friend
worker_hub
;
// -- constructors, destructors, and assignment operators --------------------
worker
();
~
worker
()
override
;
// -- implementation of resumable --------------------------------------------
subtype_t
subtype
()
const
override
;
void
intrusive_ptr_add_ref_impl
()
override
;
void
intrusive_ptr_release_impl
()
override
;
private:
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
std
::
atomic
<
worker
*>
next_
;
/// Points to our home hub.
worker_hub
*
hub_
;
/// Points to the parent system.
actor_system
*
system_
;
};
}
// namespace detail
}
// namespace caf
libcaf_core/caf/detail/worker_hub.hpp
0 → 100644
View file @
4f804d2a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include "caf/detail/abstract_worker_hub.hpp"
namespace
caf
{
namespace
detail
{
template
<
class
Worker
>
class
worker_hub
:
public
abstract_worker_hub
{
public:
// -- member types -----------------------------------------------------------
using
super
=
abstract_worker_hub
;
using
worker_type
=
Worker
;
// -- worker management ------------------------------------------------------
/// Creates a new worker and adds it to the hub.
template
<
class
...
Ts
>
void
add_new_worker
(
Ts
&&
...
xs
)
{
super
::
push_new
(
new
worker_type
(
*
this
,
std
::
forward
<
Ts
>
(
xs
)...));
}
/// Returns a worker to the hub.
void
push
(
worker_type
*
ptr
)
{
super
::
push_returning
(
ptr
);
}
/// Gets a worker from the hub.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
worker_type
*
pop
()
{
return
static_cast
<
worker_type
*>
(
super
::
pop_impl
());
}
/// Checks which worker would `pop` currently return.
/// @returns the next available worker (in LIFO order) or `nullptr` if the
/// hub is currently empty.
worker_type
*
peek
()
{
return
static_cast
<
worker_type
*>
(
super
::
peek_impl
());
}
};
}
// namespace detail
}
// namespace caf
libcaf_core/caf/fwd.hpp
View file @
4f804d2a
...
@@ -250,6 +250,8 @@ namespace detail {
...
@@ -250,6 +250,8 @@ namespace detail {
template
<
class
>
class
type_erased_value_impl
;
template
<
class
>
class
type_erased_value_impl
;
template
<
class
>
class
stream_distribution_tree
;
template
<
class
>
class
stream_distribution_tree
;
class
abstract_worker
;
class
abstract_worker_hub
;
class
disposer
;
class
disposer
;
class
dynamic_message_data
;
class
dynamic_message_data
;
class
group_manager
;
class
group_manager
;
...
...
libcaf_core/src/abstract_worker.cpp
0 → 100644
View file @
4f804d2a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/detail/abstract_worker.hpp"
namespace
caf
{
namespace
detail
{
// -- constructors, destructors, and assignment operators ----------------------
abstract_worker
::
abstract_worker
()
:
next_
(
nullptr
)
{
// nop
}
abstract_worker
::~
abstract_worker
()
{
// nop
}
// -- implementation of resumable ----------------------------------------------
resumable
::
subtype_t
abstract_worker
::
subtype
()
const
{
return
resumable
::
function_object
;
}
void
abstract_worker
::
intrusive_ptr_add_ref_impl
()
{
ref
();
}
void
abstract_worker
::
intrusive_ptr_release_impl
()
{
deref
();
}
}
// namespace detail
}
// namespace caf
libcaf_
io/src/
worker_hub.cpp
→
libcaf_
core/src/abstract_
worker_hub.cpp
View file @
4f804d2a
...
@@ -16,21 +16,20 @@
...
@@ -16,21 +16,20 @@
* http://www.boost.org/LICENSE_1_0.txt. *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
******************************************************************************/
#include "caf/
io/basp/
worker_hub.hpp"
#include "caf/
detail/abstract_
worker_hub.hpp"
#include "caf/
io/basp/
worker.hpp"
#include "caf/
detail/abstract_
worker.hpp"
namespace
caf
{
namespace
caf
{
namespace
io
{
namespace
detail
{
namespace
basp
{
// -- constructors, destructors, and assignment operators ----------------------
// -- constructors, destructors, and assignment operators ----------------------
worker_hub
::
worker_hub
()
:
head_
(
nullptr
),
running_
(
0
)
{
abstract_worker_hub
::
abstract_
worker_hub
()
:
head_
(
nullptr
),
running_
(
0
)
{
// nop
// nop
}
}
worker_hub
::~
worker_hub
()
{
abstract_worker_hub
::~
abstract_
worker_hub
()
{
await_workers
();
await_workers
();
auto
head
=
head_
.
load
();
auto
head
=
head_
.
load
();
while
(
head
!=
nullptr
)
{
while
(
head
!=
nullptr
)
{
...
@@ -40,10 +39,17 @@ worker_hub::~worker_hub() {
...
@@ -40,10 +39,17 @@ worker_hub::~worker_hub() {
}
}
}
}
// --
properties -----
----------------------------------------------------------
// --
synchronization
----------------------------------------------------------
void
worker_hub
::
add_new_worker
(
message_queue
&
queue
,
proxy_registry
&
proxies
)
{
void
abstract_worker_hub
::
await_workers
()
{
auto
ptr
=
new
worker
(
*
this
,
queue
,
proxies
);
std
::
unique_lock
<
std
::
mutex
>
guard
{
mtx_
};
while
(
running_
!=
0
)
cv_
.
wait
(
guard
);
}
// -- worker management --------------------------------------------------------
void
abstract_worker_hub
::
push_new
(
abstract_worker
*
ptr
)
{
auto
next
=
head_
.
load
();
auto
next
=
head_
.
load
();
for
(;;)
{
for
(;;)
{
ptr
->
next_
=
next
;
ptr
->
next_
=
next
;
...
@@ -52,7 +58,7 @@ void worker_hub::add_new_worker(message_queue& queue, proxy_registry& proxies) {
...
@@ -52,7 +58,7 @@ void worker_hub::add_new_worker(message_queue& queue, proxy_registry& proxies) {
}
}
}
}
void
worker_hub
::
push
(
pointer
ptr
)
{
void
abstract_worker_hub
::
push_returning
(
abstract_worker
*
ptr
)
{
auto
next
=
head_
.
load
();
auto
next
=
head_
.
load
();
for
(;;)
{
for
(;;)
{
ptr
->
next_
=
next
;
ptr
->
next_
=
next
;
...
@@ -66,7 +72,7 @@ void worker_hub::push(pointer ptr) {
...
@@ -66,7 +72,7 @@ void worker_hub::push(pointer ptr) {
}
}
}
}
worker_hub
::
pointer
worker_hub
::
pop
()
{
abstract_worker
*
abstract_worker_hub
::
pop_impl
()
{
auto
result
=
head_
.
load
();
auto
result
=
head_
.
load
();
if
(
result
==
nullptr
)
if
(
result
==
nullptr
)
return
nullptr
;
return
nullptr
;
...
@@ -80,16 +86,9 @@ worker_hub::pointer worker_hub::pop() {
...
@@ -80,16 +86,9 @@ worker_hub::pointer worker_hub::pop() {
}
}
}
}
worker_hub
::
pointer
worker_hub
::
peek
()
{
abstract_worker
*
abstract_worker_hub
::
peek_impl
()
{
return
head_
.
load
();
return
head_
.
load
();
}
}
void
worker_hub
::
await_workers
()
{
}
// namespace detail
std
::
unique_lock
<
std
::
mutex
>
guard
{
mtx_
};
while
(
running_
!=
0
)
cv_
.
wait
(
guard
);
}
}
// namespace basp
}
// namespace io
}
// namespace caf
}
// namespace caf
libcaf_io/CMakeLists.txt
View file @
4f804d2a
...
@@ -47,7 +47,6 @@ set(LIBCAF_IO_SRCS
...
@@ -47,7 +47,6 @@ set(LIBCAF_IO_SRCS
src/test_multiplexer.cpp
src/test_multiplexer.cpp
src/udp.cpp
src/udp.cpp
src/worker.cpp
src/worker.cpp
src/worker_hub.cpp
)
)
add_custom_target
(
libcaf_io
)
add_custom_target
(
libcaf_io
)
...
...
libcaf_io/caf/io/basp/instance.hpp
View file @
4f804d2a
...
@@ -23,6 +23,7 @@
...
@@ -23,6 +23,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/callback.hpp"
#include "caf/callback.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/connection_state.hpp"
...
@@ -30,7 +31,7 @@
...
@@ -30,7 +31,7 @@
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/worker
_hub
.hpp"
#include "caf/io/basp/worker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/middleman.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
...
@@ -208,7 +209,7 @@ public:
...
@@ -208,7 +209,7 @@ public:
return
this_node_
;
return
this_node_
;
}
}
worker_hub
&
hub
()
{
detail
::
worker_hub
<
worker
>
&
hub
()
{
return
hub_
;
return
hub_
;
}
}
...
@@ -236,7 +237,7 @@ private:
...
@@ -236,7 +237,7 @@ private:
node_id
this_node_
;
node_id
this_node_
;
callee
&
callee_
;
callee
&
callee_
;
message_queue
queue_
;
message_queue
queue_
;
worker_hub
hub_
;
detail
::
worker_hub
<
worker
>
hub_
;
};
};
/// @}
/// @}
...
...
libcaf_io/caf/io/basp/worker.hpp
View file @
4f804d2a
...
@@ -23,6 +23,8 @@
...
@@ -23,6 +23,8 @@
#include <vector>
#include <vector>
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/header.hpp"
...
@@ -35,26 +37,28 @@ namespace io {
...
@@ -35,26 +37,28 @@ namespace io {
namespace
basp
{
namespace
basp
{
/// Deserializes payloads for BASP messages asynchronously.
/// Deserializes payloads for BASP messages asynchronously.
class
worker
:
public
resumable
,
class
worker
:
public
detail
::
abstract_worker
,
public
remote_message_handler
<
worker
>
,
public
remote_message_handler
<
worker
>
{
public
ref_counted
{
public:
public:
// -- friends ----------------------------------------------------------------
// -- friends ----------------------------------------------------------------
friend
worker_hub
;
friend
remote_message_handler
<
worker
>
;
friend
remote_message_handler
<
worker
>
;
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
using
atomic_pointer
=
std
::
atomic
<
worker
*>
;
using
super
=
detail
::
abstract_worker
;
using
scheduler_type
=
scheduler
::
abstract_coordinator
;
using
scheduler_type
=
scheduler
::
abstract_coordinator
;
using
buffer_type
=
std
::
vector
<
char
>
;
using
buffer_type
=
std
::
vector
<
char
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker
(
hub_type
&
hub
,
message_queue
&
queue
,
proxy_registry
&
proxies
);
~
worker
()
override
;
~
worker
()
override
;
// -- management -------------------------------------------------------------
// -- management -------------------------------------------------------------
...
@@ -64,25 +68,13 @@ public:
...
@@ -64,25 +68,13 @@ public:
// -- implementation of resumable --------------------------------------------
// -- implementation of resumable --------------------------------------------
subtype_t
subtype
()
const
override
;
resume_result
resume
(
execution_unit
*
ctx
,
size_t
)
override
;
resume_result
resume
(
execution_unit
*
ctx
,
size_t
)
override
;
void
intrusive_ptr_add_ref_impl
()
override
;
void
intrusive_ptr_release_impl
()
override
;
private:
private:
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker
(
worker_hub
&
hub
,
message_queue
&
queue
,
proxy_registry
&
proxies
);
// -- constants and assertions -----------------------------------------------
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
/// Stores how many bytes the "first half" of this object requires.
static
constexpr
size_t
pointer_members_size
=
sizeof
(
atomic_pointer
)
static
constexpr
size_t
pointer_members_size
=
sizeof
(
hub_type
*
)
+
sizeof
(
worker_hub
*
)
+
sizeof
(
message_queue
*
)
+
sizeof
(
message_queue
*
)
+
sizeof
(
proxy_registry
*
)
+
sizeof
(
proxy_registry
*
)
+
sizeof
(
actor_system
*
);
+
sizeof
(
actor_system
*
);
...
@@ -92,11 +84,8 @@ private:
...
@@ -92,11 +84,8 @@ private:
// -- member variables -------------------------------------------------------
// -- member variables -------------------------------------------------------
/// Points to the next worker in the hub.
atomic_pointer
next_
;
/// Points to our home hub.
/// Points to our home hub.
worker_hub
*
hub_
;
hub_type
*
hub_
;
/// Points to the queue for establishing strict ordering.
/// Points to the queue for establishing strict ordering.
message_queue
*
queue_
;
message_queue
*
queue_
;
...
...
libcaf_io/src/worker.cpp
View file @
4f804d2a
...
@@ -20,7 +20,6 @@
...
@@ -20,7 +20,6 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
...
@@ -30,12 +29,8 @@ namespace basp {
...
@@ -30,12 +29,8 @@ namespace basp {
// -- constructors, destructors, and assignment operators ----------------------
// -- constructors, destructors, and assignment operators ----------------------
worker
::
worker
(
worker_hub
&
hub
,
message_queue
&
queue
,
proxy_registry
&
proxies
)
worker
::
worker
(
hub_type
&
hub
,
message_queue
&
queue
,
proxy_registry
&
proxies
)
:
next_
(
nullptr
),
:
hub_
(
&
hub
),
queue_
(
&
queue
),
proxies_
(
&
proxies
),
system_
(
&
proxies
.
system
())
{
hub_
(
&
hub
),
queue_
(
&
queue
),
proxies_
(
&
proxies
),
system_
(
&
proxies
.
system
())
{
CAF_IGNORE_UNUSED
(
pad_
);
CAF_IGNORE_UNUSED
(
pad_
);
}
}
...
@@ -60,10 +55,6 @@ void worker::launch(const node_id& last_hop, const basp::header& hdr,
...
@@ -60,10 +55,6 @@ void worker::launch(const node_id& last_hop, const basp::header& hdr,
// -- implementation of resumable ----------------------------------------------
// -- implementation of resumable ----------------------------------------------
resumable
::
subtype_t
worker
::
subtype
()
const
{
return
resumable
::
function_object
;
}
resumable
::
resume_result
worker
::
resume
(
execution_unit
*
ctx
,
size_t
)
{
resumable
::
resume_result
worker
::
resume
(
execution_unit
*
ctx
,
size_t
)
{
ctx
->
proxy_registry_ptr
(
proxies_
);
ctx
->
proxy_registry_ptr
(
proxies_
);
handle_remote_message
(
ctx
);
handle_remote_message
(
ctx
);
...
@@ -71,14 +62,6 @@ resumable::resume_result worker::resume(execution_unit* ctx, size_t) {
...
@@ -71,14 +62,6 @@ resumable::resume_result worker::resume(execution_unit* ctx, size_t) {
return
resumable
::
awaiting_message
;
return
resumable
::
awaiting_message
;
}
}
void
worker
::
intrusive_ptr_add_ref_impl
()
{
ref
();
}
void
worker
::
intrusive_ptr_release_impl
()
{
deref
();
}
}
// namespace basp
}
// namespace basp
}
// namespace io
}
// namespace io
}
// namespace caf
}
// namespace caf
libcaf_io/test/worker.cpp
View file @
4f804d2a
...
@@ -27,7 +27,6 @@
...
@@ -27,7 +27,6 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/proxy_registry.hpp"
...
@@ -76,7 +75,7 @@ private:
...
@@ -76,7 +75,7 @@ private:
};
};
struct
fixture
:
test_coordinator_fixture
<>
{
struct
fixture
:
test_coordinator_fixture
<>
{
io
::
basp
::
worker_hub
hub
;
detail
::
worker_hub
<
io
::
basp
::
worker
>
hub
;
io
::
basp
::
message_queue
queue
;
io
::
basp
::
message_queue
queue
;
mock_proxy_registry_backend
proxies_backend
;
mock_proxy_registry_backend
proxies_backend
;
proxy_registry
proxies
;
proxy_registry
proxies
;
...
...
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