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
6b9437d1
Commit
6b9437d1
authored
Jul 18, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Implement scaffold for new multiplexer
parent
792c3213
Changes
8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
791 additions
and
0 deletions
+791
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+3
-0
libcaf_net/caf/net/multiplexer.hpp
libcaf_net/caf/net/multiplexer.hpp
+118
-0
libcaf_net/caf/net/pollset_updater.hpp
libcaf_net/caf/net/pollset_updater.hpp
+56
-0
libcaf_net/caf/net/socket_manager.hpp
libcaf_net/caf/net/socket_manager.hpp
+96
-0
libcaf_net/src/multiplexer.cpp
libcaf_net/src/multiplexer.cpp
+228
-0
libcaf_net/src/pollset_updater.cpp
libcaf_net/src/pollset_updater.cpp
+61
-0
libcaf_net/src/socket_manager.cpp
libcaf_net/src/socket_manager.cpp
+66
-0
libcaf_net/test/multiplexer.cpp
libcaf_net/test/multiplexer.cpp
+163
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
6b9437d1
...
...
@@ -8,9 +8,12 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set
(
LIBCAF_NET_SRCS
src/host.cpp
src/multiplexer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
)
...
...
libcaf_net/caf/net/multiplexer.hpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <memory>
#include <mutex>
#include <thread>
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
extern
"C"
{
struct
pollfd
;
}
// extern "C"
namespace
caf
{
namespace
net
{
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class
multiplexer
:
public
std
::
enable_shared_from_this
<
multiplexer
>
{
public:
// -- member types -----------------------------------------------------------
using
pollfd_list
=
std
::
vector
<
pollfd
>
;
using
manager_list
=
std
::
vector
<
socket_manager_ptr
>
;
// -- constructors, destructors, and assignment operators --------------------
multiplexer
();
~
multiplexer
();
error
init
();
// -- properties -------------------------------------------------------------
/// Returns the number of currently active socket managers.
size_t
num_socket_managers
()
const
noexcept
;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t
index_of
(
const
socket_manager_ptr
&
mgr
);
// -- thread-safe signaling --------------------------------------------------
/// @thread-safe
void
update
(
const
socket_manager_ptr
&
mgr
);
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
/// ready as a result.
bool
poll_once
(
bool
blocking
);
/// Polls until no socket event handler remains.
void
run
();
void
handle_updates
();
void
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
);
protected:
// -- convenience functions --------------------------------------------------
void
add
(
socket_manager_ptr
mgr
);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
std
::
vector
<
pollfd
>
pollset_
;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list
managers_
;
/// 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.
std
::
thread
::
id
tid_
;
///
pipe_socket
write_handle_
;
///
std
::
mutex
write_lock_
;
};
/// @relates multiplexer
using
multiplexer_ptr
=
std
::
shared_ptr
<
multiplexer
>
;
/// @relates multiplexer
using
weak_multiplexer_ptr
=
std
::
weak_ptr
<
multiplexer
>
;
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/pollset_updater.hpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
namespace
caf
{
namespace
net
{
class
pollset_updater
:
public
socket_manager
{
public:
// -- member types -----------------------------------------------------------
using
super
=
socket_manager
;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater
(
pipe_socket
read_handle
,
multiplexer_ptr
parent
);
~
pollset_updater
()
override
;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
pipe_socket
handle
()
const
noexcept
{
return
socket_cast
<
pipe_socket
>
(
handle_
);
}
// -- interface functions ----------------------------------------------------
bool
handle_read_event
()
override
;
bool
handle_write_event
()
override
;
void
handle_error
(
sec
code
)
override
;
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/socket_manager.hpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
namespace
caf
{
namespace
net
{
/// Manages the lifetime of a single socket and handles any I/O events on it.
class
socket_manager
:
public
ref_counted
{
public:
// -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket`
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
);
virtual
~
socket_manager
();
socket_manager
(
const
socket_manager
&
)
=
delete
;
socket_manager
&
operator
=
(
const
socket_manager
&
)
=
delete
;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket
handle
()
const
noexcept
{
return
handle_
;
}
/// Returns for which operations the socket is registered (read or write).
operation
mask
()
const
noexcept
;
/// Adds given flag(s) to the event mask and updates the parent on success.
/// @returns `false` if `mask()` already contained `flag`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool
mask_add
(
operation
flag
)
noexcept
;
/// Deletes given flag(s) from the event mask and updates the parent on
/// success.
/// @returns `false` if `mask()` already missed `flag`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool
mask_del
(
operation
flag
)
noexcept
;
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
virtual
bool
handle_read_event
()
=
0
;
/// Called whenever the socket is allowed to send additional data.
virtual
bool
handle_write_event
()
=
0
;
/// Called when the remote side becomes unreachable due to an error.
/// @param reason The error code as reported by the operating system.
virtual
void
handle_error
(
sec
code
)
=
0
;
protected:
// -- member variables -------------------------------------------------------
socket
handle_
;
std
::
atomic
<
operation
>
mask_
;
weak_multiplexer_ptr
parent_
;
};
using
socket_manager_ptr
=
intrusive_ptr
<
socket_manager
>
;
}
// namespace net
}
// namespace caf
libcaf_net/src/multiplexer.cpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/multiplexer.hpp"
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#ifndef CAF_WINDOWS
# include <poll.h>
#else
# include "caf/detail/socket_sys_includes.hpp"
#endif // CAF_WINDOWS
namespace
caf
{
namespace
net
{
#ifndef POLLRDHUP
# define POLLRDHUP POLLHUP
#endif
#ifndef POLLPRI
# define POLLPRI POLLIN
#endif
namespace
{
#ifdef CAF_WINDOWS
// From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft
// Winsock provider, the WSAPoll function will fail.
const
short
input_mask
=
POLLIN
;
#else
const
short
input_mask
=
POLLIN
|
POLLPRI
;
#endif
const
short
error_mask
=
POLLRDHUP
|
POLLERR
|
POLLHUP
|
POLLNVAL
;
const
short
output_mask
=
POLLOUT
;
short
to_bitmask
(
operation
op
)
{
switch
(
op
)
{
case
operation
:
:
read
:
return
input_mask
;
case
operation
:
:
write
:
return
output_mask
;
case
operation
:
:
read_write
:
return
input_mask
|
output_mask
;
default:
return
0
;
}
}
}
// namespace
multiplexer
::
multiplexer
()
{
// nop
}
multiplexer
::~
multiplexer
()
{
// nop
}
error
multiplexer
::
init
()
{
auto
pipe_handles
=
make_pipe
();
if
(
!
pipe_handles
)
return
std
::
move
(
pipe_handles
.
error
());
tid_
=
std
::
this_thread
::
get_id
();
add
(
make_counted
<
pollset_updater
>
(
pipe_handles
->
first
,
shared_from_this
()));
write_handle_
=
pipe_handles
->
second
;
return
none
;
}
size_t
multiplexer
::
num_socket_managers
()
const
noexcept
{
return
managers_
.
size
();
}
ptrdiff_t
multiplexer
::
index_of
(
const
socket_manager_ptr
&
mgr
)
{
auto
max_index
=
static_cast
<
ptrdiff_t
>
(
managers_
.
size
());
for
(
ptrdiff_t
index
=
0
;
index
<
max_index
;
++
index
)
if
(
managers_
[
index
]
==
mgr
)
return
index
;
return
-
1
;
}
void
multiplexer
::
update
(
const
socket_manager_ptr
&
mgr
)
{
if
(
std
::
this_thread
::
get_id
()
==
tid_
)
{
auto
i
=
std
::
find
(
dirty_managers_
.
begin
(),
dirty_managers_
.
end
(),
mgr
);
if
(
i
==
dirty_managers_
.
end
())
dirty_managers_
.
emplace_back
(
mgr
);
}
else
{
mgr
->
ref
();
auto
value
=
reinterpret_cast
<
intptr_t
>
(
mgr
.
get
());
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
(
holds_alternative
<
sec
>
(
res
))
mgr
->
deref
();
}
}
bool
multiplexer
::
poll_once
(
bool
blocking
)
{
// We'll call poll() until poll() succeeds or fails.
for
(;;)
{
int
presult
;
#ifdef CAF_WINDOWS
presult
=
::
WSAPoll
(
pollset_
.
data
(),
static_cast
<
ULONG
>
(
pollset_
.
size
()),
blocking
?
-
1
:
0
);
#else
presult
=
::
poll
(
pollset_
.
data
(),
static_cast
<
nfds_t
>
(
pollset_
.
size
()),
blocking
?
-
1
:
0
);
#endif
if
(
presult
<
0
)
{
switch
(
last_socket_error
())
{
case
std
:
:
errc
::
interrupted
:
{
// A signal was caught. Simply try again.
CAF_LOG_DEBUG
(
"received errc::interrupted, try again"
);
break
;
}
case
std
:
:
errc
::
not_enough_memory
:
{
CAF_LOG_ERROR
(
"poll() failed due to insufficient memory"
);
// There's not much we can do other than try again in hope someone
// else releases memory.
break
;
}
default:
{
// Must not happen.
perror
(
"poll() failed"
);
CAF_CRITICAL
(
"poll() failed"
);
}
}
// Rinse and repeat.
continue
;
}
CAF_LOG_DEBUG
(
"poll() on"
<<
pollset_
.
size
()
<<
"sockets reported"
<<
presult
<<
"event(s)"
);
// No activity.
if
(
presult
==
0
)
return
false
;
// Scan pollset for events.
CAF_LOG_DEBUG
(
"scan pollset for socket events"
);
for
(
size_t
i
=
0
;
i
<
pollset_
.
size
()
&&
presult
>
0
;
++
i
)
{
auto
&
x
=
pollset_
[
i
];
if
(
x
.
revents
!=
0
)
{
handle
(
managers_
[
i
],
x
.
revents
);
--
presult
;
}
}
handle_updates
();
return
true
;
}
}
void
multiplexer
::
handle_updates
()
{
for
(
auto
mgr
:
dirty_managers_
)
{
auto
index
=
index_of
(
mgr
.
get
());
if
(
index
==
-
1
)
{
add
(
std
::
move
(
mgr
));
}
else
{
// Update or remove an existing manager in the pollset.
if
(
mgr
->
mask
()
==
operation
::
none
)
{
pollset_
.
erase
(
pollset_
.
begin
()
+
index
);
managers_
.
erase
(
managers_
.
begin
()
+
index
);
}
else
{
pollset_
[
index
].
events
=
to_bitmask
(
mgr
->
mask
());
}
}
}
dirty_managers_
.
clear
();
}
void
multiplexer
::
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
mgr
->
handle
())
<<
CAF_ARG
(
mask
));
CAF_ASSERT
(
mgr
!=
nullptr
);
bool
checkerror
=
true
;
if
((
mask
&
input_mask
)
!=
0
)
{
checkerror
=
false
;
if
(
!
mgr
->
handle_read_event
())
mgr
->
mask_del
(
operation
::
read
);
}
if
((
mask
&
output_mask
)
!=
0
)
{
checkerror
=
false
;
if
(
!
mgr
->
handle_write_event
())
mgr
->
mask_del
(
operation
::
write
);
}
if
(
checkerror
&&
((
mask
&
error_mask
)
!=
0
))
{
if
(
mask
&
POLLNVAL
)
mgr
->
handle_error
(
sec
::
socket_invalid
);
else
if
(
mask
&
POLLHUP
)
mgr
->
handle_error
(
sec
::
socket_disconnected
);
else
mgr
->
handle_error
(
sec
::
socket_operation_failed
);
mgr
->
mask_del
(
operation
::
read_write
);
}
}
void
multiplexer
::
add
(
socket_manager_ptr
mgr
)
{
pollfd
new_entry
{
socket_cast
<
socket_id
>
(
mgr
->
handle
()),
to_bitmask
(
mgr
->
mask
()),
0
};
pollset_
.
emplace_back
(
new_entry
);
managers_
.
emplace_back
(
std
::
move
(
mgr
));
}
}
// namespace net
}
// namespace caf
libcaf_net/src/pollset_updater.cpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/pollset_updater.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
pollset_updater
::
pollset_updater
(
pipe_socket
read_handle
,
multiplexer_ptr
parent
)
:
super
(
read_handle
,
std
::
move
(
parent
))
{
mask_
=
operation
::
read
;
}
pollset_updater
::~
pollset_updater
()
{
// nop
}
bool
pollset_updater
::
handle_read_event
()
{
for
(;;)
{
intptr_t
value
;
auto
res
=
read
(
handle
(),
&
value
,
sizeof
(
intptr_t
));
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
CAF_ASSERT
(
*
num_bytes
==
sizeof
(
intptr_t
));
socket_manager_ptr
mgr
{
reinterpret_cast
<
socket_manager
*>
(
value
),
false
};
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
mgr
);
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
}
bool
pollset_updater
::
handle_write_event
()
{
return
false
;
}
void
pollset_updater
::
handle_error
(
sec
)
{
// nop
}
}
// namespace net
}
// namespace caf
libcaf_net/src/socket_manager.cpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/socket_manager.hpp"
#include "caf/config.hpp"
#include "caf/net/multiplexer.hpp"
namespace
caf
{
namespace
net
{
socket_manager
::
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
)
:
handle_
(
handle
),
mask_
(
operation
::
none
),
parent_
(
parent
)
{
CAF_ASSERT
(
parent
!=
nullptr
);
CAF_ASSERT
(
handle_
!=
invalid_socket
);
}
socket_manager
::~
socket_manager
()
{
close
(
handle_
);
}
operation
socket_manager
::
mask
()
const
noexcept
{
return
mask_
.
load
();
}
bool
socket_manager
::
mask_add
(
operation
flag
)
noexcept
{
CAF_ASSERT
(
flag
!=
operation
::
none
);
auto
x
=
mask
();
while
((
x
&
flag
)
!=
flag
)
if
(
mask_
.
compare_exchange_strong
(
x
,
x
|
flag
))
{
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
this
);
return
true
;
}
return
false
;
}
bool
socket_manager
::
mask_del
(
operation
flag
)
noexcept
{
CAF_ASSERT
(
flag
!=
operation
::
none
);
auto
x
=
mask
();
while
((
x
&
flag
)
!=
operation
::
none
)
if
(
mask_
.
compare_exchange_strong
(
x
,
x
&
~
flag
))
{
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
this
);
return
true
;
}
return
false
;
}
}
// namespace net
}
// namespace caf
libcaf_net/test/multiplexer.cpp
0 → 100644
View file @
6b9437d1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 multiplexer
#include "caf/net/multiplexer.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <new>
#include <tuple>
#include <vector>
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
class
dummy_manager
:
public
socket_manager
{
public:
dummy_manager
(
size_t
&
manager_count
,
stream_socket
handle
,
multiplexer_ptr
parent
)
:
socket_manager
(
handle
,
std
::
move
(
parent
)),
count_
(
manager_count
),
rd_buf_pos_
(
0
)
{
++
count_
;
rd_buf_
.
resize
(
1024
);
}
~
dummy_manager
()
{
--
count_
;
}
stream_socket
handle
()
const
noexcept
{
return
socket_cast
<
stream_socket
>
(
socket_manager
::
handle
());
}
bool
handle_read_event
()
{
if
(
read_capacity
()
<
1024
)
rd_buf_
.
resize
(
rd_buf_
.
size
()
+
2048
);
auto
res
=
read
(
handle
(),
read_position_begin
(),
read_capacity
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
CAF_ASSERT
(
*
num_bytes
>
0
);
rd_buf_pos_
+=
*
num_bytes
;
return
true
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
bool
handle_write_event
()
{
if
(
wr_buf_
.
size
()
==
0
)
return
false
;
auto
res
=
write
(
handle
(),
wr_buf_
.
data
(),
wr_buf_
.
size
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
CAF_ASSERT
(
*
num_bytes
>
0
);
wr_buf_
.
erase
(
wr_buf_
.
begin
(),
wr_buf_
.
begin
()
+
*
num_bytes
);
return
wr_buf_
.
size
()
>
0
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
void
handle_error
(
sec
)
{
}
void
send
(
string_view
x
)
{
wr_buf_
.
insert
(
wr_buf_
.
end
(),
x
.
begin
(),
x
.
end
());
}
std
::
string
receive
()
{
std
::
string
result
(
rd_buf_
.
data
(),
read_position_begin
());
rd_buf_pos_
=
0
;
return
result
;
}
private:
char
*
read_position_begin
()
{
return
rd_buf_
.
data
()
+
rd_buf_pos_
;
}
char
*
read_position_end
()
{
return
rd_buf_
.
data
()
+
rd_buf_
.
size
();
}
size_t
read_capacity
()
const
{
return
rd_buf_
.
size
()
-
rd_buf_pos_
;
}
size_t
&
count_
;
size_t
rd_buf_pos_
;
std
::
vector
<
char
>
wr_buf_
;
std
::
vector
<
char
>
rd_buf_
;
};
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
()
{
alice
.
reset
();
bob
.
reset
();
mpx
.
reset
();
CAF_REQUIRE_EQUAL
(
manager_count
,
0u
);
}
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
)
{
alice
->
send
(
"hello bob"
);
alice
->
mask_add
(
operation
::
write
);
mpx
->
handle_updates
();
while
(
mpx
->
poll_once
(
false
))
;
// Repeat.
CAF_CHECK_EQUAL
(
bob
->
receive
(),
"hello bob"
);
}
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