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
ae4d7c58
Commit
ae4d7c58
authored
Dec 05, 2017
by
Dominik Charousset
Committed by
Dominik Charousset
Feb 06, 2018
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Redesign scatterers, implement broadcasting
parent
39601bf2
Changes
20
Expand all
Hide whitespace changes
Inline
Side-by-side
Showing
20 changed files
with
1067 additions
and
1395 deletions
+1067
-1395
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+0
-1
libcaf_core/caf/broadcast_scatterer.hpp
libcaf_core/caf/broadcast_scatterer.hpp
+114
-21
libcaf_core/caf/buffered_scatterer.hpp
libcaf_core/caf/buffered_scatterer.hpp
+21
-11
libcaf_core/caf/fused_scatterer.hpp
libcaf_core/caf/fused_scatterer.hpp
+0
-286
libcaf_core/caf/inbound_path.hpp
libcaf_core/caf/inbound_path.hpp
+18
-8
libcaf_core/caf/invalid_stream_scatterer.hpp
libcaf_core/caf/invalid_stream_scatterer.hpp
+3
-42
libcaf_core/caf/outbound_path.hpp
libcaf_core/caf/outbound_path.hpp
+61
-33
libcaf_core/caf/stream_edge_impl.hpp
libcaf_core/caf/stream_edge_impl.hpp
+0
-251
libcaf_core/caf/stream_manager.hpp
libcaf_core/caf/stream_manager.hpp
+41
-65
libcaf_core/caf/stream_scatterer.hpp
libcaf_core/caf/stream_scatterer.hpp
+88
-53
libcaf_core/caf/stream_scatterer_impl.hpp
libcaf_core/caf/stream_scatterer_impl.hpp
+0
-158
libcaf_core/caf/terminal_stream_scatterer.hpp
libcaf_core/caf/terminal_stream_scatterer.hpp
+3
-5
libcaf_core/src/inbound_path.cpp
libcaf_core/src/inbound_path.cpp
+17
-19
libcaf_core/src/invalid_stream_scatterer.cpp
libcaf_core/src/invalid_stream_scatterer.cpp
+6
-80
libcaf_core/src/outbound_path.cpp
libcaf_core/src/outbound_path.cpp
+15
-41
libcaf_core/src/stream_manager.cpp
libcaf_core/src/stream_manager.cpp
+32
-157
libcaf_core/src/stream_scatterer.cpp
libcaf_core/src/stream_scatterer.cpp
+137
-9
libcaf_core/src/stream_scatterer_impl.cpp
libcaf_core/src/stream_scatterer_impl.cpp
+0
-144
libcaf_core/src/terminal_stream_scatterer.cpp
libcaf_core/src/terminal_stream_scatterer.cpp
+5
-11
libcaf_core/test/broadcast_scatterer.cpp
libcaf_core/test/broadcast_scatterer.cpp
+506
-0
No files found.
libcaf_core/CMakeLists.txt
View file @
ae4d7c58
...
...
@@ -95,7 +95,6 @@ set (LIBCAF_CORE_SRCS
src/stream_manager.cpp
src/stream_priority.cpp
src/stream_scatterer.cpp
src/stream_scatterer_impl.cpp
src/stringification_inspector.cpp
src/sync_request_bouncer.cpp
src/term.cpp
...
...
libcaf_core/caf/broadcast_scatterer.hpp
View file @
ae4d7c58
...
...
@@ -20,47 +20,140 @@
#define CAF_BROADCAST_SCATTERER_HPP
#include "caf/buffered_scatterer.hpp"
#include "caf/outbound_path.hpp"
namespace
caf
{
template
<
class
T
>
class
broadcast_scatterer
:
public
buffered_scatterer
<
T
>
{
public:
// -- member types -----------------------------------------------------------
/// Base type.
using
super
=
buffered_scatterer
<
T
>
;
/// Container for caching `T`s per path.
using
cache_type
=
std
::
vector
<
T
>
;
/// Maps slot IDs to caches.
using
cache_map_type
=
detail
::
unordered_flat_map
<
stream_slots
,
cache_type
>
;
typename
super
::
path_ptr
add_path
(
stream_slots
slots
,
strong_actor_ptr
target
)
override
{
auto
res
=
caches_
.
emplace
(
slots
,
cache_type
{});
return
res
.
second
?
super
::
add_path
(
slots
,
target
)
:
nullptr
;
}
broadcast_scatterer
(
local_actor
*
selfptr
)
:
super
(
selfptr
)
{
// nop
}
long
credit
()
const
override
{
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return
this
->
min_credit
();
void
emit_batches
()
override
{
CAF_LOG_TRACE
(
""
);
if
(
!
this
->
delay_partial_batches
())
{
force_emit_batches
();
}
else
{
auto
f
=
[
&
](
outbound_path
&
,
cache_type
&
)
{
// Do nothing, i.e., leave the cache untouched.
};
emit_batches_impl
(
f
);
}
}
void
emit_batches
()
override
{
void
force_
emit_batches
()
override
{
CAF_LOG_TRACE
(
""
);
auto
hint
=
this
->
min_desired_batch_size
();
auto
next_chunk
=
[
&
]
{
// TODO: this iterates paths_ every time again even though we could
// easily keep track of remaining credit
return
this
->
get_chunk
(
std
::
min
(
this
->
min_credit
(),
hint
));
auto
f
=
[
&
](
outbound_path
&
p
,
cache_type
&
c
)
{
p
.
emit_batch
(
this
->
self_
,
c
.
size
(),
make_message
(
std
::
move
(
c
)));
};
for
(
auto
chunk
=
next_chunk
();
!
chunk
.
empty
();
chunk
=
next_chunk
())
{
auto
csize
=
static_cast
<
long
>
(
chunk
.
size
());
auto
wrapped_chunk
=
make_message
(
std
::
move
(
chunk
));
for
(
auto
&
x
:
this
->
paths_
)
{
CAF_ASSERT
(
x
->
open_credit
>=
csize
);
x
->
emit_batch
(
csize
,
wrapped_chunk
);
}
emit_batches_impl
(
f
);
}
protected:
void
about_to_erase
(
typename
super
::
map_type
::
iterator
i
,
bool
,
error
*
)
override
{
caches_
.
erase
(
i
->
second
->
slots
);
}
private:
/// Iterates `paths_`, applying `f` to all but the last element and `g` to
/// the last element.
template
<
class
F
,
class
G
>
void
iterate_paths
(
F
&
f
,
G
&
g
)
{
// The vector storing all paths has the same order as the vector storing
// all caches. This allows us to iterate both vectors using a single index.
auto
&
pvec
=
this
->
paths_
.
container
();
auto
&
cvec
=
caches_
.
container
();
CAF_ASSERT
(
pvec
.
size
()
==
cvec
.
size
());
size_t
l
=
pvec
.
size
()
-
1
;
for
(
size_t
i
=
0u
;
i
!=
l
;
++
i
)
{
CAF_ASSERT
(
pvec
[
i
].
first
==
cvec
[
i
].
first
);
f
(
*
pvec
[
i
].
second
,
cvec
[
i
].
second
);
}
CAF_ASSERT
(
pvec
[
l
].
first
==
cvec
[
l
].
first
);
g
(
*
pvec
[
l
].
second
,
cvec
[
l
].
second
);
}
long
desired_batch_size
()
const
override
{
// TODO: this is an O(n) computation, consider storing the result in a
// member variable for
return
super
::
min_desired_batch_size
();
template
<
class
OnPartialBatch
>
void
emit_batches_impl
(
OnPartialBatch
&
f
)
{
if
(
this
->
paths_
.
empty
())
return
;
auto
emit_impl
=
[
&
](
outbound_path
&
p
,
cache_type
&
c
)
{
if
(
c
.
empty
())
return
;
auto
dbs
=
p
.
desired_batch_size
;
CAF_ASSERT
(
dbs
>
0
);
if
(
c
.
size
()
==
dbs
)
{
p
.
emit_batch
(
this
->
self_
,
c
.
size
(),
make_message
(
std
::
move
(
c
)));
}
else
{
auto
i
=
c
.
begin
();
auto
e
=
c
.
end
();
while
(
std
::
distance
(
i
,
e
)
>=
static_cast
<
ptrdiff_t
>
(
dbs
))
{
std
::
vector
<
T
>
tmp
{
std
::
make_move_iterator
(
i
),
std
::
make_move_iterator
(
i
+
dbs
)};
p
.
emit_batch
(
this
->
self_
,
dbs
,
make_message
(
std
::
move
(
tmp
)));
i
+=
dbs
;
}
if
(
std
::
distance
(
i
,
e
)
>
0
)
{
c
.
erase
(
c
.
begin
(),
i
);
f
(
p
,
c
);
}
}
};
// Calculate how many more items we can put to our caches at the most.
auto
chunk_size
=
std
::
numeric_limits
<
size_t
>::
max
();
{
auto
&
pvec
=
this
->
paths_
.
container
();
auto
&
cvec
=
caches_
.
container
();
CAF_ASSERT
(
pvec
.
size
()
==
cvec
.
size
());
for
(
size_t
i
=
0u
;
i
!=
pvec
.
size
();
++
i
)
{
chunk_size
=
std
::
min
(
chunk_size
,
pvec
[
i
].
second
->
open_credit
-
cvec
[
i
].
second
.
size
());
}
}
auto
chunk
=
this
->
get_chunk
(
chunk_size
);
if
(
chunk
.
empty
())
{
auto
g
=
[
&
](
outbound_path
&
p
,
cache_type
&
c
)
{
emit_impl
(
p
,
c
);
};
iterate_paths
(
g
,
g
);
}
else
{
auto
copy_emit
=
[
&
](
outbound_path
&
p
,
cache_type
&
c
)
{
c
.
insert
(
c
.
end
(),
chunk
.
begin
(),
chunk
.
end
());
emit_impl
(
p
,
c
);
};
auto
move_emit
=
[
&
](
outbound_path
&
p
,
cache_type
&
c
)
{
if
(
c
.
empty
())
c
=
std
::
move
(
chunk
);
else
c
.
insert
(
c
.
end
(),
std
::
make_move_iterator
(
chunk
.
begin
()),
std
::
make_move_iterator
(
chunk
.
end
()));
emit_impl
(
p
,
c
);
};
iterate_paths
(
copy_emit
,
move_emit
);
}
}
cache_map_type
caches_
;
};
}
// namespace caf
...
...
libcaf_core/caf/buffered_scatterer.hpp
View file @
ae4d7c58
...
...
@@ -16,27 +16,29 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_
MIXIN_
BUFFERED_SCATTERER_HPP
#define CAF_
MIXIN_
BUFFERED_SCATTERER_HPP
#ifndef CAF_BUFFERED_SCATTERER_HPP
#define CAF_BUFFERED_SCATTERER_HPP
#include <deque>
#include <vector>
#include <cstddef>
#include <iterator>
#include "caf/sec.hpp"
#include "caf/stream_edge_impl.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/stream_scatterer_impl.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/stream_scatterer.hpp"
namespace
caf
{
/// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface.
template
<
class
T
>
class
buffered_scatterer
:
public
stream_scatterer
_impl
{
class
buffered_scatterer
:
public
stream_scatterer
{
public:
using
super
=
stream_scatterer_impl
;
// -- member types -----------------------------------------------------------
using
super
=
stream_scatterer
;
using
value_type
=
T
;
...
...
@@ -44,7 +46,9 @@ public:
using
chunk_type
=
std
::
vector
<
value_type
>
;
buffered_scatterer
(
local_actor
*
selfptr
)
:
super
(
selfptr
)
{
// -- constructors, destructors, and assignment operators --------------------
buffered_scatterer
(
local_actor
*
self
)
:
super
(
self
)
{
// nop
}
...
...
@@ -76,8 +80,14 @@ public:
return
get_chunk
(
buf_
,
n
);
}
long
buffered
()
const
override
{
return
static_cast
<
long
>
(
buf_
.
size
());
size_t
capacity
()
const
noexcept
override
{
// TODO: get rid of magic number
static
constexpr
size_t
max_buf_size
=
100
;
return
buf_
.
size
()
<
max_buf_size
?
buf_
.
size
()
-
max_buf_size
:
0u
;
}
size_t
buffered
()
const
noexcept
override
{
return
buf_
.
size
();
}
buffer_type
&
buf
()
{
...
...
@@ -94,4 +104,4 @@ protected:
}
// namespace caf
#endif // CAF_
MIXIN_
BUFFERED_SCATTERER_HPP
#endif // CAF_BUFFERED_SCATTERER_HPP
libcaf_core/caf/fused_scatterer.hpp
deleted
100644 → 0
View file @
39601bf2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_FUSED_SCATTERER
#define CAF_FUSED_SCATTERER
#include <tuple>
#include <cstddef>
#include "caf/logger.hpp"
#include "caf/stream_scatterer.hpp"
namespace
caf
{
namespace
detail
{
/// Utility function for repeating `x` for a given template parameter pack.
template
<
class
T
,
class
U
>
U
pack_repeat
(
U
x
)
{
return
x
;
}
template
<
class
Iter
>
class
ptr_array_initializer
{
public:
ptr_array_initializer
(
Iter
first
)
:
i_
(
first
)
{
// nop
}
void
operator
()()
{
// end of recursion
}
template
<
class
T
,
class
...
Ts
>
void
operator
()(
T
&
x
,
Ts
&
...
xs
)
{
*
i_
=
&
x
;
++
i_
;
(
*
this
)(
xs
...);
}
private:
Iter
i_
;
};
struct
scatterer_selector
{
inline
stream_scatterer
*
operator
()(
const
message
&
)
{
return
nullptr
;
}
template
<
class
T
,
class
...
Ts
>
stream_scatterer
*
operator
()(
const
message
&
msg
,
T
&
x
,
Ts
&
...
xs
)
{
if
(
msg
.
match_element
<
stream
<
typename
T
::
value_type
>>
(
0
))
return
&
x
;
return
(
*
this
)(
msg
,
xs
...);
}
};
}
// namespace detail
/// A scatterer that delegates to any number of sub-scatterers. Data is only
/// pushed to the main scatterer `T` per default.
template
<
class
T
,
class
...
Ts
>
class
fused_scatterer
:
public
stream_scatterer
{
public:
using
substreams_tuple
=
std
::
tuple
<
T
,
Ts
...
>
;
using
pointer
=
stream_scatterer
*
;
using
const_pointer
=
const
pointer
;
using
iterator
=
pointer
*
;
using
const_iterator
=
const
pointer
*
;
fused_scatterer
(
local_actor
*
self
)
:
substreams_
(
self
,
detail
::
pack_repeat
<
Ts
>
(
self
)...)
{
detail
::
ptr_array_initializer
<
pointer
*>
f
{
ptrs_
};
auto
indices
=
detail
::
get_indices
(
substreams_
);
detail
::
apply_args
(
f
,
indices
,
substreams_
);
}
path_ptr
add_path
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
origin
)
<<
CAF_ARG
(
sink_ptr
)
<<
CAF_ARG
(
stages
)
<<
CAF_ARG
(
handshake_mid
)
<<
CAF_ARG
(
handshake_data
)
<<
CAF_ARG
(
prio
)
<<
CAF_ARG
(
redeployable
));
auto
ptr
=
substream_by_handshake_type
(
handshake_data
);
if
(
!
ptr
)
return
nullptr
;
return
ptr
->
add_path
(
slot
,
std
::
move
(
origin
),
std
::
move
(
sink_ptr
),
std
::
move
(
stages
),
handshake_mid
,
std
::
move
(
handshake_data
),
prio
,
redeployable
);
}
path_ptr
confirm_path
(
stream_slot
slot
,
const
actor_addr
&
from
,
strong_actor_ptr
to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
from
)
<<
CAF_ARG
(
to
)
<<
CAF_ARG
(
initial_demand
)
<<
CAF_ARG
(
redeployable
));
return
first_hit
([
&
](
pointer
ptr
)
->
outbound_path
*
{
// Note: we cannot blindly try `confirm_path` on each scatterer, because
// this will trigger forced_close messages.
if
(
ptr
->
find
(
slot
,
from
)
==
nullptr
)
return
nullptr
;
return
ptr
->
confirm_path
(
slot
,
from
,
to
,
initial_demand
,
desired_batch_size
,
redeployable
);
});
}
bool
remove_path
(
stream_slot
slot
,
const
actor_addr
&
addr
,
error
reason
,
bool
silent
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
addr
)
<<
CAF_ARG
(
reason
)
<<
CAF_ARG
(
silent
));
return
std
::
any_of
(
begin
(),
end
(),
[
&
](
pointer
x
)
{
return
x
->
remove_path
(
slot
,
addr
,
reason
,
silent
);
});
}
bool
paths_clean
()
const
override
{
return
std
::
all_of
(
begin
(),
end
(),
[
&
](
const_pointer
x
)
{
return
x
->
paths_clean
();
});
}
void
close
()
override
{
CAF_LOG_TRACE
(
""
);
for
(
auto
ptr
:
ptrs_
)
ptr
->
close
();
}
void
abort
(
error
reason
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
reason
));
for
(
auto
ptr
:
ptrs_
)
ptr
->
abort
(
reason
);
}
long
num_paths
()
const
override
{
return
std
::
accumulate
(
begin
(),
end
(),
0l
,
[](
long
x
,
const_pointer
y
)
{
return
x
+
y
->
num_paths
();
});
}
bool
closed
()
const
override
{
return
std
::
all_of
(
begin
(),
end
(),
[
&
](
const_pointer
x
)
{
return
x
->
closed
();
});
}
bool
continuous
()
const
override
{
return
std
::
any_of
(
begin
(),
end
(),
[
&
](
const_pointer
x
)
{
return
x
->
continuous
();
});
}
void
continuous
(
bool
value
)
override
{
for
(
auto
ptr
:
ptrs_
)
ptr
->
continuous
(
value
);
}
void
emit_batches
()
override
{
CAF_LOG_TRACE
(
""
);
for
(
auto
ptr
:
ptrs_
)
ptr
->
emit_batches
();
}
path_ptr
find
(
stream_slot
slot
,
const
actor_addr
&
x
)
override
{
return
first_hit
([
&
](
const_pointer
ptr
)
{
return
ptr
->
find
(
slot
,
x
);
});
}
path_ptr
path_at
(
size_t
idx
)
override
{
auto
i
=
std
::
begin
(
ptrs_
);
auto
e
=
std
::
end
(
ptrs_
);
while
(
i
!=
e
)
{
auto
np
=
static_cast
<
size_t
>
((
*
i
)
->
num_paths
());
if
(
idx
<
np
)
return
(
*
i
)
->
path_at
(
idx
);
idx
-=
np
;
}
return
nullptr
;
}
long
credit
()
const
override
{
// TODO: only return credit of the main stream?
return
std
::
accumulate
(
begin
(),
end
(),
std
::
numeric_limits
<
long
>::
max
(),
[](
long
x
,
pointer
y
)
{
return
std
::
min
(
x
,
y
->
credit
());
});
}
long
buffered
()
const
override
{
// TODO: only return how many items are buffered at the main stream?
return
std
::
accumulate
(
begin
(),
end
(),
0l
,
[](
long
x
,
const_pointer
y
)
{
return
x
+
y
->
buffered
();
});
}
long
min_buffer_size
()
const
override
{
return
main_stream
().
min_buffer_size
();
}
duration
max_batch_delay
()
const
override
{
return
main_stream
().
max_batch_delay
();
}
void
max_batch_delay
(
duration
x
)
override
{
main_stream
().
max_batch_delay
(
x
);
}
template
<
class
...
Us
>
void
push
(
Us
&&
...
xs
)
{
main_stream
().
push
(
std
::
forward
<
Us
>
(
xs
)...);
}
iterator
begin
()
{
return
std
::
begin
(
ptrs_
);
}
const_iterator
begin
()
const
{
return
std
::
begin
(
ptrs_
);
}
iterator
end
()
{
return
std
::
end
(
ptrs_
);
}
const_iterator
end
()
const
{
return
std
::
end
(
ptrs_
);
}
T
&
main_stream
()
{
return
std
::
get
<
0
>
(
substreams_
);
}
const
T
&
main_stream
()
const
{
return
std
::
get
<
0
>
(
substreams_
);
}
template
<
size_t
I
>
typename
std
::
tuple_element
<
I
,
substreams_tuple
>::
type
&
substream
()
{
return
std
::
get
<
I
>
(
substreams_
);
}
template
<
size_t
I
>
const
typename
std
::
tuple_element
<
I
,
substreams_tuple
>::
type
&
substream
()
const
{
return
std
::
get
<
I
>
(
substreams_
);
}
stream_scatterer
*
substream_by_handshake_type
(
const
message
&
msg
)
{
detail
::
scatterer_selector
f
;
auto
indices
=
detail
::
get_indices
(
substreams_
);
return
detail
::
apply_args_prefixed
(
f
,
indices
,
substreams_
,
msg
);
}
private:
template
<
class
F
>
path_ptr
first_hit
(
F
f
)
{
for
(
auto
ptr
:
ptrs_
)
{
auto
result
=
f
(
ptr
);
if
(
result
!=
nullptr
)
return
result
;
}
return
nullptr
;
}
substreams_tuple
substreams_
;
stream_scatterer
*
ptrs_
[
sizeof
...(
Ts
)
+
1
];
};
}
// namespace caf
#endif // CAF_FUSED_SCATTERER
libcaf_core/caf/inbound_path.hpp
View file @
ae4d7c58
...
...
@@ -24,6 +24,7 @@
#include "caf/actor_control_block.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp"
...
...
@@ -44,11 +45,11 @@ public:
/// Message type for propagating errors.
using
irregular_shutdown
=
upstream_msg
::
forced_drop
;
///
Slot IDs for sender (hdl) and receiver (self)
.
stream_
slots
slots
;
///
Points to the manager responsible for incoming traffic
.
stream_
manager_ptr
mgr
;
///
Pointer to the parent actor
.
local_actor
*
self
;
///
Stores slot IDs for sender (hdl) and receiver (self)
.
stream_slots
slots
;
/// Handle to the source.
strong_actor_ptr
hdl
;
...
...
@@ -77,7 +78,8 @@ public:
error
shutdown_reason
;
/// Constructs a path for given handle and stream ID.
inbound_path
(
local_actor
*
selfptr
,
stream_slots
id
,
strong_actor_ptr
ptr
);
inbound_path
(
stream_manager_ptr
mgr_ptr
,
stream_slots
id
,
strong_actor_ptr
ptr
);
~
inbound_path
();
...
...
@@ -86,12 +88,20 @@ public:
/// Emits a `stream_msg::ack_batch` on this path and sets `assigned_credit`
/// to `initial_demand`.
void
emit_ack_open
(
actor_addr
rebind_from
,
long
initial_demand
,
void
emit_ack_open
(
local_actor
*
self
,
actor_addr
rebind_from
,
long
initial_demand
,
bool
is_redeployable
);
void
emit_ack_batch
(
long
new_demand
);
/// Sends a `stream_msg::ack_batch`.
void
emit_ack_batch
(
local_actor
*
self
,
long
new_demand
);
/// Sends a `stream_msg::close` on this path.
void
emit_regular_shutdown
(
local_actor
*
self
);
/// Sends a `stream_msg::forced_close` on this path.
void
emit_regular_shutdown
(
local_actor
*
self
,
error
reason
);
static
void
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
sid
,
/// Sends a `stream_msg::forced_close` on this path.
static
void
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
slots
,
const
strong_actor_ptr
&
hdl
,
error
reason
);
};
...
...
libcaf_core/caf/invalid_stream_scatterer.hpp
View file @
ae4d7c58
...
...
@@ -26,54 +26,15 @@ namespace caf {
/// Type-erased policy for dispatching data to sinks.
class
invalid_stream_scatterer
:
public
stream_scatterer
{
public:
invalid_stream_scatterer
(
local_actor
*
self
=
nullptr
);
invalid_stream_scatterer
(
local_actor
*
self
);
~
invalid_stream_scatterer
()
override
;
path_ptr
add_path
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
override
;
path_ptr
confirm_path
(
stream_slot
slot
,
const
actor_addr
&
from
,
strong_actor_ptr
to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
)
override
;
bool
remove_path
(
stream_slot
slot
,
const
actor_addr
&
x
,
error
reason
,
bool
silent
)
override
;
bool
paths_clean
()
const
override
;
void
close
()
override
;
void
abort
(
error
reason
)
override
;
long
num_paths
()
const
override
;
bool
closed
()
const
override
;
bool
continuous
()
const
override
;
void
continuous
(
bool
value
)
override
;
path_type
*
path_at
(
size_t
index
)
override
;
void
emit_batches
()
override
;
path_type
*
find
(
stream_slot
slot
,
const
actor_addr
&
x
)
override
;
long
credit
()
const
override
;
long
buffered
()
const
override
;
long
desired_batch_size
()
const
override
;
long
min_buffer_size
()
const
override
;
duration
max_batch_delay
()
const
override
;
size_t
capacity
()
const
noexcept
override
;
void
max_batch_delay
(
duration
x
)
override
;
size_t
buffered
()
const
noexcept
override
;
};
}
// namespace caf
...
...
libcaf_core/caf/outbound_path.hpp
View file @
ae4d7c58
...
...
@@ -29,6 +29,7 @@
#include "caf/fwd.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_slot.hpp"
#include "caf/system_messages.hpp"
#include "caf/meta/type_name.hpp"
...
...
@@ -37,27 +38,68 @@ namespace caf {
/// State for a single path to a sink on a `stream_scatterer`.
class
outbound_path
{
public:
/// Stream aborter flag to monitor a path.
static
constexpr
const
auto
aborter_type
=
stream_aborter
::
sink_aborter
;
// -- member types -----------------------------------------------------------
///
Message type for propagating
graceful shutdowns.
///
Propagates
graceful shutdowns.
using
regular_shutdown
=
downstream_msg
::
close
;
///
Message type for propagating
errors.
///
Propagates
errors.
using
irregular_shutdown
=
downstream_msg
::
forced_close
;
/// Stores batches until receiving corresponding ACKs.
using
cache_type
=
std
::
deque
<
std
::
pair
<
int64_t
,
downstream_msg
::
batch
>>
;
// -- nested classes ---------------------------------------------------------
/// Stores information about the initiator of the steam.
struct
client_data
{
strong_actor_ptr
hdl
;
message_id
mid
;
};
// -- constants --------------------------------------------------------------
/// Stream aborter flag to monitor a path.
static
constexpr
const
auto
aborter_type
=
stream_aborter
::
sink_aborter
;
// -- constructors, destructors, and assignment operators --------------------
/// Constructs a path for given handle and stream ID.
outbound_path
(
stream_slots
id
,
strong_actor_ptr
ptr
);
~
outbound_path
();
// -- downstream communication -----------------------------------------------
/// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void
handle_ack_open
(
long
initial_credit
);
/// Sends a stream handshake.
void
emit_open
(
local_actor
*
self
,
strong_actor_ptr
origin
,
mailbox_element
::
forwarding_stack
stages
,
message_id
mid
,
message
handshake_data
,
stream_priority
prio
,
bool
is_redeployable
);
/// Sends a `stream_msg::batch` on this path, decrements `open_credit` by
/// `xs_size` and increments `next_batch_id` by 1.
void
emit_batch
(
local_actor
*
self
,
long
xs_size
,
message
xs
);
/// Sends a `stream_msg::drop` on this path.
void
emit_regular_shutdown
(
local_actor
*
self
);
/// Sends a `stream_msg::forced_drop` on this path.
void
emit_irregular_shutdown
(
local_actor
*
self
,
error
reason
);
/// Sends a `stream_msg::forced_drop` on this path.
static
void
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
slots
,
const
strong_actor_ptr
&
hdl
,
error
reason
);
// -- member variables -------------------------------------------------------
/// Slot IDs for sender (self) and receiver (hdl).
stream_slots
slots
;
/// Pointer to the parent actor.
local_actor
*
self
;
/// Handle to the sink.
strong_actor_ptr
hdl
;
...
...
@@ -68,7 +110,7 @@ public:
long
open_credit
;
/// Batch size configured by the downstream actor.
long
desired_batch_size
;
uint64_t
desired_batch_size
;
/// Stores whether the downstream actor is failsafe, i.e., allows the runtime
/// to redeploy it on failure. If this field is set to `false` then
...
...
@@ -80,7 +122,7 @@ public:
int64_t
next_ack_id
;
/// Caches batches until receiving an ACK.
std
::
deque
<
std
::
pair
<
int64_t
,
downstream_msg
::
batch
>>
unacknowledged_batches
;
cache_type
unacknowledged_batches
;
/// Caches the initiator of the stream (client) with the original request ID
/// until the stream handshake is either confirmed or aborted. Once
...
...
@@ -90,35 +132,21 @@ public:
/// Stores whether an error occurred during stream processing.
error
shutdown_reason
;
/// Constructs a path for given handle and stream ID.
outbound_path
(
local_actor
*
selfptr
,
stream_slots
id
,
strong_actor_ptr
ptr
);
~
outbound_path
();
/// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void
handle_ack_open
(
long
initial_credit
);
void
emit_open
(
strong_actor_ptr
origin
,
mailbox_element
::
forwarding_stack
stages
,
message_id
mid
,
message
handshake_data
,
stream_priority
prio
,
bool
is_redeployable
);
/// Emits a `stream_msg::batch` on this path, decrements `open_credit` by
/// `xs_size` and increments `next_batch_id` by 1.
void
emit_batch
(
long
xs_size
,
message
xs
);
static
void
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
slots
,
const
strong_actor_ptr
&
hdl
,
error
reason
);
};
/// @relates outbound_path::client_data
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
outbound_path
::
client_data
&
x
)
{
return
f
(
x
.
hdl
,
x
.
mid
);
}
/// @relates outbound_path
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
outbound_path
&
x
)
{
return
f
(
meta
::
type_name
(
"outbound_path"
),
x
.
hdl
,
x
.
slots
,
x
.
next_batch_id
,
x
.
open_credit
,
x
.
redeployable
,
x
.
unacknowledged_batches
);
return
f
(
meta
::
type_name
(
"outbound_path"
),
x
.
slots
,
x
.
hdl
,
x
.
next_batch_id
,
x
.
open_credit
,
x
.
desired_batch_size
,
x
.
redeployable
,
x
.
next_ack_id
,
x
.
unacknowledged_batches
,
x
.
cd
,
x
.
shutdown_reason
);
}
}
// namespace caf
...
...
libcaf_core/caf/stream_edge_impl.hpp
deleted
100644 → 0
View file @
39601bf2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_STREAM_EDGE_IMPL_HPP
#define CAF_STREAM_EDGE_IMPL_HPP
#include <memory>
#include <vector>
#include <numeric>
#include <cstddef>
#include <algorithm>
#include "caf/fwd.hpp"
#include "caf/send.hpp"
#include "caf/config.hpp"
#include "caf/actor_addr.hpp"
#include "caf/local_actor.hpp"
#include "caf/inbound_path.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/actor_control_block.hpp"
namespace
caf
{
/// Provides a common scaffold for implementations of the `stream_gatherer` and
/// `stream_scatterer` interfaces.
template
<
class
Base
>
class
stream_edge_impl
:
public
Base
{
public:
// -- member types -----------------------------------------------------------
using
super
=
Base
;
/// Either `inbound_path` or `outbound_path`.
using
path_type
=
typename
super
::
path_type
;
/// A raw pointer to a path.
using
path_ptr
=
path_type
*
;
/// Vector of raw pointers (views) of paths.
using
path_ptr_vec
=
std
::
vector
<
path_ptr
>
;
/// Iterator to a vector of raw pointers.
using
path_ptr_iter
=
typename
path_ptr_vec
::
iterator
;
/// A unique pointer to a path.
using
path_uptr
=
std
::
unique_ptr
<
path_type
>
;
/// Vector of owning pointers of paths.
using
path_uptr_vec
=
std
::
vector
<
path_uptr
>
;
/// Iterator to a vector of owning pointers.
using
path_uptr_iter
=
typename
path_uptr_vec
::
iterator
;
/// Message type for sending graceful shutdowns along the path (either
/// `stream_msg::drop` or `stream_msg::close`).
using
regular_shutdown
=
typename
path_type
::
regular_shutdown
;
/// Message type for sending errors along the path (either
/// `stream_msg::forced_drop` or `stream_msg::forced_close`).
using
irregular_shutdown
=
typename
path_type
::
irregular_shutdown
;
/// Stream aborter flag to monitor paths.
static
constexpr
const
auto
aborter_type
=
path_type
::
aborter_type
;
// -- constructors, destructors, and assignment operators --------------------
stream_edge_impl
(
local_actor
*
selfptr
)
:
self_
(
selfptr
),
continuous_
(
false
)
{
// nop
}
~
stream_edge_impl
()
override
{
// nop
}
// -- static utility functions for path vectors ------------------------------
/// Sorts `xs` in descending order by available credit.
template
<
class
PathContainer
>
static
void
sort_by_credit
(
PathContainer
&
xs
)
{
using
value_type
=
typename
PathContainer
::
value_type
;
auto
cmp
=
[](
const
value_type
&
x
,
const
value_type
&
y
)
{
return
x
->
open_credit
>
y
->
open_credit
;
};
std
::
sort
(
xs
.
begin
(),
xs
.
end
(),
cmp
);
}
template
<
class
T
,
class
PathContainer
,
class
F
>
static
T
fold
(
PathContainer
&
xs
,
T
init
,
F
f
)
{
auto
b
=
xs
.
begin
();
auto
e
=
xs
.
end
();
return
b
!=
e
?
std
::
accumulate
(
b
,
e
,
init
,
f
)
:
static_cast
<
T
>
(
0
);
}
/// Finds the path for `ptr` and returns a pointer to it.
template
<
class
PathContainer
,
class
Handle
>
static
path_ptr
find
(
PathContainer
&
,
stream_slot
,
const
Handle
&
)
{
// TODO: implement me
return
nullptr
;
}
/// Finds the path for `ptr` and returns an iterator to it.
template
<
class
PathContainer
,
class
Handle
>
static
typename
PathContainer
::
iterator
iter_find
(
PathContainer
&
xs
,
stream_slot
,
const
Handle
&
)
{
// TODO: implement me
return
xs
.
end
();
/*
auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == x && y->slot == slot;
};
return std::find_if(xs.begin(), xs.end(), predicate);
*/
}
// -- accessors --------------------------------------------------------------
/// Returns all available paths.
inline
const
path_uptr_vec
&
paths
()
const
{
return
paths_
;
}
/// Returns a pointer to the parent actor.
inline
local_actor
*
self
()
const
{
return
self_
;
}
// -- reusable convenience functions -----------------------------------------
using
super
::
remove_path
;
bool
remove_path
(
path_uptr_iter
,
error
,
bool
)
{
// TODO: implement me
return
true
;
/*
CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(silent));
auto e = paths_.end();
if (i == e) {
CAF_LOG_DEBUG("unable to remove path");
return false;
}
auto& p = *(*i);
stream_aborter::del(p.hdl, self_->address(), p.slot, aborter_type);
if (silent)
p.hdl = nullptr;
if (reason != none)
p.shutdown_reason = std::move(reason);
if (i != paths_.end() - 1)
std::swap(*i, paths_.back());
paths_.pop_back();
return true;
*/
}
// -- implementation of common methods ---------------------------------------
bool
remove_path
(
stream_slot
slot
,
const
actor_addr
&
x
,
error
reason
,
bool
silent
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
x
)
<<
CAF_ARG
(
reason
)
<<
CAF_ARG
(
silent
));
return
remove_path
(
iter_find
(
paths_
,
slot
,
x
),
std
::
move
(
reason
),
silent
);
}
void
abort
(
error
reason
)
override
{
auto
i
=
paths_
.
begin
();
auto
e
=
paths_
.
end
();
if
(
i
!=
e
)
{
// Handle last element separately after the loop.
--
e
;
for
(;
i
!=
e
;
++
i
)
(
*
i
)
->
shutdown_reason
=
reason
;
(
*
e
)
->
shutdown_reason
=
std
::
move
(
reason
);
paths_
.
clear
();
}
}
long
num_paths
()
const
override
{
return
static_cast
<
long
>
(
paths_
.
size
());
}
bool
closed
()
const
override
{
return
!
continuous_
&&
paths_
.
empty
();
}
bool
continuous
()
const
override
{
return
continuous_
;
}
void
continuous
(
bool
value
)
override
{
continuous_
=
value
;
}
// -- implementation of identical methods in scatterer and gatherer ---------
path_ptr
path_at
(
size_t
index
)
override
{
return
paths_
[
index
].
get
();
}
using
super
::
find
;
path_ptr
find
(
stream_slot
slot
,
const
actor_addr
&
x
)
override
{
return
find
(
paths_
,
slot
,
x
);
}
protected:
/// Adds a path to the edge without emitting messages.
path_ptr
add_path_impl
(
stream_slot
,
strong_actor_ptr
)
{
// TODO: implement me
return
nullptr
;
/*
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(slot));
stream_aborter::add(x, self_->address(), slot, aborter_type);
paths_.emplace_back(new path_type(self_, slot, std::move(x)));
return paths_.back().get();
*/
}
template
<
class
F
>
void
close_impl
(
F
f
)
{
for
(
auto
&
x
:
paths_
)
{
stream_aborter
::
del
(
x
->
hdl
,
self_
->
address
(),
x
->
slot
,
aborter_type
);
f
(
*
x
);
}
paths_
.
clear
();
}
local_actor
*
self_
;
path_uptr_vec
paths_
;
bool
continuous_
;
};
}
// namespace caf
#endif // CAF_STREAM_EDGE_IMPL_HPP
libcaf_core/caf/stream_manager.hpp
View file @
ae4d7c58
...
...
@@ -23,10 +23,12 @@
#include <cstdint>
#include <cstddef>
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp"
namespace
caf
{
...
...
@@ -34,6 +36,8 @@ namespace caf {
/// @relates stream_msg
class
stream_manager
:
public
ref_counted
{
public:
stream_manager
(
local_actor
*
selfptr
);
~
stream_manager
()
override
;
/// Handles `stream_msg::open` messages by creating a new slot for incoming
...
...
@@ -51,70 +55,27 @@ public:
/// acknowledges the handshake.
/// @returns An error if the stream manager rejects the handshake.
/// @pre `hdl != nullptr`
/*
virtual error open(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage, stream_priority priority,
bool redeployable, response_promise result_cb);
*/
/// Handles `stream_msg::ack_open` messages, i.e., finalizes the stream
/// handshake.
/// @param slot Slot ID used by the sender.
/// @param rebind_from Receiver of the original `open` message.
/// @param rebind_to Sender of this confirmation.
/// @param initial_demand Credit received with this `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy
/// `rebind_to` on failure.
/// @pre `hdl != nullptr`
virtual
error
ack_open
(
stream_slot
slot
,
const
actor_addr
&
rebind_from
,
strong_actor_ptr
rebind_to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
);
virtual
error
handle
(
inbound_path
*
from
,
downstream_msg
::
batch
&
x
);
/// Handles `stream_msg::batch` messages.
/// @param hdl Handle to the sender.
/// @param xs_size Size of the vector stored in `xs`.
/// @param xs A type-erased vector of size `xs_size`.
/// @param xs_id ID of this batch (must be ACKed).
/// @pre `hdl != nullptr`
/// @pre `xs_size > 0`
virtual
error
batch
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
long
xs_size
,
message
&
xs
,
int64_t
xs_id
);
virtual
error
handle
(
inbound_path
*
from
,
downstream_msg
::
close
&
x
);
/// Handles `stream_msg::ack_batch` messages.
/// @param hdl Handle to the sender.
/// @param new_demand New credit for sending data.
/// @param cumulative_batch_id Id of last handled batch.
/// @pre `hdl != nullptr`
virtual
error
ack_batch
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
long
new_demand
,
long
desired_batch_size
,
int64_t
cumulative_batch_id
);
virtual
error
handle
(
inbound_path
*
from
,
downstream_msg
::
forced_close
&
x
);
/// Handles `stream_msg::close` messages.
/// @param hdl Handle to the sender.
/// @pre `hdl != nullptr`
virtual
error
close
(
stream_slot
slot
,
const
actor_addr
&
hdl
);
virtual
error
handle
(
outbound_path
*
from
,
upstream_msg
::
ack_open
&
x
);
/// Handles `stream_msg::drop` messages.
/// @param hdl Handle to the sender.
/// @pre `hdl != nullptr`
virtual
error
drop
(
stream_slot
slot
,
const
actor_addr
&
hdl
);
virtual
error
handle
(
outbound_path
*
from
,
upstream_msg
::
ack_batch
&
x
);
/// Handles `stream_msg::drop` messages. The default implementation calls
/// `abort(reason)` and returns `sec::unhandled_stream_error`.
/// @param hdl Handle to the sender.
/// @param reason Reported error from the source.
/// @pre `hdl != nullptr`
/// @pre `err != none`
virtual
error
forced_close
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
error
reason
);
virtual
error
handle
(
outbound_path
*
from
,
upstream_msg
::
drop
&
x
);
/// Handles `stream_msg::drop` messages. The default implementation calls
/// `abort(reason)` and returns `sec::unhandled_stream_error`.
/// @param hdl Handle to the sender.
/// @param reason Reported error from the sink.
/// @pre `hdl != nullptr`
/// @pre `err != none`
virtual
error
forced_drop
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
error
reason
);
virtual
error
handle
(
outbound_path
*
from
,
upstream_msg
::
forced_drop
&
x
);
/*
/// Adds a new sink to the stream.
virtual bool add_sink(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
...
...
@@ -127,10 +88,11 @@ public:
virtual bool add_source(stream_slot slot, strong_actor_ptr source_ptr,
strong_actor_ptr original_stage, stream_priority prio,
bool redeployable, response_promise result_cb);
*/
///
Pushes new data to downstream actors by sending batches. The amount of
///
pushed data is limited by the available credit
.
virtual
void
push
();
///
Closes the stream when the parent terminates with default exit reason or
///
the stream reached its end
.
virtual
void
close
();
/// Aborts a stream after any stream message handler returned a non-default
/// constructed error `reason` or the parent actor terminates with a
...
...
@@ -138,11 +100,18 @@ public:
/// @param reason Previous error or non-default exit reason of the parent.
virtual
void
abort
(
error
reason
);
/// Closes the stream when the parent terminates with default exit reason or
/// the stream reached its end.
virtual
void
close
();
/// Pushes new data to downstream actors by sending batches. The amount of
/// pushed data is limited by the available credit.
virtual
void
push
();
// -- implementation hooks for sources ---------------------------------------
/// Tries to generate new messages for the stream. This member function does
/// nothing on stages and sinks, but can trigger a source to produce more
/// messages.
virtual
bool
generate_messages
();
// --
implementation hooks for all children classes
--------------------------
// --
pure virtual member functions ----------------
--------------------------
/// Returns the stream edge for outgoing data.
virtual
stream_scatterer
&
out
()
=
0
;
...
...
@@ -151,12 +120,19 @@ public:
/// safely.
virtual
bool
done
()
const
=
0
;
// -- i
mplementation hooks for sources
---------------------------------------
// -- i
nput path management -----------
---------------------------------------
/// Tries to generate new messages for the stream. This member function does
/// nothing on stages and sinks, but can trigger a source to produce more
/// messages.
virtual
bool
generate_messages
();
/// Informs the manager that a new input path opens.
virtual
void
register_input_path
(
inbound_path
*
x
);
/// Informs the manager that an input path closes.
virtual
void
deregister_input_path
(
inbound_path
*
x
)
noexcept
;
// -- inline functions -------------------------------------------------------
inline
local_actor
*
self
()
{
return
self_
;
}
protected:
// -- implementation hooks for sinks -----------------------------------------
...
...
libcaf_core/caf/stream_scatterer.hpp
View file @
ae4d7c58
...
...
@@ -20,13 +20,17 @@
#define CAF_STREAM_SCATTERER_HPP
#include <cstddef>
#include <memory>
#include "caf/actor_control_block.hpp"
#include "caf/duration.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/optional.hpp"
#include "caf/stream_slot.hpp"
#include "caf/detail/unordered_flat_map.hpp"
namespace
caf
{
/// Type-erased policy for dispatching data to sinks.
...
...
@@ -38,96 +42,127 @@ public:
using
path_ptr
=
path_type
*
;
using
path_unique_ptr
=
std
::
unique_ptr
<
path_type
>
;
using
map_type
=
detail
::
unordered_flat_map
<
stream_slots
,
path_unique_ptr
>
;
// -- constructors, destructors, and assignment operators --------------------
stream_scatterer
()
=
default
;
explicit
stream_scatterer
(
local_actor
*
self
)
;
virtual
~
stream_scatterer
();
// -- pure virtual memeber functions -----------------------------------------
// -- path management --------------------------------------------------------
/// Returns the container that stores all paths.
inline
const
map_type
&
paths
()
const
noexcept
{
return
paths_
;
}
/// Adds a path to
the edge
.
/// Adds a path to
`target` to the scatterer
.
/// @returns The added path on success, `nullptr` otherwise.
virtual
path_ptr
add_path
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
=
0
;
virtual
path_ptr
add_path
(
stream_slots
slots
,
strong_actor_ptr
target
);
/// Adds a path to a sink and initiates the handshake.
virtual
path_ptr
confirm_path
(
stream_slot
slot
,
const
actor_addr
&
from
,
strong_actor_ptr
to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
)
=
0
;
/// Removes a path from the scatterer and returns it.
path_unique_ptr
take_path
(
stream_slots
slots
)
noexcept
;
/// Removes a path from the scatterer.
virtual
bool
remove_path
(
stream_slot
slot
,
const
actor_addr
&
x
,
error
reason
,
bool
silent
)
=
0
;
/// Returns the path associated to `slots` or `nullptr`.
path_ptr
path
(
stream_slots
slots
)
noexcept
;
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
path_ptr
path_at
(
size_t
index
)
noexcept
;
/// Removes a path from the scatterer and returns it.
path_unique_ptr
take_path_at
(
size_t
index
)
noexcept
;
/// Returns `true` if there is no data pending and no unacknowledged batch on
/// any path.
virtual
bool
paths_clean
()
const
=
0
;
bool
clean
()
const
noexcept
;
/// Removes all paths gracefully.
v
irtual
void
close
()
=
0
;
v
oid
close
()
;
/// Removes all paths with an error message.
v
irtual
void
abort
(
error
reason
)
=
0
;
v
oid
abort
(
error
reason
)
;
/// Returns the number of paths managed on this edge.
virtual
long
num_paths
()
const
=
0
;
/// Returns `true` if no downstream exists and `!continuous()`,
/// `false` otherwise.
inline
bool
empty
()
const
noexcept
{
return
paths_
.
empty
();
}
/// Returns
`true` if no downstream exists, `false` otherwise
.
virtual
bool
closed
()
const
=
0
;
/// Returns
the minimum amount of credit on all output paths
.
size_t
min_credit
()
const
noexcept
;
/// Returns
whether this edge remains open after the last path is removed
.
virtual
bool
continuous
()
const
=
0
;
/// Returns
the maximum amount of credit on all output paths
.
size_t
max_credit
()
const
noexcept
;
/// Sets whether this edge remains open after the last path is removed.
virtual
void
continuous
(
bool
value
)
=
0
;
/// Returns the total amount of credit on all output paths, i.e., the sum of
/// all individual credits.
size_t
total_credit
()
const
noexcept
;
/// Sends batches to sinks.
virtual
void
emit_batches
()
=
0
;
// -- configuration parameters -----------------------------------------------
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
virtual
path_ptr
find
(
stream_slot
slot
,
const
actor_addr
&
x
)
=
0
;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
inline
duration
max_batch_delay
()
const
noexcept
{
return
max_batch_delay_
;
}
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
virtual
path_ptr
path_at
(
size_t
index
)
=
0
;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
inline
void
max_batch_delay
(
duration
x
)
noexcept
{
max_batch_delay_
=
x
;
}
/// Returns the currently available credit, depending on the policy in use.
/// For example, a broadcast policy would return the minimum of all available
/// downstream credits.
virtual
long
credit
()
const
=
0
;
// -- state access -----------------------------------------------------------
/// Returns the size of the output buffer.
virtual
long
buffered
()
const
=
0
;
local_actor
*
self
()
const
{
return
self_
;
}
/// Returns the downstream-requested size for a single batch.
virtual
long
desired_batch_size
()
const
=
0
;
// -- pure virtual memeber functions -----------------------------------------
/// Minimum amount of messages we wish to store at the actor in order to emit
/// new batches immediately when receiving new downstream demand. Usually
/// dynamically adjusted based on the output rate.
virtual
long
min_buffer_size
()
const
=
0
;
/// Sends batches to sinks.
virtual
void
emit_batches
()
=
0
;
///
Forces to actor to emit a batch even if the minimum batch size was not
///
reached
.
virtual
duration
max_batch_delay
()
const
=
0
;
///
Sends batches to sinks regardless of whether or not the batches reach the
///
desired batch size
.
virtual
void
force_emit_batches
()
=
0
;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
virtual
void
max_batch_delay
(
duration
x
)
=
0
;
/// Returns the currently available capacity for the output buffer.
virtual
size_t
capacity
()
const
noexcept
=
0
;
/// Returns the size of the output buffer.
virtual
size_t
buffered
()
const
noexcept
=
0
;
// -- convenience functions --------------------------------------------------
/// Removes a path from the scatterer.
bool
remove_path
(
stream_slot
slot
,
const
strong_actor_ptr
&
x
,
bool
remove_path
(
stream_slot
s
slot
,
const
strong_actor_ptr
&
x
,
error
reason
,
bool
silent
);
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr
find
(
stream_slot
slot
,
const
strong_actor_ptr
&
x
);
inline
bool
delay_partial_batches
()
const
{
return
max_batch_delay
().
count
!=
0
;
}
protected:
// -- customization points ---------------------------------------------------
/// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`)
/// shutdown if `silent == false`.
/// @warning moves `*reason` if `reason == nullptr`
virtual
void
about_to_erase
(
map_type
::
iterator
i
,
bool
silent
,
error
*
reason
);
// -- member variables -------------------------------------------------------
map_type
paths_
;
local_actor
*
self_
;
duration
max_batch_delay_
;
};
}
// namespace caf
...
...
libcaf_core/caf/stream_scatterer_impl.hpp
deleted
100644 → 0
View file @
39601bf2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_STREAM_SCATTERER_IMPL_HPP
#define CAF_STREAM_SCATTERER_IMPL_HPP
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/duration.hpp"
#include "caf/stream_edge_impl.hpp"
#include "caf/stream_scatterer.hpp"
namespace
caf
{
/// Type-erased policy for dispatching data to sinks.
class
stream_scatterer_impl
:
public
stream_edge_impl
<
stream_scatterer
>
{
public:
// -- member types -----------------------------------------------------------
using
super
=
stream_edge_impl
<
stream_scatterer
>
;
// -- constructors, destructors, and assignment operators --------------------
stream_scatterer_impl
(
local_actor
*
selfptr
);
~
stream_scatterer_impl
()
override
;
// -- static utility functions for operating on paths ------------------------
/// Removes all paths with an error message.
void
abort
(
error
reason
)
override
;
/// Folds `paths()` by extracting the `open_credit` from each path.
template
<
class
PathContainer
,
class
F
>
static
long
fold_credit
(
const
PathContainer
&
xs
,
long
x0
,
F
f
)
{
auto
g
=
[
f
](
long
x
,
const
typename
PathContainer
::
value_type
&
y
)
{
return
f
(
x
,
y
->
open_credit
);
};
return
super
::
fold
(
xs
,
x0
,
std
::
move
(
g
));
}
/// Returns the total number (sum) of all credit in `xs`.
template
<
class
PathContainer
>
static
long
total_credit
(
const
PathContainer
&
xs
)
{
return
fold_credit
(
xs
,
0l
,
[](
long
x
,
long
y
)
{
return
x
+
y
;
});
}
/// Returns the minimum number of credit in `xs`.
template
<
class
PathContainer
>
static
long
min_credit
(
const
PathContainer
&
xs
)
{
return
!
xs
.
empty
()
?
fold_credit
(
xs
,
std
::
numeric_limits
<
long
>::
max
(),
[](
long
x
,
long
y
)
{
return
std
::
min
(
x
,
y
);
})
:
0l
;
}
/// Returns the maximum number of credit in `xs`.
template
<
class
PathContainer
>
static
long
max_credit
(
const
PathContainer
&
xs
)
{
return
fold_credit
(
xs
,
0l
,
[](
long
x
,
long
y
)
{
return
std
::
max
(
x
,
y
);
});
}
/// Folds `paths()` by extracting the `desired_batch_size` from each path.
template
<
class
PathContainer
,
class
F
>
static
long
fold_desired_batch_size
(
const
PathContainer
&
xs
,
long
x0
,
F
f
)
{
auto
g
=
[
f
](
long
x
,
const
typename
PathContainer
::
value_type
&
y
)
{
return
f
(
x
,
y
->
desired_batch_size
);
};
return
super
::
fold
(
xs
,
x0
,
std
::
move
(
g
));
}
/// Returns the total number (sum) of all desired batch sizes in `xs`.
template
<
class
PathContainer
>
static
long
total_desired_batch_size
(
const
PathContainer
&
xs
)
{
return
fold_desired_batch_size
(
xs
,
0l
,
[](
long
x
,
long
y
)
{
return
x
+
y
;
});
}
/// Returns the minimum number of desired batch sizes in `xs`.
template
<
class
PathContainer
>
static
long
min_desired_batch_size
(
const
PathContainer
&
xs
)
{
return
!
xs
.
empty
()
?
fold_desired_batch_size
(
xs
,
std
::
numeric_limits
<
long
>::
max
(),
[](
long
x
,
long
y
)
{
return
std
::
min
(
x
,
y
);
})
:
0l
;
}
/// Returns the maximum number of desired batch sizes in `xs`.
template
<
class
PathContainer
>
static
long
max_desired_batch_size
(
const
PathContainer
&
xs
)
{
return
fold_credit
(
xs
,
0l
,
[](
long
x
,
long
y
)
{
return
std
::
max
(
x
,
y
);
});
}
// -- convenience functions for children classes -----------------------------
/// Returns the total number (sum) of all credit in `paths()`.
long
total_credit
()
const
;
/// Returns the minimum number of credit in `paths()`.
long
min_credit
()
const
;
/// Returns the maximum number of credit in `paths()`.
long
max_credit
()
const
;
/// Returns the total number (sum) of all desired batch sizes in `paths()`.
long
total_desired_batch_size
()
const
;
/// Returns the minimum number of desired batch sizes in `paths()`.
long
min_desired_batch_size
()
const
;
/// Returns the maximum number of desired batch sizes in `paths()`.
long
max_desired_batch_size
()
const
;
// -- overridden functions ---------------------------------------------------
void
close
()
override
;
path_ptr
add_path
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
override
;
path_ptr
confirm_path
(
stream_slot
slot
,
const
actor_addr
&
from
,
strong_actor_ptr
to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
)
override
;
bool
paths_clean
()
const
override
;
long
min_buffer_size
()
const
override
;
duration
max_batch_delay
()
const
override
;
void
max_batch_delay
(
duration
x
)
override
;
protected:
duration
max_batch_delay_
;
};
}
// namespace caf
#endif // CAF_STREAM_SCATTERER_IMPL_HPP
libcaf_core/caf/terminal_stream_scatterer.hpp
View file @
ae4d7c58
...
...
@@ -24,18 +24,16 @@
namespace
caf
{
/// Special-purpose scatterer for sinks that terminate a stream. A terminal
/// stream scatterer generates
credit without downstream actors
.
/// stream scatterer generates
infinite credit
.
class
terminal_stream_scatterer
:
public
invalid_stream_scatterer
{
public:
using
super
=
invalid_stream_scatterer
;
terminal_stream_scatterer
(
local_actor
*
=
nullptr
);
terminal_stream_scatterer
(
local_actor
*
self
);
~
terminal_stream_scatterer
()
override
;
long
credit
()
const
override
;
long
desired_batch_size
()
const
override
;
size_t
capacity
()
const
noexcept
override
;
};
}
// namespace caf
...
...
libcaf_core/src/inbound_path.cpp
View file @
ae4d7c58
...
...
@@ -25,10 +25,10 @@
namespace
caf
{
inbound_path
::
inbound_path
(
local_actor
*
self
ptr
,
stream_slots
id
,
inbound_path
::
inbound_path
(
stream_manager_ptr
mgr_
ptr
,
stream_slots
id
,
strong_actor_ptr
ptr
)
:
slots
(
id
),
s
elf
(
selfptr
),
:
mgr
(
std
::
move
(
mgr_ptr
)
),
s
lots
(
id
),
hdl
(
std
::
move
(
ptr
)),
prio
(
stream_priority
::
normal
),
last_acked_batch_id
(
0
),
...
...
@@ -36,19 +36,11 @@ inbound_path::inbound_path(local_actor* selfptr, stream_slots id,
assigned_credit
(
0
),
desired_batch_size
(
50
),
// TODO: at least put default in some header
redeployable
(
false
)
{
// nop
mgr
->
register_input_path
(
this
);
}
inbound_path
::~
inbound_path
()
{
if
(
hdl
)
{
if
(
shutdown_reason
==
none
)
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
drop
>
(
slots
.
invert
(),
self
->
address
()));
else
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
forced_drop
>
(
slots
.
invert
(),
self
->
address
(),
shutdown_reason
));
}
mgr
->
deregister_input_path
(
this
);
}
void
inbound_path
::
handle_batch
(
long
batch_size
,
int64_t
batch_id
)
{
...
...
@@ -56,10 +48,8 @@ void inbound_path::handle_batch(long batch_size, int64_t batch_id) {
last_batch_id
=
batch_id
;
}
void
inbound_path
::
emit_ack_open
(
actor_addr
rebind_from
,
void
inbound_path
::
emit_ack_open
(
local_actor
*
self
,
actor_addr
rebind_from
,
long
initial_demand
,
bool
is_redeployable
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
rebind_from
)
<<
CAF_ARG
(
initial_demand
)
<<
CAF_ARG
(
is_redeployable
));
assigned_credit
=
initial_demand
;
redeployable
=
is_redeployable
;
auto
batch_size
=
static_cast
<
int32_t
>
(
desired_batch_size
);
...
...
@@ -70,8 +60,7 @@ void inbound_path::emit_ack_open(actor_addr rebind_from,
is_redeployable
));
}
void
inbound_path
::
emit_ack_batch
(
long
new_demand
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
new_demand
));
void
inbound_path
::
emit_ack_batch
(
local_actor
*
self
,
long
new_demand
)
{
last_acked_batch_id
=
last_batch_id
;
assigned_credit
+=
new_demand
;
auto
batch_size
=
static_cast
<
int32_t
>
(
desired_batch_size
);
...
...
@@ -81,11 +70,20 @@ void inbound_path::emit_ack_batch(long new_demand) {
batch_size
,
last_batch_id
));
}
void
inbound_path
::
emit_regular_shutdown
(
local_actor
*
self
)
{
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
drop
>
(
slots
,
self
->
address
()));
}
void
inbound_path
::
emit_regular_shutdown
(
local_actor
*
self
,
error
reason
)
{
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
forced_drop
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
}
void
inbound_path
::
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
slots
,
const
strong_actor_ptr
&
hdl
,
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
reason
));
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
forced_drop
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
...
...
libcaf_core/src/invalid_stream_scatterer.cpp
View file @
ae4d7c58
...
...
@@ -22,7 +22,8 @@
namespace
caf
{
invalid_stream_scatterer
::
invalid_stream_scatterer
(
local_actor
*
)
{
invalid_stream_scatterer
::
invalid_stream_scatterer
(
local_actor
*
self
)
:
stream_scatterer
(
self
)
{
// nop
}
...
...
@@ -30,91 +31,16 @@ invalid_stream_scatterer::~invalid_stream_scatterer() {
// nop
}
stream_scatterer
::
path_ptr
invalid_stream_scatterer
::
add_path
(
stream_slot
,
strong_actor_ptr
,
strong_actor_ptr
,
mailbox_element
::
forwarding_stack
,
message_id
,
message
,
stream_priority
,
bool
)
{
CAF_LOG_ERROR
(
"invalid_stream_scatterer::add_path called"
);
return
nullptr
;
}
stream_scatterer
::
path_ptr
invalid_stream_scatterer
::
confirm_path
(
stream_slot
,
const
actor_addr
&
,
strong_actor_ptr
,
long
,
long
,
bool
)
{
CAF_LOG_ERROR
(
"invalid_stream_scatterer::confirm_path called"
);
return
nullptr
;
}
bool
invalid_stream_scatterer
::
remove_path
(
stream_slot
,
const
actor_addr
&
,
error
,
bool
)
{
CAF_LOG_ERROR
(
"invalid_stream_scatterer::remove_path called"
);
return
false
;
}
bool
invalid_stream_scatterer
::
paths_clean
()
const
{
return
true
;
}
void
invalid_stream_scatterer
::
close
()
{
// nop
}
void
invalid_stream_scatterer
::
abort
(
error
)
{
// nop
}
long
invalid_stream_scatterer
::
num_paths
()
const
{
return
0
;
}
bool
invalid_stream_scatterer
::
closed
()
const
{
return
true
;
}
bool
invalid_stream_scatterer
::
continuous
()
const
{
return
false
;
}
void
invalid_stream_scatterer
::
continuous
(
bool
)
{
// nop
}
stream_scatterer
::
path_type
*
invalid_stream_scatterer
::
path_at
(
size_t
)
{
return
nullptr
;
}
void
invalid_stream_scatterer
::
emit_batches
()
{
// nop
}
stream_scatterer
::
path_type
*
invalid_stream_scatterer
::
find
(
stream_slot
,
const
actor_addr
&
)
{
return
nullptr
;
}
long
invalid_stream_scatterer
::
credit
()
const
{
return
0
;
size_t
invalid_stream_scatterer
::
capacity
()
const
noexcept
{
return
0u
;
}
long
invalid_stream_scatterer
::
buffered
()
const
{
return
0
;
}
long
invalid_stream_scatterer
::
desired_batch_size
()
const
{
return
0
;
}
long
invalid_stream_scatterer
::
min_buffer_size
()
const
{
return
0
;
}
duration
invalid_stream_scatterer
::
max_batch_delay
()
const
{
return
infinite
;
}
void
invalid_stream_scatterer
::
max_batch_delay
(
duration
)
{
// nop
size_t
invalid_stream_scatterer
::
buffered
()
const
noexcept
{
return
0u
;
}
}
// namespace caf
libcaf_core/src/outbound_path.cpp
View file @
ae4d7c58
...
...
@@ -25,10 +25,8 @@
namespace
caf
{
outbound_path
::
outbound_path
(
local_actor
*
selfptr
,
stream_slots
id
,
strong_actor_ptr
ptr
)
outbound_path
::
outbound_path
(
stream_slots
id
,
strong_actor_ptr
ptr
)
:
slots
(
id
),
self
(
selfptr
),
hdl
(
std
::
move
(
ptr
)),
next_batch_id
(
0
),
open_credit
(
0
),
...
...
@@ -39,45 +37,11 @@ outbound_path::outbound_path(local_actor* selfptr, stream_slots id,
}
outbound_path
::~
outbound_path
()
{
CAF_LOG_TRACE
(
CAF_ARG
(
shutdown_reason
));
if
(
hdl
)
{
if
(
shutdown_reason
==
none
)
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
close
>
(
slots
,
self
->
address
()));
else
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
forced_close
>
(
slots
,
self
->
address
(),
shutdown_reason
));
}
if
(
shutdown_reason
!=
none
)
unsafe_response
(
self
,
std
::
move
(
cd
.
hdl
),
no_stages
,
cd
.
mid
,
std
::
move
(
shutdown_reason
));
}
void
outbound_path
::
handle_ack_open
(
long
initial_credit
)
{
open_credit
=
initial_credit
;
cd
.
hdl
=
nullptr
;
}
void
outbound_path
::
emit_open
(
strong_actor_ptr
origin
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
is_redeployable
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
origin
)
<<
CAF_ARG
(
stages
)
<<
CAF_ARG
(
handshake_mid
)
<<
CAF_ARG
(
handshake_data
)
<<
CAF_ARG
(
prio
)
<<
CAF_ARG
(
is_redeployable
));
cd
=
client_data
{
origin
,
handshake_mid
};
redeployable
=
is_redeployable
;
hdl
->
enqueue
(
make_mailbox_element
(
std
::
move
(
origin
),
handshake_mid
,
std
::
move
(
stages
),
make_message
(
open_stream_msg
{
slots
.
sender
,
std
::
move
(
handshake_data
),
self
->
ctrl
(),
hdl
,
prio
,
is_redeployable
})),
self
->
context
());
// nop
}
void
outbound_path
::
emit_batch
(
long
xs_size
,
message
xs
)
{
CAF_
LOG_TRACE
(
CAF_ARG
(
xs_size
)
<<
CAF_ARG
(
xs
)
);
void
outbound_path
::
emit_batch
(
lo
cal_actor
*
self
,
lo
ng
xs_size
,
message
xs
)
{
CAF_
ASSERT
(
open_credit
>=
xs_size
);
open_credit
-=
xs_size
;
auto
bid
=
next_batch_id
++
;
downstream_msg
::
batch
batch
{
static_cast
<
int32_t
>
(
xs_size
),
std
::
move
(
xs
),
...
...
@@ -88,11 +52,21 @@ void outbound_path::emit_batch(long xs_size, message xs) {
downstream_msg
{
slots
,
self
->
address
(),
std
::
move
(
batch
)});
}
void
outbound_path
::
emit_regular_shutdown
(
local_actor
*
self
)
{
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
close
>
(
slots
,
self
->
address
()));
}
void
outbound_path
::
emit_irregular_shutdown
(
local_actor
*
self
,
error
reason
)
{
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
forced_close
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
}
void
outbound_path
::
emit_irregular_shutdown
(
local_actor
*
self
,
stream_slots
slots
,
const
strong_actor_ptr
&
hdl
,
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
reason
));
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
forced_close
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
...
...
libcaf_core/src/stream_manager.cpp
View file @
ae4d7c58
...
...
@@ -33,197 +33,72 @@
namespace
caf
{
stream_manager
::
stream_manager
(
local_actor
*
selfptr
)
:
self_
(
selfptr
)
{
// nop
}
stream_manager
::~
stream_manager
()
{
// nop
}
error
stream_manager
::
open
(
stream_slot
slot
,
strong_actor_ptr
hdl
,
strong_actor_ptr
original_stage
,
stream_priority
prio
,
bool
redeployable
,
response_promise
result_cb
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
original_stage
)
<<
CAF_ARG
(
prio
)
<<
CAF_ARG
(
redeployable
));
/*
if (hdl == nullptr)
return sec::invalid_argument;
if (in().add_path(slot, hdl, std::move(original_stage), prio,
out().credit(), redeployable, std::move(result_cb))
!= nullptr)
return none;
*/
return
sec
::
cannot_add_upstream
;
}
error
stream_manager
::
ack_open
(
stream_slot
slot
,
const
actor_addr
&
rebind_from
,
strong_actor_ptr
rebind_to
,
long
initial_demand
,
long
desired_batch_size
,
bool
redeployable
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
rebind_from
)
<<
CAF_ARG
(
rebind_to
)
<<
CAF_ARG
(
initial_demand
)
<<
CAF_ARG
(
redeployable
));
/*
if (rebind_from == nullptr)
return sec::invalid_argument;
if (rebind_to == nullptr) {
auto from_ptr = actor_cast<strong_actor_ptr>(rebind_from);
if (from_ptr)
out().remove_path(slot, from_ptr, sec::invalid_downstream, false);
return sec::invalid_downstream;
}
auto ptr = out().confirm_path(slot, rebind_from, std::move(rebind_to),
initial_demand, desired_batch_size,
redeployable);
if (ptr == nullptr)
return sec::invalid_downstream;
downstream_demand(ptr, initial_demand);
*/
error
stream_manager
::
handle
(
inbound_path
*
from
,
downstream_msg
::
batch
&
x
)
{
return
none
;
}
error
stream_manager
::
handle
(
inbound_path
*
from
,
downstream_msg
::
close
&
x
)
{
out
().
take_path
(
from
->
slots
);
return
none
;
}
error
stream_manager
::
batch
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
long
xs_size
,
message
&
xs
,
int64_t
xs_id
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
xs_size
)
<<
CAF_ARG
(
xs
)
<<
CAF_ARG
(
xs_id
));
CAF_ASSERT
(
hdl
!=
nullptr
);
/*
auto ptr = in().find(slot, hdl);
if (ptr == nullptr) {
CAF_LOG_WARNING("received batch for unknown stream");
return sec::invalid_downstream;
}
if (xs_size > ptr->assigned_credit) {
CAF_LOG_WARNING("batch size of" << xs_size << "exceeds assigned credit of"
<< ptr->assigned_credit);
return sec::invalid_stream_state;
}
ptr->handle_batch(xs_size, xs_id);
auto err = process_batch(xs);
if (err == none) {
push();
auto current_size = out().buffered();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
}
return err;
*/
error
stream_manager
::
handle
(
inbound_path
*
from
,
downstream_msg
::
forced_close
&
x
)
{
return
none
;
}
error
stream_manager
::
ack_batch
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
long
demand
,
long
batch_size
,
int64_t
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
demand
));
/*
auto ptr = out().find(slot, hdl);
if (ptr == nullptr)
return sec::invalid_downstream;
ptr->open_credit += demand;
if (ptr->desired_batch_size != batch_size)
ptr->desired_batch_size = batch_size;
downstream_demand(ptr, demand);
*/
error
stream_manager
::
handle
(
outbound_path
*
from
,
upstream_msg
::
ack_open
&
x
)
{
return
none
;
}
error
stream_manager
::
close
(
stream_slot
slot
,
const
actor_addr
&
hdl
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
hdl
));
/*
if (in().remove_path(slot, hdl, none, true) && in().closed())
input_closed(none);
*/
error
stream_manager
::
handle
(
outbound_path
*
from
,
upstream_msg
::
ack_batch
&
x
)
{
return
none
;
}
error
stream_manager
::
drop
(
stream_slot
slot
,
const
actor_addr
&
hdl
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
));
/*
if (out().remove_path(slot, hdl, none, true) && out().closed())
output_closed(none);
*/
error
stream_manager
::
handle
(
outbound_path
*
from
,
upstream_msg
::
drop
&
x
)
{
return
none
;
}
error
stream_manager
::
forced_close
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
reason
));
CAF_IGNORE_UNUSED
(
hdl
);
/*
in().remove_path(slot, hdl, reason, true);
abort(std::move(reason));
return sec::unhandled_stream_error;
*/
return
none
;
}
error
stream_manager
::
forced_drop
(
stream_slot
slot
,
const
actor_addr
&
hdl
,
error
reason
)
{
/*
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(hdl);
out().remove_path(slot, hdl, reason, true);
abort(std::move(reason));
return sec::unhandled_stream_error;
*/
return
none
;
}
void
stream_manager
::
abort
(
caf
::
error
reason
)
{
/*
CAF_LOG_TRACE(CAF_ARG(reason));
in().abort(reason);
input_closed(reason);
out().abort(reason);
output_closed(std::move(reason));
*/
error
stream_manager
::
handle
(
outbound_path
*
from
,
upstream_msg
::
forced_drop
&
x
)
{
return
none
;
}
void
stream_manager
::
close
()
{
/*
CAF_LOG_TRACE("");
in().close(make_final_result());
input_closed(none);
out
().
close
();
output_closed(none);
*/
}
bool
stream_manager
::
add_sink
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
{
/*
return out().add_path(slot, std::move(origin), std::move(sink_ptr),
std::move(stages), handshake_mid,
std::move(handshake_data), prio, redeployable)
!= nullptr;
*/
}
bool
stream_manager
::
add_source
(
stream_slot
slot
,
strong_actor_ptr
source_ptr
,
strong_actor_ptr
original_stage
,
stream_priority
prio
,
bool
redeployable
,
response_promise
result_cb
)
{
/*
// TODO: out().credit() gives the same amount of credit to any number of new
// sources -> feedback needed
return in().add_path(slot, std::move(source_ptr), std::move(original_stage),
prio, out().credit(), redeployable, std::move(result_cb))
!= nullptr;
*/
// TODO: abort all input pahts as well
}
void
stream_manager
::
abort
(
error
reason
)
{
out
().
abort
(
std
::
move
(
reason
));
// TODO: abort all input pahts as well
}
void
stream_manager
::
push
()
{
/*
CAF_LOG_TRACE
(
""
);
out
().
emit_batches
();
*/
}
bool
stream_manager
::
generate_messages
()
{
return
false
;
}
void
stream_manager
::
register_input_path
(
inbound_path
*
)
{
// nop
}
void
stream_manager
::
deregister_input_path
(
inbound_path
*
)
noexcept
{
// nop
}
message
stream_manager
::
make_final_result
()
{
return
none
;
}
...
...
libcaf_core/src/stream_scatterer.cpp
View file @
ae4d7c58
...
...
@@ -18,27 +18,155 @@
#include "caf/stream_scatterer.hpp"
#include "caf/logger.hpp"
#include <functional>
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
namespace
caf
{
namespace
{
using
pointer
=
stream_scatterer
::
path_ptr
;
using
unique_pointer
=
stream_scatterer
::
path_unique_ptr
;
}
// namespace <anonymous>
stream_scatterer
::
stream_scatterer
(
local_actor
*
self
)
:
self_
(
self
)
{
// nop
}
stream_scatterer
::~
stream_scatterer
()
{
// nop
}
bool
stream_scatterer
::
remove_path
(
stream_slot
slot
,
const
strong_actor_ptr
&
x
,
error
reason
,
bool
silent
)
{
return
remove_path
(
slot
,
actor_cast
<
actor_addr
>
(
x
),
std
::
move
(
reason
),
silent
);
pointer
stream_scatterer
::
add_path
(
stream_slots
slots
,
strong_actor_ptr
target
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
target
)
<<
CAF_ARG
(
handshake_data
));
auto
res
=
paths_
.
emplace
(
slots
,
nullptr
);
if
(
res
.
second
)
{
auto
ptr
=
new
outbound_path
(
slots
,
std
::
move
(
target
));
res
.
first
->
second
.
reset
(
ptr
);
return
ptr
;
}
return
nullptr
;
}
unique_pointer
stream_scatterer
::
take_path
(
stream_slots
slots
)
noexcept
{
unique_pointer
result
;
auto
i
=
paths_
.
find
(
slots
);
if
(
i
!=
paths_
.
end
())
{
result
.
swap
(
i
->
second
);
paths_
.
erase
(
i
);
}
return
result
;
}
pointer
stream_scatterer
::
path
(
stream_slots
slots
)
noexcept
{
auto
i
=
paths_
.
find
(
slots
);
return
i
!=
paths_
.
end
()
?
i
->
second
.
get
()
:
nullptr
;
}
pointer
stream_scatterer
::
path_at
(
size_t
index
)
noexcept
{
CAF_ASSERT
(
paths_
.
size
()
<
index
);
return
(
paths_
.
container
())[
index
].
second
.
get
();
}
unique_pointer
stream_scatterer
::
take_path_at
(
size_t
index
)
noexcept
{
CAF_ASSERT
(
paths_
.
size
()
<
index
);
unique_pointer
result
;
auto
i
=
paths_
.
begin
()
+
index
;
result
.
swap
(
i
->
second
);
paths_
.
erase
(
i
);
return
result
;
}
bool
stream_scatterer
::
clean
()
const
noexcept
{
auto
pred
=
[](
const
map_type
::
value_type
&
kvp
)
{
auto
&
p
=
*
kvp
.
second
;
return
p
.
next_batch_id
==
p
.
next_ack_id
&&
p
.
unacknowledged_batches
.
empty
();
};
return
buffered
()
==
0
&&
std
::
all_of
(
paths_
.
begin
(),
paths_
.
end
(),
pred
);
}
void
stream_scatterer
::
close
()
{
for
(
auto
i
=
paths_
.
begin
();
i
!=
paths_
.
end
();
++
i
)
about_to_erase
(
i
,
false
,
nullptr
);
paths_
.
clear
();
}
void
stream_scatterer
::
abort
(
error
reason
)
{
auto
&
vec
=
paths_
.
container
();
if
(
vec
.
empty
())
return
;
auto
i
=
paths_
.
begin
();
auto
s
=
paths_
.
end
()
-
1
;
for
(;
i
!=
s
;
++
i
)
{
auto
tmp
=
reason
;
about_to_erase
(
i
,
false
,
&
tmp
);
}
about_to_erase
(
i
,
false
,
&
reason
);
vec
.
clear
();
}
namespace
{
template
<
class
BinOp
>
struct
accumulate_helper
{
using
value_type
=
stream_scatterer
::
map_type
::
value_type
;
BinOp
op
;
size_t
operator
()(
size_t
x
,
const
value_type
&
y
)
const
noexcept
{
return
op
(
x
,
y
.
second
->
open_credit
);
}
size_t
operator
()(
const
value_type
&
x
,
size_t
y
)
const
noexcept
{
return
op
(
x
.
second
->
open_credit
,
y
);
}
};
template
<
class
BinOp
>
struct
accumulate_helper
<
BinOp
>
make_accumulate_helper
(
BinOp
f
)
{
return
{
f
};
}
template
<
class
F
,
class
T
,
class
Container
>
T
fold
(
F
fun
,
T
init
,
const
Container
&
xs
)
{
return
std
::
accumulate
(
xs
.
begin
(),
xs
.
end
(),
std
::
move
(
init
),
make_accumulate_helper
(
std
::
move
(
fun
)));
}
using
bin_op_fptr
=
const
size_t
&
(
*
)(
const
size_t
&
,
const
size_t
&
);
}
// namespace <anonymous>
size_t
stream_scatterer
::
min_credit
()
const
noexcept
{
if
(
empty
())
return
0u
;
return
fold
(
static_cast
<
bin_op_fptr
>
(
&
std
::
min
<
size_t
>
),
std
::
numeric_limits
<
size_t
>::
max
(),
paths_
);
}
size_t
stream_scatterer
::
max_credit
()
const
noexcept
{
return
fold
(
static_cast
<
bin_op_fptr
>
(
&
std
::
max
<
size_t
>
),
size_t
{
0u
},
paths_
);
}
size_t
stream_scatterer
::
total_credit
()
const
noexcept
{
return
fold
(
std
::
plus
<
size_t
>
{},
size_t
{
0u
},
paths_
);
}
stream_scatterer
::
path_type
*
stream_scatterer
::
find
(
stream_slot
slot
,
const
strong_actor_ptr
&
x
)
{
return
find
(
slot
,
actor_cast
<
actor_addr
>
(
x
));
void
stream_scatterer
::
about_to_erase
(
map_type
::
iterator
i
,
bool
silent
,
error
*
reason
)
{
if
(
!
silent
)
{
if
(
reason
==
nullptr
)
i
->
second
->
emit_regular_shutdown
(
self_
);
else
i
->
second
->
emit_irregular_shutdown
(
self_
,
std
::
move
(
*
reason
));
}
}
}
// namespace caf
libcaf_core/src/stream_scatterer_impl.cpp
deleted
100644 → 0
View file @
39601bf2
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_scatterer_impl.hpp"
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
namespace
caf
{
stream_scatterer_impl
::
stream_scatterer_impl
(
local_actor
*
selfptr
)
:
super
(
selfptr
),
max_batch_delay_
(
infinite
)
{
// nop
}
stream_scatterer_impl
::~
stream_scatterer_impl
()
{
// nop
}
stream_scatterer
::
path_ptr
stream_scatterer_impl
::
add_path
(
stream_slot
slot
,
strong_actor_ptr
origin
,
strong_actor_ptr
sink_ptr
,
mailbox_element
::
forwarding_stack
stages
,
message_id
handshake_mid
,
message
handshake_data
,
stream_priority
prio
,
bool
redeployable
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
origin
)
<<
CAF_ARG
(
sink_ptr
)
<<
CAF_ARG
(
stages
)
<<
CAF_ARG
(
handshake_mid
)
<<
CAF_ARG
(
handshake_data
)
<<
CAF_ARG
(
prio
)
<<
CAF_ARG
(
redeployable
));
auto
ptr
=
add_path_impl
(
slot
,
std
::
move
(
sink_ptr
));
if
(
ptr
!=
nullptr
)
ptr
->
emit_open
(
std
::
move
(
origin
),
std
::
move
(
stages
),
handshake_mid
,
std
::
move
(
handshake_data
),
prio
,
redeployable
);
return
ptr
;
}
stream_scatterer
::
path_ptr
stream_scatterer_impl
::
confirm_path
(
stream_slot
,
const
actor_addr
&
,
strong_actor_ptr
,
long
,
long
,
bool
)
{
return
nullptr
;
// TODO: implement me
/*
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(from) << CAF_ARG(to)
<< CAF_ARG(initial_demand) << CAF_ARG(desired_batch_size)
<< CAF_ARG(redeployable));
auto ptr = find(slot, from);
if (ptr == nullptr) {
CAF_LOG_WARNING("cannot confirm unknown path");
outbound_path::emit_irregular_shutdown(self_, slot, std::move(to),
sec::invalid_downstream);
return nullptr;
}
if (from != to)
ptr->hdl = std::move(to);
ptr->redeployable = redeployable;
ptr->open_credit += initial_demand;
if (ptr->desired_batch_size != desired_batch_size)
ptr->desired_batch_size = desired_batch_size;
return ptr;
*/
}
bool
stream_scatterer_impl
::
paths_clean
()
const
{
auto
is_clean
=
[](
const
path_uptr
&
x
)
{
return
x
->
next_ack_id
==
x
->
next_batch_id
;
};
return
buffered
()
==
0
&&
std
::
all_of
(
paths_
.
begin
(),
paths_
.
end
(),
is_clean
);
}
void
stream_scatterer_impl
::
close
()
{
CAF_LOG_TRACE
(
""
);
// TODO: implement me
/*
for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
paths_.clear();
*/
}
void
stream_scatterer_impl
::
abort
(
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
reason
));
// TODO: implement me
/*
for (auto& path : paths_) {
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
path->shutdown_reason = reason;
}
paths_.clear();
*/
}
long
stream_scatterer_impl
::
total_credit
()
const
{
return
total_credit
(
paths_
);
}
long
stream_scatterer_impl
::
min_credit
()
const
{
return
min_credit
(
paths_
);
}
long
stream_scatterer_impl
::
max_credit
()
const
{
return
max_credit
(
paths_
);
}
long
stream_scatterer_impl
::
total_desired_batch_size
()
const
{
return
total_desired_batch_size
(
paths_
);
}
long
stream_scatterer_impl
::
min_desired_batch_size
()
const
{
return
min_desired_batch_size
(
paths_
);
}
long
stream_scatterer_impl
::
max_desired_batch_size
()
const
{
return
max_desired_batch_size
(
paths_
);
}
long
stream_scatterer_impl
::
min_buffer_size
()
const
{
return
50
;
// TODO: at least place the default in a header
}
duration
stream_scatterer_impl
::
max_batch_delay
()
const
{
return
max_batch_delay_
;
}
void
stream_scatterer_impl
::
max_batch_delay
(
duration
x
)
{
max_batch_delay_
=
std
::
move
(
x
);
}
}
// namespace caf
libcaf_core/src/terminal_stream_scatterer.cpp
View file @
ae4d7c58
...
...
@@ -18,12 +18,12 @@
#include "caf/terminal_stream_scatterer.hpp"
#include
"caf/logger.hpp"
#include
<limits>
namespace
caf
{
terminal_stream_scatterer
::
terminal_stream_scatterer
(
local_actor
*
ptr
)
:
super
(
ptr
)
{
terminal_stream_scatterer
::
terminal_stream_scatterer
(
local_actor
*
self
)
:
super
(
self
)
{
// nop
}
...
...
@@ -31,14 +31,8 @@ terminal_stream_scatterer::~terminal_stream_scatterer() {
// nop
}
long
terminal_stream_scatterer
::
credit
()
const
{
// TODO: do something more advanced, yes?
return
50
;
}
long
terminal_stream_scatterer
::
desired_batch_size
()
const
{
// TODO: do something more advanced, yes?
return
50
;
size_t
terminal_stream_scatterer
::
capacity
()
const
noexcept
{
return
std
::
numeric_limits
<
size_t
>::
max
();
}
}
// namespace caf
libcaf_core/test/broadcast_scatterer.cpp
0 → 100644
View file @
ae4d7c58
This diff is collapsed.
Click to expand it.
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