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
1affa212
Commit
1affa212
authored
Nov 25, 2015
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'hotfix/join-complexity'
parents
04c1e59b
1c0e6196
Changes
5
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
33 additions
and
118 deletions
+33
-118
libcaf_core/caf/abstract_group.hpp
libcaf_core/caf/abstract_group.hpp
+9
-63
libcaf_core/caf/local_actor.hpp
libcaf_core/caf/local_actor.hpp
+3
-0
libcaf_core/src/abstract_group.cpp
libcaf_core/src/abstract_group.cpp
+0
-20
libcaf_core/src/group_manager.cpp
libcaf_core/src/group_manager.cpp
+9
-11
libcaf_core/src/local_actor.cpp
libcaf_core/src/local_actor.cpp
+12
-24
No files found.
libcaf_core/caf/abstract_group.hpp
View file @
1affa212
...
@@ -23,79 +23,22 @@
...
@@ -23,79 +23,22 @@
#include <string>
#include <string>
#include <memory>
#include <memory>
#include "caf/fwd.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/attachable.hpp"
#include "caf/ref_counted.hpp"
#include "caf/ref_counted.hpp"
#include "caf/abstract_channel.hpp"
#include "caf/abstract_channel.hpp"
namespace
caf
{
namespace
caf
{
namespace
detail
{
class
group_manager
;
class
peer_connection
;
}
// namespace detail
}
// namespace caf
namespace
caf
{
class
group
;
class
serializer
;
class
local_actor
;
class
deserializer
;
/// A multicast group.
/// A multicast group.
class
abstract_group
:
public
abstract_channel
{
class
abstract_group
:
public
abstract_channel
{
public:
public:
friend
class
detail
::
group_manager
;
friend
class
detail
::
peer_connection
;
friend
class
local_actor
;
friend
class
local_actor
;
~
abstract_group
();
class
subscription
;
struct
subscription_token
{
intrusive_ptr
<
abstract_group
>
group
;
static
constexpr
size_t
token_type
=
attachable
::
token
::
subscription
;
};
class
subscription_predicate
{
public:
inline
subscription_predicate
(
intrusive_ptr
<
abstract_group
>
group
)
:
group_
(
std
::
move
(
group
))
{
// nop
}
inline
bool
operator
()(
const
attachable_ptr
&
ptr
)
{
return
ptr
->
matches
(
subscription_token
{
group_
});
}
private:
intrusive_ptr
<
abstract_group
>
group_
;
};
// needs access to unsubscribe()
friend
class
subscription
;
friend
class
subscription
;
friend
class
detail
::
group_manager
;
// unsubscribes its channel from the group on destruction
~
abstract_group
();
class
subscription
:
public
attachable
{
public:
subscription
(
const
intrusive_ptr
<
abstract_group
>&
g
);
void
actor_exited
(
abstract_actor
*
self
,
uint32_t
reason
)
override
;
bool
matches
(
const
token
&
what
)
override
;
static
inline
attachable_ptr
make
(
intrusive_ptr
<
abstract_group
>
ptr
)
{
return
attachable_ptr
{
new
subscription
(
std
::
move
(
ptr
))};
}
const
intrusive_ptr
<
abstract_group
>&
group
()
const
{
return
group_
;
}
private:
intrusive_ptr
<
abstract_group
>
group_
;
};
/// Interface for user-defined multicast implementations.
/// Interface for user-defined multicast implementations.
class
module
{
class
module
{
...
@@ -135,16 +78,19 @@ public:
...
@@ -135,16 +78,19 @@ public:
/// Returns the name of the module.
/// Returns the name of the module.
const
std
::
string
&
module_name
()
const
;
const
std
::
string
&
module_name
()
const
;
/// Subscribes `who` to this group and returns a subscription object.
/// Subscribes `who` to this group and returns `true` on success
virtual
attachable_ptr
subscribe
(
const
actor_addr
&
who
)
=
0
;
/// or `false` if `who` is already subscribed.
virtual
bool
subscribe
(
const
actor_addr
&
who
)
=
0
;
/// Stops any background actors or threads and IO handles.
/// Stops any background actors or threads and IO handles.
virtual
void
stop
()
=
0
;
virtual
void
stop
()
=
0
;
protected:
protected:
abstract_group
(
module_ptr
module
,
std
::
string
group_id
,
const
node_id
&
nid
);
abstract_group
(
module_ptr
module
,
std
::
string
group_id
,
const
node_id
&
nid
);
// called by subscription objects
// called by local_actor
virtual
void
unsubscribe
(
const
actor_addr
&
who
)
=
0
;
virtual
void
unsubscribe
(
const
actor_addr
&
who
)
=
0
;
module_ptr
module_
;
module_ptr
module_
;
std
::
string
identifier_
;
std
::
string
identifier_
;
};
};
...
...
libcaf_core/caf/local_actor.hpp
View file @
1affa212
...
@@ -667,6 +667,9 @@ protected:
...
@@ -667,6 +667,9 @@ protected:
// used by functor-based actors to implemented make_behavior() or act()
// used by functor-based actors to implemented make_behavior() or act()
std
::
function
<
behavior
(
local_actor
*
)
>
initial_behavior_fac_
;
std
::
function
<
behavior
(
local_actor
*
)
>
initial_behavior_fac_
;
// used for group management
std
::
set
<
group
>
subscriptions_
;
/// @endcond
/// @endcond
private:
private:
...
...
libcaf_core/src/abstract_group.cpp
View file @
1affa212
...
@@ -27,26 +27,6 @@
...
@@ -27,26 +27,6 @@
namespace
caf
{
namespace
caf
{
abstract_group
::
subscription
::
subscription
(
const
abstract_group_ptr
&
g
)
:
group_
(
g
)
{
// nop
}
void
abstract_group
::
subscription
::
actor_exited
(
abstract_actor
*
ptr
,
uint32_t
)
{
group_
->
unsubscribe
(
ptr
->
address
());
}
bool
abstract_group
::
subscription
::
matches
(
const
token
&
what
)
{
if
(
what
.
subtype
!=
attachable
::
token
::
subscription
)
{
return
false
;
}
if
(
what
.
ptr
)
{
auto
&
ot
=
*
reinterpret_cast
<
const
subscription_token
*>
(
what
.
ptr
);
return
ot
.
group
==
group_
;
}
return
true
;
}
abstract_group
::
module
::
module
(
std
::
string
mname
)
:
name_
(
std
::
move
(
mname
))
{
abstract_group
::
module
::
module
(
std
::
string
mname
)
:
name_
(
std
::
move
(
mname
))
{
// nop
// nop
}
}
...
...
libcaf_core/src/group_manager.cpp
View file @
1affa212
...
@@ -108,12 +108,11 @@ public:
...
@@ -108,12 +108,11 @@ public:
return
{
success
,
subscribers_
.
size
()};
return
{
success
,
subscribers_
.
size
()};
}
}
attachable_ptr
subscribe
(
const
actor_addr
&
who
)
override
{
bool
subscribe
(
const
actor_addr
&
who
)
override
{
CAF_LOG_TRACE
(
""
);
// serializing who would cause a deadlock
CAF_LOG_TRACE
(
""
);
// serializing who would cause a deadlock
if
(
add_subscriber
(
who
).
first
)
{
if
(
add_subscriber
(
who
).
first
)
return
subscription
::
make
(
this
);
return
true
;
}
return
false
;
return
{};
}
}
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
...
@@ -254,18 +253,17 @@ public:
...
@@ -254,18 +253,17 @@ public:
monitor_
=
spawn
(
broker_monitor_actor
,
this
);
monitor_
=
spawn
(
broker_monitor_actor
,
this
);
}
}
attachable_ptr
subscribe
(
const
actor_addr
&
who
)
override
{
bool
subscribe
(
const
actor_addr
&
who
)
override
{
CAF_LOG_TRACE
(
CAF_TSARG
(
who
));
CAF_LOG_TRACE
(
CAF_TSARG
(
who
));
auto
res
=
add_subscriber
(
who
);
auto
res
=
add_subscriber
(
who
);
if
(
res
.
first
)
{
if
(
res
.
first
)
{
if
(
res
.
second
==
1
)
{
// join remote source
// join the remote source
if
(
res
.
second
==
1
)
anon_send
(
broker_
,
join_atom
::
value
,
proxy_broker_
);
anon_send
(
broker_
,
join_atom
::
value
,
proxy_broker_
);
}
return
true
;
return
subscription
::
make
(
this
);
}
}
CAF_LOG_WARNING
(
"actor already joined group"
);
CAF_LOG_WARNING
(
"actor already joined group"
);
return
{}
;
return
false
;
}
}
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
...
...
libcaf_core/src/local_actor.cpp
View file @
1affa212
...
@@ -67,27 +67,16 @@ void local_actor::demonitor(const actor_addr& whom) {
...
@@ -67,27 +67,16 @@ void local_actor::demonitor(const actor_addr& whom) {
void
local_actor
::
join
(
const
group
&
what
)
{
void
local_actor
::
join
(
const
group
&
what
)
{
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
if
(
what
==
invalid_group
)
{
if
(
what
==
invalid_group
)
return
;
return
;
}
if
(
what
->
subscribe
(
address
()))
abstract_group
::
subscription_token
tk
{
what
.
ptr
()};
subscriptions_
.
emplace
(
what
);
std
::
unique_lock
<
std
::
mutex
>
guard
{
mtx_
};
if
(
detach_impl
(
tk
,
attachables_head_
,
true
,
true
)
==
0
)
{
auto
ptr
=
what
->
subscribe
(
address
());
if
(
ptr
)
{
attach_impl
(
ptr
);
}
}
}
}
void
local_actor
::
leave
(
const
group
&
what
)
{
void
local_actor
::
leave
(
const
group
&
what
)
{
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
if
(
what
==
invalid_group
)
{
if
(
subscriptions_
.
erase
(
what
)
>
0
)
return
;
}
if
(
detach
(
abstract_group
::
subscription_token
{
what
.
ptr
()})
>
0
)
{
what
->
unsubscribe
(
address
());
what
->
unsubscribe
(
address
());
}
}
}
void
local_actor
::
on_exit
()
{
void
local_actor
::
on_exit
()
{
...
@@ -96,15 +85,8 @@ void local_actor::on_exit() {
...
@@ -96,15 +85,8 @@ void local_actor::on_exit() {
std
::
vector
<
group
>
local_actor
::
joined_groups
()
const
{
std
::
vector
<
group
>
local_actor
::
joined_groups
()
const
{
std
::
vector
<
group
>
result
;
std
::
vector
<
group
>
result
;
result
.
reserve
(
20
);
for
(
auto
&
x
:
subscriptions_
)
attachable
::
token
stk
{
attachable
::
token
::
subscription
,
nullptr
};
result
.
emplace_back
(
x
);
std
::
unique_lock
<
std
::
mutex
>
guard
{
mtx_
};
for
(
attachable
*
i
=
attachables_head_
.
get
();
i
!=
0
;
i
=
i
->
next
.
get
())
{
if
(
i
->
matches
(
stk
))
{
auto
ptr
=
static_cast
<
abstract_group
::
subscription
*>
(
i
);
result
.
emplace_back
(
ptr
->
group
());
}
}
return
result
;
return
result
;
}
}
...
@@ -984,6 +966,12 @@ void local_actor::cleanup(uint32_t reason) {
...
@@ -984,6 +966,12 @@ void local_actor::cleanup(uint32_t reason) {
detail
::
sync_request_bouncer
f
{
reason
};
detail
::
sync_request_bouncer
f
{
reason
};
mailbox_
.
close
(
f
);
mailbox_
.
close
(
f
);
pending_responses_
.
clear
();
pending_responses_
.
clear
();
{
// lifetime scope of temporary
actor_addr
me
=
address
();
for
(
auto
&
subscription
:
subscriptions_
)
subscription
->
unsubscribe
(
me
);
subscriptions_
.
clear
();
}
abstract_actor
::
cleanup
(
reason
);
abstract_actor
::
cleanup
(
reason
);
// tell registry we're done
// tell registry we're done
is_registered
(
false
);
is_registered
(
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