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
ae7a2017
Commit
ae7a2017
authored
May 19, 2015
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Coding style nitpicks
parent
d95f40d6
Changes
10
Show whitespace changes
Inline
Side-by-side
Showing
10 changed files
with
157 additions
and
119 deletions
+157
-119
libcaf_core/caf/atom.hpp
libcaf_core/caf/atom.hpp
+10
-0
libcaf_core/caf/blocking_actor.hpp
libcaf_core/caf/blocking_actor.hpp
+2
-2
libcaf_core/src/abstract_coordinator.cpp
libcaf_core/src/abstract_coordinator.cpp
+2
-2
libcaf_core/src/actor_ostream.cpp
libcaf_core/src/actor_ostream.cpp
+2
-2
libcaf_core/test/atom.cpp
libcaf_core/test/atom.cpp
+14
-9
libcaf_core/test/constructor_attach.cpp
libcaf_core/test/constructor_attach.cpp
+11
-4
libcaf_core/test/serial_reply.cpp
libcaf_core/test/serial_reply.cpp
+41
-29
libcaf_core/test/spawn.cpp
libcaf_core/test/spawn.cpp
+68
-66
libcaf_io/caf/io/network/asio_multiplexer.hpp
libcaf_io/caf/io/network/asio_multiplexer.hpp
+5
-1
libcaf_io/caf/io/network/multiplexer.hpp
libcaf_io/caf/io/network/multiplexer.hpp
+2
-4
No files found.
libcaf_core/caf/atom.hpp
View file @
ae7a2017
...
@@ -69,6 +69,11 @@ struct atom_constant {
...
@@ -69,6 +69,11 @@ struct atom_constant {
template
<
atom_value
V
>
template
<
atom_value
V
>
const
atom_constant
<
V
>
atom_constant
<
V
>::
value
=
atom_constant
<
V
>
{};
const
atom_constant
<
V
>
atom_constant
<
V
>::
value
=
atom_constant
<
V
>
{};
/**
* Generic 'ADD' atom for request operations.
*/
using
add_atom
=
atom_constant
<
atom
(
"ADD"
)
>
;
/**
/**
* Generic 'GET' atom for request operations.
* Generic 'GET' atom for request operations.
*/
*/
...
@@ -117,6 +122,11 @@ using leave_atom = atom_constant<atom("LEAVE")>;
...
@@ -117,6 +122,11 @@ using leave_atom = atom_constant<atom("LEAVE")>;
*/
*/
using
forward_atom
=
atom_constant
<
atom
(
"FORWARD"
)
>
;
using
forward_atom
=
atom_constant
<
atom
(
"FORWARD"
)
>
;
/**
* Generic 'FLUSH' atom, e.g., used by `aout`.
*/
using
flush_atom
=
atom_constant
<
atom
(
"FLUSH"
)
>
;
}
// namespace caf
}
// namespace caf
#endif // CAF_ATOM_HPP
#endif // CAF_ATOM_HPP
libcaf_core/caf/blocking_actor.hpp
View file @
ae7a2017
...
@@ -132,8 +132,8 @@ class blocking_actor
...
@@ -132,8 +132,8 @@ class blocking_actor
* ~~~
* ~~~
* int i = 0;
* int i = 0;
* receive_for(i, 10) (
* receive_for(i, 10) (
*
on(atom("get")) >> [&]() -> message
{
*
[&](get_atom)
{
* return
{"result", i}
;
* return
i
;
* }
* }
* );
* );
* ~~~
* ~~~
...
...
libcaf_core/src/abstract_coordinator.cpp
View file @
ae7a2017
...
@@ -158,7 +158,7 @@ void printer_loop(blocking_actor* self) {
...
@@ -158,7 +158,7 @@ void printer_loop(blocking_actor* self) {
};
};
bool
running
=
true
;
bool
running
=
true
;
self
->
receive_while
([
&
]
{
return
running
;
})(
self
->
receive_while
([
&
]
{
return
running
;
})(
on
(
atom
(
"add"
),
arg_match
)
>>
[
&
](
std
::
string
&
str
)
{
[
&
](
add_atom
,
std
::
string
&
str
)
{
auto
s
=
self
->
current_sender
();
auto
s
=
self
->
current_sender
();
if
(
str
.
empty
()
||
s
==
invalid_actor_addr
)
{
if
(
str
.
empty
()
||
s
==
invalid_actor_addr
)
{
return
;
return
;
...
@@ -173,7 +173,7 @@ void printer_loop(blocking_actor* self) {
...
@@ -173,7 +173,7 @@ void printer_loop(blocking_actor* self) {
}
}
flush_if_needed
(
i
->
second
);
flush_if_needed
(
i
->
second
);
},
},
on
(
atom
(
"flush"
))
>>
[
&
]
{
[
&
](
flush_atom
)
{
flush_output
(
self
->
current_sender
());
flush_output
(
self
->
current_sender
());
},
},
[
&
](
const
down_msg
&
dm
)
{
[
&
](
const
down_msg
&
dm
)
{
...
...
libcaf_core/src/actor_ostream.cpp
View file @
ae7a2017
...
@@ -34,12 +34,12 @@ actor_ostream::actor_ostream(actor self) : m_self(std::move(self)) {
...
@@ -34,12 +34,12 @@ actor_ostream::actor_ostream(actor self) : m_self(std::move(self)) {
}
}
actor_ostream
&
actor_ostream
::
write
(
std
::
string
arg
)
{
actor_ostream
&
actor_ostream
::
write
(
std
::
string
arg
)
{
send_as
(
m_self
,
m_printer
,
a
tom
(
"add"
)
,
std
::
move
(
arg
));
send_as
(
m_self
,
m_printer
,
a
dd_atom
::
value
,
std
::
move
(
arg
));
return
*
this
;
return
*
this
;
}
}
actor_ostream
&
actor_ostream
::
flush
()
{
actor_ostream
&
actor_ostream
::
flush
()
{
send_as
(
m_self
,
m_printer
,
atom
(
"flush"
)
);
send_as
(
m_self
,
m_printer
,
flush_atom
::
value
);
return
*
this
;
return
*
this
;
}
}
...
...
libcaf_core/test/atom.cpp
View file @
ae7a2017
...
@@ -31,7 +31,12 @@ namespace {
...
@@ -31,7 +31,12 @@ namespace {
constexpr
auto
s_foo
=
atom
(
"FooBar"
);
constexpr
auto
s_foo
=
atom
(
"FooBar"
);
using
a_atom
=
atom_constant
<
atom
(
"a"
)
>
;
using
b_atom
=
atom_constant
<
atom
(
"b"
)
>
;
using
c_atom
=
atom_constant
<
atom
(
"c"
)
>
;
using
abc_atom
=
atom_constant
<
atom
(
"abc"
)
>
;
using
abc_atom
=
atom_constant
<
atom
(
"abc"
)
>
;
using
def_atom
=
atom_constant
<
atom
(
"def"
)
>
;
using
foo_atom
=
atom_constant
<
atom
(
"foo"
)
>
;
struct
fixture
{
struct
fixture
{
~
fixture
()
{
~
fixture
()
{
...
@@ -69,31 +74,31 @@ struct send_to_self {
...
@@ -69,31 +74,31 @@ struct send_to_self {
CAF_TEST
(
receive_atoms
)
{
CAF_TEST
(
receive_atoms
)
{
scoped_actor
self
;
scoped_actor
self
;
send_to_self
f
{
self
.
get
()};
send_to_self
f
{
self
.
get
()};
f
(
atom
(
"foo"
)
,
static_cast
<
uint32_t
>
(
42
));
f
(
foo_atom
::
value
,
static_cast
<
uint32_t
>
(
42
));
f
(
a
tom
(
"abc"
),
atom
(
"def"
)
,
"cstring"
);
f
(
a
bc_atom
::
value
,
def_atom
::
value
,
"cstring"
);
f
(
1.
f
);
f
(
1.
f
);
f
(
a
tom
(
"a"
),
atom
(
"b"
),
atom
(
"c"
)
,
23.
f
);
f
(
a
_atom
::
value
,
b_atom
::
value
,
c_atom
::
value
,
23.
f
);
bool
matched_pattern
[
3
]
=
{
false
,
false
,
false
};
bool
matched_pattern
[
3
]
=
{
false
,
false
,
false
};
int
i
=
0
;
int
i
=
0
;
CAF_MESSAGE
(
"start receive loop"
);
CAF_MESSAGE
(
"start receive loop"
);
for
(
i
=
0
;
i
<
3
;
++
i
)
{
for
(
i
=
0
;
i
<
3
;
++
i
)
{
self
->
receive
(
self
->
receive
(
on
(
atom
(
"foo"
),
arg_match
)
>>
[
&
](
uint32_t
value
)
{
[
&
](
foo_atom
,
uint32_t
value
)
{
matched_pattern
[
0
]
=
true
;
matched_pattern
[
0
]
=
true
;
CAF_CHECK_EQUAL
(
value
,
42
);
CAF_CHECK_EQUAL
(
value
,
42
);
},
},
on
(
atom
(
"abc"
),
atom
(
"def"
),
arg_match
)
>>
[
&
](
const
std
::
string
&
str
)
{
[
&
](
abc_atom
,
def_atom
,
const
std
::
string
&
str
)
{
matched_pattern
[
1
]
=
true
;
matched_pattern
[
1
]
=
true
;
CAF_CHECK_EQUAL
(
str
,
"cstring"
);
CAF_CHECK_EQUAL
(
str
,
"cstring"
);
},
},
on
(
atom
(
"a"
),
atom
(
"b"
),
atom
(
"c"
),
arg_match
)
>>
[
&
](
float
value
)
{
[
&
](
a_atom
,
b_atom
,
c_atom
,
float
value
)
{
matched_pattern
[
2
]
=
true
;
matched_pattern
[
2
]
=
true
;
CAF_CHECK_EQUAL
(
value
,
23.
f
);
CAF_CHECK_EQUAL
(
value
,
23.
f
);
});
});
}
}
CAF_CHECK
(
matched_pattern
[
0
]
&&
matched_pattern
[
1
]
&&
matched_pattern
[
2
]);
CAF_CHECK
(
matched_pattern
[
0
]
&&
matched_pattern
[
1
]
&&
matched_pattern
[
2
]);
self
->
receive
(
self
->
receive
(
// "erase" message {
atom("b"), atom("a"), atom("c")
, 23.f }
// "erase" message {
'b', 'a, 'c'
, 23.f }
others
>>
[]
{
others
>>
[]
{
CAF_MESSAGE
(
"drain mailbox"
);
CAF_MESSAGE
(
"drain mailbox"
);
},
},
...
@@ -104,10 +109,10 @@ CAF_TEST(receive_atoms) {
...
@@ -104,10 +109,10 @@ CAF_TEST(receive_atoms) {
atom_value
x
=
atom
(
"abc"
);
atom_value
x
=
atom
(
"abc"
);
atom_value
y
=
abc_atom
::
value
;
atom_value
y
=
abc_atom
::
value
;
CAF_CHECK_EQUAL
(
x
,
y
);
CAF_CHECK_EQUAL
(
x
,
y
);
auto
msg
=
make_message
(
a
bc_atom
(
));
auto
msg
=
make_message
(
a
tom
(
"abc"
));
self
->
send
(
self
,
msg
);
self
->
send
(
self
,
msg
);
self
->
receive
(
self
->
receive
(
on
(
atom
(
"abc"
))
>>
[]
{
[](
abc_atom
)
{
CAF_MESSAGE
(
"received 'abc'"
);
CAF_MESSAGE
(
"received 'abc'"
);
},
},
others
>>
[]
{
others
>>
[]
{
...
...
libcaf_core/test/constructor_attach.cpp
View file @
ae7a2017
...
@@ -24,17 +24,24 @@
...
@@ -24,17 +24,24 @@
using
namespace
caf
;
using
namespace
caf
;
namespace
{
using
die_atom
=
atom_constant
<
atom
(
"die"
)
>
;
using
done_atom
=
atom_constant
<
atom
(
"done"
)
>
;
}
// namespace <anonymous>
CAF_TEST
(
constructor_attach
)
{
CAF_TEST
(
constructor_attach
)
{
class
testee
:
public
event_based_actor
{
class
testee
:
public
event_based_actor
{
public:
public:
testee
(
actor
buddy
)
:
m_buddy
(
buddy
)
{
testee
(
actor
buddy
)
:
m_buddy
(
buddy
)
{
attach_functor
([
=
](
uint32_t
reason
)
{
attach_functor
([
=
](
uint32_t
reason
)
{
send
(
m_buddy
,
atom
(
"done"
)
,
reason
);
send
(
m_buddy
,
done_atom
::
value
,
reason
);
});
});
}
}
behavior
make_behavior
()
{
behavior
make_behavior
()
{
return
{
return
{
on
(
atom
(
"die"
))
>>
[
=
]
{
[
=
](
die_atom
)
{
quit
(
exit_reason
::
user_shutdown
);
quit
(
exit_reason
::
user_shutdown
);
}
}
};
};
...
@@ -55,7 +62,7 @@ CAF_TEST(constructor_attach) {
...
@@ -55,7 +62,7 @@ CAF_TEST(constructor_attach) {
quit
(
msg
.
reason
);
quit
(
msg
.
reason
);
}
}
},
},
on
(
atom
(
"done"
),
arg_match
)
>>
[
=
](
uint32_t
reason
)
{
[
=
](
done_atom
,
uint32_t
reason
)
{
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
if
(
++
m_downs
==
2
)
{
if
(
++
m_downs
==
2
)
{
quit
(
reason
);
quit
(
reason
);
...
@@ -70,7 +77,7 @@ CAF_TEST(constructor_attach) {
...
@@ -70,7 +77,7 @@ CAF_TEST(constructor_attach) {
int
m_downs
;
int
m_downs
;
actor
m_testee
;
actor
m_testee
;
};
};
anon_send
(
spawn
<
spawner
>
(),
atom
(
"die"
)
);
anon_send
(
spawn
<
spawner
>
(),
die_atom
::
value
);
await_all_actors_done
();
await_all_actors_done
();
shutdown
();
shutdown
();
}
}
libcaf_core/test/serial_reply.cpp
View file @
ae7a2017
...
@@ -26,6 +26,19 @@ using std::cout;
...
@@ -26,6 +26,19 @@ using std::cout;
using
std
::
endl
;
using
std
::
endl
;
using
namespace
caf
;
using
namespace
caf
;
namespace
{
using
hi_atom
=
atom_constant
<
atom
(
"hi"
)
>
;
using
ho_atom
=
atom_constant
<
atom
(
"ho"
)
>
;
using
sub0_atom
=
atom_constant
<
atom
(
"sub0"
)
>
;
using
sub1_atom
=
atom_constant
<
atom
(
"sub1"
)
>
;
using
sub2_atom
=
atom_constant
<
atom
(
"sub2"
)
>
;
using
sub3_atom
=
atom_constant
<
atom
(
"sub3"
)
>
;
using
sub4_atom
=
atom_constant
<
atom
(
"sub4"
)
>
;
}
// namespace <anonymous>
CAF_TEST
(
test_serial_reply
)
{
CAF_TEST
(
test_serial_reply
)
{
auto
mirror_behavior
=
[
=
](
event_based_actor
*
self
)
{
auto
mirror_behavior
=
[
=
](
event_based_actor
*
self
)
{
self
->
become
(
others
>>
[
=
]()
->
message
{
self
->
become
(
others
>>
[
=
]()
->
message
{
...
@@ -42,26 +55,24 @@ CAF_TEST(test_serial_reply) {
...
@@ -42,26 +55,24 @@ CAF_TEST(test_serial_reply) {
auto
c3
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
auto
c3
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
auto
c4
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
auto
c4
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
self
->
become
(
self
->
become
(
on
(
atom
(
"hi there"
))
>>
[
=
](
)
->
continue_helper
{
[
=
](
hi_atom
)
->
continue_helper
{
CAF_MESSAGE
(
"received 'hi there'"
);
CAF_MESSAGE
(
"received 'hi there'"
);
return
self
->
sync_send
(
c0
,
atom
(
"sub0"
)
).
then
(
return
self
->
sync_send
(
c0
,
sub0_atom
::
value
).
then
(
on
(
atom
(
"sub0"
))
>>
[
=
](
)
->
continue_helper
{
[
=
](
sub0_atom
)
->
continue_helper
{
CAF_MESSAGE
(
"received 'sub0'"
);
CAF_MESSAGE
(
"received 'sub0'"
);
return
self
->
sync_send
(
c1
,
atom
(
"sub1"
)
).
then
(
return
self
->
sync_send
(
c1
,
sub1_atom
::
value
).
then
(
on
(
atom
(
"sub1"
))
>>
[
=
](
)
->
continue_helper
{
[
=
](
sub1_atom
)
->
continue_helper
{
CAF_MESSAGE
(
"received 'sub1'"
);
CAF_MESSAGE
(
"received 'sub1'"
);
return
self
->
sync_send
(
c2
,
atom
(
"sub2"
)
).
then
(
return
self
->
sync_send
(
c2
,
sub2_atom
::
value
).
then
(
on
(
atom
(
"sub2"
))
>>
[
=
](
)
->
continue_helper
{
[
=
](
sub2_atom
)
->
continue_helper
{
CAF_MESSAGE
(
"received 'sub2'"
);
CAF_MESSAGE
(
"received 'sub2'"
);
return
self
->
sync_send
(
c3
,
atom
(
"sub3"
)
).
then
(
return
self
->
sync_send
(
c3
,
sub3_atom
::
value
).
then
(
on
(
atom
(
"sub3"
))
>>
[
=
](
)
->
continue_helper
{
[
=
](
sub3_atom
)
->
continue_helper
{
CAF_MESSAGE
(
"received 'sub3'"
);
CAF_MESSAGE
(
"received 'sub3'"
);
return
self
->
sync_send
(
c4
,
atom
(
"sub4"
)
).
then
(
return
self
->
sync_send
(
c4
,
sub4_atom
::
value
).
then
(
on
(
atom
(
"sub4"
))
>>
[
=
](
)
->
atom_value
{
[
=
](
sub4_atom
)
->
atom_value
{
CAF_MESSAGE
(
"received 'sub4'"
);
CAF_MESSAGE
(
"received 'sub4'"
);
return
atom
(
"hiho"
);
return
ho_atom
::
value
;
}
);
}
}
);
);
}
}
...
@@ -74,12 +85,13 @@ CAF_TEST(test_serial_reply) {
...
@@ -74,12 +85,13 @@ CAF_TEST(test_serial_reply) {
);
);
}
}
);
);
});
{
// lifetime scope of self
{
// lifetime scope of self
scoped_actor
self
;
scoped_actor
self
;
CAF_MESSAGE
(
"ID of main: "
<<
self
->
id
());
CAF_MESSAGE
(
"ID of main: "
<<
self
->
id
());
self
->
sync_send
(
master
,
atom
(
"hi there"
)
).
await
(
self
->
sync_send
(
master
,
hi_atom
::
value
).
await
(
on
(
atom
(
"hiho"
))
>>
[]
{
[](
ho_atom
)
{
CAF_MESSAGE
(
"received
`hiho` atom
"
);
CAF_MESSAGE
(
"received
'ho'
"
);
},
},
others
>>
[
&
]
{
others
>>
[
&
]
{
CAF_TEST_ERROR
(
"Unexpected message: "
CAF_TEST_ERROR
(
"Unexpected message: "
...
...
libcaf_core/test/spawn.cpp
View file @
ae7a2017
...
@@ -36,6 +36,13 @@ namespace {
...
@@ -36,6 +36,13 @@ namespace {
std
::
atomic
<
long
>
s_max_actor_instances
;
std
::
atomic
<
long
>
s_max_actor_instances
;
std
::
atomic
<
long
>
s_actor_instances
;
std
::
atomic
<
long
>
s_actor_instances
;
using
a_atom
=
atom_constant
<
atom
(
"a"
)
>
;
using
b_atom
=
atom_constant
<
atom
(
"b"
)
>
;
using
c_atom
=
atom_constant
<
atom
(
"c"
)
>
;
using
abc_atom
=
atom_constant
<
atom
(
"abc"
)
>
;
using
name_atom
=
atom_constant
<
atom
(
"name"
)
>
;
void
inc_actor_instances
()
{
void
inc_actor_instances
()
{
long
v1
=
++
s_actor_instances
;
long
v1
=
++
s_actor_instances
;
long
v2
=
s_max_actor_instances
.
load
();
long
v2
=
s_max_actor_instances
.
load
();
...
@@ -107,7 +114,7 @@ actor spawn_event_testee2(actor parent) {
...
@@ -107,7 +114,7 @@ actor spawn_event_testee2(actor parent) {
after
(
chrono
::
milliseconds
(
1
))
>>
[
=
]
{
after
(
chrono
::
milliseconds
(
1
))
>>
[
=
]
{
CAF_MESSAGE
(
CAF_ARG
(
remaining
));
CAF_MESSAGE
(
CAF_ARG
(
remaining
));
if
(
remaining
==
1
)
{
if
(
remaining
==
1
)
{
send
(
parent
,
atom
(
"t2done"
)
);
send
(
parent
,
ok_atom
::
value
);
quit
();
quit
();
}
}
else
become
(
wait4timeout
(
remaining
-
1
));
else
become
(
wait4timeout
(
remaining
-
1
));
...
@@ -121,46 +128,40 @@ actor spawn_event_testee2(actor parent) {
...
@@ -121,46 +128,40 @@ actor spawn_event_testee2(actor parent) {
return
spawn
<
impl
>
(
parent
);
return
spawn
<
impl
>
(
parent
);
}
}
struct
chopstick
:
public
sb_actor
<
chopstick
>
{
class
chopstick
:
public
event_based_actor
{
public:
behavior
make_behavior
()
override
{
return
available
;
}
behavior
taken_by
(
actor
whom
)
{
behavior
taken_by
(
actor
whom
)
{
return
{
return
{
on
<
atom
(
"take"
)
>
()
>>
[
=
]
{
[
=
](
get_atom
)
{
return
atom
(
"busy"
)
;
return
error_atom
::
value
;
},
},
on
(
atom
(
"put"
),
whom
)
>>
[
=
]
{
on
(
put_atom
::
value
,
whom
)
>>
[
=
]()
{
become
(
available
);
become
(
available
);
},
on
(
atom
(
"break"
))
>>
[
=
]
{
quit
();
}
}
};
};
}
}
behavior
available
;
chopstick
()
{
behavior
&
init_state
=
available
;
chopstick
();
~
chopstick
();
};
chopstick
::
chopstick
()
{
inc_actor_instances
();
inc_actor_instances
();
available
.
assign
(
available
.
assign
(
on
(
atom
(
"take"
),
arg_match
)
>>
[
=
](
actor
whom
)
->
atom_value
{
[
=
](
get_atom
,
actor
whom
)
->
atom_value
{
become
(
taken_by
(
whom
));
become
(
taken_by
(
whom
));
return
atom
(
"taken"
);
return
ok_atom
::
value
;
},
on
(
atom
(
"break"
))
>>
[
=
]
{
quit
();
}
}
);
);
}
}
chopstick
::
~
chopstick
()
{
~
chopstick
()
{
dec_actor_instances
();
dec_actor_instances
();
}
}
private:
behavior
available
;
};
class
testee_actor
:
public
blocking_actor
{
class
testee_actor
:
public
blocking_actor
{
public:
public:
...
@@ -292,18 +293,18 @@ behavior simple_mirror::make_behavior() {
...
@@ -292,18 +293,18 @@ behavior simple_mirror::make_behavior() {
}
}
behavior
high_priority_testee
(
event_based_actor
*
self
)
{
behavior
high_priority_testee
(
event_based_actor
*
self
)
{
self
->
send
(
self
,
atom
(
"b"
)
);
self
->
send
(
self
,
b_atom
::
value
);
self
->
send
(
message_priority
::
high
,
self
,
a
tom
(
"a"
)
);
self
->
send
(
message_priority
::
high
,
self
,
a
_atom
::
value
);
// 'a' must be self->received before 'b'
// 'a' must be self->received before 'b'
return
{
return
{
on
(
atom
(
"b"
))
>>
[
=
]
{
[
=
](
b_atom
)
{
CAF_TEST_ERROR
(
"received 'b' before 'a'"
);
CAF_TEST_ERROR
(
"received 'b' before 'a'"
);
self
->
quit
();
self
->
quit
();
},
},
on
(
atom
(
"a"
))
>>
[
=
]
{
[
=
](
a_atom
)
{
CAF_MESSAGE
(
"received
\"
a
\"
atom"
);
CAF_MESSAGE
(
"received
\"
a
\"
atom"
);
self
->
become
(
self
->
become
(
on
(
atom
(
"b"
))
>>
[
=
]
{
[
=
](
b_atom
)
{
CAF_MESSAGE
(
"received
\"
b
\"
atom, about to quit"
);
CAF_MESSAGE
(
"received
\"
b
\"
atom, about to quit"
);
self
->
quit
();
self
->
quit
();
},
},
...
@@ -329,7 +330,7 @@ struct high_priority_testee_class : event_based_actor {
...
@@ -329,7 +330,7 @@ struct high_priority_testee_class : event_based_actor {
struct
master
:
event_based_actor
{
struct
master
:
event_based_actor
{
behavior
make_behavior
()
override
{
behavior
make_behavior
()
override
{
return
(
return
(
on
(
atom
(
"done"
))
>>
[
=
]
{
[
=
](
ok_atom
)
{
CAF_MESSAGE
(
"master: received done"
);
CAF_MESSAGE
(
"master: received done"
);
quit
(
exit_reason
::
user_shutdown
);
quit
(
exit_reason
::
user_shutdown
);
}
}
...
@@ -376,11 +377,11 @@ counting_actor::~counting_actor() {
...
@@ -376,11 +377,11 @@ counting_actor::~counting_actor() {
behavior
counting_actor
::
make_behavior
()
{
behavior
counting_actor
::
make_behavior
()
{
for
(
int
i
=
0
;
i
<
100
;
++
i
)
{
for
(
int
i
=
0
;
i
<
100
;
++
i
)
{
send
(
this
,
atom
(
"dummy"
)
);
send
(
this
,
ok_atom
::
value
);
}
}
CAF_CHECK_EQUAL
(
mailbox
().
count
(),
100
);
CAF_CHECK_EQUAL
(
mailbox
().
count
(),
100
);
for
(
int
i
=
0
;
i
<
100
;
++
i
)
{
for
(
int
i
=
0
;
i
<
100
;
++
i
)
{
send
(
this
,
atom
(
"dummy"
)
);
send
(
this
,
ok_atom
::
value
);
}
}
CAF_CHECK_EQUAL
(
mailbox
().
count
(),
200
);
CAF_CHECK_EQUAL
(
mailbox
().
count
(),
200
);
return
{};
return
{};
...
@@ -417,7 +418,7 @@ CAF_TEST(detached_actors_and_schedulued_actors) {
...
@@ -417,7 +418,7 @@ CAF_TEST(detached_actors_and_schedulued_actors) {
auto
m
=
spawn
<
master
,
detached
>
();
auto
m
=
spawn
<
master
,
detached
>
();
spawn
<
slave
>
(
m
);
spawn
<
slave
>
(
m
);
spawn
<
slave
>
(
m
);
spawn
<
slave
>
(
m
);
self
->
send
(
m
,
atom
(
"done"
)
);
self
->
send
(
m
,
ok_atom
::
value
);
}
}
CAF_TEST
(
self_receive_with_zero_timeout
)
{
CAF_TEST
(
self_receive_with_zero_timeout
)
{
...
@@ -552,8 +553,8 @@ CAF_TEST(spawn_event_testee2) {
...
@@ -552,8 +553,8 @@ CAF_TEST(spawn_event_testee2) {
scoped_actor
self
;
scoped_actor
self
;
spawn_event_testee2
(
self
);
spawn_event_testee2
(
self
);
self
->
receive
(
self
->
receive
(
on
(
atom
(
"t2done"
))
>>
[]
{
[](
ok_atom
)
{
CAF_MESSAGE
(
"Received
\"
t2done
\"
"
);
CAF_MESSAGE
(
"Received
'ok'
"
);
}
}
);
);
}
}
...
@@ -561,11 +562,11 @@ CAF_TEST(spawn_event_testee2) {
...
@@ -561,11 +562,11 @@ CAF_TEST(spawn_event_testee2) {
CAF_TEST
(
chopsticks
)
{
CAF_TEST
(
chopsticks
)
{
scoped_actor
self
;
scoped_actor
self
;
auto
cstk
=
spawn
<
chopstick
>
();
auto
cstk
=
spawn
<
chopstick
>
();
self
->
send
(
cstk
,
atom
(
"take"
)
,
self
);
self
->
send
(
cstk
,
get_atom
::
value
,
self
);
self
->
receive
(
self
->
receive
(
on
(
atom
(
"taken"
))
>>
[
&
]
{
[
&
](
ok_atom
)
{
self
->
send
(
cstk
,
atom
(
"put"
)
,
self
);
self
->
send
(
cstk
,
put_atom
::
value
,
self
);
self
->
send
(
cstk
,
atom
(
"break"
)
);
self
->
send
_exit
(
cstk
,
exit_reason
::
kill
);
},
},
others
>>
[
&
]
{
others
>>
[
&
]
{
CAF_TEST_ERROR
(
"Unexpected message: "
<<
CAF_TEST_ERROR
(
"Unexpected message: "
<<
...
@@ -649,7 +650,7 @@ CAF_TEST(inflater) {
...
@@ -649,7 +650,7 @@ CAF_TEST(inflater) {
[
=
](
int
n
,
const
string
&
str
)
{
[
=
](
int
n
,
const
string
&
str
)
{
send
(
m_buddy
,
n
*
2
,
str
+
" from "
+
m_name
);
send
(
m_buddy
,
n
*
2
,
str
+
" from "
+
m_name
);
},
},
on
(
atom
(
"done"
))
>>
[
=
]
{
[
=
](
ok_atom
)
{
quit
();
quit
();
}
}
};
};
...
@@ -671,9 +672,9 @@ CAF_TEST(inflater) {
...
@@ -671,9 +672,9 @@ CAF_TEST(inflater) {
}
}
);
);
// kill joe and bob
// kill joe and bob
auto
poison_pill
=
make_message
(
atom
(
"done"
)
);
auto
ok_message
=
make_message
(
ok_atom
::
value
);
anon_send
(
joe
,
poison_pill
);
anon_send
(
joe
,
ok_message
);
anon_send
(
bob
,
poison_pill
);
anon_send
(
bob
,
ok_message
);
}
}
CAF_TEST
(
kr34t0r
)
{
CAF_TEST
(
kr34t0r
)
{
...
@@ -709,29 +710,29 @@ CAF_TEST(kr34t0r) {
...
@@ -709,29 +710,29 @@ CAF_TEST(kr34t0r) {
};
};
scoped_actor
self
;
scoped_actor
self
;
auto
joe_the_second
=
spawn
<
kr34t0r
>
(
"Joe"
,
invalid_actor
);
auto
joe_the_second
=
spawn
<
kr34t0r
>
(
"Joe"
,
invalid_actor
);
self
->
send
(
joe_the_second
,
atom
(
"done"
)
);
self
->
send
(
joe_the_second
,
ok_atom
::
value
);
}
}
CAF_TEST
(
function_spawn
)
{
CAF_TEST
(
function_spawn
)
{
scoped_actor
self
;
scoped_actor
self
;
auto
f
=
[](
const
string
&
name
)
->
behavior
{
auto
f
=
[](
const
string
&
name
)
->
behavior
{
return
(
return
(
on
(
atom
(
"get_name"
))
>>
[
name
]
{
[
name
](
get_atom
)
{
return
make_message
(
atom
(
"name"
)
,
name
);
return
std
::
make_tuple
(
name_atom
::
value
,
name
);
}
}
);
);
};
};
auto
a1
=
spawn
(
f
,
"alice"
);
auto
a1
=
spawn
(
f
,
"alice"
);
auto
a2
=
spawn
(
f
,
"bob"
);
auto
a2
=
spawn
(
f
,
"bob"
);
self
->
send
(
a1
,
atom
(
"get_name"
)
);
self
->
send
(
a1
,
get_atom
::
value
);
self
->
receive
(
self
->
receive
(
on
(
atom
(
"name"
),
arg_match
)
>>
[
&
](
const
string
&
name
)
{
[
&
](
name_atom
,
const
string
&
name
)
{
CAF_CHECK_EQUAL
(
name
,
"alice"
);
CAF_CHECK_EQUAL
(
name
,
"alice"
);
}
}
);
);
self
->
send
(
a2
,
atom
(
"get_name"
)
);
self
->
send
(
a2
,
get_atom
::
value
);
self
->
receive
(
self
->
receive
(
on
(
atom
(
"name"
),
arg_match
)
>>
[
&
](
const
string
&
name
)
{
[
&
](
name_atom
,
const
string
&
name
)
{
CAF_CHECK_EQUAL
(
name
,
"bob"
);
CAF_CHECK_EQUAL
(
name
,
"bob"
);
}
}
);
);
...
@@ -739,14 +740,12 @@ CAF_TEST(function_spawn) {
...
@@ -739,14 +740,12 @@ CAF_TEST(function_spawn) {
self
->
send_exit
(
a2
,
exit_reason
::
user_shutdown
);
self
->
send_exit
(
a2
,
exit_reason
::
user_shutdown
);
}
}
using
abc_atom
=
atom_constant
<
atom
(
"abc"
)
>
;
using
typed_testee
=
typed_actor
<
replies_to
<
abc_atom
>::
with
<
std
::
string
>>
;
using
typed_testee
=
typed_actor
<
replies_to
<
abc_atom
>::
with
<
std
::
string
>>
;
typed_testee
::
behavior_type
testee
()
{
typed_testee
::
behavior_type
testee
()
{
return
{
return
{
[](
abc_atom
)
{
[](
abc_atom
)
{
CAF_MESSAGE
(
"received
abc_atom
"
);
CAF_MESSAGE
(
"received
'abc'
"
);
return
"abc"
;
return
"abc"
;
}
}
};
};
...
@@ -755,7 +754,7 @@ typed_testee::behavior_type testee() {
...
@@ -755,7 +754,7 @@ typed_testee::behavior_type testee() {
CAF_TEST
(
typed_await
)
{
CAF_TEST
(
typed_await
)
{
scoped_actor
self
;
scoped_actor
self
;
auto
x
=
spawn_typed
(
testee
);
auto
x
=
spawn_typed
(
testee
);
self
->
sync_send
(
x
,
abc_atom
()
).
await
(
self
->
sync_send
(
x
,
abc_atom
::
value
).
await
(
[](
const
std
::
string
&
str
)
{
[](
const
std
::
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
"abc"
);
CAF_CHECK_EQUAL
(
str
,
"abc"
);
}
}
...
@@ -769,14 +768,15 @@ CAF_TEST(constructor_attach) {
...
@@ -769,14 +768,15 @@ CAF_TEST(constructor_attach) {
public:
public:
testee
(
actor
buddy
)
:
m_buddy
(
buddy
)
{
testee
(
actor
buddy
)
:
m_buddy
(
buddy
)
{
attach_functor
([
=
](
uint32_t
reason
)
{
attach_functor
([
=
](
uint32_t
reason
)
{
send
(
buddy
,
atom
(
"done"
)
,
reason
);
send
(
buddy
,
ok_atom
::
value
,
reason
);
});
});
}
}
behavior
make_behavior
()
{
behavior
make_behavior
()
{
return
{
return
{
on
(
atom
(
"die"
))
>>
[
=
]
{
others
>>
[
=
]
{
quit
(
exit_reason
::
user_shutdown
);
CAF_TEST_ERROR
(
"Unexpected message: "
<<
to_string
(
current_message
()));
}
}
};
};
}
}
...
@@ -794,6 +794,7 @@ CAF_TEST(constructor_attach) {
...
@@ -794,6 +794,7 @@ CAF_TEST(constructor_attach) {
// nop
// nop
}
}
behavior
make_behavior
()
{
behavior
make_behavior
()
{
trap_exit
(
true
);
m_testee
=
spawn
<
testee
,
monitored
>
(
this
);
m_testee
=
spawn
<
testee
,
monitored
>
(
this
);
return
{
return
{
[
=
](
const
down_msg
&
msg
)
{
[
=
](
const
down_msg
&
msg
)
{
...
@@ -802,7 +803,7 @@ CAF_TEST(constructor_attach) {
...
@@ -802,7 +803,7 @@ CAF_TEST(constructor_attach) {
quit
(
msg
.
reason
);
quit
(
msg
.
reason
);
}
}
},
},
on
(
atom
(
"done"
),
arg_match
)
>>
[
=
](
uint32_t
reason
)
{
[
=
](
ok_atom
,
uint32_t
reason
)
{
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
if
(
++
m_downs
==
2
)
{
if
(
++
m_downs
==
2
)
{
quit
(
reason
);
quit
(
reason
);
...
@@ -822,7 +823,7 @@ CAF_TEST(constructor_attach) {
...
@@ -822,7 +823,7 @@ CAF_TEST(constructor_attach) {
int
m_downs
;
int
m_downs
;
actor
m_testee
;
actor
m_testee
;
};
};
anon_send
(
spawn
<
spawner
>
(),
atom
(
"die"
)
);
anon_send
_exit
(
spawn
<
spawner
>
(),
exit_reason
::
user_shutdown
);
}
}
namespace
{
namespace
{
...
@@ -893,8 +894,9 @@ CAF_TEST(kill_the_immortal) {
...
@@ -893,8 +894,9 @@ CAF_TEST(kill_the_immortal) {
auto
wannabe_immortal
=
spawn
([](
event_based_actor
*
self
)
->
behavior
{
auto
wannabe_immortal
=
spawn
([](
event_based_actor
*
self
)
->
behavior
{
self
->
trap_exit
(
true
);
self
->
trap_exit
(
true
);
return
{
return
{
others
>>
[]
{
others
>>
[
=
]
{
CAF_TEST_ERROR
(
"Unexpected message"
);
CAF_TEST_ERROR
(
"Unexpected message: "
<<
to_string
(
self
->
current_message
()));
}
}
};
};
});
});
...
...
libcaf_io/caf/io/network/asio_multiplexer.hpp
View file @
ae7a2017
...
@@ -21,6 +21,8 @@
...
@@ -21,6 +21,8 @@
#ifndef CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#ifndef CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#include "caf/config.hpp"
CAF_PUSH_WARNINGS
CAF_PUSH_WARNINGS
#include "boost/asio.hpp"
#include "boost/asio.hpp"
CAF_POP_WARNINGS
CAF_POP_WARNINGS
...
@@ -100,7 +102,9 @@ class asio_multiplexer : public multiplexer {
...
@@ -100,7 +102,9 @@ class asio_multiplexer : public multiplexer {
void
run
()
override
;
void
run
()
override
;
private:
private:
inline
boost
::
asio
::
io_service
&
backend
()
{
return
m_backend
;
}
inline
boost
::
asio
::
io_service
&
backend
()
{
return
m_backend
;
}
io_backend
m_backend
;
io_backend
m_backend
;
std
::
mutex
m_mtx_sockets
;
std
::
mutex
m_mtx_sockets
;
...
...
libcaf_io/caf/io/network/multiplexer.hpp
View file @
ae7a2017
...
@@ -25,6 +25,7 @@
...
@@ -25,6 +25,7 @@
#include <functional>
#include <functional>
#include "caf/extend.hpp"
#include "caf/extend.hpp"
#include "caf/make_counted.hpp"
#include "caf/memory_managed.hpp"
#include "caf/memory_managed.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/fwd.hpp"
...
@@ -34,8 +35,6 @@
...
@@ -34,8 +35,6 @@
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/detail/memory.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace
boost
{
namespace
boost
{
...
@@ -176,8 +175,7 @@ class multiplexer {
...
@@ -176,8 +175,7 @@ class multiplexer {
f
();
f
();
}
}
};
};
dispatch_runnable
(
runnable_ptr
{
detail
::
memory
::
create
<
impl
>
(
std
::
move
(
fun
)),
dispatch_runnable
(
make_counted
<
impl
>
(
std
::
move
(
fun
)));
false
});
}
}
/**
/**
...
...
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