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
5a0ece9e
Commit
5a0ece9e
authored
Jul 24, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Implement endpoint managers
parent
1516ac5f
Changes
6
Show whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
502 additions
and
0 deletions
+502
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/net/endpoint_manager.hpp
libcaf_net/caf/net/endpoint_manager.hpp
+148
-0
libcaf_net/caf/net/endpoint_manager_impl.hpp
libcaf_net/caf/net/endpoint_manager_impl.hpp
+101
-0
libcaf_net/caf/net/make_endpoint_manager.hpp
libcaf_net/caf/net/make_endpoint_manager.hpp
+37
-0
libcaf_net/src/endpoint_manager.cpp
libcaf_net/src/endpoint_manager.cpp
+67
-0
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+148
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
5a0ece9e
...
...
@@ -7,6 +7,7 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set
(
LIBCAF_NET_SRCS
src/endpoint_manager.cpp
src/host.cpp
src/multiplexer.cpp
src/network_socket.cpp
...
...
libcaf_net/caf/net/endpoint_manager.hpp
0 → 100644
View file @
5a0ece9e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstddef>
#include <cstdint>
#include <memory>
#include "caf/actor.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
/// Manages a communication endpoint.
class
endpoint_manager
:
public
socket_manager
{
public:
// -- member types -----------------------------------------------------------
using
super
=
socket_manager
;
struct
event
:
intrusive
::
singly_linked
<
event
>
{
struct
resolve_request
{
std
::
string
path
;
actor
listener
;
};
struct
timeout
{
atom_value
type
;
uint64_t
id
;
};
event
(
std
::
string
path
,
actor
listener
);
event
(
atom_value
type
,
uint64_t
id
);
/// Either contains a string for `resolve` requests or an `atom_value`
variant
<
resolve_request
,
timeout
>
value
;
};
struct
event_policy
{
using
deficit_type
=
size_t
;
using
task_size_type
=
size_t
;
using
mapped_type
=
event
;
using
unique_pointer
=
std
::
unique_ptr
<
event
>
;
using
queue_type
=
intrusive
::
drr_queue
<
event_policy
>
;
task_size_type
task_size
(
const
event
&
)
const
noexcept
{
return
1
;
}
};
using
event_queue_type
=
intrusive
::
fifo_inbox
<
event_policy
>
;
struct
message
:
intrusive
::
singly_linked
<
message
>
{
/// Original message to a remote actor.
mailbox_element_ptr
msg
;
/// Serialized representation of of `msg->content()`.
std
::
vector
<
char
>
payload
;
};
struct
message_policy
{
using
deficit_type
=
size_t
;
using
task_size_type
=
size_t
;
using
mapped_type
=
message
;
using
unique_pointer
=
std
::
unique_ptr
<
message
>
;
using
queue_type
=
intrusive
::
drr_queue
<
message_policy
>
;
task_size_type
task_size
(
const
message
&
x
)
const
noexcept
{
return
x
.
payload
.
size
();
}
};
using
message_queue_type
=
intrusive
::
fifo_inbox
<
message_policy
>
;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
,
actor_system
&
sys
);
~
endpoint_manager
()
override
;
// -- properties -------------------------------------------------------------
event_queue_type
&
event_queue
()
{
return
events_
;
}
message_queue_type
&
message_queue
()
{
return
messages_
;
}
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor
void
resolve
(
std
::
string
path
,
actor
listener
);
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual
error
init
()
=
0
;
protected:
/// Points to the hosting actor system.
actor_system
&
sys_
;
/// Stores control events.
event_queue_type
events_
;
/// Stores outbound messages.
message_queue_type
messages_
;
};
using
endpoint_manager_ptr
=
intrusive_ptr
<
endpoint_manager
>
;
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/endpoint_manager_impl.hpp
0 → 100644
View file @
5a0ece9e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/endpoint_manager.hpp"
namespace
caf
{
namespace
net
{
template
<
class
Transport
,
class
Application
>
class
endpoint_manager_impl
:
public
endpoint_manager
{
public:
// -- member types -----------------------------------------------------------
using
super
=
endpoint_manager
;
using
transport_type
=
Transport
;
using
application_type
=
Application
;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl
(
const
multiplexer_ptr
&
parent
,
actor_system
&
sys
,
Transport
trans
,
Application
app
)
:
super
(
trans
.
handle
(),
parent
,
sys
),
transport_
(
std
::
move
(
trans
)),
application_
(
std
::
move
(
app
))
{
// nop
}
~
endpoint_manager_impl
()
override
{
// nop
}
// -- properties -------------------------------------------------------------
transport_type
&
transport
()
{
return
transport_
;
}
application_type
&
application
()
{
return
application_
;
}
// -- interface functions ----------------------------------------------------
error
init
()
override
{
return
transport_
.
init
(
*
this
);
}
bool
handle_read_event
()
override
{
return
transport_
.
handle_read_event
(
*
this
);
}
bool
handle_write_event
()
override
{
if
(
!
this
->
events_
.
empty
())
{
this
->
events_
.
fetch_more
();
auto
&
q
=
this
->
events_
.
queue
();
q
.
inc_deficit
(
q
.
total_task_size
());
for
(
auto
ptr
=
q
.
next
();
ptr
!=
nullptr
;
ptr
=
q
.
next
())
{
using
timeout
=
endpoint_manager
::
event
::
timeout
;
using
resolve_request
=
endpoint_manager
::
event
::
resolve_request
;
if
(
auto
rr
=
get_if
<
resolve_request
>
(
&
ptr
->
value
))
{
transport_
.
resolve
(
*
this
,
std
::
move
(
rr
->
path
),
std
::
move
(
rr
->
listener
));
}
else
{
auto
&
t
=
get
<
timeout
>
(
ptr
->
value
);
transport_
.
timeout
(
*
this
,
t
.
type
,
t
.
id
);
}
}
}
return
transport_
.
handle_write_event
(
*
this
);
}
void
handle_error
(
sec
code
)
override
{
transport_
.
handle_error
(
application_
,
code
);
}
private:
Transport
transport_
;
Application
application_
;
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/make_endpoint_manager.hpp
0 → 100644
View file @
5a0ece9e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/make_counted.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
namespace
caf
{
namespace
net
{
template
<
class
Transport
,
class
Application
>
endpoint_manager_ptr
make_endpoint_manager
(
const
multiplexer_ptr
&
mpx
,
actor_system
&
sys
,
Transport
trans
,
Application
app
)
{
using
impl
=
endpoint_manager_impl
<
Transport
,
Application
>
;
return
make_counted
<
impl
>
(
mpx
,
sys
,
std
::
move
(
trans
),
std
::
move
(
app
));
}
}
// namespace net
}
// namespace caf
libcaf_net/src/endpoint_manager.cpp
0 → 100644
View file @
5a0ece9e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/endpoint_manager.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace
caf
{
namespace
net
{
endpoint_manager
::
event
::
event
(
std
::
string
path
,
actor
listener
)
:
value
(
resolve_request
{
std
::
move
(
path
),
std
::
move
(
listener
)})
{
// nop
}
endpoint_manager
::
event
::
event
(
atom_value
type
,
uint64_t
id
)
:
value
(
timeout
{
type
,
id
})
{
// nop
}
endpoint_manager
::
endpoint_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
,
actor_system
&
sys
)
:
super
(
handle
,
parent
),
sys_
(
sys
),
events_
(
event_policy
{}),
messages_
(
message_policy
{})
{
// nop
}
endpoint_manager
::~
endpoint_manager
()
{
// nop
}
void
endpoint_manager
::
resolve
(
std
::
string
path
,
actor
listener
)
{
using
intrusive
::
inbox_result
;
auto
ptr
=
new
event
(
std
::
move
(
path
),
std
::
move
(
listener
));
switch
(
events_
.
push_back
(
ptr
))
{
default:
break
;
case
inbox_result
:
:
unblocked_reader
:
mask_add
(
operation
::
write
);
break
;
case
inbox_result
:
:
queue_closed
:
anon_send
(
listener
,
resolve_atom
::
value
,
make_error
(
sec
::
request_receiver_down
));
}
}
}
// namespace net
}
// namespace caf
libcaf_net/test/endpoint_manager.cpp
0 → 100644
View file @
5a0ece9e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 endpoint_manager
#include "caf/net/endpoint_manager.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
string_view
hello_manager
{
"hello manager!"
};
string_view
hello_test
{
"hello test!"
};
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
}
multiplexer_ptr
mpx
;
};
class
dummy_application
{};
class
dummy_transport
{
public:
dummy_transport
(
stream_socket
handle
,
std
::
shared_ptr
<
std
::
vector
<
char
>>
data
)
:
handle_
(
handle
),
data_
(
data
),
read_buf_
(
1024
)
{
// nop
}
stream_socket
handle
()
{
return
handle_
;
}
template
<
class
Manager
>
error
init
(
Manager
&
manager
)
{
write_buf_
.
insert
(
write_buf_
.
end
(),
hello_test
.
begin
(),
hello_test
.
end
());
CAF_CHECK
(
manager
.
mask_add
(
operation
::
read_write
));
return
none
;
}
template
<
class
Manager
>
bool
handle_read_event
(
Manager
&
)
{
auto
res
=
read
(
handle_
,
read_buf_
.
data
(),
read_buf_
.
size
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
data_
->
insert
(
data_
->
end
(),
read_buf_
.
begin
(),
read_buf_
.
begin
()
+
*
num_bytes
);
return
true
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
template
<
class
Manager
>
bool
handle_write_event
(
Manager
&
)
{
auto
res
=
write
(
handle_
,
write_buf_
.
data
(),
write_buf_
.
size
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
write_buf_
.
erase
(
write_buf_
.
begin
(),
write_buf_
.
begin
()
+
*
num_bytes
);
return
write_buf_
.
size
()
>
0
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
template
<
class
Manager
>
void
handle_error
(
Manager
&
,
sec
)
{
// nop
}
template
<
class
Manager
>
void
resolve
(
Manager
&
,
std
::
string
path
,
actor
listener
)
{
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
make_error
(
sec
::
feature_disabled
));
}
template
<
class
Manager
>
void
timeout
(
Manager
&
,
atom_value
,
uint64_t
)
{
// nop
}
private:
stream_socket
handle_
;
std
::
shared_ptr
<
std
::
vector
<
char
>>
data_
;
std
::
vector
<
char
>
read_buf_
;
std
::
vector
<
char
>
write_buf_
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
send
and
receive
)
{
std
::
vector
<
char
>
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
char
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
dummy_transport
{
sockets
.
first
,
buf
},
dummy_application
{});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
hello_manager
.
data
(),
hello_manager
.
size
()),
hello_manager
.
size
());
while
(
mpx
->
poll_once
(
false
))
;
// Repeat.
CAF_CHECK_EQUAL
(
string_view
(
buf
->
data
(),
buf
->
size
()),
hello_manager
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
hello_test
.
size
());
CAF_CHECK_EQUAL
(
string_view
(
read_buf
.
data
(),
hello_test
.
size
()),
hello_test
);
}
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