Skip to content

Move message queue to Consumer #87

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jakobod wants to merge 28 commits into
base: master
Choose a base branch
from
Draft

Move message queue to Consumer #87

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
b60f85b
Bump CAF tag
Neverlord Sep 24, 2020
b702230
Add message_oriented_layer_ptr
jakobod Sep 22, 2020
16b709f
WIP: Refactoring BASP
jakobod Sep 22, 2020
16ba14f
WIP: message rework of BASP
jakobod Sep 23, 2020
ec4d770
Add convenience function for writing messages
jakobod Sep 23, 2020
6a5ab93
Complete Interfaces of BASP and length_prefixing
jakobod Sep 23, 2020
8c1ec37
Fix length prefixing
jakobod Sep 23, 2020
e090e0a
Move message queue to consumer
jakobod Sep 23, 2020
e6245bf
Fix prefix_framing test
jakobod Sep 24, 2020
b6dd051
Fix BASP tests
jakobod Sep 24, 2020
47c1f07
WIP: Fix full-setup test
jakobod Sep 24, 2020
b319da5
Implement proper event handling in BASP
jakobod Sep 25, 2020
9ee1ee8
Fix ping_pong test
jakobod Sep 25, 2020
5b638bc
Fix debug build and actor_shell test
jakobod Sep 25, 2020
7f85a0c
Fix example/web-socket after rebase
jakobod Sep 28, 2020
6cdded1
Check other TAG version
jakobod Sep 28, 2020
12437c5
Add mutex to application
jakobod Sep 28, 2020
b8adfe4
Simplify BASP state tracking
jakobod Sep 29, 2020
c9c4161
Wait for fulfilled handshake in BASP
jakobod Oct 1, 2020
4016a88
Use correct config value in basp
jakobod Oct 1, 2020
d58e004
Cleanup
jakobod Oct 1, 2020
8c592d2
Add missing functions after rebase
jakobod Oct 2, 2020
ab2eb4d
Remove basp specifics from socket_manager
jakobod Oct 2, 2020
417cea0
Remove endpoint_manager(impl)
jakobod Oct 2, 2020
f68d37b
Reenable worker and message-queue tests
jakobod Oct 2, 2020
228826c
Remove unnecessary resolve functions
jakobod Oct 2, 2020
784c96e
Remove consumer interface
jakobod Oct 4, 2020
a51fef6
Honor can_send_more result in application
jakobod Oct 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare(
actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG 4f8609b
GIT_TAG 963ef5fe
)
FetchContent_Populate(actor_framework)
set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE)
Expand Down
2 changes: 2 additions & 0 deletions examples/net/web-socket-calculator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
// asyncio.get_event_loop().run_until_complete(hello())
// ~~~

#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte_span.hpp"
Expand All @@ -47,6 +48,7 @@
#include "caf/net/socket_manager.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket_server.hpp"
#include "caf/string_view.hpp"
#include "caf/tag/mixed_message_oriented.hpp"

#include <cstdint>
Expand Down
26 changes: 10 additions & 16 deletions libcaf_net/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,6 @@ file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")

# -- add consistency checks for enum to_string implementations -----------------

caf_incubator_add_enum_consistency_check("caf/net/basp/connection_state.hpp"
"src/basp/connection_state_strings.cpp")
caf_incubator_add_enum_consistency_check("caf/net/basp/ec.hpp"
"src/basp/ec_strings.cpp")
caf_incubator_add_enum_consistency_check("caf/net/basp/message_type.hpp"
Expand Down Expand Up @@ -34,13 +32,11 @@ endfunction()
# -- add library targets -------------------------------------------------------

add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
#src/actor_proxy_impl.cpp
#src/basp/application.cpp
#src/endpoint_manager.cpp
src/actor_proxy_impl.cpp
#src/net/backend/tcp.cpp
#src/net/backend/test.cpp
#src/net/endpoint_manager_queue.cpp
src/basp/connection_state_strings.cpp
src/net/backend/test.cpp
src/net/consumer_queue.cpp
src/basp/application.cpp
src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp
src/basp/operation_strings.cpp
Expand Down Expand Up @@ -81,8 +77,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_net_set_default_properties(libcaf_net_obj libcaf_net)

target_include_directories(libcaf_net INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>)
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>)

add_library(CAF::net ALIAS libcaf_net)

Expand Down Expand Up @@ -126,21 +122,19 @@ target_link_libraries(caf-net-test PRIVATE CAF::test)

caf_incubator_add_test_suites(caf-net-test
accept_socket
#application
application
convert_ip_endpoint
datagram_socket
detail.rfc6455
#datagram_transport
#doorman
#endpoint_manager
header
ip
multiplexer
net.actor_shell
#net.backend.tcp
#net.basp.message_queue
#net.basp.ping_pong
#net.basp.worker
net.basp.message_queue
net.basp.ping_pong
net.basp.worker
net.length_prefix_framing
net.web_socket_server
network_socket
Expand Down
6 changes: 3 additions & 3 deletions libcaf_net/caf/net/actor_proxy_impl.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#pragma once

#include "caf/actor_proxy.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"

namespace caf::net {

Expand All @@ -28,7 +28,7 @@ class actor_proxy_impl : public actor_proxy {
public:
using super = actor_proxy;

actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst);
actor_proxy_impl(actor_config& cfg, basp::application* app);

~actor_proxy_impl() override;

Expand All @@ -37,7 +37,7 @@ class actor_proxy_impl : public actor_proxy {
void kill_proxy(execution_unit* ctx, error rsn) override;

private:
endpoint_manager_ptr dst_;
basp::application* app_;
Copy link
Member

@Neverlord Neverlord Oct 4, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a very serious shift in semantics. Previously, the proxy participated in shared ownership of the endpoint manager. Now, it only keeps a raw pointer to the BASP app directly. How is CAF making sure that the proxy never accesses a pointer past the lifetime of the pointed-to object now?

I couldn't find any comments going in to detail on this, and actor_proxy_impl::enqueue calls into the object without any safeguarding. Furthermore, what is the rationale behind keeping this proxy implementation in the public API when it's clearly bound to BASP?

Copy link
Collaborator Author

@jakobod jakobod Oct 4, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How is CAF making sure that the proxy never accesses a pointer past the lifetime of the pointed-to object now

It doesn't. The proxy would have to hold a socket_manager_ptr to ensure the lifetime of the BASP app, which is definitely possible. I'll add that.

};

} // namespace caf::net
4 changes: 1 addition & 3 deletions libcaf_net/caf/net/all.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,13 @@
#pragma once

#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/consumer_queue.hpp"
#include "caf/net/datagram_socket.hpp"
#include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/host.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
Expand Down
18 changes: 12 additions & 6 deletions libcaf_net/caf/net/backend/test.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,26 @@
#pragma once

#include <map>
#include <mutex>
#include <tuple>

#include "caf/detail/net_export.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/node_id.hpp"

namespace caf::net::backend {

/// Minimal backend for unit testing.
/// @warning this backend is *not* thread safe.
class CAF_NET_EXPORT test : public middleman_backend {
public:
// -- member types -----------------------------------------------------------

using peer_entry = std::pair<stream_socket, endpoint_manager_ptr>;
using peer_entry
= std::tuple<stream_socket, socket_manager_ptr, basp::application*>;

// -- constructors, destructors, and assignment operators --------------------

Expand All @@ -49,9 +52,9 @@ class CAF_NET_EXPORT test : public middleman_backend {

void stop() override;

endpoint_manager_ptr peer(const node_id& id) override;
socket_manager_ptr peer(const node_id& id) override;

expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
expected<socket_manager_ptr> get_or_connect(const uri& locator) override;

void resolve(const uri& locator, const actor& listener) override;

Expand All @@ -62,7 +65,8 @@ class CAF_NET_EXPORT test : public middleman_backend {
// -- properties -------------------------------------------------------------

stream_socket socket(const node_id& peer_id) {
return get_peer(peer_id).first;
auto& entry = get_peer(peer_id);
return std::get<stream_socket>(entry);
}

uint16_t port() const noexcept override;
Expand All @@ -78,6 +82,8 @@ class CAF_NET_EXPORT test : public middleman_backend {
std::map<node_id, peer_entry> peers_;

proxy_registry proxies_;

std::mutex lock_;
};

} // namespace caf::net::backend
Loading