Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
actor-incubator
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
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-incubator
Commits
58cfcbd2
Commit
58cfcbd2
authored
Oct 17, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Port basp::worker to caf::net
parent
3320d23e
Changes
5
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
391 additions
and
0 deletions
+391
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/net/basp/remote_message_handler.hpp
libcaf_net/caf/net/basp/remote_message_handler.hpp
+85
-0
libcaf_net/caf/net/basp/worker.hpp
libcaf_net/caf/net/basp/worker.hpp
+118
-0
libcaf_net/src/worker.cpp
libcaf_net/src/worker.cpp
+64
-0
libcaf_net/test/net/basp/worker.cpp
libcaf_net/test/net/basp/worker.cpp
+123
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
58cfcbd2
...
@@ -34,6 +34,7 @@ set(LIBCAF_NET_SRCS
...
@@ -34,6 +34,7 @@ set(LIBCAF_NET_SRCS
src/udp_datagram_socket.cpp
src/udp_datagram_socket.cpp
src/defaults.cpp
src/defaults.cpp
src/message_queue.cpp
src/message_queue.cpp
src/worker.cpp
)
)
add_custom_target
(
libcaf_net
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/net/basp/remote_message_handler.hpp
0 → 100644
View file @
58cfcbd2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <vector>
#include "caf/actor_control_block.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/config.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/node_id.hpp"
namespace
caf
{
namespace
net
{
namespace
basp
{
template
<
class
Subtype
>
class
remote_message_handler
{
public:
void
handle_remote_message
(
execution_unit
*
ctx
)
{
// Local variables.
auto
&
dref
=
static_cast
<
Subtype
&>
(
*
this
);
auto
&
payload
=
dref
.
payload_
;
auto
&
hdr
=
dref
.
hdr_
;
auto
&
registry
=
dref
.
system_
->
registry
();
auto
&
proxies
=
*
dref
.
proxies_
;
CAF_LOG_TRACE
(
CAF_ARG
(
hdr
)
<<
CAF_ARG2
(
"payload.size"
,
payload
.
size
()));
// Deserialize payload.
actor_id
src_id
=
0
;
node_id
src_node
;
actor_id
dst_id
=
0
;
std
::
vector
<
strong_actor_ptr
>
fwd_stack
;
message
content
;
binary_deserializer
source
{
ctx
,
payload
};
if
(
auto
err
=
source
(
src_node
,
src_id
,
dst_id
,
fwd_stack
,
content
))
{
CAF_LOG_ERROR
(
"could not deserialize payload: "
<<
CAF_ARG
(
err
));
return
;
}
// Sanity checks.
if
(
dst_id
==
0
)
return
;
// Try to fetch the receiver.
auto
dst_hdl
=
registry
.
get
(
dst_id
);
if
(
dst_hdl
==
nullptr
)
{
CAF_LOG_DEBUG
(
"no actor found for given ID, drop message"
);
return
;
}
// Try to fetch the sender.
strong_actor_ptr
src_hdl
;
if
(
src_node
!=
none
&&
src_id
!=
0
)
src_hdl
=
proxies
.
get_or_put
(
src_node
,
src_id
);
// Ship the message.
auto
ptr
=
make_mailbox_element
(
std
::
move
(
src_hdl
),
make_message_id
(
hdr
.
operation_data
),
std
::
move
(
fwd_stack
),
std
::
move
(
content
));
dst_hdl
->
get
()
->
enqueue
(
std
::
move
(
ptr
),
nullptr
);
}
};
}
// namespace basp
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/basp/worker.hpp
0 → 100644
View file @
58cfcbd2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <atomic>
#include <cstdint>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
#include "caf/net/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/resumable.hpp"
namespace
caf
{
namespace
net
{
namespace
basp
{
/// Deserializes payloads for BASP messages asynchronously.
class
worker
:
public
detail
::
abstract_worker
,
public
remote_message_handler
<
worker
>
{
public:
// -- friends ----------------------------------------------------------------
friend
remote_message_handler
<
worker
>
;
// -- member types -----------------------------------------------------------
using
super
=
detail
::
abstract_worker
;
using
scheduler_type
=
scheduler
::
abstract_coordinator
;
using
buffer_type
=
std
::
vector
<
char
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
// -- 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
;
// -- management -------------------------------------------------------------
void
launch
(
const
node_id
&
last_hop
,
const
basp
::
header
&
hdr
,
const
buffer_type
&
payload
);
// -- implementation of resumable --------------------------------------------
resume_result
resume
(
execution_unit
*
ctx
,
size_t
)
override
;
private:
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
static
constexpr
size_t
pointer_members_size
=
sizeof
(
hub_type
*
)
+
sizeof
(
message_queue
*
)
+
sizeof
(
proxy_registry
*
)
+
sizeof
(
actor_system
*
);
static_assert
(
CAF_CACHE_LINE_SIZE
>
pointer_members_size
,
"invalid cache line size"
);
// -- member variables -------------------------------------------------------
/// Points to our home hub.
hub_type
*
hub_
;
/// Points to the queue for establishing strict ordering.
message_queue
*
queue_
;
/// Points to our proxy registry / factory.
proxy_registry
*
proxies_
;
/// Points to the parent system.
actor_system
*
system_
;
/// Prevents false sharing when writing to `next`.
char
pad_
[
CAF_CACHE_LINE_SIZE
-
pointer_members_size
];
/// ID for local ordering.
uint64_t
msg_id_
;
/// Identifies the node that sent us `hdr_` and `payload_`.
node_id
last_hop_
;
/// The header for the next message. Either a direct_message or a
/// routed_message.
header
hdr_
;
/// Contains whatever this worker deserializes next.
buffer_type
payload_
;
};
}
// namespace basp
}
// namespace net
}
// namespace caf
libcaf_net/src/worker.cpp
0 → 100644
View file @
58cfcbd2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/net/basp/worker.hpp"
#include "caf/actor_system.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace
caf
{
namespace
net
{
namespace
basp
{
// -- constructors, destructors, and assignment operators ----------------------
worker
::
worker
(
hub_type
&
hub
,
message_queue
&
queue
,
proxy_registry
&
proxies
)
:
hub_
(
&
hub
),
queue_
(
&
queue
),
proxies_
(
&
proxies
),
system_
(
&
proxies
.
system
())
{
CAF_IGNORE_UNUSED
(
pad_
);
}
worker
::~
worker
()
{
// nop
}
// -- management ---------------------------------------------------------------
void
worker
::
launch
(
const
node_id
&
last_hop
,
const
basp
::
header
&
hdr
,
const
buffer_type
&
payload
)
{
msg_id_
=
queue_
->
new_id
();
last_hop_
=
last_hop
;
memcpy
(
&
hdr_
,
&
hdr
,
sizeof
(
basp
::
header
));
payload_
.
assign
(
payload
.
begin
(),
payload
.
end
());
ref
();
system_
->
scheduler
().
enqueue
(
this
);
}
// -- implementation of resumable ----------------------------------------------
resumable
::
resume_result
worker
::
resume
(
execution_unit
*
ctx
,
size_t
)
{
ctx
->
proxy_registry_ptr
(
proxies_
);
handle_remote_message
(
ctx
);
hub_
->
push
(
this
);
return
resumable
::
awaiting_message
;
}
}
// namespace basp
}
// namespace io
}
// namespace caf
libcaf_net/test/net/basp/worker.cpp
0 → 100644
View file @
58cfcbd2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE basp.worker
#include "caf/net/basp/worker.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp"
using
namespace
caf
;
namespace
{
behavior
testee_impl
()
{
return
{[](
ok_atom
)
{
// nop
}};
}
class
mock_actor_proxy
:
public
actor_proxy
{
public:
explicit
mock_actor_proxy
(
actor_config
&
cfg
)
:
actor_proxy
(
cfg
)
{
// nop
}
void
enqueue
(
mailbox_element_ptr
,
execution_unit
*
)
override
{
CAF_FAIL
(
"mock_actor_proxy::enqueue called"
);
}
void
kill_proxy
(
execution_unit
*
,
error
)
override
{
// nop
}
};
class
mock_proxy_registry_backend
:
public
proxy_registry
::
backend
{
public:
mock_proxy_registry_backend
(
actor_system
&
sys
)
:
sys_
(
sys
)
{
// nop
}
strong_actor_ptr
make_proxy
(
node_id
nid
,
actor_id
aid
)
override
{
actor_config
cfg
;
return
make_actor
<
mock_actor_proxy
,
strong_actor_ptr
>
(
aid
,
nid
,
&
sys_
,
cfg
);
}
void
set_last_hop
(
node_id
*
)
override
{
// nop
}
private:
actor_system
&
sys_
;
};
struct
fixture
:
test_coordinator_fixture
<>
{
detail
::
worker_hub
<
net
::
basp
::
worker
>
hub
;
net
::
basp
::
message_queue
queue
;
mock_proxy_registry_backend
proxies_backend
;
proxy_registry
proxies
;
node_id
last_hop
;
actor
testee
;
fixture
()
:
proxies_backend
(
sys
),
proxies
(
sys
,
proxies_backend
)
{
auto
tmp
=
make_node_id
(
123
,
"0011223344556677889900112233445566778899"
);
last_hop
=
unbox
(
std
::
move
(
tmp
));
testee
=
sys
.
spawn
<
lazy_init
>
(
testee_impl
);
sys
.
registry
().
put
(
testee
.
id
(),
testee
);
}
~
fixture
()
{
sys
.
registry
().
erase
(
testee
.
id
());
}
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
worker_tests
,
fixture
)
CAF_TEST
(
deliver
serialized
message
)
{
CAF_MESSAGE
(
"create the BASP worker"
);
CAF_REQUIRE_EQUAL
(
hub
.
peek
(),
nullptr
);
hub
.
add_new_worker
(
queue
,
proxies
);
CAF_REQUIRE_NOT_EQUAL
(
hub
.
peek
(),
nullptr
);
auto
w
=
hub
.
pop
();
CAF_MESSAGE
(
"create a fake message + BASP header"
);
std
::
vector
<
char
>
payload
;
std
::
vector
<
strong_actor_ptr
>
stages
;
binary_serializer
sink
{
sys
,
payload
};
if
(
auto
err
=
sink
(
node_id
{},
self
->
id
(),
testee
.
id
(),
stages
,
make_message
(
ok_atom
::
value
)))
CAF_FAIL
(
"unable to serialize message: "
<<
sys
.
render
(
err
));
net
::
basp
::
header
hdr
{
net
::
basp
::
message_type
::
actor_message
,
static_cast
<
uint32_t
>
(
payload
.
size
()),
make_message_id
().
integer_value
()};
CAF_MESSAGE
(
"launch worker"
);
w
->
launch
(
last_hop
,
hdr
,
payload
);
sched
.
run_once
();
expect
((
ok_atom
),
from
(
_
).
to
(
testee
));
}
CAF_TEST_FIXTURE_SCOPE_END
()
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