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
ca2c0a57
Commit
ca2c0a57
authored
Nov 25, 2015
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'master' into develop
parents
a6585202
02bc07c8
Changes
6
Show whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
34 additions
and
119 deletions
+34
-119
libcaf_core/caf/abstract_group.hpp
libcaf_core/caf/abstract_group.hpp
+9
-63
libcaf_core/caf/config.hpp
libcaf_core/caf/config.hpp
+1
-1
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 @
ca2c0a57
...
...
@@ -23,79 +23,22 @@
#include <string>
#include <memory>
#include "caf/fwd.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/ref_counted.hpp"
#include "caf/abstract_channel.hpp"
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.
class
abstract_group
:
public
abstract_channel
{
public:
friend
class
detail
::
group_manager
;
friend
class
detail
::
peer_connection
;
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
detail
::
group_manager
;
// unsubscribes its channel from the group on destruction
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_
;
};
~
abstract_group
();
/// Interface for user-defined multicast implementations.
class
module
{
...
...
@@ -135,16 +78,19 @@ public:
/// Returns the name of the module.
const
std
::
string
&
module_name
()
const
;
/// Subscribes `who` to this group and returns a subscription object.
virtual
attachable_ptr
subscribe
(
const
actor_addr
&
who
)
=
0
;
/// Subscribes `who` to this group and returns `true` on success
/// or `false` if `who` is already subscribed.
virtual
bool
subscribe
(
const
actor_addr
&
who
)
=
0
;
/// Stops any background actors or threads and IO handles.
virtual
void
stop
()
=
0
;
protected:
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
;
module_ptr
module_
;
std
::
string
identifier_
;
};
...
...
libcaf_core/caf/config.hpp
View file @
ca2c0a57
...
...
@@ -35,7 +35,7 @@
/// Denotes version of CAF in the format {MAJOR}{MINOR}{PATCH},
/// whereas each number is a two-digit decimal number without
/// leading zeros (e.g. 900 is version 0.9.0).
#define CAF_VERSION 140
3
#define CAF_VERSION 140
4
/// Defined to the major version number of CAF.
#define CAF_MAJOR_VERSION (CAF_VERSION / 10000)
...
...
libcaf_core/caf/local_actor.hpp
View file @
ca2c0a57
...
...
@@ -667,6 +667,9 @@ protected:
// used by functor-based actors to implemented make_behavior() or act()
std
::
function
<
behavior
(
local_actor
*
)
>
initial_behavior_fac_
;
// used for group management
std
::
set
<
group
>
subscriptions_
;
/// @endcond
private:
...
...
libcaf_core/src/abstract_group.cpp
View file @
ca2c0a57
...
...
@@ -27,26 +27,6 @@
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
))
{
// nop
}
...
...
libcaf_core/src/group_manager.cpp
View file @
ca2c0a57
...
...
@@ -108,12 +108,11 @@ public:
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
if
(
add_subscriber
(
who
).
first
)
{
return
subscription
::
make
(
this
);
}
return
{};
if
(
add_subscriber
(
who
).
first
)
return
true
;
return
false
;
}
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
...
...
@@ -254,18 +253,17 @@ public:
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
));
auto
res
=
add_subscriber
(
who
);
if
(
res
.
first
)
{
if
(
res
.
second
==
1
)
{
// join the remote source
// join remote source
if
(
res
.
second
==
1
)
anon_send
(
broker_
,
join_atom
::
value
,
proxy_broker_
);
}
return
subscription
::
make
(
this
);
return
true
;
}
CAF_LOG_WARNING
(
"actor already joined group"
);
return
{}
;
return
false
;
}
void
unsubscribe
(
const
actor_addr
&
who
)
override
{
...
...
libcaf_core/src/local_actor.cpp
View file @
ca2c0a57
...
...
@@ -67,27 +67,16 @@ void local_actor::demonitor(const actor_addr& whom) {
void
local_actor
::
join
(
const
group
&
what
)
{
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
if
(
what
==
invalid_group
)
{
if
(
what
==
invalid_group
)
return
;
}
abstract_group
::
subscription_token
tk
{
what
.
ptr
()};
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
);
}
}
if
(
what
->
subscribe
(
address
()))
subscriptions_
.
emplace
(
what
);
}
void
local_actor
::
leave
(
const
group
&
what
)
{
CAF_LOG_TRACE
(
CAF_TSARG
(
what
));
if
(
what
==
invalid_group
)
{
return
;
}
if
(
detach
(
abstract_group
::
subscription_token
{
what
.
ptr
()})
>
0
)
{
if
(
subscriptions_
.
erase
(
what
)
>
0
)
what
->
unsubscribe
(
address
());
}
}
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
>
result
;
result
.
reserve
(
20
);
attachable
::
token
stk
{
attachable
::
token
::
subscription
,
nullptr
};
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
());
}
}
for
(
auto
&
x
:
subscriptions_
)
result
.
emplace_back
(
x
);
return
result
;
}
...
...
@@ -984,6 +966,12 @@ void local_actor::cleanup(uint32_t reason) {
detail
::
sync_request_bouncer
f
{
reason
};
mailbox_
.
close
(
f
);
pending_responses_
.
clear
();
{
// lifetime scope of temporary
actor_addr
me
=
address
();
for
(
auto
&
subscription
:
subscriptions_
)
subscription
->
unsubscribe
(
me
);
subscriptions_
.
clear
();
}
abstract_actor
::
cleanup
(
reason
);
// tell registry we're done
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