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
ddef551c
Unverified
Commit
ddef551c
authored
Nov 14, 2019
by
Joseph Noir
Committed by
GitHub
Nov 14, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #964
Implement new fan_out_request function
parents
b91102ed
09da4942
Changes
12
Hide whitespace changes
Inline
Side-by-side
Showing
12 changed files
with
884 additions
and
144 deletions
+884
-144
doc/tex/MessagePassing.tex
doc/tex/MessagePassing.tex
+18
-0
examples/CMakeLists.txt
examples/CMakeLists.txt
+1
-0
examples/message_passing/fan_out_request.cpp
examples/message_passing/fan_out_request.cpp
+156
-0
libcaf_core/caf/blocking_actor.hpp
libcaf_core/caf/blocking_actor.hpp
+0
-9
libcaf_core/caf/mixin/requester.hpp
libcaf_core/caf/mixin/requester.hpp
+87
-38
libcaf_core/caf/policy/fan_in_responses.hpp
libcaf_core/caf/policy/fan_in_responses.hpp
+221
-0
libcaf_core/caf/policy/single_response.hpp
libcaf_core/caf/policy/single_response.hpp
+81
-0
libcaf_core/caf/response_handle.hpp
libcaf_core/caf/response_handle.hpp
+84
-90
libcaf_core/caf/typed_actor.hpp
libcaf_core/caf/typed_actor.hpp
+3
-3
libcaf_core/test/mixin/requester.cpp
libcaf_core/test/mixin/requester.cpp
+28
-0
libcaf_core/test/policy/fan_in_responses.cpp
libcaf_core/test/policy/fan_in_responses.cpp
+201
-0
libcaf_test/caf/test/dsl.hpp
libcaf_test/caf/test/dsl.hpp
+4
-4
No files found.
doc/tex/MessagePassing.tex
View file @
ddef551c
...
@@ -186,6 +186,24 @@ Both event-based approaches send all requests, install a series of one-shot
...
@@ -186,6 +186,24 @@ Both event-based approaches send all requests, install a series of one-shot
handlers, and then return from the implementing function. In contrast, the
handlers, and then return from the implementing function. In contrast, the
blocking function waits for a response before sending another request.
blocking function waits for a response before sending another request.
\subsubsection
{
Sending Multiple Requests
}
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.
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
average over a row, we use
\lstinline
`fan
_
out
_
request`. The result handler
passed to
\lstinline
`then` now gets called only once with a
\lstinline
`vector`
holding all collected results. Using a response promise
\see
{
promise
}
further
allows us to delay responding to the client until we have collected all worker
results.
\cppexample
[86-98]
{
message
_
passing/fan
_
out
_
request
}
\clearpage
\clearpage
\subsubsection
{
Error Handling in Requests
}
\subsubsection
{
Error Handling in Requests
}
\label
{
error-response
}
\label
{
error-response
}
...
...
examples/CMakeLists.txt
View file @
ddef551c
...
@@ -24,6 +24,7 @@ add(message_passing cell)
...
@@ -24,6 +24,7 @@ add(message_passing cell)
add
(
message_passing dancing_kirby
)
add
(
message_passing dancing_kirby
)
add
(
message_passing delegating
)
add
(
message_passing delegating
)
add
(
message_passing divider
)
add
(
message_passing divider
)
add
(
message_passing fan_out_request
)
add
(
message_passing fixed_stack
)
add
(
message_passing fixed_stack
)
add
(
message_passing promises
)
add
(
message_passing promises
)
add
(
message_passing request
)
add
(
message_passing request
)
...
...
examples/message_passing/fan_out_request.cpp
0 → 100644
View file @
ddef551c
// This file is partially included in the manual, do not modify
// without updating the references in the *.tex files!
// Manual references: lines 86-98 (MessagePassing.tex)
#include <cassert>
#include <chrono>
#include <cmath>
#include <iomanip>
#include <iostream>
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#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/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_event_based_actor.hpp"
using
std
::
endl
;
using
std
::
chrono
::
seconds
;
using
namespace
caf
;
using
row_atom
=
atom_constant
<
atom
(
"row"
)
>
;
using
column_atom
=
atom_constant
<
atom
(
"column"
)
>
;
using
average_atom
=
atom_constant
<
atom
(
"average"
)
>
;
/// A simple actor for storing an integer value.
using
cell
=
typed_actor
<
// Writes a new value.
reacts_to
<
put_atom
,
int
>
,
// Reads the value.
replies_to
<
get_atom
>::
with
<
int
>>
;
/// An for storing a 2-dimensional matrix of integers.
using
matrix
=
typed_actor
<
// Writes a new value to given cell (x-coordinate, y-coordinate, new-value).
reacts_to
<
put_atom
,
int
,
int
,
int
>
,
// Reads from given cell.
replies_to
<
get_atom
,
int
,
int
>::
with
<
int
>
,
// Computes the average for given row.
replies_to
<
get_atom
,
average_atom
,
row_atom
,
int
>::
with
<
double
>
,
// Computes the average for given column.
replies_to
<
get_atom
,
average_atom
,
column_atom
,
int
>::
with
<
double
>>
;
struct
cell_state
{
int
value
=
0
;
static
constexpr
const
char
*
name
=
"cell"
;
};
cell
::
behavior_type
cell_actor
(
cell
::
stateful_pointer
<
cell_state
>
self
)
{
return
{
[
=
](
put_atom
,
int
val
)
{
self
->
state
.
value
=
val
;
},
[
=
](
get_atom
)
{
return
self
->
state
.
value
;
},
};
}
struct
matrix_state
{
using
row_type
=
std
::
vector
<
cell
>
;
std
::
vector
<
row_type
>
rows
;
static
constexpr
const
char
*
name
=
"matrix"
;
};
matrix
::
behavior_type
matrix_actor
(
matrix
::
stateful_pointer
<
matrix_state
>
self
,
int
rows
,
int
columns
)
{
// Spawn all cells and return our behavior.
self
->
state
.
rows
.
resize
(
rows
);
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
{
auto
&
row_vec
=
self
->
state
.
rows
[
row
];
row_vec
.
resize
(
columns
);
for
(
int
column
=
0
;
column
<
columns
;
++
column
)
row_vec
[
column
]
=
self
->
spawn
(
cell_actor
);
}
return
{
[
=
](
put_atom
put
,
int
row
,
int
column
,
int
val
)
{
assert
(
row
<
rows
&&
column
<
columns
);
return
self
->
delegate
(
self
->
state
.
rows
[
row
][
column
],
put
,
val
);
},
[
=
](
get_atom
get
,
int
row
,
int
column
)
{
assert
(
row
<
rows
&&
column
<
columns
);
return
self
->
delegate
(
self
->
state
.
rows
[
row
][
column
],
get
);
},
[
=
](
get_atom
get
,
average_atom
,
row_atom
,
int
row
)
{
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
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
columns
));
rp
.
deliver
(
std
::
accumulate
(
xs
.
begin
(),
xs
.
end
(),
0.0
)
/
columns
);
},
[
=
](
error
&
err
)
mutable
{
rp
.
deliver
(
std
::
move
(
err
));
});
return
rp
;
},
[
=
](
get_atom
get
,
average_atom
,
column_atom
,
int
column
)
{
assert
(
column
<
columns
);
std
::
vector
<
cell
>
columns
;
columns
.
reserve
(
rows
);
auto
&
rows_vec
=
self
->
state
.
rows
;
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
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
rows
));
rp
.
deliver
(
std
::
accumulate
(
xs
.
begin
(),
xs
.
end
(),
0.0
)
/
rows
);
},
[
=
](
error
&
err
)
mutable
{
rp
.
deliver
(
std
::
move
(
err
));
});
return
rp
;
},
};
}
std
::
ostream
&
operator
<<
(
std
::
ostream
&
out
,
const
expected
<
int
>&
x
)
{
if
(
x
)
return
out
<<
*
x
;
return
out
<<
to_string
(
x
.
error
());
}
void
caf_main
(
actor_system
&
sys
)
{
// Spawn our matrix.
static
constexpr
int
rows
=
3
;
static
constexpr
int
columns
=
6
;
auto
mx
=
sys
.
spawn
(
matrix_actor
,
rows
,
columns
);
auto
f
=
make_function_view
(
mx
);
// Set cells in our matrix to these values:
// 2 4 8 16 32 64
// 3 9 27 81 243 729
// 4 16 64 256 1024 4096
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
for
(
int
column
=
0
;
column
<
columns
;
++
column
)
f
(
put_atom
::
value
,
row
,
column
,
(
int
)
pow
(
row
+
2
,
column
+
1
));
// Print out matrix.
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
{
for
(
int
column
=
0
;
column
<
columns
;
++
column
)
std
::
cout
<<
std
::
setw
(
4
)
<<
f
(
get_atom
::
value
,
row
,
column
)
<<
' '
;
std
::
cout
<<
'\n'
;
}
// Print out AVG for each row and column.
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
std
::
cout
<<
"AVG(row "
<<
row
<<
") = "
<<
f
(
get_atom
::
value
,
average_atom
::
value
,
row_atom
::
value
,
row
)
<<
'\n'
;
for
(
int
column
=
0
;
column
<
columns
;
++
column
)
std
::
cout
<<
"AVG(column "
<<
column
<<
") = "
<<
f
(
get_atom
::
value
,
average_atom
::
value
,
column_atom
::
value
,
column
)
<<
'\n'
;
}
CAF_MAIN
()
libcaf_core/caf/blocking_actor.hpp
View file @
ddef551c
...
@@ -57,15 +57,6 @@
...
@@ -57,15 +57,6 @@
#include "caf/mixin/sender.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp"
#include "caf/mixin/subscriber.hpp"
namespace
caf
{
namespace
mixin
{
template
<
>
struct
is_blocking_requester
<
blocking_actor
>
:
std
::
true_type
{
};
}
// namespace caf
}
// namespace mixin
namespace
caf
{
namespace
caf
{
/// A thread-mapped or context-switching actor using a blocking
/// A thread-mapped or context-switching actor using a blocking
...
...
libcaf_core/caf/mixin/requester.hpp
View file @
ddef551c
...
@@ -18,8 +18,9 @@
...
@@ -18,8 +18,9 @@
#pragma once
#pragma once
#include <tuple>
#include <chrono>
#include <chrono>
#include <tuple>
#include <vector>
#include "caf/actor.hpp"
#include "caf/actor.hpp"
#include "caf/check_typed_input.hpp"
#include "caf/check_typed_input.hpp"
...
@@ -29,15 +30,13 @@
...
@@ -29,15 +30,13 @@
#include "caf/message.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/message_id.hpp"
#include "caf/message_priority.hpp"
#include "caf/message_priority.hpp"
#include "caf/policy/single_response.hpp"
#include "caf/response_handle.hpp"
#include "caf/response_handle.hpp"
#include "caf/response_type.hpp"
#include "caf/response_type.hpp"
namespace
caf
{
namespace
caf
{
namespace
mixin
{
namespace
mixin
{
template
<
class
T
>
struct
is_blocking_requester
:
std
::
false_type
{
};
/// A `requester` is an actor that supports
/// A `requester` is an actor that supports
/// `self->request(...).{then|await|receive}`.
/// `self->request(...).{then|await|receive}`.
template
<
class
Base
,
class
Subtype
>
template
<
class
Base
,
class
Subtype
>
...
@@ -60,22 +59,12 @@ public:
...
@@ -60,22 +59,12 @@ public:
/// @returns A handle identifying a future-like handle to the response.
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
/// sent message cannot be received by another actor.
template
<
message_priority
P
=
message_priority
::
normal
,
template
<
message_priority
P
=
message_priority
::
normal
,
class
Handle
=
actor
,
class
Handle
=
actor
,
class
...
Ts
>
class
...
Ts
>
response_handle
<
Subtype
,
auto
request
(
const
Handle
&
dest
,
const
duration
&
timeout
,
Ts
&&
...
xs
)
{
response_type_t
<
using
namespace
detail
;
typename
Handle
::
signatures
,
typename
detail
::
implicit_conversions
<
typename
std
::
decay
<
Ts
>::
type
>::
type
...
>
,
is_blocking_requester
<
Subtype
>::
value
>
request
(
const
Handle
&
dest
,
const
duration
&
timeout
,
Ts
&&
...
xs
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
using
token
=
using
token
=
type_list
<
implicit_conversions_t
<
decay_t
<
Ts
>>
...
>
;
detail
::
type_list
<
typename
detail
::
implicit_conversions
<
typename
std
::
decay
<
Ts
>::
type
>::
type
...
>
;
static_assert
(
response_type_unbox
<
signatures_of_t
<
Handle
>
,
token
>::
valid
,
static_assert
(
response_type_unbox
<
signatures_of_t
<
Handle
>
,
token
>::
valid
,
"receiver does not accept given message"
);
"receiver does not accept given message"
);
auto
self
=
static_cast
<
Subtype
*>
(
this
);
auto
self
=
static_cast
<
Subtype
*>
(
this
);
...
@@ -88,29 +77,89 @@ public:
...
@@ -88,29 +77,89 @@ public:
self
->
eq_impl
(
req_id
.
response_id
(),
self
->
ctrl
(),
self
->
context
(),
self
->
eq_impl
(
req_id
.
response_id
(),
self
->
ctrl
(),
self
->
context
(),
make_error
(
sec
::
invalid_argument
));
make_error
(
sec
::
invalid_argument
));
}
}
return
{
req_id
.
response_id
(),
self
};
using
response_type
}
=
response_type_t
<
typename
Handle
::
signatures
,
detail
::
implicit_conversions_t
<
detail
::
decay_t
<
Ts
>>
...
>
;
using
handle_type
=
response_handle
<
Subtype
,
policy
::
single_response
<
response_type
>>
;
return
handle_type
{
self
,
req_id
.
response_id
()};
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @copydoc request
/// @returns A handle identifying a future-like handle to the response.
template
<
message_priority
P
=
message_priority
::
normal
,
class
Rep
=
int
,
/// @warning The returned handle is actor specific and the response to the
class
Period
=
std
::
ratio
<
1
>,
class
Handle
=
actor
,
class
...
Ts
>
/// sent message cannot be received by another actor.
auto
request
(
const
Handle
&
dest
,
std
::
chrono
::
duration
<
Rep
,
Period
>
timeout
,
template
<
message_priority
P
=
message_priority
::
normal
,
Ts
&&
...
xs
)
{
class
Rep
=
int
,
class
Period
=
std
::
ratio
<
1
>,
class
Handle
=
actor
,
class
...
Ts
>
response_handle
<
Subtype
,
response_type_t
<
typename
Handle
::
signatures
,
typename
detail
::
implicit_conversions
<
typename
std
::
decay
<
Ts
>::
type
>::
type
...
>
,
is_blocking_requester
<
Subtype
>::
value
>
request
(
const
Handle
&
dest
,
std
::
chrono
::
duration
<
Rep
,
Period
>
timeout
,
Ts
&&
...
xs
)
{
return
request
(
dest
,
duration
{
timeout
},
std
::
forward
<
Ts
>
(
xs
)...);
return
request
(
dest
,
duration
{
timeout
},
std
::
forward
<
Ts
>
(
xs
)...);
}
}
/// Sends `{xs...}` to each actor in the range `destinations` as a synchronous
/// message. Response messages get combined into a single result according to
/// the `MergePolicy`.
/// @tparam MergePolicy Configures how individual response messages get
/// combined by the actor. The policy makes sure that the
/// response handler gets invoked at most once. In case of
/// one or more errors, the policy calls the error handler
/// exactly once, with the first error occurred.
/// @tparam Prio Specifies the priority of the synchronous messages.
/// @tparam Container A container type for holding actor handles. Must provide
/// the type alias `value_type` as well as the member
/// functions `begin()`, `end()`, and `size()`.
/// @param destinations A container holding handles to all destination actors.
/// @param timeout Maximum duration before dropping the request. The runtime
/// system will send an error message to the actor in case the
/// receiver does not respond in time.
/// @returns A helper object that takes response handlers via `.await()`,
/// `.then()`, or `.receive()`.
/// @note The returned handle is actor-specific. Only the actor that called
/// `request` can use it for setting response handlers.
template
<
template
<
class
>
class
MergePolicy
,
message_priority
Prio
=
message_priority
::
normal
,
class
Container
,
class
...
Ts
>
auto
fan_out_request
(
const
Container
&
destinations
,
const
duration
&
timeout
,
Ts
&&
...
xs
)
{
using
handle_type
=
typename
Container
::
value_type
;
using
namespace
detail
;
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
using
token
=
type_list
<
implicit_conversions_t
<
decay_t
<
Ts
>>
...
>
;
static_assert
(
response_type_unbox
<
signatures_of_t
<
handle_type
>
,
token
>::
valid
,
"receiver does not accept given message"
);
auto
dptr
=
static_cast
<
Subtype
*>
(
this
);
std
::
vector
<
message_id
>
ids
;
ids
.
reserve
(
destinations
.
size
());
for
(
const
auto
&
dest
:
destinations
)
{
if
(
!
dest
)
continue
;
auto
req_id
=
dptr
->
new_request_id
(
Prio
);
dest
->
eq_impl
(
req_id
,
dptr
->
ctrl
(),
dptr
->
context
(),
std
::
forward
<
Ts
>
(
xs
)...);
dptr
->
request_response_timeout
(
timeout
,
req_id
);
ids
.
emplace_back
(
req_id
.
response_id
());
}
if
(
ids
.
empty
())
{
auto
req_id
=
dptr
->
new_request_id
(
Prio
);
dptr
->
eq_impl
(
req_id
.
response_id
(),
dptr
->
ctrl
(),
dptr
->
context
(),
make_error
(
sec
::
invalid_argument
));
ids
.
emplace_back
(
req_id
.
response_id
());
}
using
response_type
=
response_type_t
<
typename
handle_type
::
signatures
,
detail
::
implicit_conversions_t
<
detail
::
decay_t
<
Ts
>>
...
>
;
using
result_type
=
response_handle
<
Subtype
,
MergePolicy
<
response_type
>>
;
return
result_type
{
dptr
,
std
::
move
(
ids
)};
}
/// @copydoc fan_out_request
template
<
template
<
class
>
class
MergePolicy
,
message_priority
Prio
=
message_priority
::
normal
,
class
Rep
,
class
Period
,
class
Container
,
class
...
Ts
>
auto
fan_out_request
(
const
Container
&
destinations
,
std
::
chrono
::
duration
<
Rep
,
Period
>
timeout
,
Ts
&&
...
xs
)
{
return
request
<
MergePolicy
,
Prio
>
(
destinations
,
duration
{
timeout
},
std
::
forward
<
Ts
>
(
xs
)...);
}
};
};
}
// namespace mixin
}
// namespace mixin
}
// namespace caf
}
// namespace caf
libcaf_core/caf/policy/fan_in_responses.hpp
0 → 100644
View file @
ddef551c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <functional>
#include <limits>
#include <memory>
#include <vector>
#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/error.hpp"
#include "caf/logger.hpp"
#include "caf/message_id.hpp"
namespace
caf
{
namespace
detail
{
template
<
class
F
,
class
T
>
struct
fan_in_responses_helper
{
std
::
vector
<
T
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
F
f
;
void
operator
()(
T
&
x
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"pending"
,
*
pending
));
if
(
*
pending
>
0
)
{
results
.
emplace_back
(
std
::
move
(
x
));
if
(
--*
pending
==
0
)
f
(
std
::
move
(
results
));
}
}
template
<
class
Fun
>
fan_in_responses_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
auto
wrap
()
{
return
[
this
](
T
&
x
)
{
(
*
this
)(
x
);
};
}
};
template
<
class
F
,
class
...
Ts
>
struct
fan_in_responses_tuple_helper
{
using
value_type
=
std
::
tuple
<
Ts
...
>
;
std
::
vector
<
value_type
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
F
f
;
void
operator
()(
Ts
&
...
xs
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"pending"
,
*
pending
));
if
(
*
pending
>
0
)
{
results
.
emplace_back
(
std
::
move
(
xs
)...);
if
(
--*
pending
==
0
)
f
(
std
::
move
(
results
));
}
}
template
<
class
Fun
>
fan_in_responses_tuple_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
auto
wrap
()
{
return
[
this
](
Ts
&
...
xs
)
{
(
*
this
)(
xs
...);
};
}
};
template
<
class
F
,
class
=
typename
detail
::
get_callable_trait
<
F
>
::
arg_types
>
struct
select_fan_in_responses_helper
;
template
<
class
F
,
class
...
Ts
>
struct
select_fan_in_responses_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
std
::
tuple
<
Ts
...
>>>>
{
using
type
=
fan_in_responses_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
>
;
};
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
;
};
}
// namespace detail
}
// namespace caf
namespace
caf
{
namespace
policy
{
/// Enables a `response_handle` to fan-in multiple responses into a single
/// result (a `vector` of individual values) for the client.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
fan_in_responses
{
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
::
fan_in_responses_helper_t
<
detail
::
decay_t
<
Fun
>>>
;
explicit
fan_in_responses
(
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
;
fan_in_responses
&
operator
=
(
fan_in_responses
&&
)
noexcept
=
default
;
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
helper_type
=
detail
::
fan_in_responses_helper_t
<
detail
::
decay_t
<
F
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
forward
<
F
>
(
f
)};
auto
error_handler
=
[
&
](
error
&
err
)
{
if
(
*
helper
.
pending
>
0
)
{
*
helper
.
pending
=
0
;
helper
.
results
.
clear
();
g
(
err
);
}
};
for
(
auto
id
:
ids_
)
{
typename
Self
::
accept_one_cond
rc
;
self
->
varargs_receive
(
rc
,
id
,
helper
.
wrap
(),
error_handler
);
}
}
const
message_id_list
&
ids
()
const
noexcept
{
return
ids_
;
}
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
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
move
(
f
)};
error_handler_type
err_helper
{
std
::
forward
<
OnError
>
(
g
),
helper
.
pending
};
return
{
std
::
move
(
helper
),
std
::
move
(
err_helper
),
};
}
message_id_list
ids_
;
};
}
// namespace policy
}
// namespace caf
libcaf_core/caf/policy/single_response.hpp
0 → 100644
View file @
ddef551c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/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/error.hpp"
#include "caf/message_id.hpp"
namespace
caf
{
namespace
policy
{
/// Trivial policy for handling a single result in a `response_handler`.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
single_response
{
public:
static
constexpr
bool
is_trivial
=
true
;
using
response_type
=
ResponseType
;
template
<
class
Fun
>
using
type_checker
=
detail
::
type_checker
<
response_type
,
Fun
>
;
explicit
single_response
(
message_id
mid
)
noexcept
:
mid_
(
mid
)
{
// nop
}
single_response
(
single_response
&&
)
noexcept
=
default
;
single_response
&
operator
=
(
single_response
&&
)
noexcept
=
default
;
template
<
class
Self
,
class
F
,
class
OnError
>
void
await
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
behavior
bhvr
{
std
::
forward
<
F
>
(
f
),
std
::
forward
<
OnError
>
(
g
)};
self
->
add_awaited_response_handler
(
mid_
,
std
::
move
(
bhvr
));
}
template
<
class
Self
,
class
F
,
class
OnError
>
void
then
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
behavior
bhvr
{
std
::
forward
<
F
>
(
f
),
std
::
forward
<
OnError
>
(
g
)};
self
->
add_multiplexed_response_handler
(
mid_
,
std
::
move
(
bhvr
));
}
template
<
class
Self
,
class
F
,
class
OnError
>
void
receive
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
typename
Self
::
accept_one_cond
rc
;
self
->
varargs_receive
(
rc
,
mid_
,
std
::
forward
<
F
>
(
f
),
std
::
forward
<
OnError
>
(
g
));
}
message_id
id
()
const
noexcept
{
return
mid_
;
}
private:
message_id
mid_
;
};
}
// namespace policy
}
// namespace caf
libcaf_core/caf/response_handle.hpp
View file @
ddef551c
...
@@ -20,8 +20,10 @@
...
@@ -20,8 +20,10 @@
#include <type_traits>
#include <type_traits>
#include "caf/actor_traits.hpp"
#include "caf/catch_all.hpp"
#include "caf/catch_all.hpp"
#include "caf/message_id.hpp"
#include "caf/message_id.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_behavior.hpp"
...
@@ -33,145 +35,137 @@ namespace caf {
...
@@ -33,145 +35,137 @@ namespace caf {
/// This helper class identifies an expected response message and enables
/// This helper class identifies an expected response message and enables
/// `request(...).then(...)`.
/// `request(...).then(...)`.
template
<
class
Self
,
class
Output
,
bool
IsBlocking
>
template
<
class
ActorType
,
class
Policy
>
class
response_handle
;
class
response_handle
{
/******************************************************************************
* nonblocking *
******************************************************************************/
template
<
class
Self
,
class
Output
>
class
response_handle
<
Self
,
Output
,
false
>
{
public:
public:
response_handle
()
=
delete
;
// -- member types -----------------------------------------------------------
response_handle
(
const
response_handle
&
)
=
default
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
default
;
response_handle
(
message_id
mid
,
Self
*
self
)
:
mid_
(
mid
),
self_
(
self
)
{
using
actor_type
=
ActorType
;
// nop
}
template
<
class
F
,
class
E
=
detail
::
is_callable_t
<
F
>
>
using
traits
=
actor_traits
<
actor_type
>
;
void
await
(
F
f
)
const
{
await_impl
(
f
);
}
template
<
class
F
,
class
OnError
,
class
E1
=
detail
::
is_callable_t
<
F
>,
using
policy_type
=
Policy
;
class
E2
=
detail
::
is_handler_for_ef
<
OnError
,
error
>>
void
await
(
F
f
,
OnError
e
)
const
{
await_impl
(
f
,
e
);
}
template
<
class
F
,
class
E
=
detail
::
is_callable_t
<
F
>
>
using
response_type
=
typename
policy_type
::
response_type
;
void
then
(
F
f
)
const
{
then_impl
(
f
);
}
template
<
class
F
,
class
OnError
,
class
E1
=
detail
::
is_callable_t
<
F
>,
// -- constructors, destructors, and assignment operators --------------------
class
E2
=
detail
::
is_handler_for_ef
<
OnError
,
error
>>
void
then
(
F
f
,
OnError
e
)
const
{
then_impl
(
f
,
e
);
}
message_id
id
()
const
noexcept
{
response_handle
()
=
delete
;
return
mid_
;
}
Self
*
self
()
noexcept
{
response_handle
(
const
response_handle
&
)
=
delete
;
return
self_
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
delete
;
response_handle
(
response_handle
&&
)
noexcept
=
default
;
response_handle
&
operator
=
(
response_handle
&&
)
noexcept
=
delete
;
template
<
class
...
Ts
>
explicit
response_handle
(
actor_type
*
self
,
Ts
&&
...
xs
)
:
self_
(
self
),
policy_
(
std
::
forward
<
Ts
>
(
xs
)...)
{
// nop
}
}
private:
// -- non-blocking API -------------------------------------------------------
template
<
class
F
,
class
OnError
>
void
await_impl
(
F
&
f
,
OnError
&
g
)
const
{
template
<
class
T
=
traits
,
class
F
=
none_t
,
class
OnError
=
none_t
,
class
=
detail
::
enable_if_t
<
T
::
is_non_blocking
>
>
void
await
(
F
f
,
OnError
g
)
const
{
static_assert
(
detail
::
is_callable
<
F
>::
value
,
"F must provide a single, "
"non-template operator()"
);
static_assert
(
detail
::
is_callable_with
<
OnError
,
error
&>::
value
,
"OnError must provide an operator() that takes a caf::error"
);
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
"response handlers are not allowed to have a return "
"response handlers are not allowed to have a return "
"type other than void"
);
"type other than void"
);
detail
::
type_checker
<
Output
,
F
>::
check
();
policy_type
::
template
type_checker
<
F
>
::
check
();
self_
->
add_awaited_response_handler
(
mid_
,
policy_
.
await
(
self_
,
std
::
move
(
f
),
std
::
move
(
g
));
behavior
{
std
::
move
(
f
),
std
::
move
(
g
)});
}
}
template
<
class
F
>
template
<
class
T
=
traits
,
class
F
=
none_t
,
void
await_impl
(
F
&
f
)
const
{
class
=
detail
::
enable_if_t
<
T
::
is_non_blocking
>
>
void
await
(
F
f
)
const
{
auto
self
=
self_
;
auto
self
=
self_
;
auto
on_error
=
[
self
](
error
&
err
)
{
self
->
call_error_handler
(
err
);
};
await
(
std
::
move
(
f
),
[
self
](
error
&
err
)
{
self
->
call_error_handler
(
err
);
});
return
await_impl
(
f
,
on_error
);
}
}
template
<
class
F
,
class
OnError
>
template
<
class
T
=
traits
,
class
F
=
none_t
,
class
OnError
=
none_t
,
void
then_impl
(
F
&
f
,
OnError
&
g
)
const
{
class
=
detail
::
enable_if_t
<
T
::
is_non_blocking
>
>
void
then
(
F
f
,
OnError
g
)
const
{
static_assert
(
detail
::
is_callable
<
F
>::
value
,
"F must provide a single, "
"non-template operator()"
);
static_assert
(
detail
::
is_callable_with
<
OnError
,
error
&>::
value
,
"OnError must provide an operator() that takes a caf::error"
);
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
"response handlers are not allowed to have a return "
"response handlers are not allowed to have a return "
"type other than void"
);
"type other than void"
);
detail
::
type_checker
<
Output
,
F
>::
check
();
policy_type
::
template
type_checker
<
F
>
::
check
();
self_
->
add_multiplexed_response_handler
(
mid_
,
behavior
{
std
::
move
(
f
),
policy_
.
then
(
self_
,
std
::
move
(
f
),
std
::
move
(
g
));
std
::
move
(
g
)});
}
}
template
<
class
F
>
template
<
class
T
=
traits
,
class
F
=
none_t
,
void
then_impl
(
F
&
f
)
const
{
class
=
detail
::
enable_if_t
<
T
::
is_non_blocking
>
>
void
then
(
F
f
)
const
{
auto
self
=
self_
;
auto
self
=
self_
;
auto
on_error
=
[
self
](
error
&
err
)
{
self
->
call_error_handler
(
err
);
};
then
(
std
::
move
(
f
),
[
self
](
error
&
err
)
{
self
->
call_error_handler
(
err
);
});
return
then_impl
(
f
,
on_error
);
}
message_id
mid_
;
Self
*
self_
;
};
/******************************************************************************
* blocking *
******************************************************************************/
template
<
class
Self
,
class
Output
>
class
response_handle
<
Self
,
Output
,
true
>
{
public:
response_handle
()
=
delete
;
response_handle
(
const
response_handle
&
)
=
default
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
default
;
response_handle
(
message_id
mid
,
Self
*
self
)
:
mid_
(
mid
),
self_
(
self
)
{
// nop
}
}
using
error_handler
=
std
::
function
<
void
(
error
&
)
>
;
// -- blocking API -----------------------------------------------------------
template
<
class
F
,
class
OnError
,
template
<
class
T
=
traits
,
class
F
=
none_t
,
class
OnError
=
none_t
,
class
E
=
detail
::
is_handler_for_ef
<
OnError
,
error
>
>
class
=
detail
::
enable_if_t
<
T
::
is_blocking
>
>
detail
::
is_callable_t
<
F
>
receive
(
F
f
,
OnError
g
)
{
detail
::
is_handler_for_ef
<
OnError
,
error
>
receive
(
F
f
,
OnError
g
)
{
static_assert
(
detail
::
is_callable
<
F
>::
value
,
"F must provide a single, "
"non-template operator()"
);
static_assert
(
detail
::
is_callable_with
<
OnError
,
error
&>::
value
,
"OnError must provide an operator() that takes a caf::error"
);
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
using
result_type
=
typename
detail
::
get_callable_trait
<
F
>::
result_type
;
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
static_assert
(
std
::
is_same
<
void
,
result_type
>::
value
,
"response handlers are not allowed to have a return "
"response handlers are not allowed to have a return "
"type other than void"
);
"type other than void"
);
detail
::
type_checker
<
Output
,
F
>::
check
();
policy_type
::
template
type_checker
<
F
>
::
check
();
typename
Self
::
accept_one_cond
rc
;
policy_
.
receive
(
self_
,
std
::
move
(
f
),
std
::
move
(
g
));
self_
->
varargs_receive
(
rc
,
mid_
,
std
::
move
(
f
),
std
::
move
(
g
));
}
}
template
<
class
OnError
,
class
F
,
class
E
=
detail
::
is_callable_t
<
F
>
>
template
<
class
T
=
traits
,
class
OnError
=
none_t
,
class
F
=
none_t
,
class
=
detail
::
enable_if_t
<
T
::
is_blocking
>
>
detail
::
is_handler_for_ef
<
OnError
,
error
>
receive
(
OnError
g
,
F
f
)
{
detail
::
is_handler_for_ef
<
OnError
,
error
>
receive
(
OnError
g
,
F
f
)
{
// TODO: allowing blocking actors to pass the error handler in first may be
// more flexible, but it makes the API asymmetric. Consider
// deprecating / removing this member function.
receive
(
std
::
move
(
f
),
std
::
move
(
g
));
receive
(
std
::
move
(
f
),
std
::
move
(
g
));
}
}
template
<
class
OnError
,
class
F
,
template
<
class
T
=
policy_type
,
class
OnError
=
none_t
,
class
F
=
none_t
,
class
E
=
detail
::
is_handler_for_ef
<
OnError
,
error
>
>
class
E
=
detail
::
is_handler_for_ef
<
OnError
,
error
>,
class
=
detail
::
enable_if_t
<
T
::
is_trivial
>>
void
receive
(
OnError
g
,
catch_all
<
F
>
f
)
{
void
receive
(
OnError
g
,
catch_all
<
F
>
f
)
{
typename
Self
::
accept_one_cond
rc
;
// TODO: this bypasses the policy. Either we deprecate `catch_all` or *all*
self_
->
varargs_receive
(
rc
,
mid_
,
std
::
move
(
g
),
std
::
move
(
f
));
// policies must support it. Currently, we only enable this member
// function on the trivial policy for backwards compatibility.
typename
actor_type
::
accept_one_cond
rc
;
self_
->
varargs_receive
(
rc
,
id
(),
std
::
move
(
g
),
std
::
move
(
f
));
}
}
// -- properties -------------------------------------------------------------
template
<
class
T
=
policy_type
,
class
=
detail
::
enable_if_t
<
T
::
is_trivial
>
>
message_id
id
()
const
noexcept
{
message_id
id
()
const
noexcept
{
return
mid_
;
return
policy_
.
id
()
;
}
}
Self
*
self
()
noexcept
{
actor_type
*
self
()
noexcept
{
return
self_
;
return
self_
;
}
}
private:
private:
message_id
mid_
;
/// Points to the parent actor.
Self
*
self_
;
actor_type
*
self_
;
/// Configures how the actor wants to process an incoming response.
policy_type
policy_
;
};
};
}
// namespace caf
}
// namespace caf
libcaf_core/caf/typed_actor.hpp
View file @
ddef551c
...
@@ -144,9 +144,9 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
...
@@ -144,9 +144,9 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
template
<
class
T
,
template
<
class
T
,
class
=
detail
::
enable_if_t
<
actor_traits
<
T
>
::
is_statically_typed
>>
class
=
detail
::
enable_if_t
<
actor_traits
<
T
>
::
is_statically_typed
>>
typed_actor
(
T
*
ptr
)
:
ptr_
(
ptr
->
ctrl
())
{
typed_actor
(
T
*
ptr
)
:
ptr_
(
ptr
->
ctrl
())
{
static_assert
(
detail
::
tl_subset_of
<
signatures
,
static_assert
(
typename
T
::
signatures
>::
value
,
detail
::
tl_subset_of
<
signatures
,
typename
T
::
signatures
>::
value
,
"Cannot assign T* to incompatible handle type"
);
"Cannot assign T* to incompatible handle type"
);
CAF_ASSERT
(
ptr
!=
nullptr
);
CAF_ASSERT
(
ptr
!=
nullptr
);
}
}
...
...
libcaf_core/test/mixin/requester.cpp
View file @
ddef551c
...
@@ -23,6 +23,7 @@
...
@@ -23,6 +23,7 @@
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/policy/fan_in_responses.hpp"
using
namespace
caf
;
using
namespace
caf
;
...
@@ -150,4 +151,31 @@ CAF_TEST(delegated request with integer result) {
...
@@ -150,4 +151,31 @@ CAF_TEST(delegated request with integer result) {
CAF_CHECK_EQUAL
(
*
result
,
3
);
CAF_CHECK_EQUAL
(
*
result
,
3
);
}
}
CAF_TEST
(
requesters
support
fan_out_request
)
{
using
policy
::
fan_in_responses
;
std
::
vector
<
adding_server_type
>
workers
{
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
};
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
)
.
then
([
=
](
std
::
vector
<
int
>
results
)
{
for
(
auto
result
:
results
)
CAF_CHECK_EQUAL
(
result
,
3
);
*
sum
=
std
::
accumulate
(
results
.
begin
(),
results
.
end
(),
0
);
});
});
run_once
();
expect
((
int
,
int
),
from
(
client
).
to
(
workers
[
0
]).
with
(
1
,
2
));
expect
((
int
),
from
(
workers
[
0
]).
to
(
client
).
with
(
3
));
expect
((
int
,
int
),
from
(
client
).
to
(
workers
[
1
]).
with
(
1
,
2
));
expect
((
int
),
from
(
workers
[
1
]).
to
(
client
).
with
(
3
));
expect
((
int
,
int
),
from
(
client
).
to
(
workers
[
2
]).
with
(
1
,
2
));
expect
((
int
),
from
(
workers
[
2
]).
to
(
client
).
with
(
3
));
CAF_CHECK_EQUAL
(
*
sum
,
9
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/policy/fan_in_responses.cpp
0 → 100644
View file @
ddef551c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 policy.fan_in_responses
#include "caf/policy/fan_in_responses.hpp"
#include "caf/test/dsl.hpp"
#include <tuple>
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using
caf
::
policy
::
fan_in_responses
;
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
);
}
std
::
function
<
void
(
const
error
&
)
>
make_error_handler
()
{
return
[
this
](
const
error
&
err
)
{
CAF_FAIL
(
"unexpected error: "
<<
sys
.
render
(
err
));
};
}
std
::
function
<
void
(
const
error
&
)
>
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
(
fan_in_responses_tests
,
fixture
)
CAF_TEST
(
fan_in_responses
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
);
auto
server2
=
make_server
(
f
);
SUBTEST
(
"request.receive"
)
{
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
()}};
run
();
merge
.
receive
(
self
.
ptr
(),
[](
int_list
results
)
{
std
::
sort
(
results
.
begin
(),
results
.
end
());
CAF_CHECK_EQUAL
(
results
,
int_list
({
3
,
5
}));
},
make_error_handler
());
}
SUBTEST
(
"vector of tuples"
)
{
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
()}};
run
();
using
results_vector
=
std
::
vector
<
std
::
tuple
<
int
>>
;
merge
.
receive
(
self
.
ptr
(),
[](
results_vector
results
)
{
std
::
sort
(
results
.
begin
(),
results
.
end
());
CAF_CHECK_EQUAL
(
results
,
results_vector
({
make_tuple
(
3
),
make_tuple
(
5
)}));
},
make_error_handler
());
}
}
SUBTEST
(
"request.then"
)
{
int_list
results
;
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
()}};
merge
.
then
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
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 stores results in arrival order"
);
CAF_CHECK_EQUAL
(
results
,
int_list
({
3
,
5
}));
}
SUBTEST
(
"request.await"
)
{
int_list
results
;
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
()}};
merge
.
await
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
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
(
results
,
int_list
({
5
,
3
}));
}
}
CAF_TEST
(
fan_in_responses
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
);
auto
server2
=
make_server
(
f
);
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
()}};
run
();
size_t
errors
=
0
;
merge
.
receive
(
self
.
ptr
(),
[](
int_list
)
{
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
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
then
(
client_ptr
,
[](
int_list
)
{
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
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
await
(
client_ptr
,
[](
int_list
)
{
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
()
libcaf_test/caf/test/dsl.hpp
View file @
ddef551c
...
@@ -645,9 +645,9 @@ protected:
...
@@ -645,9 +645,9 @@ protected:
template
<
class
...
Ts
>
template
<
class
...
Ts
>
struct
test_coordinator_fixture_fetch_helper
{
struct
test_coordinator_fixture_fetch_helper
{
template
<
class
Self
,
class
Interface
>
template
<
class
Self
,
template
<
class
>
class
Policy
,
class
Interface
>
std
::
tuple
<
Ts
...
>
std
::
tuple
<
Ts
...
>
operator
()(
caf
::
response_handle
<
Self
,
Interface
,
true
>&
from
)
const
{
operator
()(
caf
::
response_handle
<
Self
,
Policy
<
Interface
>
>&
from
)
const
{
std
::
tuple
<
Ts
...
>
result
;
std
::
tuple
<
Ts
...
>
result
;
from
.
receive
([
&
](
Ts
&
...
xs
)
{
result
=
std
::
make_tuple
(
std
::
move
(
xs
)...);
},
from
.
receive
([
&
](
Ts
&
...
xs
)
{
result
=
std
::
make_tuple
(
std
::
move
(
xs
)...);
},
[
&
](
caf
::
error
&
err
)
{
[
&
](
caf
::
error
&
err
)
{
...
@@ -659,8 +659,8 @@ struct test_coordinator_fixture_fetch_helper {
...
@@ -659,8 +659,8 @@ struct test_coordinator_fixture_fetch_helper {
template
<
class
T
>
template
<
class
T
>
struct
test_coordinator_fixture_fetch_helper
<
T
>
{
struct
test_coordinator_fixture_fetch_helper
<
T
>
{
template
<
class
Self
,
class
Interface
>
template
<
class
Self
,
template
<
class
>
class
Policy
,
class
Interface
>
T
operator
()(
caf
::
response_handle
<
Self
,
Interface
,
true
>&
from
)
const
{
T
operator
()(
caf
::
response_handle
<
Self
,
Policy
<
Interface
>
>&
from
)
const
{
T
result
;
T
result
;
from
.
receive
([
&
](
T
&
x
)
{
result
=
std
::
move
(
x
);
},
from
.
receive
([
&
](
T
&
x
)
{
result
=
std
::
move
(
x
);
},
[
&
](
caf
::
error
&
err
)
{
[
&
](
caf
::
error
&
err
)
{
...
...
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