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
956433d1
Commit
956433d1
authored
Sep 28, 2022
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge topic/neverlord/flow-batching
parents
1aa64f65
3e0bf390
Changes
9
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
437 additions
and
1 deletion
+437
-1
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+1
-0
libcaf_core/caf/cow_vector.hpp
libcaf_core/caf/cow_vector.hpp
+5
-0
libcaf_core/caf/flow/observable.hpp
libcaf_core/caf/flow/observable.hpp
+28
-0
libcaf_core/caf/flow/observable_decl.hpp
libcaf_core/caf/flow/observable_decl.hpp
+7
-0
libcaf_core/caf/flow/op/buffer.hpp
libcaf_core/caf/flow/op/buffer.hpp
+274
-0
libcaf_core/caf/flow/scoped_coordinator.hpp
libcaf_core/caf/flow/scoped_coordinator.hpp
+2
-0
libcaf_core/src/flow/scoped_coordinator.cpp
libcaf_core/src/flow/scoped_coordinator.cpp
+14
-0
libcaf_core/test/flow/buffer.cpp
libcaf_core/test/flow/buffer.cpp
+105
-0
libcaf_core/test/flow/interval.cpp
libcaf_core/test/flow/interval.cpp
+1
-1
No files found.
libcaf_core/CMakeLists.txt
View file @
956433d1
...
...
@@ -275,6 +275,7 @@ caf_add_component(
dynamic_spawn
error
expected
flow.buffer
flow.concat
flow.concat_map
flow.defer
...
...
libcaf_core/caf/cow_vector.hpp
View file @
956433d1
...
...
@@ -9,6 +9,7 @@
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <initializer_list>
#include <vector>
namespace
caf
{
...
...
@@ -41,6 +42,10 @@ public:
impl_
=
make_counted
<
impl
>
(
std
::
move
(
std
));
}
explicit
cow_vector
(
std
::
initializer_list
<
T
>
values
)
{
impl_
=
make_counted
<
impl
>
(
std_type
{
values
});
}
cow_vector
(
cow_vector
&&
)
noexcept
=
default
;
cow_vector
(
const
cow_vector
&
)
noexcept
=
default
;
...
...
libcaf_core/caf/flow/observable.hpp
View file @
956433d1
...
...
@@ -19,10 +19,13 @@
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/op/buffer.hpp"
#include "caf/flow/op/concat.hpp"
#include "caf/flow/op/from_resource.hpp"
#include "caf/flow/op/from_steps.hpp"
#include "caf/flow/op/interval.hpp"
#include "caf/flow/op/merge.hpp"
#include "caf/flow/op/never.hpp"
#include "caf/flow/op/prefetch.hpp"
#include "caf/flow/op/publish.hpp"
#include "caf/flow/step/all.hpp"
...
...
@@ -219,6 +222,15 @@ public:
return
add_step
(
step
::
take
<
output_type
>
{
n
});
}
/// @copydoc observable::buffer
auto
buffer
(
size_t
count
)
&&
{
return
materialize
().
buffer
(
count
);
}
auto
buffer
(
size_t
count
,
timespan
period
)
{
return
materialize
().
buffer
(
count
,
period
);
}
template
<
class
Predicate
>
auto
filter
(
Predicate
predicate
)
&&
{
return
add_step
(
step
::
filter
<
Predicate
>
{
std
::
move
(
predicate
)});
...
...
@@ -566,6 +578,22 @@ observable<T>::take_while(Predicate predicate) {
return
transform
(
step
::
take_while
{
std
::
move
(
predicate
)});
}
template
<
class
T
>
observable
<
cow_vector
<
T
>>
observable
<
T
>::
buffer
(
size_t
count
)
{
using
trait_t
=
op
::
buffer_default_trait
<
T
>
;
using
impl_t
=
op
::
buffer
<
trait_t
>
;
return
make_observable
<
impl_t
>
(
ctx
(),
count
,
*
this
,
make_observable
<
op
::
never
<
unit_t
>>
(
ctx
()));
}
template
<
class
T
>
observable
<
cow_vector
<
T
>>
observable
<
T
>::
buffer
(
size_t
count
,
timespan
period
)
{
using
trait_t
=
op
::
buffer_interval_trait
<
T
>
;
using
impl_t
=
op
::
buffer
<
trait_t
>
;
return
make_observable
<
impl_t
>
(
ctx
(),
count
,
*
this
,
make_observable
<
op
::
interval
>
(
ctx
(),
period
,
period
));
}
// -- observable: combining ----------------------------------------------------
template
<
class
T
>
...
...
libcaf_core/caf/flow/observable_decl.hpp
View file @
956433d1
...
...
@@ -141,6 +141,13 @@ public:
.
filter
([](
const
vector_type
&
xs
)
{
return
!
xs
.
empty
();
});
}
/// Emits items in buffers of size @p count.
observable
<
cow_vector
<
T
>>
buffer
(
size_t
count
);
/// Emits items in buffers of size up to @p count and forces an item at
/// regular intervals .
observable
<
cow_vector
<
T
>>
buffer
(
size_t
count
,
timespan
period
);
// -- combining --------------------------------------------------------------
/// Combines the output of multiple @ref observable objects into one by
...
...
libcaf_core/caf/flow/op/buffer.hpp
0 → 100644
View file @
956433d1
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/cow_vector.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/unit.hpp"
#include <vector>
namespace
caf
::
flow
::
op
{
struct
buffer_input_t
{};
struct
buffer_emit_t
{};
template
<
class
T
>
struct
buffer_default_trait
{
static
constexpr
bool
skip_empty
=
false
;
using
input_type
=
T
;
using
output_type
=
cow_vector
<
T
>
;
using
select_token_type
=
unit_t
;
output_type
operator
()(
const
std
::
vector
<
input_type
>&
xs
)
{
return
output_type
{
xs
};
}
};
template
<
class
T
>
struct
buffer_interval_trait
{
static
constexpr
bool
skip_empty
=
false
;
using
input_type
=
T
;
using
output_type
=
cow_vector
<
T
>
;
using
select_token_type
=
int64_t
;
output_type
operator
()(
const
std
::
vector
<
input_type
>&
xs
)
{
return
output_type
{
xs
};
}
};
///
template
<
class
Trait
>
class
buffer_sub
:
public
subscription
::
impl_base
{
public:
// -- member types -----------------------------------------------------------
using
input_type
=
typename
Trait
::
input_type
;
using
output_type
=
typename
Trait
::
output_type
;
using
select_token_type
=
typename
Trait
::
select_token_type
;
// -- constants --------------------------------------------------------------
static
constexpr
size_t
val_id
=
0
;
static
constexpr
size_t
ctrl_id
=
1
;
// -- constructors, destructors, and assignment operators --------------------
buffer_sub
(
coordinator
*
ctx
,
size_t
max_buf_size
,
observer
<
output_type
>
out
)
:
ctx_
(
ctx
),
max_buf_size_
(
max_buf_size
),
out_
(
std
::
move
(
out
))
{
// nop
}
// -- callbacks for the parent -----------------------------------------------
void
init
(
observable
<
input_type
>
vals
,
observable
<
select_token_type
>
ctrl
)
{
using
val_fwd_t
=
forwarder
<
input_type
,
buffer_sub
,
buffer_input_t
>
;
using
ctrl_fwd_t
=
forwarder
<
select_token_type
,
buffer_sub
,
buffer_emit_t
>
;
vals
.
subscribe
(
make_counted
<
val_fwd_t
>
(
this
,
buffer_input_t
{})
->
as_observer
());
ctrl
.
subscribe
(
make_counted
<
ctrl_fwd_t
>
(
this
,
buffer_emit_t
{})
->
as_observer
());
}
// -- callbacks for the forwarders -------------------------------------------
void
fwd_on_subscribe
(
buffer_input_t
,
subscription
sub
)
{
if
(
!
value_sub_
&&
out_
)
{
value_sub_
=
std
::
move
(
sub
);
if
(
pending_demand_
>
0
)
{
value_sub_
.
request
(
pending_demand_
);
pending_demand_
=
0
;
}
}
else
{
sub
.
dispose
();
}
}
void
fwd_on_complete
(
buffer_input_t
)
{
CAF_ASSERT
(
value_sub_
.
valid
());
CAF_ASSERT
(
out_
.
valid
());
value_sub_
=
nullptr
;
if
(
!
buf_
.
empty
())
do_emit
();
out_
.
on_complete
();
out_
=
nullptr
;
if
(
control_sub_
)
{
control_sub_
.
dispose
();
control_sub_
=
nullptr
;
}
}
void
fwd_on_error
(
buffer_input_t
,
const
error
&
what
)
{
value_sub_
=
nullptr
;
do_abort
(
what
);
}
void
fwd_on_next
(
buffer_input_t
,
const
input_type
&
item
)
{
buf_
.
push_back
(
item
);
if
(
buf_
.
size
()
==
max_buf_size_
)
do_emit
();
}
void
fwd_on_subscribe
(
buffer_emit_t
,
subscription
sub
)
{
if
(
!
control_sub_
&&
out_
)
{
control_sub_
=
std
::
move
(
sub
);
control_sub_
.
request
(
1
);
}
else
{
sub
.
dispose
();
}
}
void
fwd_on_complete
(
buffer_emit_t
)
{
do_abort
(
make_error
(
sec
::
end_of_stream
,
"buffer: unexpected end of the control stream"
));
}
void
fwd_on_error
(
buffer_emit_t
,
const
error
&
what
)
{
control_sub_
=
nullptr
;
do_abort
(
what
);
}
void
fwd_on_next
(
buffer_emit_t
,
select_token_type
)
{
if
constexpr
(
Trait
::
skip_empty
)
{
if
(
!
buf_
.
empty
())
do_emit
();
}
else
{
do_emit
();
}
control_sub_
.
request
(
1
);
}
// -- implementation of subscription -----------------------------------------
bool
disposed
()
const
noexcept
override
{
return
!
out_
;
}
void
dispose
()
override
{
if
(
out_
)
{
ctx_
->
delay_fn
([
strong_this
=
intrusive_ptr
<
buffer_sub
>
{
this
}]
{
strong_this
->
do_dispose
();
});
}
}
void
request
(
size_t
n
)
override
{
CAF_ASSERT
(
out_
.
valid
());
if
(
value_sub_
)
value_sub_
.
request
(
n
);
else
pending_demand_
+=
n
;
}
private:
void
do_emit
()
{
Trait
f
;
out_
.
on_next
(
f
(
buf_
));
buf_
.
clear
();
}
void
do_dispose
()
{
if
(
value_sub_
)
{
value_sub_
.
dispose
();
value_sub_
=
nullptr
;
}
if
(
control_sub_
)
{
control_sub_
.
dispose
();
control_sub_
=
nullptr
;
}
if
(
out_
)
{
out_
.
on_complete
();
}
}
void
do_abort
(
const
error
&
reason
)
{
if
(
value_sub_
)
{
value_sub_
.
dispose
();
value_sub_
=
nullptr
;
}
if
(
control_sub_
)
{
control_sub_
.
dispose
();
control_sub_
=
nullptr
;
}
if
(
out_
)
{
out_
.
on_error
(
reason
);
}
}
/// Stores the context (coordinator) that runs this flow.
coordinator
*
ctx_
;
/// Stores the maximum buffer size before forcing a batch.
size_t
max_buf_size_
;
/// Stores the elements until we can emit them.
std
::
vector
<
input_type
>
buf_
;
/// Stores a handle to the subscribed observer.
observer
<
output_type
>
out_
;
/// Our subscription for the values.
subscription
value_sub_
;
/// Our subscription for the control tokens.
subscription
control_sub_
;
///
size_t
pending_demand_
=
0
;
};
template
<
class
Trait
>
class
buffer
:
public
cold
<
typename
Trait
::
output_type
>
{
public:
// -- member types -----------------------------------------------------------
using
input_type
=
typename
Trait
::
input_type
;
using
output_type
=
typename
Trait
::
output_type
;
using
super
=
cold
<
output_type
>
;
using
input
=
observable
<
input_type
>
;
using
selector
=
observable
<
typename
Trait
::
select_token_type
>
;
// -- constructors, destructors, and assignment operators --------------------
buffer
(
coordinator
*
ctx
,
size_t
max_items
,
input
in
,
selector
select
)
:
super
(
ctx
),
max_items_
(
max_items
),
in_
(
std
::
move
(
in
)),
select_
(
std
::
move
(
select
))
{
// nop
}
// -- implementation of observable<T> -----------------------------------
disposable
subscribe
(
observer
<
output_type
>
out
)
override
{
auto
ptr
=
make_counted
<
buffer_sub
<
Trait
>>
(
super
::
ctx_
,
max_items_
,
out
);
ptr
->
init
(
in_
,
select_
);
out
.
on_subscribe
(
subscription
{
ptr
});
return
ptr
->
as_disposable
();
}
private:
/// Configures how many items get pushed into one buffer at most.
size_t
max_items_
;
/// Sequence of input values.
input
in_
;
/// Sequence of control messages to select previous inputs.
selector
select_
;
};
}
// namespace caf::flow::op
libcaf_core/caf/flow/scoped_coordinator.hpp
View file @
956433d1
...
...
@@ -25,6 +25,8 @@ public:
void
run
();
size_t
run_some
();
// -- reference counting -----------------------------------------------------
void
ref_execution_context
()
const
noexcept
override
;
...
...
libcaf_core/src/flow/scoped_coordinator.cpp
View file @
956433d1
...
...
@@ -26,6 +26,20 @@ void scoped_coordinator::run() {
}
}
size_t
scoped_coordinator
::
run_some
()
{
size_t
result
=
0
;
for
(;;)
{
auto
f
=
next
(
false
);
if
(
f
.
ptr
()
!=
nullptr
)
{
++
result
;
f
.
run
();
drop_disposed_flows
();
}
else
{
return
result
;
}
}
}
// -- reference counting -------------------------------------------------------
void
scoped_coordinator
::
ref_execution_context
()
const
noexcept
{
...
...
libcaf_core/test/flow/buffer.cpp
0 → 100644
View file @
956433d1
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.buffer
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/item_publisher.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using
namespace
caf
;
using
namespace
std
::
literals
;
namespace
{
struct
fixture
:
test_coordinator_fixture
<>
{
flow
::
scoped_coordinator_ptr
ctx
=
flow
::
make_scoped_coordinator
();
};
}
// namespace
BEGIN_FIXTURE_SCOPE
(
fixture
)
SCENARIO
(
"the buffer operator groups items together"
)
{
GIVEN
(
"an observable"
)
{
WHEN
(
"calling .buffer(3)"
)
{
THEN
(
"the observer receives values in groups of three"
)
{
auto
inputs
=
std
::
vector
<
int
>
{
1
,
2
,
4
,
8
,
16
,
32
,
64
,
128
};
auto
outputs
=
std
::
vector
<
cow_vector
<
int
>>
{};
auto
expected
=
std
::
vector
<
cow_vector
<
int
>>
{
cow_vector
<
int
>
{
1
,
2
,
4
},
cow_vector
<
int
>
{
8
,
16
,
32
},
cow_vector
<
int
>
{
64
,
128
},
};
ctx
->
make_observable
()
.
from_container
(
inputs
)
//
.
buffer
(
3
)
.
for_each
([
&
outputs
](
const
cow_vector
<
int
>&
xs
)
{
outputs
.
emplace_back
(
xs
);
});
ctx
->
run
();
CHECK_EQ
(
outputs
,
expected
);
}
}
}
}
SCENARIO
(
"the buffer operator forces items at regular intervals"
)
{
GIVEN
(
"an observable"
)
{
WHEN
(
"calling .buffer(3, 1s)"
)
{
THEN
(
"the observer receives values in groups of three or after 1s"
)
{
auto
outputs
=
std
::
make_shared
<
std
::
vector
<
cow_vector
<
int
>>>
();
auto
expected
=
std
::
vector
<
cow_vector
<
int
>>
{
cow_vector
<
int
>
{
1
,
2
,
4
},
cow_vector
<
int
>
{
8
,
16
,
32
},
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
64
},
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
128
,
256
,
512
},
};
auto
pub
=
flow
::
item_publisher
<
int
>
{
ctx
.
get
()};
sys
.
spawn
([
&
pub
,
outputs
](
caf
::
event_based_actor
*
self
)
{
pub
//
.
as_observable
()
.
observe_on
(
self
)
.
buffer
(
3
,
1s
)
.
for_each
([
outputs
](
const
cow_vector
<
int
>&
xs
)
{
outputs
->
emplace_back
(
xs
);
});
});
sched
.
run
();
MESSAGE
(
"emit the first six items"
);
pub
.
push
({
1
,
2
,
4
,
8
,
16
,
32
});
ctx
->
run_some
();
sched
.
run
();
MESSAGE
(
"force an empty buffer"
);
advance_time
(
1s
);
sched
.
run
();
MESSAGE
(
"force a buffer with a single element"
);
pub
.
push
(
64
);
ctx
->
run_some
();
sched
.
run
();
advance_time
(
1s
);
sched
.
run
();
MESSAGE
(
"force an empty buffer"
);
advance_time
(
1s
);
sched
.
run
();
MESSAGE
(
"emit the last items and close the source"
);
pub
.
push
({
128
,
256
,
512
});
pub
.
close
();
ctx
->
run_some
();
sched
.
run
();
advance_time
(
1s
);
sched
.
run
();
CHECK_EQ
(
*
outputs
,
expected
);
}
}
}
}
END_FIXTURE_SCOPE
()
libcaf_core/test/flow/interval.cpp
View file @
956433d1
...
...
@@ -47,7 +47,7 @@ SCENARIO("scheduled actors schedule observable intervals on the actor clock") {
WHEN
(
"an observer subscribes to it"
)
{
THEN
(
"the actor uses the actor clock to schedule flow processing"
)
{
auto
outputs
=
i64_list
{};
sys
.
spawn
([
&
outputs
](
caf
::
event_based_actor
*
self
)
{
sys
.
spawn
([
&
outputs
](
event_based_actor
*
self
)
{
self
->
make_observable
()
.
interval
(
50ms
,
25ms
)
.
take
(
3
)
...
...
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