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
633d6f92
Commit
633d6f92
authored
Sep 26, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix deserialization of actor handles
parent
2e465866
Changes
2
Show whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
62 additions
and
34 deletions
+62
-34
libcaf_net/caf/net/basp/application.hpp
libcaf_net/caf/net/basp/application.hpp
+17
-24
libcaf_net/src/application.cpp
libcaf_net/src/application.cpp
+45
-10
No files found.
libcaf_net/caf/net/basp/application.hpp
View file @
633d6f92
...
@@ -38,6 +38,7 @@
...
@@ -38,6 +38,7 @@
#include "caf/node_id.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/response_promise.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
#include "caf/unit.hpp"
...
@@ -69,6 +70,8 @@ public:
...
@@ -69,6 +70,8 @@ public:
error
init
(
Parent
&
parent
)
{
error
init
(
Parent
&
parent
)
{
// Initialize member variables.
// Initialize member variables.
system_
=
&
parent
.
system
();
system_
=
&
parent
.
system
();
executor_
.
system_ptr
(
system_
);
executor_
.
proxy_registry_ptr
(
&
proxies_
);
// TODO: use `if constexpr` when switching to C++17.
// TODO: use `if constexpr` when switching to C++17.
// Allow unit tests to run the application without endpoint manager.
// Allow unit tests to run the application without endpoint manager.
if
(
!
std
::
is_base_of
<
test_tag
,
Parent
>::
value
)
if
(
!
std
::
is_base_of
<
test_tag
,
Parent
>::
value
)
...
@@ -86,30 +89,11 @@ public:
...
@@ -86,30 +89,11 @@ public:
template
<
class
Parent
>
template
<
class
Parent
>
error
write_message
(
Parent
&
parent
,
error
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
ptr
)
{
std
::
unique_ptr
<
endpoint_manager
::
message
>
ptr
)
{
CAF_ASSERT
(
ptr
!=
nullptr
);
auto
write_packet
=
make_callback
([
&
](
byte_span
hdr
,
byte_span
payload
)
{
CAF_ASSERT
(
ptr
->
msg
!=
nullptr
);
parent
.
write_packet
(
hdr
,
payload
);
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
(),
buf_
};
const
auto
&
src
=
ptr
->
msg
->
sender
;
const
auto
&
dst
=
ptr
->
receiver
;
if
(
dst
==
nullptr
)
{
// TODO: valid?
return
none
;
}
if
(
src
!=
nullptr
)
{
if
(
auto
err
=
sink
(
src
->
node
(),
src
->
id
(),
dst
->
id
(),
ptr
->
msg
->
stages
))
return
err
;
}
else
{
if
(
auto
err
=
sink
(
node_id
{},
actor_id
{
0
},
dst
->
id
(),
ptr
->
msg
->
stages
))
return
err
;
}
// TODO: avoid extra copy of the payload
buf_
.
insert
(
buf_
.
end
(),
ptr
->
payload
.
begin
(),
ptr
->
payload
.
end
());
header
hdr
{
message_type
::
actor_message
,
static_cast
<
uint32_t
>
(
buf_
.
size
()),
ptr
->
msg
->
mid
.
integer_value
()};
auto
bytes
=
to_bytes
(
hdr
);
parent
.
write_packet
(
make_span
(
bytes
),
make_span
(
buf_
));
return
none
;
return
none
;
});
return
write
(
write_packet
,
std
::
move
(
ptr
));
}
}
template
<
class
Parent
>
template
<
class
Parent
>
...
@@ -183,7 +167,12 @@ public:
...
@@ -183,7 +167,12 @@ public:
}
}
private:
private:
// -- message handling -------------------------------------------------------
// -- handling of outgoing messages ------------------------------------------
error
write
(
write_packet_callback
&
write_packet
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
ptr
);
// -- handling of incoming messages ------------------------------------------
error
handle
(
size_t
&
next_read_size
,
write_packet_callback
&
write_packet
,
error
handle
(
size_t
&
next_read_size
,
write_packet_callback
&
write_packet
,
byte_span
bytes
);
byte_span
bytes
);
...
@@ -246,6 +235,10 @@ private:
...
@@ -246,6 +235,10 @@ private:
/// Points to the endpoint manager that owns this applications.
/// Points to the endpoint manager that owns this applications.
endpoint_manager
*
manager_
=
nullptr
;
endpoint_manager
*
manager_
=
nullptr
;
/// Provides pointers to the actor system as well as the registry to
/// serializers and deserializer.
scoped_execution_unit
executor_
;
};
};
}
// namespace basp
}
// namespace basp
...
...
libcaf_net/src/application.cpp
View file @
633d6f92
...
@@ -82,7 +82,7 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
...
@@ -82,7 +82,7 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
string_view
path
,
actor
listener
)
{
string_view
path
,
actor
listener
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
path
)
<<
CAF_ARG
(
listener
));
CAF_LOG_TRACE
(
CAF_ARG
(
path
)
<<
CAF_ARG
(
listener
));
buf_
.
clear
();
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
()
,
buf_
};
serializer_impl
<
buffer_type
>
sink
{
&
executor_
,
buf_
};
if
(
auto
err
=
sink
(
path
))
{
if
(
auto
err
=
sink
(
path
))
{
CAF_LOG_ERROR
(
"unable to serialize path"
);
CAF_LOG_ERROR
(
"unable to serialize path"
);
return
;
return
;
...
@@ -99,6 +99,35 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
...
@@ -99,6 +99,35 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
pending_resolves_
.
emplace
(
req_id
,
std
::
move
(
rp
));
pending_resolves_
.
emplace
(
req_id
,
std
::
move
(
rp
));
}
}
error
application
::
write
(
write_packet_callback
&
write_packet
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
ptr
)
{
CAF_ASSERT
(
ptr
!=
nullptr
);
CAF_ASSERT
(
ptr
->
msg
!=
nullptr
);
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
(),
buf_
};
const
auto
&
src
=
ptr
->
msg
->
sender
;
const
auto
&
dst
=
ptr
->
receiver
;
if
(
dst
==
nullptr
)
{
// TODO: valid?
return
none
;
}
if
(
src
!=
nullptr
)
{
auto
src_id
=
src
->
id
();
system
().
registry
().
put
(
src_id
,
src
);
if
(
auto
err
=
sink
(
src
->
node
(),
src_id
,
dst
->
id
(),
ptr
->
msg
->
stages
))
return
err
;
}
else
{
if
(
auto
err
=
sink
(
node_id
{},
actor_id
{
0
},
dst
->
id
(),
ptr
->
msg
->
stages
))
return
err
;
}
// TODO: avoid extra copy of the payload
buf_
.
insert
(
buf_
.
end
(),
ptr
->
payload
.
begin
(),
ptr
->
payload
.
end
());
header
hdr
{
message_type
::
actor_message
,
static_cast
<
uint32_t
>
(
buf_
.
size
()),
ptr
->
msg
->
mid
.
integer_value
()};
auto
bytes
=
to_bytes
(
hdr
);
return
write_packet
(
make_span
(
bytes
),
make_span
(
buf_
));
}
error
application
::
handle
(
size_t
&
next_read_size
,
error
application
::
handle
(
size_t
&
next_read_size
,
write_packet_callback
&
write_packet
,
write_packet_callback
&
write_packet
,
byte_span
bytes
)
{
byte_span
bytes
)
{
...
@@ -177,7 +206,7 @@ error application::handle_handshake(write_packet_callback&, header hdr,
...
@@ -177,7 +206,7 @@ error application::handle_handshake(write_packet_callback&, header hdr,
return
ec
::
version_mismatch
;
return
ec
::
version_mismatch
;
node_id
peer_id
;
node_id
peer_id
;
std
::
vector
<
std
::
string
>
app_ids
;
std
::
vector
<
std
::
string
>
app_ids
;
binary_deserializer
source
{
system
()
,
payload
};
binary_deserializer
source
{
&
executor_
,
payload
};
if
(
auto
err
=
source
(
peer_id
,
app_ids
))
if
(
auto
err
=
source
(
peer_id
,
app_ids
))
return
err
;
return
err
;
if
(
!
peer_id
||
app_ids
.
empty
())
if
(
!
peer_id
||
app_ids
.
empty
())
...
@@ -203,7 +232,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
...
@@ -203,7 +232,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
actor_id
dst_id
;
actor_id
dst_id
;
std
::
vector
<
strong_actor_ptr
>
fwd_stack
;
std
::
vector
<
strong_actor_ptr
>
fwd_stack
;
message
content
;
message
content
;
binary_deserializer
source
{
system
()
,
payload
};
binary_deserializer
source
{
&
executor_
,
payload
};
if
(
auto
err
=
source
(
src_node
,
src_id
,
dst_id
,
fwd_stack
,
content
))
if
(
auto
err
=
source
(
src_node
,
src_id
,
dst_id
,
fwd_stack
,
content
))
return
err
;
return
err
;
// Sanity checks.
// Sanity checks.
...
@@ -232,7 +261,7 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
...
@@ -232,7 +261,7 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
CAF_LOG_TRACE
(
CAF_ARG
(
hdr
)
<<
CAF_ARG2
(
"payload.size"
,
payload
.
size
()));
CAF_LOG_TRACE
(
CAF_ARG
(
hdr
)
<<
CAF_ARG2
(
"payload.size"
,
payload
.
size
()));
CAF_ASSERT
(
hdr
.
type
==
message_type
::
resolve_request
);
CAF_ASSERT
(
hdr
.
type
==
message_type
::
resolve_request
);
size_t
path_size
=
0
;
size_t
path_size
=
0
;
binary_deserializer
source
{
system
()
,
payload
};
binary_deserializer
source
{
&
executor_
,
payload
};
if
(
auto
err
=
source
.
begin_sequence
(
path_size
))
if
(
auto
err
=
source
.
begin_sequence
(
path_size
))
return
err
;
return
err
;
// We expect the payload to consist only of the path.
// We expect the payload to consist only of the path.
...
@@ -244,10 +273,16 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
...
@@ -244,10 +273,16 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
// Write result.
// Write result.
auto
result
=
resolve_local_path
(
path
);
auto
result
=
resolve_local_path
(
path
);
buf_
.
clear
();
buf_
.
clear
();
actor_id
aid
=
result
?
result
->
id
()
:
0
;
actor_id
aid
;
std
::
set
<
std
::
string
>
ifs
;
std
::
set
<
std
::
string
>
ifs
;
if
(
result
)
{
aid
=
result
->
id
();
system
().
registry
().
put
(
aid
,
result
);
}
else
{
aid
=
0
;
}
// TODO: figure out how to obtain messaging interface.
// TODO: figure out how to obtain messaging interface.
serializer_impl
<
buffer_type
>
sink
{
system
()
,
buf_
};
serializer_impl
<
buffer_type
>
sink
{
&
executor_
,
buf_
};
if
(
auto
err
=
sink
(
aid
,
ifs
))
if
(
auto
err
=
sink
(
aid
,
ifs
))
return
err
;
return
err
;
auto
out_hdr
=
to_bytes
(
header
{
message_type
::
resolve_response
,
auto
out_hdr
=
to_bytes
(
header
{
message_type
::
resolve_response
,
...
@@ -272,7 +307,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
...
@@ -272,7 +307,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
});
});
actor_id
aid
;
actor_id
aid
;
std
::
set
<
std
::
string
>
ifs
;
std
::
set
<
std
::
string
>
ifs
;
binary_deserializer
source
{
system
()
,
payload
};
binary_deserializer
source
{
&
executor_
,
payload
};
if
(
auto
err
=
source
(
aid
,
ifs
))
if
(
auto
err
=
source
(
aid
,
ifs
))
return
err
;
return
err
;
if
(
aid
==
0
)
{
if
(
aid
==
0
)
{
...
@@ -299,7 +334,7 @@ error application::handle_monitor_message(write_packet_callback& write_packet,
...
@@ -299,7 +334,7 @@ error application::handle_monitor_message(write_packet_callback& write_packet,
}
else
{
}
else
{
error
reason
=
exit_reason
::
unknown
;
error
reason
=
exit_reason
::
unknown
;
buf_
.
clear
();
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
()
,
buf_
};
serializer_impl
<
buffer_type
>
sink
{
&
executor_
,
buf_
};
if
(
auto
err
=
sink
(
reason
))
if
(
auto
err
=
sink
(
reason
))
return
err
;
return
err
;
auto
out_hdr
=
to_bytes
(
header
{
message_type
::
down_message
,
auto
out_hdr
=
to_bytes
(
header
{
message_type
::
down_message
,
...
@@ -314,7 +349,7 @@ error application::handle_down_message(write_packet_callback&, header hdr,
...
@@ -314,7 +349,7 @@ error application::handle_down_message(write_packet_callback&, header hdr,
byte_span
payload
)
{
byte_span
payload
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdr
)
<<
CAF_ARG2
(
"payload.size"
,
payload
.
size
()));
CAF_LOG_TRACE
(
CAF_ARG
(
hdr
)
<<
CAF_ARG2
(
"payload.size"
,
payload
.
size
()));
error
reason
;
error
reason
;
binary_deserializer
source
{
system
()
,
payload
};
binary_deserializer
source
{
&
executor_
,
payload
};
if
(
auto
err
=
source
(
reason
))
if
(
auto
err
=
source
(
reason
))
return
err
;
return
err
;
proxies_
.
erase
(
peer_id_
,
hdr
.
operation_data
,
std
::
move
(
reason
));
proxies_
.
erase
(
peer_id_
,
hdr
.
operation_data
,
std
::
move
(
reason
));
...
@@ -323,7 +358,7 @@ error application::handle_down_message(write_packet_callback&, header hdr,
...
@@ -323,7 +358,7 @@ error application::handle_down_message(write_packet_callback&, header hdr,
error
application
::
generate_handshake
()
{
error
application
::
generate_handshake
()
{
buf_
.
clear
();
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
()
,
buf_
};
serializer_impl
<
buffer_type
>
sink
{
&
executor_
,
buf_
};
return
sink
(
system
().
node
(),
return
sink
(
system
().
node
(),
get_or
(
system
().
config
(),
"middleman.app-identifiers"
,
get_or
(
system
().
config
(),
"middleman.app-identifiers"
,
defaults
::
middleman
::
app_identifiers
));
defaults
::
middleman
::
app_identifiers
));
...
...
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