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
d1f11390
Commit
d1f11390
authored
Apr 11, 2018
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix sign and conversion warnings
parent
e75871fc
Changes
18
Show whitespace changes
Inline
Side-by-side
Showing
18 changed files
with
85 additions
and
74 deletions
+85
-74
libcaf_core/caf/buffered_downstream_manager.hpp
libcaf_core/caf/buffered_downstream_manager.hpp
+4
-4
libcaf_core/caf/detail/tick_emitter.hpp
libcaf_core/caf/detail/tick_emitter.hpp
+5
-5
libcaf_core/caf/outbound_path.hpp
libcaf_core/caf/outbound_path.hpp
+9
-6
libcaf_core/caf/scheduled_actor.hpp
libcaf_core/caf/scheduled_actor.hpp
+2
-2
libcaf_core/src/blocking_actor.cpp
libcaf_core/src/blocking_actor.cpp
+1
-1
libcaf_core/src/downstream_messages.cpp
libcaf_core/src/downstream_messages.cpp
+1
-1
libcaf_core/src/outbound_path.cpp
libcaf_core/src/outbound_path.cpp
+1
-1
libcaf_core/src/scheduled_actor.cpp
libcaf_core/src/scheduled_actor.cpp
+3
-2
libcaf_core/src/tick_emitter.cpp
libcaf_core/src/tick_emitter.cpp
+9
-8
libcaf_core/test/broadcast_downstream_manager.cpp
libcaf_core/test/broadcast_downstream_manager.cpp
+5
-4
libcaf_core/test/continuous_streaming.cpp
libcaf_core/test/continuous_streaming.cpp
+3
-3
libcaf_core/test/fused_streaming.cpp
libcaf_core/test/fused_streaming.cpp
+2
-2
libcaf_core/test/mock_streaming_classes.cpp
libcaf_core/test/mock_streaming_classes.cpp
+2
-2
libcaf_core/test/native_streaming_classes.cpp
libcaf_core/test/native_streaming_classes.cpp
+10
-6
libcaf_core/test/pipeline_streaming.cpp
libcaf_core/test/pipeline_streaming.cpp
+9
-9
libcaf_core/test/tick_emitter.cpp
libcaf_core/test/tick_emitter.cpp
+16
-15
libcaf_core/test/unordered_flat_map.cpp
libcaf_core/test/unordered_flat_map.cpp
+1
-1
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
+2
-2
No files found.
libcaf_core/caf/buffered_downstream_manager.hpp
View file @
d1f11390
...
...
@@ -55,12 +55,12 @@ public:
}
/// @pre `n <= buf_.size()`
static
chunk_type
get_chunk
(
buffer_type
&
buf
,
long
n
)
{
static
chunk_type
get_chunk
(
buffer_type
&
buf
,
size_t
n
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
buf
)
<<
CAF_ARG
(
n
));
chunk_type
xs
;
if
(
!
buf
.
empty
()
&&
n
>
0
)
{
xs
.
reserve
(
std
::
min
(
static_cast
<
size_t
>
(
n
)
,
buf
.
size
()));
if
(
static_cast
<
size_t
>
(
n
)
<
buf
.
size
())
{
xs
.
reserve
(
std
::
min
(
n
,
buf
.
size
()));
if
(
n
<
buf
.
size
())
{
auto
first
=
buf
.
begin
();
auto
last
=
first
+
static_cast
<
ptrdiff_t
>
(
n
);
std
::
move
(
first
,
last
,
std
::
back_inserter
(
xs
));
...
...
@@ -73,7 +73,7 @@ public:
return
xs
;
}
chunk_type
get_chunk
(
long
n
)
{
chunk_type
get_chunk
(
size_t
n
)
{
return
get_chunk
(
buf_
,
n
);
}
...
...
libcaf_core/caf/detail/tick_emitter.hpp
View file @
d1f11390
...
...
@@ -70,24 +70,24 @@ public:
template
<
class
F
>
void
update
(
time_point
now
,
F
&
consumer
)
{
CAF_ASSERT
(
interval_
.
count
()
!=
0
);
auto
d
iff
=
now
-
start_
;
auto
current_tick_id
=
diff
.
count
()
/
interval_
.
count
(
);
auto
d
=
now
-
start_
;
auto
current_tick_id
=
static_cast
<
size_t
>
(
d
.
count
()
/
interval_
.
count
()
);
while
(
last_tick_id_
<
current_tick_id
)
consumer
(
++
last_tick_id_
);
}
/// Advances time by `t` and returns all triggered periods as bitmask.
long
timeouts
(
time_point
t
,
std
::
initializer_list
<
long
>
periods
);
size_t
timeouts
(
time_point
t
,
std
::
initializer_list
<
size_t
>
periods
);
/// Returns the next time point after `t` that would trigger any of the tick
/// periods, i.e., returns the next time where any of the tick periods
/// triggers `tick id % period == 0`.
time_point
next_timeout
(
time_point
t
,
std
::
initializer_list
<
long
>
periods
);
time_point
next_timeout
(
time_point
t
,
std
::
initializer_list
<
size_t
>
periods
);
private:
time_point
start_
;
duration_type
interval_
;
long
last_tick_id_
;
size_t
last_tick_id_
;
};
}
// namespace detail
...
...
libcaf_core/caf/outbound_path.hpp
View file @
d1f11390
...
...
@@ -72,14 +72,15 @@ public:
/// Sends a `downstream_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
);
void
emit_batch
(
local_actor
*
self
,
int32_t
xs_size
,
message
xs
);
template
<
class
Iterator
>
Iterator
emit_batches_impl
(
local_actor
*
self
,
Iterator
i
,
Iterator
e
,
bool
force_underfull
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
force_underfull
));
using
type
=
detail
::
decay_t
<
decltype
(
*
i
)
>
;
while
(
std
::
distance
(
i
,
e
)
>=
static_cast
<
ptrdiff_t
>
(
desired_batch_size
))
{
// Signed Desired Batch Size.
while
(
std
::
distance
(
i
,
e
)
>=
desired_batch_size
)
{
std
::
vector
<
type
>
tmp
{
std
::
make_move_iterator
(
i
),
std
::
make_move_iterator
(
i
+
desired_batch_size
)};
emit_batch
(
self
,
desired_batch_size
,
make_message
(
std
::
move
(
tmp
)));
...
...
@@ -88,7 +89,7 @@ public:
if
(
i
!=
e
&&
force_underfull
)
{
std
::
vector
<
type
>
tmp
{
std
::
make_move_iterator
(
i
),
std
::
make_move_iterator
(
e
)};
auto
tmp_size
=
static_cast
<
long
>
(
tmp
.
size
());
auto
tmp_size
=
static_cast
<
int32_t
>
(
tmp
.
size
());
emit_batch
(
self
,
tmp_size
,
make_message
(
std
::
move
(
tmp
)));
return
e
;
}
...
...
@@ -104,8 +105,10 @@ public:
if
(
pending
())
return
;
CAF_ASSERT
(
desired_batch_size
>
0
);
CAF_ASSERT
(
cache
.
size
()
<
std
::
numeric_limits
<
int32_t
>::
max
());
auto
first
=
cache
.
begin
();
auto
last
=
first
+
std
::
min
(
open_credit
,
static_cast
<
long
>
(
cache
.
size
()));
auto
last
=
first
+
std
::
min
(
open_credit
,
static_cast
<
int32_t
>
(
cache
.
size
()));
if
(
first
==
last
)
return
;
auto
i
=
emit_batches_impl
(
self
,
first
,
last
,
force_underfull
);
...
...
@@ -147,10 +150,10 @@ public:
int64_t
next_batch_id
;
/// Currently available credit on this path.
long
open_credit
;
int32_t
open_credit
;
/// Ideal batch size. Configured by the sink.
uint64
_t
desired_batch_size
;
int32
_t
desired_batch_size
;
/// ID of the first unacknowledged batch. Note that CAF uses accumulative
/// ACKs, i.e., receiving an ACK with a higher ID is not an error.
...
...
libcaf_core/caf/scheduled_actor.hpp
View file @
d1f11390
...
...
@@ -885,10 +885,10 @@ protected:
detail
::
tick_emitter
stream_ticks_
;
/// Number of ticks per batch delay.
long
max_batch_delay_ticks_
;
size_t
max_batch_delay_ticks_
;
/// Number of ticks of each credit round.
long
credit_round_ticks_
;
size_t
credit_round_ticks_
;
/// Pointer to a private thread object associated with a detached actor.
detail
::
private_thread
*
private_thread_
;
...
...
libcaf_core/src/blocking_actor.cpp
View file @
d1f11390
...
...
@@ -94,7 +94,7 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
// actor lives in its own thread
ptr
->
home_system
->
thread_started
();
auto
this_ptr
=
ptr
->
get
();
CAF_ASSERT
(
dynamic_cast
<
blocking_actor
*>
(
this_ptr
)
!=
0
);
CAF_ASSERT
(
dynamic_cast
<
blocking_actor
*>
(
this_ptr
)
!=
nullptr
);
auto
self
=
static_cast
<
blocking_actor
*>
(
this_ptr
);
CAF_SET_LOGGER_SYS
(
ptr
->
home_system
);
CAF_PUSH_AID_FROM_PTR
(
self
);
...
...
libcaf_core/src/downstream_messages.cpp
View file @
d1f11390
...
...
@@ -65,7 +65,7 @@ bool downstream_messages::enabled(const nested_queue_type& q) noexcept {
auto
downstream_messages
::
quantum
(
const
nested_queue_type
&
q
,
deficit_type
x
)
noexcept
->
deficit_type
{
// TODO: adjust quantum based on the stream priority
return
x
*
q
.
policy
().
handler
->
desired_batch_size
;
return
x
*
static_cast
<
deficit_type
>
(
q
.
policy
().
handler
->
desired_batch_size
)
;
}
}
// namespace policy
...
...
libcaf_core/src/outbound_path.cpp
View file @
d1f11390
...
...
@@ -56,7 +56,7 @@ void outbound_path::emit_open(local_actor* self, stream_slot slot,
nullptr
,
prio
});
}
void
outbound_path
::
emit_batch
(
local_actor
*
self
,
long
xs_size
,
message
xs
)
{
void
outbound_path
::
emit_batch
(
local_actor
*
self
,
int32_t
xs_size
,
message
xs
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
xs_size
)
<<
CAF_ARG
(
xs
));
CAF_ASSERT
(
xs_size
>
0
);
CAF_ASSERT
(
xs_size
<=
std
::
numeric_limits
<
int32_t
>::
max
());
...
...
libcaf_core/src/scheduled_actor.cpp
View file @
d1f11390
...
...
@@ -1099,13 +1099,14 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
// Fill up credit on each input path.
if
((
bitmask
&
0x02
)
!=
0
)
{
auto
cycle
=
stream_ticks_
.
interval
();
cycle
*=
credit_round_ticks_
;
cycle
*=
static_cast
<
decltype
(
cycle
)
>
credit_round_ticks_
;
auto
bc_us
=
home_system
().
config
().
streaming_desired_batch_complexity_us
;
auto
bc
=
std
::
chrono
::
microseconds
(
bc_us
);
auto
&
qs
=
get
<
2
>
(
mailbox_
.
queue
().
queues
()).
queues
();
for
(
auto
&
kvp
:
qs
)
{
auto
inptr
=
kvp
.
second
.
policy
().
handler
.
get
();
inptr
->
emit_ack_batch
(
this
,
kvp
.
second
.
total_task_size
(),
cycle
,
bc
);
auto
bs
=
static_cast
<
int32_t
>
(
kvp
.
second
.
total_task_size
());
inptr
->
emit_ack_batch
(
this
,
bs
,
cycle
,
bc
);
}
}
return
stream_ticks_
.
next_timeout
(
now
,
{
max_batch_delay_ticks_
,
...
...
libcaf_core/src/tick_emitter.cpp
View file @
d1f11390
...
...
@@ -54,12 +54,12 @@ void tick_emitter::interval(duration_type x) {
interval_
=
x
;
}
long
tick_emitter
::
timeouts
(
time_point
now
,
std
::
initializer_list
<
long
>
periods
)
{
size_t
tick_emitter
::
timeouts
(
time_point
now
,
std
::
initializer_list
<
size_t
>
periods
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
now
)
<<
CAF_ARG
(
periods
));
long
result
=
0
;
auto
f
=
[
&
](
long
tick
)
{
long
n
=
0
;
size_t
result
=
0
;
auto
f
=
[
&
](
size_t
tick
)
{
size_t
n
=
0
;
for
(
auto
p
:
periods
)
{
if
(
tick
%
p
==
0
)
result
|=
1l
<<
n
;
...
...
@@ -71,16 +71,17 @@ long tick_emitter::timeouts(time_point now,
}
tick_emitter
::
time_point
tick_emitter
::
next_timeout
(
time_point
t
,
std
::
initializer_list
<
long
>
periods
)
{
tick_emitter
::
next_timeout
(
time_point
t
,
std
::
initializer_list
<
size_t
>
periods
)
{
CAF_ASSERT
(
interval_
.
count
()
!=
0
);
auto
is_trigger
=
[
&
](
long
tick_id
)
{
auto
is_trigger
=
[
&
](
size_t
tick_id
)
{
for
(
auto
period
:
periods
)
if
(
tick_id
%
period
==
0
)
return
true
;
return
false
;
};
auto
diff
=
t
-
start_
;
auto
this_tick
=
diff
.
count
()
/
interval_
.
count
(
);
auto
this_tick
=
static_cast
<
size_t
>
(
diff
.
count
()
/
interval_
.
count
()
);
auto
tick_id
=
this_tick
+
1
;
while
(
!
is_trigger
(
tick_id
))
++
tick_id
;
...
...
libcaf_core/test/broadcast_downstream_manager.cpp
View file @
d1f11390
...
...
@@ -19,6 +19,7 @@
#define CAF_SUITE broadcast_downstream_manager
#include <cstdint>
#include <memory>
#include "caf/actor_system.hpp"
...
...
@@ -85,7 +86,7 @@ public:
// nop
}
void
add_path_to
(
entity
&
x
,
int
desired_batch_size
)
{
void
add_path_to
(
entity
&
x
,
int
32_t
desired_batch_size
)
{
auto
ptr
=
bs
.
add_path
(
next_slot
++
,
x
.
ctrl
());
CAF_REQUIRE
(
ptr
!=
nullptr
);
ptr
->
desired_batch_size
=
desired_batch_size
;
...
...
@@ -93,7 +94,7 @@ public:
paths
.
emplace_back
(
ptr
);
}
size_t
credit_for
(
entity
&
x
)
{
long
credit_for
(
entity
&
x
)
{
auto
pred
=
[
&
](
outbound_path
*
ptr
)
{
return
ptr
->
hdl
==
&
x
;
};
...
...
@@ -198,7 +199,7 @@ struct fixture {
batch_type
make_batch
(
int
first
,
int
last
)
{
batch_type
result
;
result
.
resize
(
(
last
+
1
)
-
first
);
result
.
resize
(
static_cast
<
size_t
>
((
last
+
1
)
-
first
)
);
std
::
iota
(
result
.
begin
(),
result
.
end
(),
first
);
return
result
;
}
...
...
@@ -298,7 +299,7 @@ receive_checker<F> operator<<(receive_checker<F> xs, not_empty_t) {
#define FOR \
struct CONCAT(for_helper_, __LINE__) { \
entity& who; \
size_t amount;
\
long amount;
\
void operator<<(entity& x) const { \
CAF_CHECK_EQUAL(who.credit_for(x), amount); \
} \
...
...
libcaf_core/test/continuous_streaming.cpp
View file @
d1f11390
...
...
@@ -144,7 +144,7 @@ using fixture = test_coordinator_fixture<>;
CAF_TEST_FIXTURE_SCOPE
(
local_streaming_tests
,
fixture
)
CAF_TEST
(
depth_3_pipeline_with_fork
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
stg
=
sys
.
spawn
(
stream_multiplexer
);
auto
snk1
=
sys
.
spawn
(
sum_up
);
auto
snk2
=
sys
.
spawn
(
sum_up
);
...
...
@@ -168,8 +168,8 @@ CAF_TEST(depth_3_pipeline_with_fork) {
}
CAF_TEST
(
depth_3_pipeline_with_join
)
{
auto
src1
=
sys
.
spawn
(
file_reader
,
50
);
auto
src2
=
sys
.
spawn
(
file_reader
,
50
);
auto
src1
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
src2
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
stg
=
sys
.
spawn
(
stream_multiplexer
);
auto
snk
=
sys
.
spawn
(
sum_up
);
auto
&
st
=
deref
<
stream_multiplexer_actor
>
(
stg
).
state
;
...
...
libcaf_core/test/fused_streaming.cpp
View file @
d1f11390
...
...
@@ -261,8 +261,8 @@ using fixture = test_coordinator_fixture<>;
CAF_TEST_FIXTURE_SCOPE
(
fused_streaming_tests
,
fixture
)
CAF_TEST
(
depth_3_pipeline_with_fork
)
{
auto
src1
=
sys
.
spawn
(
int_file_reader
,
50
);
auto
src2
=
sys
.
spawn
(
string_file_reader
,
50
);
auto
src1
=
sys
.
spawn
(
int_file_reader
,
50
u
);
auto
src2
=
sys
.
spawn
(
string_file_reader
,
50
u
);
auto
stg
=
sys
.
spawn
(
stream_multiplexer
);
auto
snk1
=
sys
.
spawn
(
sum_up
);
auto
snk2
=
sys
.
spawn
(
collect
);
...
...
libcaf_core/test/mock_streaming_classes.cpp
View file @
d1f11390
...
...
@@ -129,10 +129,10 @@ struct dmsg {
struct
umsg
{
struct
ack_handshake
{
long
credit
;
int32_t
credit
;
};
struct
ack_batch
{
long
credit
;
int32_t
credit
;
};
struct
drop
{
// nop
...
...
libcaf_core/test/native_streaming_classes.cpp
View file @
d1f11390
...
...
@@ -172,7 +172,7 @@ public:
using
duration_type
=
time_point
::
duration
;
/// The type of a single tick.
using
tick_type
=
long
;
using
tick_type
=
size_t
;
// -- constructors, destructors, and assignment operators --------------------
...
...
@@ -186,8 +186,10 @@ public:
:
*
global_time
)
{
auto
cycle
=
detail
::
gcd
(
credit_interval
.
count
(),
force_batches_interval
.
count
());
ticks_per_force_batches_interval
=
force_batches_interval
.
count
()
/
cycle
;
ticks_per_credit_interval
=
credit_interval
.
count
()
/
cycle
;
ticks_per_force_batches_interval
=
static_cast
<
size_t
>
(
force_batches_interval
.
count
()
/
cycle
);
ticks_per_credit_interval
=
static_cast
<
size_t
>
(
credit_interval
.
count
()
/
cycle
);
tick_emitter_
.
interval
(
duration_type
{
cycle
});
}
...
...
@@ -342,8 +344,8 @@ public:
auto
&
qs
=
get
<
2
>
(
mbox
.
queues
()).
queues
();
for
(
auto
&
kvp
:
qs
)
{
auto
inptr
=
kvp
.
second
.
policy
().
handler
.
get
();
inptr
->
emit_ack_batch
(
this
,
kvp
.
second
.
total_task_size
(),
cycle
,
desired_batch_complexity
);
auto
bs
=
static_cast
<
int32_t
>
(
kvp
.
second
.
total_task_size
());
inptr
->
emit_ack_batch
(
this
,
bs
,
cycle
,
desired_batch_complexity
);
}
}
};
...
...
@@ -604,7 +606,9 @@ struct fixture {
};
vector
<
int
>
make_iota
(
int
first
,
int
last
)
{
vector
<
int
>
result
(
last
-
first
);
CAF_ASSERT
(
first
<
last
);
vector
<
int
>
result
;
result
.
resize
(
static_cast
<
size_t
>
(
last
-
first
));
std
::
iota
(
result
.
begin
(),
result
.
end
(),
first
);
return
result
;
}
...
...
libcaf_core/test/pipeline_streaming.cpp
View file @
d1f11390
...
...
@@ -220,7 +220,7 @@ using fixture = test_coordinator_fixture<>;
CAF_TEST_FIXTURE_SCOPE
(
local_streaming_tests
,
fixture
)
CAF_TEST
(
depth_2_pipeline_50_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
snk
=
sys
.
spawn
(
sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -241,7 +241,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
}
CAF_TEST
(
depth_2_pipeline_setup2_50_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
snk
=
sys
.
spawn
(
sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -262,7 +262,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
}
CAF_TEST
(
delayed_depth_2_pipeline_50_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
snk
=
sys
.
spawn
(
delayed_sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -289,7 +289,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
}
CAF_TEST
(
depth_2_pipeline_500_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
500
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
u
);
auto
snk
=
sys
.
spawn
(
sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -318,7 +318,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_TEST
(
depth_2_pipeline_error_during_handshake
)
{
CAF_MESSAGE
(
"streams must abort if a sink fails to initialize its state"
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
snk
=
sys
.
spawn
(
broken_sink
);
CAF_MESSAGE
(
"initiate stream handshake"
);
self
->
send
(
snk
*
src
,
"numbers.txt"
);
...
...
@@ -330,7 +330,7 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) {
CAF_TEST
(
depth_2_pipeline_error_at_source
)
{
CAF_MESSAGE
(
"streams must abort if a source fails at runtime"
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
u
);
auto
snk
=
sys
.
spawn
(
sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -348,7 +348,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
CAF_TEST
(
depth_2_pipelin_error_at_sink
)
{
CAF_MESSAGE
(
"streams must abort if a sink fails at runtime"
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
u
);
auto
snk
=
sys
.
spawn
(
sum_up
);
CAF_MESSAGE
(
CAF_ARG
(
self
)
<<
CAF_ARG
(
src
)
<<
CAF_ARG
(
snk
));
CAF_MESSAGE
(
"initiate stream handshake"
);
...
...
@@ -364,7 +364,7 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
}
CAF_TEST
(
depth_3_pipeline_50_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
50
);
auto
src
=
sys
.
spawn
(
file_reader
,
50
u
);
auto
stg
=
sys
.
spawn
(
filter
);
auto
snk
=
sys
.
spawn
(
sum_up
);
auto
next_cycle
=
[
&
]
{
...
...
@@ -399,7 +399,7 @@ CAF_TEST(depth_3_pipeline_50_items) {
}
CAF_TEST
(
depth_4_pipeline_500_items
)
{
auto
src
=
sys
.
spawn
(
file_reader
,
500
);
auto
src
=
sys
.
spawn
(
file_reader
,
500
u
);
auto
stg1
=
sys
.
spawn
(
filter
);
auto
stg2
=
sys
.
spawn
(
doubler
);
auto
snk
=
sys
.
spawn
(
sum_up
);
...
...
libcaf_core/test/tick_emitter.cpp
View file @
d1f11390
...
...
@@ -72,14 +72,15 @@ CAF_TEST(ticks) {
auto
cycle
=
detail
::
gcd
(
credit_interval
.
count
(),
force_batch_interval
.
count
());
CAF_CHECK_EQUAL
(
cycle
,
50
);
auto
force_batch_frequency
=
force_batch_interval
.
count
()
/
cycle
;
auto
credit_frequency
=
credit_interval
.
count
()
/
cycle
;
auto
force_batch_frequency
=
static_cast
<
size_t
>
(
force_batch_interval
.
count
()
/
cycle
);
auto
credit_frequency
=
static_cast
<
size_t
>
(
credit_interval
.
count
()
/
cycle
);
detail
::
tick_emitter
tctrl
{
time_point
{
timespan
{
100
}}};
tctrl
.
interval
(
timespan
{
cycle
});
vector
<
long
>
ticks
;
in
t
force_batch_triggers
=
0
;
in
t
credit_triggers
=
0
;
auto
f
=
[
&
](
long
tick_id
)
{
vector
<
size_t
>
ticks
;
size_
t
force_batch_triggers
=
0
;
size_
t
credit_triggers
=
0
;
auto
f
=
[
&
](
size_t
tick_id
)
{
ticks
.
push_back
(
tick_id
);
if
(
tick_id
%
force_batch_frequency
==
0
)
++
force_batch_triggers
;
...
...
@@ -89,13 +90,13 @@ CAF_TEST(ticks) {
CAF_MESSAGE
(
"trigger 4 ticks"
);
tctrl
.
update
(
time_point
{
timespan
{
300
}},
f
);
CAF_CHECK_EQUAL
(
deep_to_string
(
ticks
),
"[1, 2, 3, 4]"
);
CAF_CHECK_EQUAL
(
force_batch_triggers
,
4
);
CAF_CHECK_EQUAL
(
credit_triggers
,
1
);
CAF_CHECK_EQUAL
(
force_batch_triggers
,
4
lu
);
CAF_CHECK_EQUAL
(
credit_triggers
,
1
lu
);
CAF_MESSAGE
(
"trigger 3 more ticks"
);
tctrl
.
update
(
time_point
{
timespan
{
475
}},
f
);
CAF_CHECK_EQUAL
(
deep_to_string
(
ticks
),
"[1, 2, 3, 4, 5, 6, 7]"
);
CAF_CHECK_EQUAL
(
force_batch_triggers
,
7
);
CAF_CHECK_EQUAL
(
credit_triggers
,
1
);
CAF_CHECK_EQUAL
(
force_batch_triggers
,
7
lu
);
CAF_CHECK_EQUAL
(
credit_triggers
,
1
lu
);
}
CAF_TEST
(
timeouts
)
{
...
...
@@ -107,23 +108,23 @@ CAF_TEST(timeouts) {
CAF_MESSAGE
(
"advance until the first 5-tick-period ends"
);
now
+=
interval
*
5
;
auto
bitmask
=
tctrl
.
timeouts
(
now
,
{
5
,
7
});
CAF_CHECK_EQUAL
(
bitmask
,
0x01
);
CAF_CHECK_EQUAL
(
bitmask
,
0x01
u
);
CAF_MESSAGE
(
"advance until the first 7-tick-period ends"
);
now
+=
interval
*
2
;
bitmask
=
tctrl
.
timeouts
(
now
,
{
5
,
7
});
CAF_CHECK_EQUAL
(
bitmask
,
0x02
);
CAF_CHECK_EQUAL
(
bitmask
,
0x02
u
);
CAF_MESSAGE
(
"advance until both tick period ends"
);
now
+=
interval
*
7
;
bitmask
=
tctrl
.
timeouts
(
now
,
{
5
,
7
});
CAF_CHECK_EQUAL
(
bitmask
,
0x03
);
CAF_CHECK_EQUAL
(
bitmask
,
0x03
u
);
CAF_MESSAGE
(
"advance until both tick period end multiple times"
);
now
+=
interval
*
21
;
bitmask
=
tctrl
.
timeouts
(
now
,
{
5
,
7
});
CAF_CHECK_EQUAL
(
bitmask
,
0x03
);
CAF_CHECK_EQUAL
(
bitmask
,
0x03
u
);
CAF_MESSAGE
(
"advance without any timeout"
);
now
+=
interval
*
1
;
bitmask
=
tctrl
.
timeouts
(
now
,
{
5
,
7
});
CAF_CHECK_EQUAL
(
bitmask
,
0x00
);
CAF_CHECK_EQUAL
(
bitmask
,
0x00
u
);
}
CAF_TEST
(
next_timeout
)
{
...
...
libcaf_core/test/unordered_flat_map.cpp
View file @
d1f11390
...
...
@@ -81,7 +81,7 @@ struct fixture {
void
fill_ys
()
{
char
buf
[]
=
{
'\0'
,
'\0'
};
for
(
int
i
=
0
;
i
<
4
;
++
i
)
{
buf
[
0
]
=
'a'
+
i
;
buf
[
0
]
=
static_cast
<
char
>
(
'a'
+
i
)
;
ys
.
emplace
(
i
+
1
,
buf
);
}
}
...
...
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
View file @
d1f11390
...
...
@@ -98,7 +98,7 @@ struct fetch_helper {
inode
&
x
)
{
if
(
!
result
.
empty
())
result
+=
','
;
result
+=
to_string
(
I
);
result
+=
std
::
to_string
(
I
);
result
+=
':'
;
result
+=
to_string
(
x
);
return
task_result
::
resume
;
...
...
@@ -124,7 +124,7 @@ struct fixture {
auto
f
=
[
&
](
size_t
id
,
drr_queue
<
inode_policy
>&
,
inode
&
x
)
{
if
(
!
result
.
empty
())
result
+=
','
;
result
+=
to_string
(
id
);
result
+=
std
::
to_string
(
id
);
result
+=
':'
;
result
+=
to_string
(
x
);
return
task_result
::
resume
;
...
...
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