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
e1308bfa
Commit
e1308bfa
authored
Jan 16, 2021
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix regressions and GCC debug mode findings
parent
b26c7d6a
Changes
7
Hide whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
99 additions
and
144 deletions
+99
-144
libcaf_core/caf/typed_response_promise.hpp
libcaf_core/caf/typed_response_promise.hpp
+1
-1
libcaf_core/src/response_promise.cpp
libcaf_core/src/response_promise.cpp
+20
-20
libcaf_core/src/scheduler/abstract_coordinator.cpp
libcaf_core/src/scheduler/abstract_coordinator.cpp
+34
-82
libcaf_core/src/string_algorithms.cpp
libcaf_core/src/string_algorithms.cpp
+3
-5
libcaf_core/test/actor_system_config.cpp
libcaf_core/test/actor_system_config.cpp
+22
-24
libcaf_core/test/constructor_attach.cpp
libcaf_core/test/constructor_attach.cpp
+17
-10
libcaf_io/test/io/receive_buffer.cpp
libcaf_io/test/io/receive_buffer.cpp
+2
-2
No files found.
libcaf_core/caf/typed_response_promise.hpp
View file @
e1308bfa
...
...
@@ -88,7 +88,7 @@ public:
template
<
class
...
Us
>
std
::
enable_if_t
<
(
std
::
is_constructible
<
Ts
,
Us
>::
value
&&
...)
>
deliver
(
Us
...
xs
)
{
promise_
.
deliver
(
make_message
(
Ts
{
std
::
forward
<
Us
>
(
xs
)}...)
);
promise_
.
deliver
(
Ts
{
std
::
forward
<
Us
>
(
xs
)}...
);
}
/// Satisfies the promise by sending an empty response message.
...
...
libcaf_core/src/response_promise.cpp
View file @
e1308bfa
...
...
@@ -16,15 +16,16 @@ namespace caf {
namespace
{
bool
requires_response
(
const
strong_actor_ptr
&
source
,
const
mailbox_element
::
forwarding_stack
&
stages
,
message_id
mid
)
{
return
!
mid
.
is_response
()
&&
!
mid
.
is_answered
()
&&
(
source
||
!
stages
.
empty
());
bool
requires_response
(
message_id
mid
)
{
return
!
mid
.
is_response
()
&&
!
mid
.
is_answered
();
}
bool
requires_response
(
const
mailbox_element
&
src
)
{
return
requires_response
(
src
.
sender
,
src
.
stages
,
src
.
mid
);
return
requires_response
(
src
.
mid
);
}
bool
has_response_receiver
(
const
mailbox_element
&
src
)
{
return
src
.
sender
||
!
src
.
stages
.
empty
();
}
}
// namespace
...
...
@@ -37,7 +38,7 @@ response_promise::response_promise(local_actor* self, strong_actor_ptr source,
// Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case. Also don't create promises for
// anonymous messages since there's nowhere to send the message to anyway.
if
(
requires_response
(
source
,
stages
,
mid
))
{
if
(
requires_response
(
mid
))
{
state_
=
make_counted
<
state
>
();
state_
->
self
=
self
;
state_
->
source
.
swap
(
source
);
...
...
@@ -118,7 +119,8 @@ void response_promise::deliver() {
void
response_promise
::
respond_to
(
local_actor
*
self
,
mailbox_element
*
request
,
message
&
response
)
{
if
(
request
&&
requires_response
(
*
request
))
{
if
(
request
&&
requires_response
(
*
request
)
&&
has_response_receiver
(
*
request
))
{
state
tmp
;
tmp
.
self
=
self
;
tmp
.
source
.
swap
(
request
->
sender
);
...
...
@@ -131,7 +133,8 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request,
void
response_promise
::
respond_to
(
local_actor
*
self
,
mailbox_element
*
request
,
error
&
response
)
{
if
(
request
&&
requires_response
(
*
request
))
{
if
(
request
&&
requires_response
(
*
request
)
&&
has_response_receiver
(
*
request
))
{
state
tmp
;
tmp
.
self
=
self
;
tmp
.
source
.
swap
(
request
->
sender
);
...
...
@@ -159,17 +162,14 @@ void response_promise::state::deliver_impl(message msg) {
CAF_LOG_TRACE
(
CAF_ARG
(
msg
));
if
(
msg
.
empty
()
&&
id
.
is_async
())
{
CAF_LOG_DEBUG
(
"drop response: empty response to asynchronous input"
);
}
else
{
if
(
stages
.
empty
())
{
detail
::
profiled_send
(
self
,
self
->
ctrl
(),
source
,
id
.
response_id
(),
forwarding_stack
{},
self
->
context
(),
std
::
move
(
msg
));
}
else
{
auto
next
=
std
::
move
(
stages
.
back
());
stages
.
pop_back
();
detail
::
profiled_send
(
self
,
std
::
move
(
source
),
next
,
id
,
std
::
move
(
stages
),
self
->
context
(),
std
::
move
(
msg
));
}
}
else
if
(
!
stages
.
empty
())
{
auto
next
=
std
::
move
(
stages
.
back
());
stages
.
pop_back
();
detail
::
profiled_send
(
self
,
std
::
move
(
source
),
next
,
id
,
std
::
move
(
stages
),
self
->
context
(),
std
::
move
(
msg
));
}
else
if
(
source
!=
nullptr
)
{
detail
::
profiled_send
(
self
,
self
->
ctrl
(),
source
,
id
.
response_id
(),
forwarding_stack
{},
self
->
context
(),
std
::
move
(
msg
));
}
cancel
();
}
...
...
libcaf_core/src/scheduler/abstract_coordinator.cpp
View file @
e1308bfa
...
...
@@ -11,6 +11,7 @@
#include <fstream>
#include <ios>
#include <iostream>
#include <memory>
#include <thread>
#include <unordered_map>
...
...
@@ -38,95 +39,46 @@ namespace {
using
string_sink
=
std
::
function
<
void
(
std
::
string
&&
)
>
;
// the first value is the use count, the last ostream_handle that
// decrements it to 0 removes the ostream pointer from the map
using
counted_sink
=
std
::
pair
<
size_t
,
string_sink
>
;
using
string_sink_ptr
=
std
::
shared_ptr
<
string_sink
>
;
using
sink_cache
=
std
::
map
<
std
::
string
,
counted_sink
>
;
class
sink_handle
{
public:
using
iterator
=
sink_cache
::
iterator
;
sink_handle
()
:
cache_
(
nullptr
)
{
// nop
}
sink_handle
(
sink_cache
*
fc
,
iterator
iter
)
:
cache_
(
fc
),
iter_
(
iter
)
{
if
(
cache_
!=
nullptr
)
++
iter_
->
second
.
first
;
}
sink_handle
(
const
sink_handle
&
other
)
:
cache_
(
nullptr
)
{
*
this
=
other
;
}
sink_handle
&
operator
=
(
const
sink_handle
&
other
)
{
if
(
cache_
!=
other
.
cache_
||
iter_
!=
other
.
iter_
)
{
clear
();
cache_
=
other
.
cache_
;
if
(
cache_
!=
nullptr
)
{
iter_
=
other
.
iter_
;
++
iter_
->
second
.
first
;
}
}
return
*
this
;
}
~
sink_handle
()
{
clear
();
}
explicit
operator
bool
()
const
{
return
cache_
!=
nullptr
;
}
string_sink
&
operator
*
()
{
CAF_ASSERT
(
iter_
->
second
.
second
!=
nullptr
);
return
iter_
->
second
.
second
;
}
private:
void
clear
()
{
if
(
cache_
!=
nullptr
&&
--
iter_
->
second
.
first
==
0
)
{
cache_
->
erase
(
iter_
);
cache_
=
nullptr
;
}
}
sink_cache
*
cache_
;
sink_cache
::
iterator
iter_
;
};
using
sink_cache
=
std
::
map
<
std
::
string
,
string_sink_ptr
>
;
string_sink
make_sink
(
actor_system
&
sys
,
const
std
::
string
&
fn
,
int
flags
)
{
if
(
fn
.
empty
())
if
(
fn
.
empty
())
{
return
nullptr
;
if
(
fn
.
front
()
==
':'
)
{
}
else
if
(
fn
.
front
()
==
':'
)
{
// "virtual file" name given, translate this to group communication
auto
grp
=
sys
.
groups
().
get_local
(
fn
);
return
[
grp
,
fn
](
std
::
string
&&
out
)
{
anon_send
(
grp
,
fn
,
std
::
move
(
out
));
};
}
else
{
auto
append
=
(
flags
&
actor_ostream
::
append
)
!=
0
;
auto
fs
=
std
::
make_shared
<
std
::
ofstream
>
();
fs
->
open
(
fn
,
append
?
std
::
ios_base
::
out
|
std
::
ios_base
::
app
:
std
::
ios_base
::
out
);
if
(
fs
->
is_open
())
{
return
[
fs
](
std
::
string
&&
out
)
{
*
fs
<<
out
;
};
}
else
{
std
::
cerr
<<
"cannot open file: "
<<
fn
<<
std
::
endl
;
return
nullptr
;
}
}
auto
append
=
(
flags
&
actor_ostream
::
append
)
!=
0
;
auto
fs
=
std
::
make_shared
<
std
::
ofstream
>
();
fs
->
open
(
fn
,
append
?
std
::
ios_base
::
out
|
std
::
ios_base
::
app
:
std
::
ios_base
::
out
);
if
(
fs
->
is_open
())
return
[
fs
](
std
::
string
&&
out
)
{
*
fs
<<
out
;
};
std
::
cerr
<<
"cannot open file: "
<<
fn
<<
std
::
endl
;
return
nullptr
;
}
sink_handle
get_sink_handle
(
actor_system
&
sys
,
sink_cache
&
fc
,
const
std
::
string
&
fn
,
int
flags
)
{
auto
i
=
fc
.
find
(
fn
);
if
(
i
!=
fc
.
end
())
return
{
&
fc
,
i
};
auto
fs
=
make_sink
(
sys
,
fn
,
flags
);
if
(
fs
)
{
i
=
fc
.
emplace
(
fn
,
sink_cache
::
mapped_type
{
0
,
std
::
move
(
fs
)}).
first
;
return
{
&
fc
,
i
};
string_sink_ptr
get_or_add_sink_ptr
(
actor_system
&
sys
,
sink_cache
&
fc
,
const
std
::
string
&
fn
,
int
flags
)
{
if
(
auto
i
=
fc
.
find
(
fn
);
i
!=
fc
.
end
())
{
return
i
->
second
;
}
else
if
(
auto
fs
=
make_sink
(
sys
,
fn
,
flags
))
{
if
(
fs
)
{
auto
ptr
=
std
::
make_shared
<
string_sink
>
(
std
::
move
(
fs
));
fc
.
emplace
(
fn
,
ptr
);
return
ptr
;
}
else
{
return
nullptr
;
}
}
else
{
return
nullptr
;
}
return
{};
}
class
printer_actor
:
public
blocking_actor
{
...
...
@@ -138,14 +90,14 @@ public:
void
act
()
override
{
struct
actor_data
{
std
::
string
current_line
;
s
ink_handle
redirect
;
s
tring_sink_ptr
redirect
;
actor_data
()
{
// nop
}
};
using
data_map
=
std
::
unordered_map
<
actor_id
,
actor_data
>
;
sink_cache
fcache
;
s
ink_handle
global_redirect
;
s
tring_sink_ptr
global_redirect
;
data_map
data
;
auto
get_data
=
[
&
](
actor_id
addr
,
bool
insert_missing
)
->
actor_data
*
{
if
(
addr
==
invalid_actor_id
)
...
...
@@ -191,12 +143,12 @@ public:
}
},
[
&
](
redirect_atom
,
const
std
::
string
&
fn
,
int
flag
)
{
global_redirect
=
get_
sink_handle
(
system
(),
fcache
,
fn
,
flag
);
global_redirect
=
get_
or_add_sink_ptr
(
system
(),
fcache
,
fn
,
flag
);
},
[
&
](
redirect_atom
,
actor_id
aid
,
const
std
::
string
&
fn
,
int
flag
)
{
auto
d
=
get_data
(
aid
,
true
);
if
(
d
!=
nullptr
)
d
->
redirect
=
get_
sink_handle
(
system
(),
fcache
,
fn
,
flag
);
d
->
redirect
=
get_
or_add_sink_ptr
(
system
(),
fcache
,
fn
,
flag
);
},
[
&
](
exit_msg
&
em
)
{
fail_state
(
std
::
move
(
em
.
reason
));
...
...
libcaf_core/src/string_algorithms.cpp
View file @
e1308bfa
...
...
@@ -59,13 +59,11 @@ void replace_all(std::string& str, string_view what, string_view with) {
};
auto
i
=
next
(
str
.
begin
());
while
(
i
!=
str
.
end
())
{
auto
before
=
std
::
distance
(
str
.
begin
(),
i
);
CAF_ASSERT
(
before
>=
0
);
auto
ws
=
static_cast
<
decltype
(
before
)
>
(
what
.
size
());
str
.
replace
(
i
,
i
+
ws
,
with
.
begin
(),
with
.
end
());
auto
before
=
static_cast
<
size_t
>
(
std
::
distance
(
str
.
begin
(),
i
));
str
.
replace
(
i
,
i
+
what
.
size
(),
with
.
begin
(),
with
.
end
());
// Iterator i became invalidated -> use new iterator pointing to the first
// character after the replaced text.
i
=
next
(
str
.
begin
()
+
before
+
w
s
);
i
=
next
(
str
.
begin
()
+
before
+
w
ith
.
size
()
);
}
}
...
...
libcaf_core/test/actor_system_config.cpp
View file @
e1308bfa
...
...
@@ -152,8 +152,10 @@ CAF_TEST(file input overrides defaults but CLI args always win) {
// Checks whether both a synced variable and the corresponding entry in
// content(cfg) are equal to `value`.
#define CHECK_SYNCED(var,
value)
\
#define CHECK_SYNCED(var,
...)
\
do { \
using ref_value_type = std::decay_t<decltype(var)>; \
ref_value_type value{__VA_ARGS__}; \
CAF_CHECK_EQUAL(var, value); \
if (auto maybe_val = get_as<decltype(var)>(cfg, #var)) { \
CAF_CHECK_EQUAL(*maybe_val, value); \
...
...
@@ -206,14 +208,12 @@ CAF_TEST(integers and integer containers options) {
CHECK_SYNCED
(
some_int
,
42
);
CHECK_SYNCED
(
some_other_int
,
23
);
CHECK_TEXT_ONLY
(
int
,
yet_another_int
,
123
);
CHECK_SYNCED
(
some_int_list
,
int_list
({
1
,
2
,
3
}));
CHECK_SYNCED
(
some_int_list_list
,
int_list_list
({{
1
,
2
,
3
},
{
4
,
5
,
6
}}));
CHECK_SYNCED
(
some_int_map
,
int_map
({{{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
}}}));
CHECK_SYNCED
(
some_int_list_map
,
int_list_map
({{{
"a"
,
{
1
,
2
,
3
}},
{
"b"
,
{
4
,
5
,
6
}}}}));
CHECK_SYNCED
(
some_int_map_list
,
int_map_list
({{{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
}},
{{
"d"
,
4
},
{
"e"
,
5
},
{
"f"
,
6
}}}));
CHECK_SYNCED
(
some_int_list
,
1
,
2
,
3
);
CHECK_SYNCED
(
some_int_list_list
,
{
1
,
2
,
3
},
{
4
,
5
,
6
});
CHECK_SYNCED
(
some_int_map
,
{{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
}});
CHECK_SYNCED
(
some_int_list_map
,
{{
"a"
,
{
1
,
2
,
3
}},
{
"b"
,
{
4
,
5
,
6
}}});
CHECK_SYNCED
(
some_int_map_list
,
{{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
}},
{{
"d"
,
4
},
{
"e"
,
5
},
{
"f"
,
6
}});
}
CAF_TEST
(
basic
and
basic
containers
options
)
{
...
...
@@ -279,22 +279,20 @@ CAF_TEST(basic and basic containers options) {
CHECK_SYNCED
(
some_uri
,
"foo:bar"
_u
);
CHECK_SYNCED
(
some_string
,
"string"
s
);
CAF_MESSAGE
(
"check list types"
);
CHECK_SYNCED
(
some_int_list
,
int_list
({
1
,
2
,
3
})
);
CHECK_SYNCED
(
some_bool_list
,
bool_list
({
false
,
true
})
);
CHECK_SYNCED
(
some_double_list
,
double_list
({
1.
,
2.
,
3.
})
);
CHECK_SYNCED
(
some_timespan_list
,
timespan_list
({
123
_ms
,
234
_ms
,
345
_ms
})
);
CHECK_SYNCED
(
some_uri_list
,
uri_list
({
"foo:a"
_u
,
"foo:b"
_u
,
"foo:c"
_u
})
);
CHECK_SYNCED
(
some_string_list
,
string_list
({
"a"
,
"b"
,
"c"
})
);
CHECK_SYNCED
(
some_int_list
,
1
,
2
,
3
);
CHECK_SYNCED
(
some_bool_list
,
false
,
true
);
CHECK_SYNCED
(
some_double_list
,
1.
,
2.
,
3.
);
CHECK_SYNCED
(
some_timespan_list
,
123
_ms
,
234
_ms
,
345
_ms
);
CHECK_SYNCED
(
some_uri_list
,
"foo:a"
_u
,
"foo:b"
_u
,
"foo:c"
_u
);
CHECK_SYNCED
(
some_string_list
,
"a"
,
"b"
,
"c"
);
CAF_MESSAGE
(
"check dictionary types"
);
CHECK_SYNCED
(
some_int_map
,
int_map
({{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
}}));
CHECK_SYNCED
(
some_bool_map
,
bool_map
({{
"a"
,
true
},
{
"b"
,
false
}}));
CHECK_SYNCED
(
some_double_map
,
double_map
({{
"a"
,
1.
},
{
"b"
,
2.
},
{
"c"
,
3.
}}));
CHECK_SYNCED
(
some_timespan_map
,
timespan_map
({{
"a"
,
123
_ms
},
{
"b"
,
234
_ms
},
{
"c"
,
345
_ms
}}));
CHECK_SYNCED
(
some_uri_map
,
uri_map
({{
"a"
,
"foo:a"
_u
},
{
"b"
,
"foo:b"
_u
},
{
"c"
,
"foo:c"
_u
}}));
CHECK_SYNCED
(
some_string_map
,
string_map
({{
"a"
,
"1"
},
{
"b"
,
"2"
},
{
"c"
,
"3"
}}));
CHECK_SYNCED
(
some_int_map
,
{
"a"
,
1
},
{
"b"
,
2
},
{
"c"
,
3
});
CHECK_SYNCED
(
some_bool_map
,
{
"a"
,
true
},
{
"b"
,
false
});
CHECK_SYNCED
(
some_double_map
,
{
"a"
,
1.
},
{
"b"
,
2.
},
{
"c"
,
3.
});
CHECK_SYNCED
(
some_timespan_map
,
{
"a"
,
123
_ms
},
{
"b"
,
234
_ms
},
{
"c"
,
345
_ms
});
CHECK_SYNCED
(
some_uri_map
,
{
"a"
,
"foo:a"
_u
},
{
"b"
,
"foo:b"
_u
},
{
"c"
,
"foo:c"
_u
});
CHECK_SYNCED
(
some_string_map
,
{
"a"
,
"1"
},
{
"b"
,
"2"
},
{
"c"
,
"3"
});
}
SCENARIO
(
"config files allow both nested and dot-separated values"
)
{
...
...
libcaf_core/test/constructor_attach.cpp
View file @
e1308bfa
...
...
@@ -20,17 +20,17 @@ public:
behavior
make_behavior
()
override
{
return
{
[
=
](
delete_atom
)
{
quit
(
exit_reason
::
user_shutdown
);
},
[
=
](
delete_atom
)
{
CAF_MESSAGE
(
"testee received delete"
);
quit
(
exit_reason
::
user_shutdown
);
},
};
}
};
class
spawner
:
public
event_based_actor
{
public:
spawner
(
actor_config
&
cfg
)
:
event_based_actor
(
cfg
),
downs_
(
0
),
testee_
(
spawn
<
testee
,
monitored
>
(
this
))
{
spawner
(
actor_config
&
cfg
)
:
event_based_actor
(
cfg
),
downs_
(
0
)
{
set_down_handler
([
=
](
down_msg
&
msg
)
{
CAF_CHECK_EQUAL
(
msg
.
reason
,
exit_reason
::
user_shutdown
);
CAF_CHECK_EQUAL
(
msg
.
source
,
testee_
.
address
());
...
...
@@ -40,6 +40,7 @@ public:
}
behavior
make_behavior
()
override
{
testee_
=
spawn
<
testee
,
monitored
>
(
this
);
return
{
[
=
](
ok_atom
,
const
error
&
reason
)
{
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
...
...
@@ -47,12 +48,15 @@ public:
quit
(
reason
);
}
},
[
=
](
delete_atom
x
)
{
return
delegate
(
testee_
,
x
);
},
[
=
](
delete_atom
x
)
{
CAF_MESSAGE
(
"spawner received delete"
);
return
delegate
(
testee_
,
x
);
},
};
}
void
on_exit
()
override
{
destroy
(
testee_
)
;
testee_
=
nullptr
;
}
private:
...
...
@@ -62,8 +66,11 @@ private:
}
// namespace
BEGIN_FIXTURE_SCOPE
(
test_coordinator_fixture
<>
)
CAF_TEST
(
constructor_attach
)
{
actor_system_config
cfg
;
actor_system
system
{
cfg
};
anon_send
(
system
.
spawn
<
spawner
>
(),
delete_atom_v
);
anon_send
(
sys
.
spawn
<
spawner
>
(),
delete_atom_v
);
run
();
}
END_FIXTURE_SCOPE
()
libcaf_io/test/io/receive_buffer.cpp
View file @
e1308bfa
...
...
@@ -63,13 +63,13 @@ CAF_TEST(reserve) {
CAF_CHECK_EQUAL
(
a
.
size
(),
0ul
);
CAF_CHECK_EQUAL
(
a
.
capacity
(),
1024ul
);
CAF_CHECK
(
a
.
data
()
!=
nullptr
);
CAF_CHECK
_EQUAL
(
a
.
begin
(),
a
.
end
());
CAF_CHECK
(
a
.
begin
()
==
a
.
end
());
CAF_CHECK
(
a
.
empty
());
a
.
reserve
(
512
);
CAF_CHECK_EQUAL
(
a
.
size
(),
0ul
);
CAF_CHECK_EQUAL
(
a
.
capacity
(),
1024ul
);
CAF_CHECK
(
a
.
data
()
!=
nullptr
);
CAF_CHECK
_EQUAL
(
a
.
begin
(),
a
.
end
());
CAF_CHECK
(
a
.
begin
()
==
a
.
end
());
CAF_CHECK
(
a
.
empty
());
}
...
...
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