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
d2bb88a6
Unverified
Commit
d2bb88a6
authored
Jan 20, 2020
by
Joseph Noir
Committed by
GitHub
Jan 20, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #1012
Add new select_any policy
parents
35c5d3e1
cd03a6b0
Changes
9
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
374 additions
and
68 deletions
+374
-68
doc/tex/MessagePassing.tex
doc/tex/MessagePassing.tex
+4
-1
examples/message_passing/fan_out_request.cpp
examples/message_passing/fan_out_request.cpp
+3
-3
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+2
-1
libcaf_core/caf/policy/select_all.hpp
libcaf_core/caf/policy/select_all.hpp
+31
-47
libcaf_core/caf/policy/select_any.hpp
libcaf_core/caf/policy/select_any.hpp
+141
-0
libcaf_core/caf/sec.hpp
libcaf_core/caf/sec.hpp
+2
-0
libcaf_core/test/mixin/requester.cpp
libcaf_core/test/mixin/requester.cpp
+3
-3
libcaf_core/test/policy/select_all.cpp
libcaf_core/test/policy/select_all.cpp
+13
-13
libcaf_core/test/policy/select_any.cpp
libcaf_core/test/policy/select_any.cpp
+175
-0
No files found.
doc/tex/MessagePassing.tex
View file @
d2bb88a6
...
...
@@ -192,7 +192,7 @@ Sending the same message to a group of workers is a common work flow in actor
applications. Usually, a manager maintains a set of workers. On request, the
manager fans-out the request to all of its workers and then collects the
results. The function
\lstinline
`fan
_
out
_
request` combined with the merge policy
\lstinline
`
fan
_
in
_
responses
` streamlines this exact use case.
\lstinline
`
select
_
all
` streamlines this exact use case.
In the following snippet, we have a matrix actor (
\lstinline
`self`) that stores
worker actors for each cell (each simply storing an integer). For computing the
...
...
@@ -204,6 +204,9 @@ results.
\cppexample
[86-98]
{
message
_
passing/fan
_
out
_
request
}
The policy
\lstinline
`select
_
any` models a second common use case: sending a
request to multiple receivers but only caring for the first arriving response.
\clearpage
\subsubsection
{
Error Handling in Requests
}
\label
{
error-response
}
...
...
examples/message_passing/fan_out_request.cpp
View file @
d2bb88a6
...
...
@@ -14,7 +14,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/function_view.hpp"
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_actor.hpp"
...
...
@@ -87,7 +87,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
assert
(
row
<
rows
);
auto
rp
=
self
->
make_response_promise
<
double
>
();
auto
&
row_vec
=
self
->
state
.
rows
[
row
];
self
->
fan_out_request
<
policy
::
fan_in_responses
>
(
row_vec
,
infinite
,
get
)
self
->
fan_out_request
<
policy
::
select_all
>
(
row_vec
,
infinite
,
get
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
columns
));
...
...
@@ -104,7 +104,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
columns
.
emplace_back
(
rows_vec
[
row
][
column
]);
auto
rp
=
self
->
make_response_promise
<
double
>
();
self
->
fan_out_request
<
policy
::
fan_in_responses
>
(
columns
,
infinite
,
get
)
self
->
fan_out_request
<
policy
::
select_all
>
(
columns
,
infinite
,
get
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
rows
));
...
...
libcaf_core/CMakeLists.txt
View file @
d2bb88a6
...
...
@@ -251,7 +251,8 @@ set(CAF_CORE_TEST_SOURCES
test/or_else.cpp
test/pipeline_streaming.cpp
test/policy/categorized.cpp
test/policy/fan_in_responses.cpp
test/policy/select_all.cpp
test/policy/select_any.cpp
test/request_timeout.cpp
test/result.cpp
test/rtti_pair.cpp
...
...
libcaf_core/caf/policy/
fan_in_responses
.hpp
→
libcaf_core/caf/policy/
select_all
.hpp
View file @
d2bb88a6
...
...
@@ -36,7 +36,7 @@
namespace
caf
::
detail
{
template
<
class
F
,
class
T
>
struct
fan_in_responses
_helper
{
struct
select_all
_helper
{
std
::
vector
<
T
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
F
f
;
...
...
@@ -51,7 +51,7 @@ struct fan_in_responses_helper {
}
template
<
class
Fun
>
fan_in_responses
_helper
(
size_t
pending
,
Fun
&&
f
)
select_all
_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
...
...
@@ -62,7 +62,7 @@ struct fan_in_responses_helper {
};
template
<
class
F
,
class
...
Ts
>
struct
fan_in_responses
_tuple_helper
{
struct
select_all
_tuple_helper
{
using
value_type
=
std
::
tuple
<
Ts
...
>
;
std
::
vector
<
value_type
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
...
...
@@ -78,7 +78,7 @@ struct fan_in_responses_tuple_helper {
}
template
<
class
Fun
>
fan_in_responses
_tuple_helper
(
size_t
pending
,
Fun
&&
f
)
select_all
_tuple_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
...
...
@@ -89,56 +89,32 @@ struct fan_in_responses_tuple_helper {
};
template
<
class
F
,
class
=
typename
detail
::
get_callable_trait
<
F
>
::
arg_types
>
struct
select_
fan_in_responses
_helper
;
struct
select_
select_all
_helper
;
template
<
class
F
,
class
...
Ts
>
struct
select_
fan_in_responses
_helper
<
struct
select_
select_all
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
std
::
tuple
<
Ts
...
>>>>
{
using
type
=
fan_in_responses
_tuple_helper
<
F
,
Ts
...
>
;
using
type
=
select_all
_tuple_helper
<
F
,
Ts
...
>
;
};
template
<
class
F
,
class
T
>
struct
select_
fan_in_responses
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
T
>>>
{
using
type
=
fan_in_responses
_helper
<
F
,
T
>
;
struct
select_
select_all
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
T
>>>
{
using
type
=
select_all
_helper
<
F
,
T
>
;
};
template
<
class
F
>
using
fan_in_responses_helper_t
=
typename
select_fan_in_responses_helper
<
F
>::
type
;
// TODO: Replace with a lambda when switching to C++17 (move g into lambda).
template
<
class
G
>
class
fan_in_responses_error_handler
{
public:
template
<
class
Fun
>
fan_in_responses_error_handler
(
Fun
&&
fun
,
std
::
shared_ptr
<
size_t
>
pending
)
:
handler
(
std
::
forward
<
Fun
>
(
fun
)),
pending
(
std
::
move
(
pending
))
{
// nop
}
void
operator
()(
error
&
err
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"pending"
,
*
pending
));
if
(
*
pending
>
0
)
{
*
pending
=
0
;
handler
(
err
);
}
}
private:
G
handler
;
std
::
shared_ptr
<
size_t
>
pending
;
};
using
select_all_helper_t
=
typename
select_select_all_helper
<
F
>::
type
;
}
// namespace caf::detail
namespace
caf
::
policy
{
/// Enables a `response_handle` to fan-in
multiple respons
es into a single
/// result (a `vector`
of individual values) for the client
.
/// Enables a `response_handle` to fan-in
all responses messag
es into a single
/// result (a `vector`
that stores all received results)
.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
fan_in_responses
{
class
select_all
{
public:
static
constexpr
bool
is_trivial
=
false
;
...
...
@@ -147,17 +123,18 @@ public:
using
message_id_list
=
std
::
vector
<
message_id
>
;
template
<
class
Fun
>
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
fan_in_responses_helper_t
<
detail
::
decay_t
<
Fun
>>>
;
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
select_all_helper_t
<
detail
::
decay_t
<
Fun
>>>
;
explicit
fan_in_responses
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
explicit
select_all
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
CAF_ASSERT
(
ids_
.
size
()
<=
static_cast
<
size_t
>
(
std
::
numeric_limits
<
int
>::
max
()));
}
fan_in_responses
(
fan_in_responses
&&
)
noexcept
=
default
;
select_all
(
select_all
&&
)
noexcept
=
default
;
fan_in_responses
&
operator
=
(
fan_in_responses
&&
)
noexcept
=
default
;
select_all
&
operator
=
(
select_all
&&
)
noexcept
=
default
;
template
<
class
Self
,
class
F
,
class
OnError
>
void
await
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
...
...
@@ -178,7 +155,7 @@ public:
template
<
class
Self
,
class
F
,
class
G
>
void
receive
(
Self
*
self
,
F
&&
f
,
G
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
using
helper_type
=
detail
::
fan_in_responses
_helper_t
<
detail
::
decay_t
<
F
>>
;
using
helper_type
=
detail
::
select_all
_helper_t
<
detail
::
decay_t
<
F
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
forward
<
F
>
(
f
)};
auto
error_handler
=
[
&
](
error
&
err
)
{
if
(
*
helper
.
pending
>
0
)
{
...
...
@@ -202,13 +179,20 @@ private:
template
<
class
F
,
class
OnError
>
behavior
make_behavior
(
F
&&
f
,
OnError
&&
g
)
const
{
using
namespace
detail
;
using
helper_type
=
fan_in_responses_helper_t
<
decay_t
<
F
>>
;
using
error_handler_type
=
fan_in_responses_error_handler
<
decay_t
<
OnError
>>
;
using
helper_type
=
select_all_helper_t
<
decay_t
<
F
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
move
(
f
)};
error_handler_type
err_helper
{
std
::
forward
<
OnError
>
(
g
),
helper
.
pending
};
auto
pending
=
helper
.
pending
;
auto
error_handler
=
[
pending
{
std
::
move
(
pending
)},
g
{
std
::
forward
<
OnError
>
(
g
)}](
error
&
err
)
mutable
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"pending"
,
*
pending
));
if
(
*
pending
>
0
)
{
*
pending
=
0
;
g
(
err
);
}
};
return
{
std
::
move
(
helper
),
std
::
move
(
err
_help
er
),
std
::
move
(
err
or_handl
er
),
};
}
...
...
libcaf_core/caf/policy/select_any.hpp
0 → 100644
View file @
d2bb88a6
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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 <memory>
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
namespace
caf
::
detail
{
template
<
class
F
,
class
=
typename
get_callable_trait
<
F
>
::
arg_types
>
struct
select_any_factory
;
template
<
class
F
,
class
...
Ts
>
struct
select_any_factory
<
F
,
type_list
<
Ts
...
>>
{
template
<
class
Fun
>
static
auto
make
(
std
::
shared_ptr
<
size_t
>
pending
,
Fun
&&
fun
)
{
return
[
pending
,
f
{
std
::
forward
<
Fun
>
(
fun
)}](
Ts
...
xs
)
mutable
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"pending"
,
*
pending
));
if
(
*
pending
>
0
)
{
f
(
xs
...);
*
pending
=
0
;
}
};
}
};
}
// namespace caf::detail
namespace
caf
::
policy
{
/// Enables a `response_handle` to pick the first arriving response, ignoring
/// all other results.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
select_any
{
public:
static
constexpr
bool
is_trivial
=
false
;
using
response_type
=
ResponseType
;
using
message_id_list
=
std
::
vector
<
message_id
>
;
template
<
class
Fun
>
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
decay_t
<
Fun
>>
;
explicit
select_any
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
CAF_ASSERT
(
ids_
.
size
()
<=
static_cast
<
size_t
>
(
std
::
numeric_limits
<
int
>::
max
()));
}
template
<
class
Self
,
class
F
,
class
OnError
>
void
await
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
auto
bhvr
=
make_behavior
(
std
::
forward
<
F
>
(
f
),
std
::
forward
<
OnError
>
(
g
));
for
(
auto
id
:
ids_
)
self
->
add_awaited_response_handler
(
id
,
bhvr
);
}
template
<
class
Self
,
class
F
,
class
OnError
>
void
then
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
auto
bhvr
=
make_behavior
(
std
::
forward
<
F
>
(
f
),
std
::
forward
<
OnError
>
(
g
));
for
(
auto
id
:
ids_
)
self
->
add_multiplexed_response_handler
(
id
,
bhvr
);
}
template
<
class
Self
,
class
F
,
class
G
>
void
receive
(
Self
*
self
,
F
&&
f
,
G
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
using
factory
=
detail
::
select_any_factory
<
std
::
decay_t
<
F
>>
;
auto
pending
=
std
::
make_shared
<
size_t
>
(
ids_
.
size
());
auto
fw
=
factory
::
make
(
pending
,
std
::
forward
<
F
>
(
f
));
auto
gw
=
make_error_handler
(
std
::
move
(
pending
),
std
::
forward
<
G
>
(
g
));
for
(
auto
id
:
ids_
)
{
typename
Self
::
accept_one_cond
rc
;
auto
fcopy
=
fw
;
auto
gcopy
=
gw
;
self
->
varargs_receive
(
rc
,
id
,
fcopy
,
gcopy
);
}
}
const
message_id_list
&
ids
()
const
noexcept
{
return
ids_
;
}
private:
template
<
class
OnError
>
auto
make_error_handler
(
std
::
shared_ptr
<
size_t
>
p
,
OnError
&&
g
)
const
{
return
[
p
{
std
::
move
(
p
)},
g
{
std
::
forward
<
OnError
>
(
g
)}](
error
&
)
mutable
{
if
(
*
p
==
0
)
{
// nop
}
else
if
(
*
p
==
1
)
{
auto
err
=
make_error
(
sec
::
all_requests_failed
);
g
(
err
);
}
else
{
--*
p
;
}
};
}
template
<
class
F
,
class
OnError
>
behavior
make_behavior
(
F
&&
f
,
OnError
&&
g
)
const
{
using
factory
=
detail
::
select_any_factory
<
std
::
decay_t
<
F
>>
;
auto
pending
=
std
::
make_shared
<
size_t
>
(
ids_
.
size
());
auto
result_handler
=
factory
::
make
(
pending
,
std
::
forward
<
F
>
(
f
));
return
{
std
::
move
(
result_handler
),
make_error_handler
(
std
::
move
(
pending
),
std
::
forward
<
OnError
>
(
g
)),
};
}
message_id_list
ids_
;
};
}
// namespace caf::policy
libcaf_core/caf/sec.hpp
View file @
d2bb88a6
...
...
@@ -130,6 +130,8 @@ enum class sec : uint8_t {
remote_lookup_failed
,
/// Serialization failed because actor_system::tracing_context is null.
no_tracing_context
,
/// No request produced a valid result.
all_requests_failed
,
};
/// @relates sec
...
...
libcaf_core/test/mixin/requester.cpp
View file @
d2bb88a6
...
...
@@ -23,7 +23,7 @@
#include "caf/test/dsl.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
using
namespace
caf
;
...
...
@@ -152,7 +152,7 @@ CAF_TEST(delegated request with integer result) {
}
CAF_TEST
(
requesters
support
fan_out_request
)
{
using
policy
::
fan_in_responses
;
using
policy
::
select_all
;
std
::
vector
<
adding_server_type
>
workers
{
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
...
...
@@ -161,7 +161,7 @@ CAF_TEST(requesters support fan_out_request) {
run
();
auto
sum
=
std
::
make_shared
<
int
>
(
0
);
auto
client
=
sys
.
spawn
([
=
](
event_based_actor
*
self
)
{
self
->
fan_out_request
<
fan_in_responses
>
(
workers
,
infinite
,
1
,
2
)
self
->
fan_out_request
<
select_all
>
(
workers
,
infinite
,
1
,
2
)
.
then
([
=
](
std
::
vector
<
int
>
results
)
{
for
(
auto
result
:
results
)
CAF_CHECK_EQUAL
(
result
,
3
);
...
...
libcaf_core/test/policy/
fan_in_responses
.cpp
→
libcaf_core/test/policy/
select_all
.cpp
View file @
d2bb88a6
...
...
@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE policy.
fan_in_responses
#define CAF_SUITE policy.
select_all
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
#include "caf/test/dsl.hpp"
...
...
@@ -28,7 +28,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using
caf
::
policy
::
fan_in_responses
;
using
caf
::
policy
::
select_all
;
using
namespace
caf
;
...
...
@@ -63,9 +63,9 @@ struct fixture : test_coordinator_fixture<> {
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE
(
fan_in_responses
_tests
,
fixture
)
CAF_TEST_FIXTURE_SCOPE
(
select_all
_tests
,
fixture
)
CAF_TEST
(
fan_in_responses
combines
two
integer
results
into
one
vector
)
{
CAF_TEST
(
select_all
combines
two
integer
results
into
one
vector
)
{
using
int_list
=
std
::
vector
<
int
>
;
auto
f
=
[](
int
x
,
int
y
)
{
return
x
+
y
;
};
auto
server1
=
make_server
(
f
);
...
...
@@ -74,7 +74,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
SUBTEST
(
"vector of int"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
merge
.
receive
(
self
.
ptr
(),
...
...
@@ -88,7 +88,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
using
std
::
make_tuple
;
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
using
results_vector
=
std
::
vector
<
std
::
tuple
<
int
>>
;
merge
.
receive
(
...
...
@@ -106,7 +106,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto
client
=
sys
.
spawn
([
=
,
&
results
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
then
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
make_error_handler
());
...
...
@@ -124,7 +124,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto
client
=
sys
.
spawn
([
=
,
&
results
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
await
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
make_error_handler
());
...
...
@@ -141,7 +141,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
}
}
CAF_TEST
(
fan_in_responses
calls
the
error
handler
at
most
once
)
{
CAF_TEST
(
select_all
calls
the
error
handler
at
most
once
)
{
using
int_list
=
std
::
vector
<
int
>
;
auto
f
=
[](
int
,
int
)
->
result
<
int
>
{
return
sec
::
invalid_argument
;
};
auto
server1
=
make_server
(
f
);
...
...
@@ -149,7 +149,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
SUBTEST
(
"request.receive"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
size_t
errors
=
0
;
merge
.
receive
(
...
...
@@ -163,7 +163,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
then
(
client_ptr
,
[](
int_list
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
@@ -181,7 +181,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
await
(
client_ptr
,
[](
int_list
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
libcaf_core/test/policy/select_any.cpp
0 → 100644
View file @
d2bb88a6
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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 policy.select_any
#include "caf/policy/select_any.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using
caf
::
policy
::
select_any
;
using
namespace
caf
;
namespace
{
struct
fixture
:
test_coordinator_fixture
<>
{
template
<
class
F
>
actor
make_server
(
F
f
)
{
auto
init
=
[
f
]()
->
behavior
{
return
{
[
f
](
int
x
,
int
y
)
{
return
f
(
x
,
y
);
},
};
};
return
sys
.
spawn
(
init
);
}
auto
make_error_handler
()
{
return
[
this
](
const
error
&
err
)
{
CAF_FAIL
(
"unexpected error: "
<<
sys
.
render
(
err
));
};
}
auto
make_counting_error_handler
(
size_t
*
count
)
{
return
[
count
](
const
error
&
)
{
*
count
+=
1
;
};
}
};
}
// namespace
#define SUBTEST(message) \
run(); \
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE
(
select_any_tests
,
fixture
)
CAF_TEST
(
select_any
picks
the
first
arriving
integer
)
{
auto
f
=
[](
int
x
,
int
y
)
{
return
x
+
y
;
};
auto
server1
=
make_server
(
f
);
auto
server2
=
make_server
(
f
);
SUBTEST
(
"request.receive"
)
{
SUBTEST
(
"single integer"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
run
();
choose
.
receive
(
self
.
ptr
(),
[](
int
result
)
{
CAF_CHECK_EQUAL
(
result
,
3
);
},
make_error_handler
());
}
}
SUBTEST
(
"request.then"
)
{
int
result
=
0
;
auto
client
=
sys
.
spawn
([
=
,
&
result
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
then
(
client_ptr
,
[
&
result
](
int
x
)
{
result
=
x
;
},
make_error_handler
());
});
run_once
();
expect
((
int
,
int
),
from
(
client
).
to
(
server1
).
with
(
1
,
2
));
expect
((
int
,
int
),
from
(
client
).
to
(
server2
).
with
(
2
,
3
));
expect
((
int
),
from
(
server1
).
to
(
client
).
with
(
3
));
expect
((
int
),
from
(
server2
).
to
(
client
).
with
(
5
));
CAF_MESSAGE
(
"request.then picks the first arriving result"
);
CAF_CHECK_EQUAL
(
result
,
3
);
}
SUBTEST
(
"request.await"
)
{
int
result
=
0
;
auto
client
=
sys
.
spawn
([
=
,
&
result
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
await
(
client_ptr
,
[
&
result
](
int
x
)
{
result
=
x
;
},
make_error_handler
());
});
run_once
();
expect
((
int
,
int
),
from
(
client
).
to
(
server1
).
with
(
1
,
2
));
expect
((
int
,
int
),
from
(
client
).
to
(
server2
).
with
(
2
,
3
));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run
();
CAF_MESSAGE
(
"request.await froces responses into reverse request order"
);
CAF_CHECK_EQUAL
(
result
,
5
);
}
}
CAF_TEST
(
select_any
calls
the
error
handler
at
most
once
)
{
auto
f
=
[](
int
,
int
)
->
result
<
int
>
{
return
sec
::
invalid_argument
;
};
auto
server1
=
make_server
(
f
);
auto
server2
=
make_server
(
f
);
SUBTEST
(
"request.receive"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
run
();
size_t
errors
=
0
;
choose
.
receive
(
self
.
ptr
(),
[](
int
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
make_counting_error_handler
(
&
errors
));
CAF_CHECK_EQUAL
(
errors
,
1u
);
}
SUBTEST
(
"request.then"
)
{
size_t
errors
=
0
;
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
then
(
client_ptr
,
[](
int
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
make_counting_error_handler
(
&
errors
));
});
run_once
();
expect
((
int
,
int
),
from
(
client
).
to
(
server1
).
with
(
1
,
2
));
expect
((
int
,
int
),
from
(
client
).
to
(
server2
).
with
(
2
,
3
));
expect
((
error
),
from
(
server1
).
to
(
client
).
with
(
sec
::
invalid_argument
));
expect
((
error
),
from
(
server2
).
to
(
client
).
with
(
sec
::
invalid_argument
));
CAF_CHECK_EQUAL
(
errors
,
1u
);
}
SUBTEST
(
"request.await"
)
{
size_t
errors
=
0
;
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
await
(
client_ptr
,
[](
int
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
make_counting_error_handler
(
&
errors
));
});
run_once
();
expect
((
int
,
int
),
from
(
client
).
to
(
server1
).
with
(
1
,
2
));
expect
((
int
,
int
),
from
(
client
).
to
(
server2
).
with
(
2
,
3
));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run
();
CAF_CHECK_EQUAL
(
errors
,
1u
);
}
}
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