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
2f6aebe2
Commit
2f6aebe2
authored
May 06, 2014
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
fixed GCC build
parent
f375a2d9
Changes
3
Hide whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
53 additions
and
126 deletions
+53
-126
cppa/cppa.hpp
cppa/cppa.hpp
+6
-5
cppa/io/broker.hpp
cppa/io/broker.hpp
+23
-44
src/broker.cpp
src/broker.cpp
+24
-77
No files found.
cppa/cppa.hpp
View file @
2f6aebe2
...
@@ -662,8 +662,9 @@ actor spawn_io_client(F fun,
...
@@ -662,8 +662,9 @@ actor spawn_io_client(F fun,
io
::
input_stream_ptr
in
,
io
::
input_stream_ptr
in
,
io
::
output_stream_ptr
out
,
io
::
output_stream_ptr
out
,
Ts
&&
...
args
)
{
Ts
&&
...
args
)
{
auto
ptr
=
io
::
broker
::
from
(
std
::
move
(
fun
),
std
::
move
(
in
),
std
::
move
(
out
),
auto
hdl
=
io
::
connection_handle
::
from_int
(
in
->
read_handle
());
std
::
forward
<
Ts
>
(
args
)...);
auto
ptr
=
io
::
broker
::
from
(
std
::
move
(
fun
),
hdl
,
std
::
forward
<
Ts
>
(
args
)...);
ptr
->
add_connection
(
std
::
move
(
in
),
std
::
move
(
out
));
return
{
io
::
init_and_launch
(
std
::
move
(
ptr
))};
return
{
io
::
init_and_launch
(
std
::
move
(
ptr
))};
}
}
...
@@ -684,9 +685,9 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
...
@@ -684,9 +685,9 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
static_assert
(
is_unbound
(
Os
),
static_assert
(
is_unbound
(
Os
),
"top-level spawns cannot have monitor or link flag"
);
"top-level spawns cannot have monitor or link flag"
);
using
namespace
std
;
using
namespace
std
;
auto
ptr
=
io
::
broker
::
from
(
move
(
fun
),
auto
aptr
=
io
::
tcp_acceptor
::
create
(
port
);
io
::
tcp_acceptor
::
create
(
port
),
auto
ptr
=
io
::
broker
::
from
(
move
(
fun
),
forward
<
Ts
>
(
args
)...);
forward
<
Ts
>
(
args
)...
);
ptr
->
add_acceptor
(
std
::
move
(
aptr
)
);
return
{
io
::
init_and_launch
(
move
(
ptr
))};
return
{
io
::
init_and_launch
(
move
(
ptr
))};
}
}
...
...
cppa/io/broker.hpp
View file @
2f6aebe2
...
@@ -125,11 +125,15 @@ class broker : public extend<local_actor>::
...
@@ -125,11 +125,15 @@ class broker : public extend<local_actor>::
template
<
typename
F
,
typename
...
Ts
>
template
<
typename
F
,
typename
...
Ts
>
actor
fork
(
F
fun
,
connection_handle
hdl
,
Ts
&&
...
args
)
{
actor
fork
(
F
fun
,
connection_handle
hdl
,
Ts
&&
...
args
)
{
return
this
->
fork_impl
(
std
::
bind
(
std
::
move
(
fun
),
auto
f
=
std
::
bind
(
std
::
move
(
fun
),
std
::
placeholders
::
_1
,
std
::
placeholders
::
_1
,
hdl
,
hdl
,
std
::
forward
<
Ts
>
(
args
)...),
std
::
forward
<
Ts
>
(
args
)...);
hdl
);
// transform to STD function here, because GCC is unable
// to select proper overload otherwise ...
typedef
decltype
(
f
((
broker
*
)
nullptr
))
fres
;
std
::
function
<
fres
(
broker
*
)
>
stdfun
{
std
::
move
(
f
)};
return
this
->
fork_impl
(
std
::
move
(
stdfun
),
hdl
);
}
}
inline
size_t
num_connections
()
const
{
inline
size_t
num_connections
()
const
{
...
@@ -156,43 +160,24 @@ class broker : public extend<local_actor>::
...
@@ -156,43 +160,24 @@ class broker : public extend<local_actor>::
bool
initialized
()
const
;
bool
initialized
()
const
;
template
<
typename
F
,
typename
...
Ts
>
template
<
typename
F
>
static
broker_ptr
from
(
F
fun
,
static
broker_ptr
from
(
F
fun
)
{
input_stream_ptr
in
,
// transform to STD function here, because GCC is unable
output_stream_ptr
out
,
// to select proper overload otherwise ...
Ts
&&
...
args
)
{
typedef
decltype
(
fun
((
broker
*
)
nullptr
))
fres
;
auto
hdl
=
connection_handle
::
from_int
(
in
->
read_handle
());
std
::
function
<
fres
(
broker
*
)
>
stdfun
{
std
::
move
(
fun
)};
return
from_impl
(
std
::
bind
(
std
::
move
(
fun
),
return
from_impl
(
std
::
move
(
stdfun
));
std
::
placeholders
::
_1
,
hdl
,
std
::
forward
<
Ts
>
(
args
)...),
std
::
move
(
in
),
std
::
move
(
out
));
}
}
static
broker_ptr
from
(
std
::
function
<
void
(
broker
*
)
>
fun
);
template
<
typename
F
,
typename
T
,
typename
...
Ts
>
static
broker_ptr
from
(
F
fun
,
T
&&
v
,
Ts
&&
...
vs
)
{
static
broker_ptr
from
(
std
::
function
<
behavior
(
broker
*
)
>
fun
);
return
from
(
std
::
bind
(
fun
,
std
::
placeholders
::
_1
,
std
::
forward
<
T
>
(
v
),
static
broker_ptr
from
(
std
::
function
<
void
(
broker
*
)
>
fun
,
acceptor_uptr
in
);
std
::
forward
<
Ts
>
(
vs
)...));
static
broker_ptr
from
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
acceptor_uptr
in
);
template
<
typename
F
,
typename
T0
,
typename
...
Ts
>
static
broker_ptr
from
(
F
fun
,
acceptor_uptr
in
,
T0
&&
arg0
,
Ts
&&
...
args
)
{
return
from
(
std
::
bind
(
std
::
move
(
fun
),
std
::
placeholders
::
_1
,
std
::
forward
<
T0
>
(
arg0
),
std
::
forward
<
Ts
>
(
args
)...),
std
::
move
(
in
));
}
}
protected:
protected:
broker
(
input_stream_ptr
in
,
output_stream_ptr
out
);
broker
(
acceptor_uptr
in
);
broker
();
broker
();
void
cleanup
(
std
::
uint32_t
reason
)
override
;
void
cleanup
(
std
::
uint32_t
reason
)
override
;
...
@@ -201,8 +186,6 @@ class broker : public extend<local_actor>::
...
@@ -201,8 +186,6 @@ class broker : public extend<local_actor>::
typedef
std
::
unique_ptr
<
broker
::
doorman
>
doorman_pointer
;
typedef
std
::
unique_ptr
<
broker
::
doorman
>
doorman_pointer
;
explicit
broker
(
scribe_pointer
);
virtual
behavior
make_behavior
()
=
0
;
virtual
behavior
make_behavior
()
=
0
;
/** @endcond */
/** @endcond */
...
@@ -215,13 +198,9 @@ class broker : public extend<local_actor>::
...
@@ -215,13 +198,9 @@ class broker : public extend<local_actor>::
actor
fork_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
actor
fork_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
connection_handle
hdl
);
connection_handle
hdl
);
static
broker_ptr
from_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
,
static
broker_ptr
from_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
);
input_stream_ptr
in
,
output_stream_ptr
out
);
static
broker_ptr
from_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
static
broker_ptr
from_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
);
input_stream_ptr
in
,
output_stream_ptr
out
);
void
invoke_message
(
msg_hdr_cref
hdr
,
any_tuple
msg
);
void
invoke_message
(
msg_hdr_cref
hdr
,
any_tuple
msg
);
...
...
src/broker.cpp
View file @
2f6aebe2
...
@@ -68,15 +68,6 @@ class default_broker : public broker {
...
@@ -68,15 +68,6 @@ class default_broker : public broker {
default_broker
(
function_type
f
)
:
m_fun
(
std
::
move
(
f
))
{
}
default_broker
(
function_type
f
)
:
m_fun
(
std
::
move
(
f
))
{
}
default_broker
(
function_type
f
,
input_stream_ptr
in
,
output_stream_ptr
out
)
:
broker
(
std
::
move
(
in
),
std
::
move
(
out
)),
m_fun
(
std
::
move
(
f
))
{
}
default_broker
(
function_type
f
,
scribe_pointer
ptr
)
:
broker
(
std
::
move
(
ptr
)),
m_fun
(
std
::
move
(
f
))
{
}
default_broker
(
function_type
f
,
acceptor_uptr
ptr
)
:
broker
(
std
::
move
(
ptr
)),
m_fun
(
std
::
move
(
f
))
{
}
behavior
make_behavior
()
override
{
behavior
make_behavior
()
override
{
CPPA_PUSH_AID
(
id
());
CPPA_PUSH_AID
(
id
());
CPPA_LOG_TRACE
(
""
);
CPPA_LOG_TRACE
(
""
);
...
@@ -422,25 +413,6 @@ broker::broker() {
...
@@ -422,25 +413,6 @@ broker::broker() {
init_broker
();
init_broker
();
}
}
broker
::
broker
(
input_stream_ptr
in
,
output_stream_ptr
out
)
{
using
namespace
std
;
init_broker
();
add_connection
(
move
(
in
),
move
(
out
));
}
broker
::
broker
(
scribe_pointer
ptr
)
{
using
namespace
std
;
init_broker
();
auto
id
=
ptr
->
id
();
m_io
.
insert
(
make_pair
(
id
,
move
(
ptr
)));
}
broker
::
broker
(
acceptor_uptr
ptr
)
{
using
namespace
std
;
init_broker
();
add_acceptor
(
move
(
ptr
));
}
void
broker
::
cleanup
(
std
::
uint32_t
reason
)
{
void
broker
::
cleanup
(
std
::
uint32_t
reason
)
{
super
::
cleanup
(
reason
);
super
::
cleanup
(
reason
);
get_actor_registry
()
->
dec_running
();
get_actor_registry
()
->
dec_running
();
...
@@ -462,23 +434,7 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) {
...
@@ -462,23 +434,7 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) {
broker_ptr
init_and_launch
(
broker_ptr
ptr
)
{
broker_ptr
init_and_launch
(
broker_ptr
ptr
)
{
CPPA_PUSH_AID
(
ptr
->
id
());
CPPA_PUSH_AID
(
ptr
->
id
());
CPPA_LOGF_TRACE
(
"init and launch actor with id "
<<
ptr
->
id
());
CPPA_LOGF_TRACE
(
"init and launch broker with id "
<<
ptr
->
id
());
// continue reader only if not inherited from default_broker_impl
auto
mm
=
get_middleman
();
mm
->
run_later
([
=
]
{
CPPA_LOGC_TRACE
(
"NONE"
,
"init_and_launch::run_later_functor"
,
""
);
CPPA_LOGF_WARNING_IF
(
ptr
->
m_io
.
empty
()
&&
ptr
->
m_accept
.
empty
(),
"both m_io and m_accept are empty"
);
// 'launch' all backends
CPPA_LOGC_DEBUG
(
"NONE"
,
"init_and_launch::run_later_functor"
,
"add "
<<
ptr
->
m_io
.
size
()
<<
" IO servants"
);
for
(
auto
&
kvp
:
ptr
->
m_io
)
mm
->
continue_reader
(
kvp
.
second
.
get
());
CPPA_LOGC_DEBUG
(
"NONE"
,
"init_and_launch::run_later_functor"
,
"add "
<<
ptr
->
m_accept
.
size
()
<<
" acceptors"
);
for
(
auto
&
kvp
:
ptr
->
m_accept
)
mm
->
continue_reader
(
kvp
.
second
.
get
());
});
// exec initialization code
// exec initialization code
auto
bhvr
=
ptr
->
make_behavior
();
auto
bhvr
=
ptr
->
make_behavior
();
if
(
bhvr
)
ptr
->
become
(
std
::
move
(
bhvr
));
if
(
bhvr
)
ptr
->
become
(
std
::
move
(
bhvr
));
...
@@ -486,41 +442,17 @@ broker_ptr init_and_launch(broker_ptr ptr) {
...
@@ -486,41 +442,17 @@ broker_ptr init_and_launch(broker_ptr ptr) {
return
ptr
;
return
ptr
;
}
}
broker_ptr
broker
::
from_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
broker_ptr
broker
::
from_impl
(
std
::
function
<
behavior
(
broker
*
)
>
fun
)
{
input_stream_ptr
in
,
output_stream_ptr
out
)
{
return
make_counted
<
default_broker
>
(
move
(
fun
),
move
(
in
),
move
(
out
));
}
broker_ptr
broker
::
from_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
,
input_stream_ptr
in
,
output_stream_ptr
out
)
{
auto
f
=
[
=
](
broker
*
ptr
)
->
behavior
{
fun
(
ptr
);
return
behavior
{};
};
return
make_counted
<
default_broker
>
(
f
,
move
(
in
),
move
(
out
));
}
broker_ptr
broker
::
from
(
std
::
function
<
behavior
(
broker
*
)
>
fun
)
{
return
make_counted
<
default_broker
>
(
fun
);
return
make_counted
<
default_broker
>
(
fun
);
}
}
broker_ptr
broker
::
from
(
std
::
function
<
void
(
broker
*
)
>
fun
)
{
broker_ptr
broker
::
from
_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
)
{
return
from
([
=
](
broker
*
self
)
->
behavior
{
return
from
([
=
](
broker
*
self
)
->
behavior
{
fun
(
self
);
fun
(
self
);
return
{};
return
{};
});
});
}
}
broker_ptr
broker
::
from
(
std
::
function
<
behavior
(
broker
*
)
>
fun
,
acceptor_uptr
in
)
{
return
make_counted
<
default_broker
>
(
move
(
fun
),
move
(
in
));
}
broker_ptr
broker
::
from
(
std
::
function
<
void
(
broker
*
)
>
fun
,
acceptor_uptr
in
)
{
auto
f
=
[
=
](
broker
*
ptr
)
->
behavior
{
fun
(
ptr
);
return
behavior
{};
};
return
make_counted
<
default_broker
>
(
f
,
move
(
in
));
}
void
broker
::
erase_io
(
int
id
)
{
void
broker
::
erase_io
(
int
id
)
{
m_io
.
erase
(
connection_handle
::
from_int
(
id
));
m_io
.
erase
(
connection_handle
::
from_int
(
id
));
}
}
...
@@ -532,14 +464,28 @@ void broker::erase_acceptor(int id) {
...
@@ -532,14 +464,28 @@ void broker::erase_acceptor(int id) {
connection_handle
broker
::
add_connection
(
input_stream_ptr
in
,
output_stream_ptr
out
)
{
connection_handle
broker
::
add_connection
(
input_stream_ptr
in
,
output_stream_ptr
out
)
{
using
namespace
std
;
using
namespace
std
;
auto
id
=
connection_handle
::
from_int
(
in
->
read_handle
());
auto
id
=
connection_handle
::
from_int
(
in
->
read_handle
());
m_io
.
insert
(
make_pair
(
id
,
create_unique
<
scribe
>
(
this
,
move
(
in
),
move
(
out
))));
auto
ires
=
m_io
.
insert
(
make_pair
(
id
,
create_unique
<
scribe
>
(
this
,
move
(
in
),
move
(
out
))));
CPPA_REQUIRE
(
ires
.
second
==
true
);
auto
sptr
=
ires
.
first
->
second
.
get
();
auto
mm
=
get_middleman
();
mm
->
run_later
([
=
]
{
// 'launch' backends
mm
->
continue_reader
(
sptr
);
});
return
id
;
return
id
;
}
}
accept_handle
broker
::
add_acceptor
(
acceptor_uptr
ptr
)
{
accept_handle
broker
::
add_acceptor
(
acceptor_uptr
ptr
)
{
using
namespace
std
;
using
namespace
std
;
auto
id
=
accept_handle
::
from_int
(
ptr
->
file_handle
());
auto
id
=
accept_handle
::
from_int
(
ptr
->
file_handle
());
m_accept
.
insert
(
make_pair
(
id
,
create_unique
<
doorman
>
(
this
,
move
(
ptr
))));
auto
ires
=
m_accept
.
insert
(
make_pair
(
id
,
create_unique
<
doorman
>
(
this
,
move
(
ptr
))));
CPPA_REQUIRE
(
ires
.
second
==
true
);
auto
aptr
=
ires
.
first
->
second
.
get
();
auto
mm
=
get_middleman
();
mm
->
run_later
([
=
]
{
// 'launch' backends
mm
->
continue_reader
(
aptr
);
});
return
id
;
return
id
;
}
}
...
@@ -551,8 +497,9 @@ actor broker::fork_impl(std::function<behavior (broker*)> fun,
...
@@ -551,8 +497,9 @@ actor broker::fork_impl(std::function<behavior (broker*)> fun,
CPPA_LOG_ERROR
(
"invalid handle"
);
CPPA_LOG_ERROR
(
"invalid handle"
);
throw
std
::
invalid_argument
(
"invalid handle"
);
throw
std
::
invalid_argument
(
"invalid handle"
);
}
}
scribe
*
sptr
=
i
->
second
.
get
();
// non-owning pointer
scribe
*
sptr
=
i
->
second
.
get
();
// keep non-owning pointer
auto
result
=
make_counted
<
default_broker
>
(
fun
,
move
(
i
->
second
));
auto
result
=
make_counted
<
default_broker
>
(
fun
);
result
->
m_io
.
insert
(
make_pair
(
sptr
->
id
(),
move
(
i
->
second
)));
init_and_launch
(
result
);
init_and_launch
(
result
);
sptr
->
set_broker
(
result
);
// set new broker
sptr
->
set_broker
(
result
);
// set new broker
m_io
.
erase
(
i
);
m_io
.
erase
(
i
);
...
@@ -561,10 +508,10 @@ actor broker::fork_impl(std::function<behavior (broker*)> fun,
...
@@ -561,10 +508,10 @@ actor broker::fork_impl(std::function<behavior (broker*)> fun,
actor
broker
::
fork_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
,
actor
broker
::
fork_impl
(
std
::
function
<
void
(
broker
*
)
>
fun
,
connection_handle
hdl
)
{
connection_handle
hdl
)
{
return
fork_impl
([
=
](
broker
*
self
)
->
behavior
{
return
fork_impl
(
std
::
function
<
behavior
(
broker
*
)
>
{
[
=
](
broker
*
self
)
->
behavior
{
fun
(
self
);
fun
(
self
);
return
behavior
{};
return
behavior
{};
},
hdl
);
}
}
,
hdl
);
}
}
void
broker
::
receive_policy
(
const
connection_handle
&
hdl
,
void
broker
::
receive_policy
(
const
connection_handle
&
hdl
,
...
...
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