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
4a4c0276
Commit
4a4c0276
authored
May 29, 2023
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Improve performance of the mcast operator
parent
4883df85
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
229 additions
and
93 deletions
+229
-93
libcaf_core/caf/flow/op/mcast.hpp
libcaf_core/caf/flow/op/mcast.hpp
+50
-27
libcaf_core/caf/flow/op/prefix_and_tail.hpp
libcaf_core/caf/flow/op/prefix_and_tail.hpp
+38
-19
libcaf_core/caf/flow/op/publish.hpp
libcaf_core/caf/flow/op/publish.hpp
+51
-7
libcaf_core/caf/flow/op/ucast.hpp
libcaf_core/caf/flow/op/ucast.hpp
+79
-28
libcaf_core/src/flow/scoped_coordinator.cpp
libcaf_core/src/flow/scoped_coordinator.cpp
+7
-11
libcaf_core/test/flow/op/buffer.cpp
libcaf_core/test/flow/op/buffer.cpp
+4
-1
No files found.
libcaf_core/caf/flow/op/mcast.hpp
View file @
4a4c0276
...
@@ -16,6 +16,7 @@
...
@@ -16,6 +16,7 @@
#include <algorithm>
#include <algorithm>
#include <deque>
#include <deque>
#include <memory>
#include <memory>
#include <numeric>
namespace
caf
::
flow
::
op
{
namespace
caf
::
flow
::
op
{
...
@@ -67,7 +68,7 @@ private:
...
@@ -67,7 +68,7 @@ private:
// Base type for *hot* operators that multicast data to subscribed observers.
// Base type for *hot* operators that multicast data to subscribed observers.
template
<
class
T
>
template
<
class
T
>
class
mcast
:
public
hot
<
T
>
{
class
mcast
:
public
hot
<
T
>
,
public
ucast_sub_state_listener
<
T
>
{
public:
public:
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
...
@@ -85,18 +86,30 @@ public:
...
@@ -85,18 +86,30 @@ public:
// nop
// nop
}
}
~
mcast
()
override
{
close
();
}
// -- broadcasting -----------------------------------------------------------
/// Pushes @p item to all subscribers.
/// Pushes @p item to all subscribers.
void
push_all
(
const
T
&
item
)
{
/// @returns `true` if all observers consumed the item immediately without
for
(
auto
&
state
:
states_
)
/// buffering it, `false` otherwise.
state
->
push
(
item
);
bool
push_all
(
const
T
&
item
)
{
return
std
::
accumulate
(
states_
.
begin
(),
states_
.
end
(),
true
,
[
&
item
](
bool
res
,
const
state_ptr_type
&
ptr
)
{
return
res
&
ptr
->
push
(
item
);
});
}
}
/// Closes the operator, eventually emitting on_complete on all observers.
/// Closes the operator, eventually emitting on_complete on all observers.
void
close
()
{
void
close
()
{
if
(
!
closed_
)
{
if
(
!
closed_
)
{
closed_
=
true
;
closed_
=
true
;
for
(
auto
&
state
:
states_
)
for
(
auto
&
state
:
states_
)
{
state
->
listener
=
nullptr
;
state
->
close
();
state
->
close
();
}
states_
.
clear
();
states_
.
clear
();
}
}
}
}
...
@@ -105,13 +118,17 @@ public:
...
@@ -105,13 +118,17 @@ public:
void
abort
(
const
error
&
reason
)
{
void
abort
(
const
error
&
reason
)
{
if
(
!
closed_
)
{
if
(
!
closed_
)
{
closed_
=
true
;
closed_
=
true
;
for
(
auto
&
state
:
states_
)
for
(
auto
&
state
:
states_
)
{
state
->
listener
=
nullptr
;
state
->
abort
(
reason
);
state
->
abort
(
reason
);
}
states_
.
clear
();
states_
.
clear
();
err_
=
reason
;
err_
=
reason
;
}
}
}
}
// -- properties -------------------------------------------------------------
size_t
max_demand
()
const
noexcept
{
size_t
max_demand
()
const
noexcept
{
if
(
states_
.
empty
())
{
if
(
states_
.
empty
())
{
return
0
;
return
0
;
...
@@ -166,20 +183,20 @@ public:
...
@@ -166,20 +183,20 @@ public:
return
states_
.
size
();
return
states_
.
size
();
}
}
// -- state management -------------------------------------------------------
/// Adds state for a new observer to the operator.
state_ptr_type
add_state
(
observer_type
out
)
{
state_ptr_type
add_state
(
observer_type
out
)
{
auto
state
=
make_counted
<
state_type
>
(
super
::
ctx_
,
std
::
move
(
out
));
auto
state
=
make_counted
<
state_type
>
(
super
::
ctx_
,
std
::
move
(
out
));
auto
mc
=
strong_this
();
state
->
listener
=
this
;
state
->
when_disposed
=
make_action
([
mc
,
state
]()
mutable
{
//
mc
->
do_dispose
(
state
);
});
state
->
when_consumed_some
=
make_action
([
mc
,
state
]()
mutable
{
//
mc
->
on_consumed_some
(
*
state
);
});
states_
.
push_back
(
state
);
states_
.
push_back
(
state
);
return
state
;
return
state
;
}
}
disposable
subscribe
(
observer
<
T
>
out
)
override
{
// -- implementation of observable -------------------------------------------
/// Adds a new observer to the operator.
disposable
subscribe
(
observer_type
out
)
override
{
if
(
!
closed_
)
{
if
(
!
closed_
)
{
auto
ptr
=
make_counted
<
mcast_sub
<
T
>>
(
super
::
ctx_
,
add_state
(
out
));
auto
ptr
=
make_counted
<
mcast_sub
<
T
>>
(
super
::
ctx_
,
add_state
(
out
));
out
.
on_subscribe
(
subscription
{
ptr
});
out
.
on_subscribe
(
subscription
{
ptr
});
...
@@ -192,6 +209,23 @@ public:
...
@@ -192,6 +209,23 @@ public:
}
}
}
}
// -- implementation of ucast_sub_state_listener -----------------------------
void
on_disposed
(
state_type
*
ptr
)
final
{
super
::
ctx_
->
delay_fn
([
mc
=
strong_this
(),
sptr
=
state_ptr_type
{
ptr
}]
{
if
(
auto
i
=
std
::
find
(
mc
->
states_
.
begin
(),
mc
->
states_
.
end
(),
sptr
);
i
!=
mc
->
states_
.
end
())
{
// We don't care about preserving the order of elements in the vector.
// Hence, we can swap the element to the back and then pop it.
auto
last
=
mc
->
states_
.
end
()
-
1
;
if
(
i
!=
last
)
std
::
swap
(
*
i
,
*
last
);
mc
->
states_
.
pop_back
();
mc
->
do_dispose
(
sptr
);
}
});
}
protected:
protected:
bool
closed_
=
false
;
bool
closed_
=
false
;
error
err_
;
error
err_
;
...
@@ -202,19 +236,8 @@ private:
...
@@ -202,19 +236,8 @@ private:
return
{
this
};
return
{
this
};
}
}
void
do_dispose
(
state_ptr_type
&
state
)
{
/// Called whenever a state is disposed.
auto
e
=
states_
.
end
();
virtual
void
do_dispose
(
const
state_ptr_type
&
)
{
if
(
auto
i
=
std
::
find
(
states_
.
begin
(),
e
,
state
);
i
!=
e
)
{
states_
.
erase
(
i
);
on_dispose
(
*
state
);
}
}
virtual
void
on_dispose
(
state_type
&
)
{
// nop
}
virtual
void
on_consumed_some
(
state_type
&
)
{
// nop
// nop
}
}
};
};
...
...
libcaf_core/caf/flow/op/prefix_and_tail.hpp
View file @
4a4c0276
...
@@ -23,16 +23,32 @@ namespace caf::flow::op {
...
@@ -23,16 +23,32 @@ namespace caf::flow::op {
template
<
class
T
>
template
<
class
T
>
class
prefix_and_tail_sub
:
public
detail
::
plain_ref_counted
,
class
prefix_and_tail_sub
:
public
detail
::
plain_ref_counted
,
public
observer_impl
<
T
>
,
public
observer_impl
<
T
>
,
public
subscription_impl
{
public
subscription_impl
,
public
ucast_sub_state_listener
<
T
>
{
public:
public:
// -- member types -----------------------------------------------------------
using
tuple_t
=
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>
;
using
tuple_t
=
cow_tuple
<
cow_vector
<
T
>
,
observable
<
T
>>
;
using
state_type
=
ucast_sub_state
<
T
>
;
// -- constructors, destructors, and assignment operators --------------------
prefix_and_tail_sub
(
coordinator
*
ctx
,
observer
<
tuple_t
>
out
,
prefix_and_tail_sub
(
coordinator
*
ctx
,
observer
<
tuple_t
>
out
,
size_t
prefix_size
)
size_t
prefix_size
)
:
ctx_
(
ctx
),
out_
(
std
::
move
(
out
)),
prefix_size_
(
prefix_size
)
{
:
ctx_
(
ctx
),
out_
(
std
::
move
(
out
)),
prefix_size_
(
prefix_size
)
{
prefix_buf_
.
reserve
(
prefix_size
);
prefix_buf_
.
reserve
(
prefix_size
);
}
}
~
prefix_and_tail_sub
()
{
if
(
sink_
)
{
sink_
->
state
().
listener
=
nullptr
;
sink_
->
close
();
}
}
// -- implementation of observer ---------------------------------------------
void
ref_coordinated
()
const
noexcept
override
{
void
ref_coordinated
()
const
noexcept
override
{
ref
();
ref
();
}
}
...
@@ -53,7 +69,7 @@ public:
...
@@ -53,7 +69,7 @@ public:
if
(
prefix_buf_
.
size
()
==
prefix_size_
)
{
if
(
prefix_buf_
.
size
()
==
prefix_size_
)
{
// Create the sink to deliver to tail lazily and deliver the prefix.
// Create the sink to deliver to tail lazily and deliver the prefix.
sink_
=
make_counted
<
ucast
<
T
>>
(
ctx_
);
sink_
=
make_counted
<
ucast
<
T
>>
(
ctx_
);
s
et_callbacks
()
;
s
ink_
->
state
().
listener
=
this
;
// Force member to be null before calling on_next / on_complete.
// Force member to be null before calling on_next / on_complete.
auto
out
=
std
::
move
(
out_
);
auto
out
=
std
::
move
(
out_
);
auto
tup
=
make_cow_tuple
(
cow_vector
<
T
>
{
std
::
move
(
prefix_buf_
)},
auto
tup
=
make_cow_tuple
(
cow_vector
<
T
>
{
std
::
move
(
prefix_buf_
)},
...
@@ -66,7 +82,7 @@ public:
...
@@ -66,7 +82,7 @@ public:
void
on_error
(
const
error
&
reason
)
override
{
void
on_error
(
const
error
&
reason
)
override
{
if
(
sink_
)
{
if
(
sink_
)
{
sink_
->
state
().
when_demand_changed
=
nullptr
;
sink_
->
state
().
listener
=
nullptr
;
sink_
->
abort
(
reason
);
sink_
->
abort
(
reason
);
sub_
=
nullptr
;
sub_
=
nullptr
;
}
else
if
(
out_
)
{
}
else
if
(
out_
)
{
...
@@ -77,7 +93,7 @@ public:
...
@@ -77,7 +93,7 @@ public:
void
on_complete
()
override
{
void
on_complete
()
override
{
if
(
sink_
)
{
if
(
sink_
)
{
sink_
->
state
().
when_demand_changed
=
nullptr
;
sink_
->
state
().
listener
=
nullptr
;
sink_
->
close
();
sink_
->
close
();
sub_
=
nullptr
;
sub_
=
nullptr
;
}
else
if
(
out_
)
{
}
else
if
(
out_
)
{
...
@@ -86,6 +102,8 @@ public:
...
@@ -86,6 +102,8 @@ public:
}
}
}
}
// -- implementation of observable -------------------------------------------
void
on_subscribe
(
flow
::
subscription
sub
)
override
{
void
on_subscribe
(
flow
::
subscription
sub
)
override
{
if
(
!
sub_
&&
out_
)
{
if
(
!
sub_
&&
out_
)
{
sub_
=
std
::
move
(
sub
);
sub_
=
std
::
move
(
sub
);
...
@@ -98,6 +116,8 @@ public:
...
@@ -98,6 +116,8 @@ public:
}
}
}
}
// -- implementation of disposable -------------------------------------------
void
dispose
()
override
{
void
dispose
()
override
{
if
(
out_
)
{
if
(
out_
)
{
out_
=
nullptr
;
out_
=
nullptr
;
...
@@ -129,20 +149,13 @@ public:
...
@@ -129,20 +149,13 @@ public:
}
}
}
}
private:
// -- implementation of ucast_sub_state_listener -----------------------------
intrusive_ptr
<
prefix_and_tail_sub
>
strong_this
()
{
return
{
this
};
}
void
set_callbacks
()
{
void
on_disposed
(
state_type
*
)
override
{
auto
sptr
=
strong_this
();
ctx_
->
delay_fn
([
sptr
=
strong_this
()]
{
sptr
->
do_dispose
();
});
auto
demand_cb
=
[
sptr
]
{
sptr
->
on_sink_demand_change
();
};
sink_
->
state
().
when_demand_changed
=
make_action
(
std
::
move
(
demand_cb
));
auto
disposed_cb
=
[
sptr
]
{
sptr
->
on_sink_dispose
();
};
sink_
->
state
().
when_disposed
=
make_action
(
std
::
move
(
disposed_cb
));
}
}
void
on_
sink_demand_change
()
{
void
on_
demand_changed
(
state_type
*
)
override
{
if
(
sink_
&&
sub_
)
{
if
(
sink_
&&
sub_
)
{
auto
&
st
=
sink_
->
state
();
auto
&
st
=
sink_
->
state
();
auto
pending
=
in_flight_
+
st
.
buf
.
size
();
auto
pending
=
in_flight_
+
st
.
buf
.
size
();
...
@@ -154,14 +167,20 @@ private:
...
@@ -154,14 +167,20 @@ private:
}
}
}
}
void
on_sink_dispose
()
{
private:
intrusive_ptr
<
prefix_and_tail_sub
>
strong_this
()
{
return
{
this
};
}
void
do_dispose
()
{
sink_
=
nullptr
;
sink_
=
nullptr
;
if
(
sub
_
)
{
if
(
out
_
)
{
auto
tmp
=
std
::
move
(
sub
_
);
auto
tmp
=
std
::
move
(
out
_
);
tmp
.
dispos
e
();
tmp
.
on_complet
e
();
}
}
}
}
/// Our scheduling context.
/// Our scheduling context.
coordinator
*
ctx_
;
coordinator
*
ctx_
;
...
...
libcaf_core/caf/flow/op/publish.hpp
View file @
4a4c0276
...
@@ -21,6 +21,8 @@ public:
...
@@ -21,6 +21,8 @@ public:
using
state_type
=
typename
super
::
state_type
;
using
state_type
=
typename
super
::
state_type
;
using
state_ptr_type
=
mcast_sub_state_ptr
<
T
>
;
using
src_ptr
=
intrusive_ptr
<
base
<
T
>>
;
using
src_ptr
=
intrusive_ptr
<
base
<
T
>>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
...
@@ -28,7 +30,11 @@ public:
...
@@ -28,7 +30,11 @@ public:
publish
(
coordinator
*
ctx
,
src_ptr
src
,
publish
(
coordinator
*
ctx
,
src_ptr
src
,
size_t
max_buf_size
=
defaults
::
flow
::
buffer_size
)
size_t
max_buf_size
=
defaults
::
flow
::
buffer_size
)
:
super
(
ctx
),
max_buf_size_
(
max_buf_size
),
source_
(
std
::
move
(
src
))
{
:
super
(
ctx
),
max_buf_size_
(
max_buf_size
),
source_
(
std
::
move
(
src
))
{
// nop
try_request_more_
=
make_action
([
this
]
{
this
->
try_request_more
();
});
}
~
publish
()
override
{
try_request_more_
.
dispose
();
}
}
// -- ref counting (and disambiguation due to multiple base types) -----------
// -- ref counting (and disambiguation due to multiple base types) -----------
...
@@ -86,7 +92,16 @@ public:
...
@@ -86,7 +92,16 @@ public:
void
on_next
(
const
T
&
item
)
override
{
void
on_next
(
const
T
&
item
)
override
{
--
in_flight_
;
--
in_flight_
;
this
->
push_all
(
item
);
if
(
this
->
push_all
(
item
))
{
if
(
in_
&&
this
->
has_observers
())
{
// If push_all returns `true`, it means that all observers have consumed
// the item without buffering it. Hence, we know that
// this->max_buffered() is 0 and we can request more items from the
// source right away.
++
in_flight_
;
in_
.
request
(
1
);
}
}
}
}
void
on_complete
()
override
{
void
on_complete
()
override
{
...
@@ -107,8 +122,18 @@ public:
...
@@ -107,8 +122,18 @@ public:
}
}
}
}
// -- implementation of ucast_sub_state_listener -----------------------------
void
on_consumed_some
(
state_type
*
,
size_t
,
size_t
)
override
{
if
(
!
try_request_more_pending_
)
{
try_request_more_pending_
=
true
;
super
::
ctx_
->
delay
(
try_request_more_
);
}
}
protected:
protected:
void
try_request_more
()
{
void
try_request_more
()
{
try_request_more_pending_
=
false
;
if
(
in_
&&
this
->
has_observers
())
{
if
(
in_
&&
this
->
has_observers
())
{
if
(
auto
buf_size
=
this
->
max_buffered
()
+
in_flight_
;
if
(
auto
buf_size
=
this
->
max_buffered
()
+
in_flight_
;
max_buf_size_
>
buf_size
)
{
max_buf_size_
>
buf_size
)
{
...
@@ -120,7 +145,7 @@ protected:
...
@@ -120,7 +145,7 @@ protected:
}
}
private:
private:
void
on_dispose
(
state
_type
&
)
override
{
void
do_dispose
(
const
state_ptr
_type
&
)
override
{
try_request_more
();
try_request_more
();
if
(
auto_disconnect_
&&
connected_
&&
super
::
observer_count
()
==
0
)
{
if
(
auto_disconnect_
&&
connected_
&&
super
::
observer_count
()
==
0
)
{
connected_
=
false
;
connected_
=
false
;
...
@@ -129,17 +154,36 @@ private:
...
@@ -129,17 +154,36 @@ private:
}
}
}
}
void
on_consumed_some
(
state_type
&
)
override
{
/// Keeps track of the number of items that have been requested but that have
try_request_more
();
/// not yet been delivered.
}
size_t
in_flight_
=
0
;
size_t
in_flight_
=
0
;
/// Maximum number of items to buffer.
size_t
max_buf_size_
;
size_t
max_buf_size_
;
/// Our subscription for fetching items.
subscription
in_
;
subscription
in_
;
/// The source operator we subscribe to lazily.
src_ptr
source_
;
src_ptr
source_
;
/// Keeps track of whether we are connected to the source operator.
bool
connected_
=
false
;
bool
connected_
=
false
;
/// The number of observers that need to connect before we connect to the
/// source operator.
size_t
auto_connect_threshold_
=
std
::
numeric_limits
<
size_t
>::
max
();
size_t
auto_connect_threshold_
=
std
::
numeric_limits
<
size_t
>::
max
();
/// Whether to disconnect from the source operator when the last observer
/// unsubscribes.
bool
auto_disconnect_
=
false
;
bool
auto_disconnect_
=
false
;
/// Scheduled when on_consumed_some() is called. Having this as a member
/// variable avoids allocating a new action object for each call.
action
try_request_more_
;
/// Guards against scheduling `try_request_more_` while it is already pending.
bool
try_request_more_pending_
=
false
;
};
};
}
// namespace caf::flow::op
}
// namespace caf::flow::op
libcaf_core/caf/flow/op/ucast.hpp
View file @
4a4c0276
...
@@ -17,10 +17,13 @@
...
@@ -17,10 +17,13 @@
namespace
caf
::
flow
::
op
{
namespace
caf
::
flow
::
op
{
/// State shared between one multicast operator and one subscribed observer.
/// Shared state between an operator that emits values and the subscribed
/// observer.
template
<
class
T
>
template
<
class
T
>
class
ucast_sub_state
:
public
detail
::
plain_ref_counted
{
class
ucast_sub_state
:
public
detail
::
plain_ref_counted
{
public:
public:
// -- friends ----------------------------------------------------------------
friend
void
intrusive_ptr_add_ref
(
const
ucast_sub_state
*
ptr
)
noexcept
{
friend
void
intrusive_ptr_add_ref
(
const
ucast_sub_state
*
ptr
)
noexcept
{
ptr
->
ref
();
ptr
->
ref
();
}
}
...
@@ -29,6 +32,39 @@ public:
...
@@ -29,6 +32,39 @@ public:
ptr
->
deref
();
ptr
->
deref
();
}
}
// -- member types -----------------------------------------------------------
/// Interface for listeners that want to be notified when a `ucast_sub_state`
/// is disposed, has consumed some items, or when its demand hast changed.
class
abstract_listener
{
public:
virtual
~
abstract_listener
()
{
// nop
}
/// Called when the `ucast_sub_state` is disposed.
virtual
void
on_disposed
(
ucast_sub_state
*
)
=
0
;
/// Called when the `ucast_sub_state` receives new demand.
virtual
void
on_demand_changed
(
ucast_sub_state
*
)
{
// nop
}
/// Called when the `ucast_sub_state` has consumed some items.
/// @param state The `ucast_sub_state` that consumed items.
/// @param old_buffer_size The number of items in the buffer before
/// consuming items.
/// @param new_buffer_size The number of items in the buffer after
/// consuming items.
virtual
void
on_consumed_some
([[
maybe_unused
]]
ucast_sub_state
*
state
,
[[
maybe_unused
]]
size_t
old_buffer_size
,
[[
maybe_unused
]]
size_t
new_buffer_size
)
{
// nop
}
};
// -- constructors, destructors, and assignment operators --------------------
explicit
ucast_sub_state
(
coordinator
*
ptr
)
:
ctx
(
ptr
)
{
explicit
ucast_sub_state
(
coordinator
*
ptr
)
:
ctx
(
ptr
)
{
// nop
// nop
}
}
...
@@ -38,31 +74,49 @@ public:
...
@@ -38,31 +74,49 @@ public:
// nop
// nop
}
}
/// The coordinator for scheduling delayed function calls.
coordinator
*
ctx
;
coordinator
*
ctx
;
/// The buffer for storing items until the observer requests them.
std
::
deque
<
T
>
buf
;
std
::
deque
<
T
>
buf
;
/// The number items that the observer has requested but not yet received.
size_t
demand
=
0
;
size_t
demand
=
0
;
/// The observer to send items to.
observer
<
T
>
out
;
observer
<
T
>
out
;
/// Keeps track of whether this object has been disposed.
bool
disposed
=
false
;
bool
disposed
=
false
;
/// Keeps track of whether this object has been closed.
bool
closed
=
false
;
bool
closed
=
false
;
/// Keeps track of whether `do_run` is currently running.
bool
running
=
false
;
bool
running
=
false
;
/// The error to pass to the observer after the last `on_next` call. If this
/// error is default-constructed, then the observer receives `on_complete`.
/// Otherwise, the observer receives `on_error`.
error
err
;
error
err
;
action
when_disposed
;
/// The listener for state changes. We hold a non-owning pointer to the
action
when_consumed_some
;
/// listener, because the listener owns the state.
a
ction
when_demand_changed
;
a
bstract_listener
*
listener
=
nullptr
;
void
push
(
const
T
&
item
)
{
/// Returns `true` if `item` was consumed, `false` when it was buffered.
[[
nodiscard
]]
bool
push
(
const
T
&
item
)
{
if
(
disposed
)
{
if
(
disposed
)
{
// nop
return
true
;
}
else
if
(
demand
>
0
&&
!
running
)
{
}
else
if
(
demand
>
0
&&
!
running
)
{
CAF_ASSERT
(
out
);
CAF_ASSERT
(
out
);
CAF_ASSERT
(
buf
.
empty
());
CAF_ASSERT
(
buf
.
empty
());
--
demand
;
--
demand
;
out
.
on_next
(
item
);
out
.
on_next
(
item
);
if
(
when_consumed_some
)
return
true
;
ctx
->
delay
(
when_consumed_some
);
}
else
{
}
else
{
buf
.
push_back
(
item
);
buf
.
push_back
(
item
);
return
false
;
}
}
}
}
...
@@ -71,9 +125,7 @@ public:
...
@@ -71,9 +125,7 @@ public:
closed
=
true
;
closed
=
true
;
if
(
!
running
&&
buf
.
empty
())
{
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
disposed
=
true
;
when_disposed
=
nullptr
;
listener
=
nullptr
;
when_consumed_some
=
nullptr
;
when_demand_changed
=
nullptr
;
if
(
out
)
{
if
(
out
)
{
auto
tmp
=
std
::
move
(
out
);
auto
tmp
=
std
::
move
(
out
);
tmp
.
on_complete
();
tmp
.
on_complete
();
...
@@ -88,9 +140,7 @@ public:
...
@@ -88,9 +140,7 @@ public:
err
=
reason
;
err
=
reason
;
if
(
!
running
&&
buf
.
empty
())
{
if
(
!
running
&&
buf
.
empty
())
{
disposed
=
true
;
disposed
=
true
;
when_disposed
=
nullptr
;
listener
=
nullptr
;
when_consumed_some
=
nullptr
;
when_demand_changed
=
nullptr
;
if
(
out
)
{
if
(
out
)
{
auto
tmp
=
std
::
move
(
out
);
auto
tmp
=
std
::
move
(
out
);
tmp
.
on_error
(
reason
);
tmp
.
on_error
(
reason
);
...
@@ -100,17 +150,14 @@ public:
...
@@ -100,17 +150,14 @@ public:
}
}
void
dispose
()
{
void
dispose
()
{
if
(
when_disposed
)
{
ctx
->
delay
(
std
::
move
(
when_disposed
));
}
if
(
when_consumed_some
)
{
auto
tmp
=
std
::
move
(
when_consumed_some
);
tmp
.
dispose
();
}
when_demand_changed
=
nullptr
;
buf
.
clear
();
buf
.
clear
();
demand
=
0
;
demand
=
0
;
disposed
=
true
;
disposed
=
true
;
if
(
listener
)
{
auto
*
lptr
=
listener
;
listener
=
nullptr
;
lptr
->
on_disposed
(
this
);
}
if
(
out
)
{
if
(
out
)
{
auto
tmp
=
std
::
move
(
out
);
auto
tmp
=
std
::
move
(
out
);
tmp
.
on_complete
();
tmp
.
on_complete
();
...
@@ -120,6 +167,7 @@ public:
...
@@ -120,6 +167,7 @@ public:
void
do_run
()
{
void
do_run
()
{
auto
guard
=
detail
::
make_scope_guard
([
this
]
{
running
=
false
;
});
auto
guard
=
detail
::
make_scope_guard
([
this
]
{
running
=
false
;
});
if
(
!
disposed
)
{
if
(
!
disposed
)
{
auto
old_buf_size
=
buf
.
size
();
auto
got_some
=
demand
>
0
&&
!
buf
.
empty
();
auto
got_some
=
demand
>
0
&&
!
buf
.
empty
();
for
(
bool
run
=
got_some
;
run
;
run
=
demand
>
0
&&
!
buf
.
empty
())
{
for
(
bool
run
=
got_some
;
run
;
run
=
demand
>
0
&&
!
buf
.
empty
())
{
out
.
on_next
(
buf
.
front
());
out
.
on_next
(
buf
.
front
());
...
@@ -136,13 +184,16 @@ public:
...
@@ -136,13 +184,16 @@ public:
else
else
tmp
.
on_complete
();
tmp
.
on_complete
();
dispose
();
dispose
();
}
else
if
(
got_some
&&
when_consumed_some
)
{
}
else
if
(
got_some
&&
listener
)
{
ctx
->
delay
(
when_consumed_some
);
listener
->
on_consumed_some
(
this
,
old_buf_size
,
buf
.
size
()
);
}
}
}
}
}
}
};
};
template
<
class
T
>
using
ucast_sub_state_listener
=
typename
ucast_sub_state
<
T
>::
abstract_listener
;
template
<
class
T
>
template
<
class
T
>
using
ucast_sub_state_ptr
=
intrusive_ptr
<
ucast_sub_state
<
T
>>
;
using
ucast_sub_state_ptr
=
intrusive_ptr
<
ucast_sub_state
<
T
>>
;
...
@@ -173,9 +224,9 @@ public:
...
@@ -173,9 +224,9 @@ public:
if
(
!
state_
)
if
(
!
state_
)
return
;
return
;
state_
->
demand
+=
n
;
state_
->
demand
+=
n
;
if
(
state_
->
when_demand_changed
)
if
(
state_
->
listener
)
state_
->
when_demand_changed
.
run
(
);
state_
->
listener
->
on_demand_changed
(
state_
.
get
()
);
if
(
!
state_
->
running
)
{
if
(
!
state_
->
running
&&
!
state_
->
buf
.
empty
()
)
{
state_
->
running
=
true
;
state_
->
running
=
true
;
ctx_
->
delay_fn
([
state
=
state_
]
{
state
->
do_run
();
});
ctx_
->
delay_fn
([
state
=
state_
]
{
state
->
do_run
();
});
}
}
...
@@ -211,7 +262,7 @@ public:
...
@@ -211,7 +262,7 @@ public:
/// Pushes @p item to the subscriber or buffers them until subscribed.
/// Pushes @p item to the subscriber or buffers them until subscribed.
void
push
(
const
T
&
item
)
{
void
push
(
const
T
&
item
)
{
state_
->
push
(
item
);
st
d
::
ignore
=
st
ate_
->
push
(
item
);
}
}
/// Closes the operator, eventually emitting on_complete on all observers.
/// Closes the operator, eventually emitting on_complete on all observers.
...
...
libcaf_core/src/flow/scoped_coordinator.cpp
View file @
4a4c0276
...
@@ -16,27 +16,23 @@ intrusive_ptr<scoped_coordinator> scoped_coordinator::make() {
...
@@ -16,27 +16,23 @@ intrusive_ptr<scoped_coordinator> scoped_coordinator::make() {
void
scoped_coordinator
::
run
()
{
void
scoped_coordinator
::
run
()
{
for
(;;)
{
for
(;;)
{
drop_disposed_flows
();
auto
f
=
next
(
!
watched_disposables_
.
empty
());
auto
f
=
next
(
!
watched_disposables_
.
empty
());
if
(
f
.
ptr
()
!=
nullptr
)
{
if
(
!
f
)
f
.
run
();
drop_disposed_flows
();
}
else
{
return
;
return
;
}
f
.
run
();
}
}
}
}
size_t
scoped_coordinator
::
run_some
()
{
size_t
scoped_coordinator
::
run_some
()
{
size_t
result
=
0
;
size_t
result
=
0
;
for
(;;)
{
for
(;;)
{
drop_disposed_flows
();
auto
f
=
next
(
false
);
auto
f
=
next
(
false
);
if
(
f
.
ptr
()
!=
nullptr
)
{
if
(
!
f
)
++
result
;
f
.
run
();
drop_disposed_flows
();
}
else
{
return
result
;
return
result
;
}
++
result
;
f
.
run
();
}
}
}
}
...
...
libcaf_core/test/flow/op/buffer.cpp
View file @
4a4c0276
...
@@ -118,11 +118,13 @@ SCENARIO("the buffer operator forces items at regular intervals") {
...
@@ -118,11 +118,13 @@ SCENARIO("the buffer operator forces items at regular intervals") {
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
64
},
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
64
},
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
128
,
256
,
512
},
cow_vector
<
int
>
{},
cow_vector
<
int
>
{
128
,
256
,
512
},
};
};
auto
closed
=
std
::
make_shared
<
bool
>
(
false
);
auto
pub
=
flow
::
item_publisher
<
int
>
{
ctx
.
get
()};
auto
pub
=
flow
::
item_publisher
<
int
>
{
ctx
.
get
()};
sys
.
spawn
([
&
pub
,
outputs
](
caf
::
event_based_actor
*
self
)
{
sys
.
spawn
([
&
pub
,
outputs
,
closed
](
caf
::
event_based_actor
*
self
)
{
pub
.
as_observable
()
pub
.
as_observable
()
.
observe_on
(
self
)
//
.
observe_on
(
self
)
//
.
buffer
(
3
,
1s
)
.
buffer
(
3
,
1s
)
.
do_on_complete
([
closed
]
{
*
closed
=
true
;
})
.
for_each
([
outputs
](
const
cow_vector
<
int
>&
xs
)
{
.
for_each
([
outputs
](
const
cow_vector
<
int
>&
xs
)
{
outputs
->
emplace_back
(
xs
);
outputs
->
emplace_back
(
xs
);
});
});
...
@@ -152,6 +154,7 @@ SCENARIO("the buffer operator forces items at regular intervals") {
...
@@ -152,6 +154,7 @@ SCENARIO("the buffer operator forces items at regular intervals") {
advance_time
(
1s
);
advance_time
(
1s
);
sched
.
run
();
sched
.
run
();
CHECK_EQ
(
*
outputs
,
expected
);
CHECK_EQ
(
*
outputs
,
expected
);
CHECK
(
*
closed
);
}
}
}
}
}
}
...
...
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