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
6602cb18
Commit
6602cb18
authored
Feb 22, 2018
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Improve logging / unit test output
parent
9fd715d9
Changes
4
Hide whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
24 additions
and
11 deletions
+24
-11
libcaf_core/src/inbound_path.cpp
libcaf_core/src/inbound_path.cpp
+6
-2
libcaf_core/src/outbound_path.cpp
libcaf_core/src/outbound_path.cpp
+6
-0
libcaf_core/src/stream_manager.cpp
libcaf_core/src/stream_manager.cpp
+1
-0
libcaf_core/test/local_streaming.cpp
libcaf_core/test/local_streaming.cpp
+11
-9
No files found.
libcaf_core/src/inbound_path.cpp
View file @
6602cb18
...
...
@@ -73,7 +73,7 @@ inbound_path::~inbound_path() {
}
void
inbound_path
::
handle
(
downstream_msg
::
batch
&
x
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
));
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
x
));
auto
batch_size
=
x
.
xs_size
;
assigned_credit
-=
batch_size
;
last_batch_id
=
x
.
id
;
...
...
@@ -87,6 +87,7 @@ void inbound_path::handle(downstream_msg::batch& x) {
}
void
inbound_path
::
emit_ack_open
(
local_actor
*
self
,
actor_addr
rebind_from
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
rebind_from
));
assigned_credit
=
50
;
// TODO: put constant in some header
int32_t
desired_batch_size
=
50
;
// TODO: put constant in some header
unsafe_send_as
(
self
,
hdl
,
...
...
@@ -98,7 +99,8 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
void
inbound_path
::
emit_ack_batch
(
local_actor
*
self
,
long
queued_items
,
timespan
cycle
,
timespan
complexity
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
queued_items
)
<<
CAF_ARG
(
cycle
)
<<
CAF_ARG
(
complexity
));
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
queued_items
)
<<
CAF_ARG
(
cycle
)
<<
CAF_ARG
(
complexity
));
if
(
last_acked_batch_id
==
last_batch_id
)
return
;
auto
x
=
stats
.
calculate
(
cycle
,
complexity
);
...
...
@@ -118,10 +120,12 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
}
void
inbound_path
::
emit_regular_shutdown
(
local_actor
*
self
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
));
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
drop
>
(
slots
,
self
->
address
()));
}
void
inbound_path
::
emit_irregular_shutdown
(
local_actor
*
self
,
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
reason
));
unsafe_send_as
(
self
,
hdl
,
make
<
upstream_msg
::
forced_drop
>
(
slots
.
invert
(),
self
->
address
(),
std
::
move
(
reason
)));
...
...
libcaf_core/src/outbound_path.cpp
View file @
6602cb18
...
...
@@ -42,6 +42,8 @@ outbound_path::~outbound_path() {
void
outbound_path
::
emit_open
(
local_actor
*
self
,
stream_slot
slot
,
strong_actor_ptr
to
,
message
handshake_data
,
stream_priority
prio
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slot
)
<<
CAF_ARG
(
to
)
<<
CAF_ARG
(
handshake_data
)
<<
CAF_ARG
(
prio
));
CAF_ASSERT
(
self
!=
nullptr
);
CAF_ASSERT
(
to
!=
nullptr
);
// TODO: attach an aborter to `to`
...
...
@@ -51,6 +53,7 @@ void outbound_path::emit_open(local_actor* self, stream_slot slot,
}
void
outbound_path
::
emit_batch
(
local_actor
*
self
,
long
xs_size
,
message
xs
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
xs_size
)
<<
CAF_ARG
(
xs
));
CAF_ASSERT
(
open_credit
>=
xs_size
);
open_credit
-=
xs_size
;
auto
bid
=
next_batch_id
++
;
...
...
@@ -61,11 +64,13 @@ void outbound_path::emit_batch(local_actor* self, long xs_size, message xs) {
}
void
outbound_path
::
emit_regular_shutdown
(
local_actor
*
self
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
));
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
close
>
(
slots
,
self
->
address
()));
}
void
outbound_path
::
emit_irregular_shutdown
(
local_actor
*
self
,
error
reason
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
slots
)
<<
CAF_ARG
(
reason
));
unsafe_send_as
(
self
,
hdl
,
make
<
downstream_msg
::
forced_close
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
...
...
@@ -75,6 +80,7 @@ void outbound_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
<
downstream_msg
::
forced_close
>
(
slots
,
self
->
address
(),
std
::
move
(
reason
)));
...
...
libcaf_core/src/stream_manager.cpp
View file @
6602cb18
...
...
@@ -165,6 +165,7 @@ void stream_manager::deregister_input_path(inbound_path* ptr) noexcept {
swap
(
*
i
,
inbound_paths_
.
back
());
}
inbound_paths_
.
pop_back
();
CAF_LOG_DEBUG
(
inbound_paths_
.
size
()
<<
"paths remaining"
);
}
void
stream_manager
::
add_promise
(
response_promise
x
)
{
...
...
libcaf_core/test/local_streaming.cpp
View file @
6602cb18
...
...
@@ -73,9 +73,12 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
},
// check whether we reached the end
[
=
](
const
buf
&
xs
)
{
return
xs
.
empty
();
}
);
if
(
xs
.
empty
())
{
CAF_MESSAGE
(
"sum_up is done"
);
return
true
;
}
return
false
;
});
}
};
}
...
...
@@ -97,6 +100,7 @@ TESTEE(sum_up) {
},
// cleanup and produce result message
[](
int
&
x
)
->
int
{
CAF_MESSAGE
(
"sum_up is done"
);
return
x
;
}
);
...
...
@@ -124,6 +128,7 @@ TESTEE(delayed_sum_up) {
},
// cleanup and produce result message
[](
int
&
x
)
->
int
{
CAF_MESSAGE
(
"delayed_sum_up is done"
);
return
x
;
}
);
...
...
@@ -163,7 +168,7 @@ TESTEE(filter) {
},
// cleanup
[
=
](
unit_t
&
)
{
// nop
CAF_MESSAGE
(
"filter is done"
);
}
);
}
...
...
@@ -359,18 +364,15 @@ CAF_TEST(depth_3_pipeline_50_items) {
CAF_MESSAGE
(
"the stage should delay its first batch since its underfull"
);
disallow
((
downstream_msg
::
batch
),
from
(
stg
).
to
(
snk
));
next_cycle
();
CAF_MESSAGE
(
"
expect close message from src and batch from stage to sink
"
);
CAF_MESSAGE
(
"
the source shuts down and the stage sends the final batch
"
);
expect
((
upstream_msg
::
ack_batch
),
from
(
stg
).
to
(
src
));
expect
((
downstream_msg
::
close
),
from
(
src
).
to
(
stg
));
expect
((
downstream_msg
::
batch
),
from
(
stg
).
to
(
snk
));
next_cycle
();
CAF_MESSAGE
(
"
expect close message from stage to sink
"
);
CAF_MESSAGE
(
"
the stage shuts down and the sink produces its final result
"
);
expect
((
upstream_msg
::
ack_batch
),
from
(
snk
).
to
(
stg
));
expect
((
downstream_msg
::
close
),
from
(
stg
).
to
(
snk
));
expect
((
int
),
from
(
snk
).
to
(
self
).
with
(
625
));
CAF_CHECK_EQUAL
(
fail_state
(
snk
),
exit_reason
::
normal
);
CAF_CHECK_EQUAL
(
fail_state
(
stg
),
exit_reason
::
normal
);
CAF_CHECK_EQUAL
(
fail_state
(
src
),
exit_reason
::
normal
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
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