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
b295946e
Commit
b295946e
authored
Jul 19, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add missing functions, docs, and tests
parent
6b9437d1
Changes
3
Hide whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
71 additions
and
34 deletions
+71
-34
libcaf_net/caf/net/multiplexer.hpp
libcaf_net/caf/net/multiplexer.hpp
+17
-8
libcaf_net/src/multiplexer.cpp
libcaf_net/src/multiplexer.cpp
+17
-1
libcaf_net/test/multiplexer.cpp
libcaf_net/test/multiplexer.cpp
+37
-25
No files found.
libcaf_net/caf/net/multiplexer.hpp
View file @
b295946e
...
...
@@ -64,9 +64,15 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`.
/// @thread-safe
void
update
(
const
socket_manager_ptr
&
mgr
);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
/// @thread-safe
void
close_pipe
();
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
...
...
@@ -76,19 +82,22 @@ public:
/// Polls until no socket event handler remains.
void
run
();
/// Processes all updates on the socket managers.
void
handle_updates
();
void
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
);
protected:
// -- convenience functions --------------------------------------------------
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
);
/// Adds a new socket manager to the pollset.
void
add
(
socket_manager_ptr
mgr
);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
std
::
vector
<
pollfd
>
pollset_
;
pollfd_list
pollset_
;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
...
...
@@ -97,14 +106,14 @@ protected:
/// Managers that updated their event mask and need updating.
manager_list
dirty_managers_
;
/// Stores the ID of the thread this multiplexer is running in.
Must be set
///
by the subclass
.
/// Stores the ID of the thread this multiplexer is running in.
Set when
///
calling `init()`
.
std
::
thread
::
id
tid_
;
///
///
Used for pushing updates to the multiplexer's thread.
pipe_socket
write_handle_
;
///
///
Guards `write_handle_`.
std
::
mutex
write_lock_
;
};
...
...
libcaf_net/src/multiplexer.cpp
View file @
b295946e
...
...
@@ -115,13 +115,24 @@ void multiplexer::update(const socket_manager_ptr& mgr) {
variant
<
size_t
,
sec
>
res
;
{
// Lifetime scope of guard.
std
::
lock_guard
<
std
::
mutex
>
guard
{
write_lock_
};
res
=
write
(
write_handle_
,
&
value
,
sizeof
(
intptr_t
));
if
(
write_handle_
!=
invalid_socket
)
res
=
write
(
write_handle_
,
&
value
,
sizeof
(
intptr_t
));
else
res
=
sec
::
socket_invalid
;
}
if
(
holds_alternative
<
sec
>
(
res
))
mgr
->
deref
();
}
}
void
multiplexer
::
close_pipe
()
{
std
::
lock_guard
<
std
::
mutex
>
guard
{
write_lock_
};
if
(
write_handle_
!=
invalid_socket
)
{
close
(
write_handle_
);
write_handle_
=
pipe_socket
{};
}
}
bool
multiplexer
::
poll_once
(
bool
blocking
)
{
// We'll call poll() until poll() succeeds or fails.
for
(;;)
{
...
...
@@ -174,6 +185,11 @@ bool multiplexer::poll_once(bool blocking) {
}
}
void
multiplexer
::
run
()
{
while
(
!
pollset_
.
empty
())
poll_once
(
true
);
}
void
multiplexer
::
handle_updates
()
{
for
(
auto
mgr
:
dirty_managers_
)
{
auto
index
=
index_of
(
mgr
.
get
());
...
...
libcaf_net/test/multiplexer.cpp
View file @
b295946e
...
...
@@ -52,10 +52,10 @@ public:
}
stream_socket
handle
()
const
noexcept
{
return
socket_cast
<
stream_socket
>
(
socket_manager
::
handle
()
);
return
socket_cast
<
stream_socket
>
(
handle_
);
}
bool
handle_read_event
()
{
bool
handle_read_event
()
override
{
if
(
read_capacity
()
<
1024
)
rd_buf_
.
resize
(
rd_buf_
.
size
()
+
2048
);
auto
res
=
read
(
handle
(),
read_position_begin
(),
read_capacity
());
...
...
@@ -67,7 +67,7 @@ public:
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
bool
handle_write_event
()
{
bool
handle_write_event
()
override
{
if
(
wr_buf_
.
size
()
==
0
)
return
false
;
auto
res
=
write
(
handle
(),
wr_buf_
.
data
(),
wr_buf_
.
size
());
...
...
@@ -79,7 +79,8 @@ public:
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
void
handle_error
(
sec
)
{
void
handle_error
(
sec
code
)
override
{
CAF_FAIL
(
"handle_error called with code "
<<
code
);
}
void
send
(
string_view
x
)
{
...
...
@@ -117,46 +118,57 @@ private:
using
dummy_manager_ptr
=
intrusive_ptr
<
dummy_manager
>
;
struct
fixture
:
host_fixture
{
fixture
()
:
manager_count
(
0
)
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
CAF_REQUIRE_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
CAF_REQUIRE_EQUAL
(
mpx
->
init
(),
none
);
CAF_REQUIRE_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
sockets
=
unbox
(
make_stream_socket_pair
());
alice
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
first
,
mpx
);
bob
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
second
,
mpx
);
alice
->
mask_add
(
operation
::
read
);
bob
->
mask_add
(
operation
::
read
);
mpx
->
handle_updates
();
CAF_REQUIRE_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
fixture
()
:
manager_count
(
0
),
mpx
(
std
::
make_shared
<
multiplexer
>
())
{
// nop
}
~
fixture
()
{
alice
.
reset
();
bob
.
reset
();
mpx
.
reset
();
CAF_REQUIRE_EQUAL
(
manager_count
,
0u
);
}
void
exhaust
()
{
while
(
mpx
->
poll_once
(
false
))
;
// Repeat.
}
size_t
manager_count
;
multiplexer_ptr
mpx
;
dummy_manager_ptr
alice
;
dummy_manager_ptr
bob
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
multiplexer_tests
,
fixture
)
CAF_TEST
(
poll
once
)
{
CAF_TEST
(
default
construction
)
{
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
}
CAF_TEST
(
init
)
{
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
CAF_REQUIRE_EQUAL
(
mpx
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
mpx
->
close_pipe
();
exhaust
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
// Calling run must have no effect now.
mpx
->
run
();
}
CAF_TEST
(
send
and
receive
)
{
CAF_REQUIRE_EQUAL
(
mpx
->
init
(),
none
);
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
alice
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
first
,
mpx
);
auto
bob
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
second
,
mpx
);
alice
->
mask_add
(
operation
::
read
);
bob
->
mask_add
(
operation
::
read
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
alice
->
send
(
"hello bob"
);
alice
->
mask_add
(
operation
::
write
);
mpx
->
handle_updates
();
while
(
mpx
->
poll_once
(
false
))
;
// Repeat.
exhaust
();
CAF_CHECK_EQUAL
(
bob
->
receive
(),
"hello bob"
);
}
...
...
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