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
473f0df8
Unverified
Commit
473f0df8
authored
Jan 13, 2020
by
Dominik Charousset
Committed by
GitHub
Jan 13, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #51
Implement proper shutdown of multiplexer
parents
1e800f77
1e2f8d43
Changes
8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
84 additions
and
19 deletions
+84
-19
libcaf_net/caf/net/backend/test.hpp
libcaf_net/caf/net/backend/test.hpp
+3
-1
libcaf_net/caf/net/middleman_backend.hpp
libcaf_net/caf/net/middleman_backend.hpp
+4
-1
libcaf_net/caf/net/multiplexer.hpp
libcaf_net/caf/net/multiplexer.hpp
+10
-0
libcaf_net/caf/net/operation.hpp
libcaf_net/caf/net/operation.hpp
+1
-0
libcaf_net/src/multiplexer.cpp
libcaf_net/src/multiplexer.cpp
+40
-10
libcaf_net/src/net/backend/test.cpp
libcaf_net/src/net/backend/test.cpp
+6
-0
libcaf_net/src/net/middleman.cpp
libcaf_net/src/net/middleman.cpp
+6
-2
libcaf_net/src/pollset_updater.cpp
libcaf_net/src/pollset_updater.cpp
+14
-5
No files found.
libcaf_net/caf/net/backend/test.hpp
View file @
473f0df8
/******************************************************************************
_opt
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
...
...
@@ -47,6 +47,8 @@ public:
error
init
()
override
;
void
stop
()
override
;
endpoint_manager_ptr
peer
(
const
node_id
&
id
)
override
;
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
override
;
...
...
libcaf_net/caf/net/middleman_backend.hpp
View file @
473f0df8
...
...
@@ -20,6 +20,7 @@
#include <string>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp"
...
...
@@ -29,7 +30,7 @@ namespace caf::net {
/// Technology-specific backend for connecting to and managing peer
/// connections.
/// @relates middleman
class
middleman_backend
:
public
proxy_registry
::
backend
{
class
CAF_NET_EXPORT
middleman_backend
:
public
proxy_registry
::
backend
{
public:
// -- constructors, destructors, and assignment operators --------------------
...
...
@@ -48,6 +49,8 @@ public:
/// Resolves a path to a remote actor.
virtual
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
=
0
;
virtual
void
stop
()
=
0
;
// -- properties -------------------------------------------------------------
const
std
::
string
&
id
()
const
noexcept
{
...
...
libcaf_net/caf/net/multiplexer.hpp
View file @
473f0df8
...
...
@@ -88,6 +88,10 @@ public:
/// Polls until no socket event handler remains.
void
run
();
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
void
shutdown
();
protected:
// -- utility functions ------------------------------------------------------
...
...
@@ -97,6 +101,9 @@ protected:
/// Adds a new socket manager to the pollset.
void
add
(
socket_manager_ptr
mgr
);
/// Deletes a known socket manager from the pollset.
void
del
(
ptrdiff_t
index
);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
void
write_to_pipe
(
uint8_t
opcode
,
const
socket_manager_ptr
&
mgr
);
...
...
@@ -119,6 +126,9 @@ protected:
/// Guards `write_handle_`.
std
::
mutex
write_lock_
;
/// Signals shutdown has been requested.
bool
shutting_down_
;
};
/// @relates multiplexer
...
...
libcaf_net/caf/net/operation.hpp
View file @
473f0df8
...
...
@@ -28,6 +28,7 @@ enum class operation {
read
=
0x01
,
write
=
0x02
,
read_write
=
0x03
,
shutdown
=
0x04
,
};
constexpr
operation
operator
|
(
operation
x
,
operation
y
)
{
...
...
libcaf_net/src/multiplexer.cpp
View file @
473f0df8
...
...
@@ -29,7 +29,6 @@
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#ifndef CAF_WINDOWS
...
...
@@ -77,7 +76,7 @@ short to_bitmask(operation op) {
}
// namespace
multiplexer
::
multiplexer
()
{
multiplexer
::
multiplexer
()
:
shutting_down_
(
false
)
{
// nop
}
...
...
@@ -107,7 +106,9 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
void
multiplexer
::
register_reading
(
const
socket_manager_ptr
&
mgr
)
{
if
(
std
::
this_thread
::
get_id
()
==
tid_
)
{
if
(
mgr
->
mask
()
!=
operation
::
none
)
{
if
(
shutting_down_
)
{
// discard
}
else
if
(
mgr
->
mask
()
!=
operation
::
none
)
{
CAF_ASSERT
(
index_of
(
mgr
)
!=
-
1
);
if
(
mgr
->
mask_add
(
operation
::
read
))
{
auto
&
fd
=
pollset_
[
index_of
(
mgr
)];
...
...
@@ -123,7 +124,9 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
void
multiplexer
::
register_writing
(
const
socket_manager_ptr
&
mgr
)
{
if
(
std
::
this_thread
::
get_id
()
==
tid_
)
{
if
(
mgr
->
mask
()
!=
operation
::
none
)
{
if
(
shutting_down_
)
{
// discard
}
else
if
(
mgr
->
mask
()
!=
operation
::
none
)
{
CAF_ASSERT
(
index_of
(
mgr
)
!=
-
1
);
if
(
mgr
->
mask_add
(
operation
::
write
))
{
auto
&
fd
=
pollset_
[
index_of
(
mgr
)];
...
...
@@ -199,8 +202,7 @@ bool multiplexer::poll_once(bool blocking) {
auto
new_events
=
handle
(
mgr
,
events
,
revents
);
--
presult
;
if
(
new_events
==
0
)
{
pollset_
.
erase
(
pollset_
.
begin
()
+
i
);
managers_
.
erase
(
managers_
.
begin
()
+
i
);
del
(
i
);
continue
;
}
else
if
(
new_events
!=
events
)
{
pollset_
[
i
].
events
=
new_events
;
...
...
@@ -222,6 +224,27 @@ void multiplexer::run() {
poll_once
(
true
);
}
void
multiplexer
::
shutdown
()
{
if
(
std
::
this_thread
::
get_id
()
==
tid_
)
{
shutting_down_
=
true
;
// First manager is the pollset_updater. Skip it and delete later.
for
(
size_t
i
=
1
;
i
<
managers_
.
size
();)
{
auto
&
mgr
=
managers_
[
i
];
if
(
mgr
->
mask_del
(
operation
::
read
))
{
auto
&
fd
=
pollset_
[
index_of
(
mgr
)];
fd
.
events
&=
~
input_mask
;
}
if
(
mgr
->
mask
()
==
operation
::
none
)
del
(
i
);
else
++
i
;
}
close_pipe
();
}
else
{
write_to_pipe
(
5
,
nullptr
);
}
}
short
multiplexer
::
handle
(
const
socket_manager_ptr
&
mgr
,
short
events
,
short
revents
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
mgr
->
handle
()));
...
...
@@ -262,11 +285,18 @@ void multiplexer::add(socket_manager_ptr mgr) {
managers_
.
emplace_back
(
std
::
move
(
mgr
));
}
void
multiplexer
::
del
(
ptrdiff_t
index
)
{
CAF_ASSERT
(
index
!=
-
1
);
pollset_
.
erase
(
pollset_
.
begin
()
+
index
);
managers_
.
erase
(
managers_
.
begin
()
+
index
);
}
void
multiplexer
::
write_to_pipe
(
uint8_t
opcode
,
const
socket_manager_ptr
&
mgr
)
{
CAF_ASSERT
(
opcode
==
0
||
opcode
==
1
);
CAF_ASSERT
(
mgr
!=
nullptr
);
CAF_ASSERT
(
opcode
==
0
||
opcode
==
1
||
opcode
==
4
||
opcode
==
5
);
CAF_ASSERT
(
mgr
!=
nullptr
||
opcode
==
4
);
pollset_updater
::
msg_buf
buf
;
mgr
->
ref
();
if
(
opcode
!=
5
)
mgr
->
ref
();
buf
[
0
]
=
static_cast
<
byte
>
(
opcode
);
auto
value
=
reinterpret_cast
<
intptr_t
>
(
mgr
.
get
());
memcpy
(
buf
.
data
()
+
1
,
&
value
,
sizeof
(
intptr_t
));
...
...
@@ -278,7 +308,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
else
res
=
sec
::
socket_invalid
;
}
if
(
holds_alternative
<
sec
>
(
res
))
if
(
holds_alternative
<
sec
>
(
res
)
&&
opcode
!=
4
)
mgr
->
deref
();
}
...
...
libcaf_net/src/net/backend/test.cpp
View file @
473f0df8
...
...
@@ -44,6 +44,12 @@ error test::init() {
return
none
;
}
void
test
::
stop
()
{
for
(
const
auto
&
p
:
peers_
)
proxies_
.
erase
(
p
.
first
);
peers_
.
clear
();
}
endpoint_manager_ptr
test
::
peer
(
const
node_id
&
id
)
{
return
get_peer
(
id
).
second
;
}
...
...
libcaf_net/src/net/middleman.cpp
View file @
473f0df8
...
...
@@ -53,15 +53,19 @@ void middleman::start() {
}
void
middleman
::
stop
()
{
mpx_
->
close_pipe
();
for
(
const
auto
&
backend
:
backends_
)
backend
->
stop
();
mpx_
->
shutdown
();
if
(
mpx_thread_
.
joinable
())
mpx_thread_
.
join
();
else
mpx_
->
run
();
}
void
middleman
::
init
(
actor_system_config
&
cfg
)
{
if
(
auto
err
=
mpx_
->
init
())
{
CAF_LOG_ERROR
(
"mgr->init() failed: "
<<
system
().
render
(
err
));
CAF_RAISE_ERROR
(
"mpx->init failed"
);
CAF_RAISE_ERROR
(
"mpx->init
()
failed"
);
}
if
(
auto
node_uri
=
get_if
<
uri
>
(
&
cfg
,
"middleman.this-node"
))
{
auto
this_node
=
make_node_id
(
std
::
move
(
*
node_uri
));
...
...
libcaf_net/src/pollset_updater.cpp
View file @
473f0df8
...
...
@@ -20,6 +20,7 @@
#include <cstring>
#include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
...
...
@@ -51,11 +52,19 @@ bool pollset_updater::handle_read_event() {
memcpy
(
&
value
,
buf_
.
data
()
+
1
,
sizeof
(
intptr_t
));
socket_manager_ptr
mgr
{
reinterpret_cast
<
socket_manager
*>
(
value
),
false
};
if
(
auto
ptr
=
parent_
.
lock
())
{
if
(
opcode
==
0
)
{
ptr
->
register_reading
(
mgr
);
}
else
{
CAF_ASSERT
(
opcode
==
1
);
ptr
->
register_writing
(
mgr
);
switch
(
opcode
)
{
case
0
:
ptr
->
register_reading
(
mgr
);
break
;
case
1
:
ptr
->
register_writing
(
mgr
);
break
;
case
4
:
ptr
->
shutdown
();
break
;
default:
CAF_LOG_DEBUG
(
"opcode not recognized: "
<<
CAF_ARG
(
opcode
));
break
;
}
}
}
...
...
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