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
114dfc65
Unverified
Commit
114dfc65
authored
Jun 12, 2019
by
Joseph Noir
Committed by
GitHub
Jun 12, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #849
Fix performance of thread-safe actor clock
parents
f4c7ebae
3fc3b630
Changes
9
Expand all
Show whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
592 additions
and
240 deletions
+592
-240
libcaf_core/caf/detail/make_unique.hpp
libcaf_core/caf/detail/make_unique.hpp
+32
-0
libcaf_core/caf/detail/ringbuffer.hpp
libcaf_core/caf/detail/ringbuffer.hpp
+38
-0
libcaf_core/caf/detail/simple_actor_clock.hpp
libcaf_core/caf/detail/simple_actor_clock.hpp
+239
-45
libcaf_core/caf/detail/test_actor_clock.hpp
libcaf_core/caf/detail/test_actor_clock.hpp
+0
-4
libcaf_core/caf/detail/thread_safe_actor_clock.hpp
libcaf_core/caf/detail/thread_safe_actor_clock.hpp
+19
-6
libcaf_core/src/simple_actor_clock.cpp
libcaf_core/src/simple_actor_clock.cpp
+129
-83
libcaf_core/src/test_actor_clock.cpp
libcaf_core/src/test_actor_clock.cpp
+6
-22
libcaf_core/src/thread_safe_actor_clock.cpp
libcaf_core/src/thread_safe_actor_clock.cpp
+95
-80
libcaf_core/test/ringbuffer.cpp
libcaf_core/test/ringbuffer.cpp
+34
-0
No files found.
libcaf_core/caf/detail/make_unique.hpp
0 → 100644
View file @
114dfc65
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <memory>
namespace
caf
{
namespace
detail
{
template
<
class
T
,
class
...
Ts
>
std
::
unique_ptr
<
T
>
make_unique
(
Ts
&&
...
xs
)
{
return
std
::
unique_ptr
<
T
>
{
new
T
(
std
::
forward
<
Ts
>
(
xs
)...)};
}
}
// namespace detail
}
// namespace caf
libcaf_core/caf/detail/ringbuffer.hpp
View file @
114dfc65
...
@@ -23,6 +23,8 @@
...
@@ -23,6 +23,8 @@
#include <condition_variable>
#include <condition_variable>
#include <mutex>
#include <mutex>
#include "caf/config.hpp"
namespace
caf
{
namespace
caf
{
namespace
detail
{
namespace
detail
{
...
@@ -46,6 +48,15 @@ public:
...
@@ -46,6 +48,15 @@ public:
cv_empty_
.
wait
(
guard
);
cv_empty_
.
wait
(
guard
);
}
}
template
<
class
TimePoint
>
bool
wait_nonempty
(
TimePoint
timeout
)
{
if
(
!
empty
())
return
true
;
auto
pred
=
[
&
]
{
return
!
empty
();
};
guard_type
guard
{
mtx_
};
return
cv_empty_
.
wait_until
(
guard
,
timeout
,
pred
);
}
T
&
front
()
{
T
&
front
()
{
// Safe to access without lock, because we assume a single consumer.
// Safe to access without lock, because we assume a single consumer.
return
buf_
[
rd_pos_
];
return
buf_
[
rd_pos_
];
...
@@ -60,6 +71,33 @@ public:
...
@@ -60,6 +71,33 @@ public:
cv_full_
.
notify_all
();
cv_full_
.
notify_all
();
}
}
template
<
class
OutputIterator
>
OutputIterator
get_all
(
OutputIterator
i
)
{
// No lock needed, again because of single-consumer assumption.
auto
first
=
rd_pos_
.
load
();
auto
last
=
wr_pos_
.
load
();
size_t
n
;
CAF_ASSERT
(
first
!=
last
);
// Move buffer content to the output iterator.
if
(
first
<
last
)
{
n
=
last
-
first
;
for
(
auto
j
=
first
;
j
!=
last
;
++
j
)
*
i
++
=
std
::
move
(
buf_
[
j
]);
}
else
{
n
=
(
Size
-
first
)
+
last
;
for
(
size_t
j
=
first
;
j
!=
Size
;
++
j
)
*
i
++
=
std
::
move
(
buf_
[
j
]);
for
(
size_t
j
=
0
;
j
!=
last
;
++
j
)
*
i
++
=
std
::
move
(
buf_
[
j
]);
}
guard_type
guard
{
mtx_
};
rd_pos_
=
(
first
+
n
)
%
Size
;
// Wakeup a waiting producers if the queue became non-full.
if
(
first
==
next
(
wr_pos_
))
cv_full_
.
notify_all
();
return
i
;
}
void
push_back
(
T
&&
x
)
{
void
push_back
(
T
&&
x
)
{
guard_type
guard
{
mtx_
};
guard_type
guard
{
mtx_
};
while
(
full
())
while
(
full
())
...
...
libcaf_core/caf/detail/simple_actor_clock.hpp
View file @
114dfc65
This diff is collapsed.
Click to expand it.
libcaf_core/caf/detail/test_actor_clock.hpp
View file @
114dfc65
...
@@ -51,10 +51,6 @@ public:
...
@@ -51,10 +51,6 @@ public:
/// @returns The number of triggered timeouts.
/// @returns The number of triggered timeouts.
size_t
trigger_timeouts
();
size_t
trigger_timeouts
();
/// Triggers all timeouts with timestamp <= now.
/// @returns The number of triggered timeouts.
size_t
trigger_expired_timeouts
();
/// Advances the time by `x` and dispatches timeouts and delayed messages.
/// Advances the time by `x` and dispatches timeouts and delayed messages.
/// @returns The number of triggered timeouts.
/// @returns The number of triggered timeouts.
size_t
advance_time
(
duration_type
x
);
size_t
advance_time
(
duration_type
x
);
...
...
libcaf_core/caf/detail/thread_safe_actor_clock.hpp
View file @
114dfc65
...
@@ -18,10 +18,14 @@
...
@@ -18,10 +18,14 @@
#pragma once
#pragma once
#include <
mutex
>
#include <
array
>
#include <atomic>
#include <atomic>
#include <condition_variable>
#include <condition_variable>
#include <memory>
#include <mutex>
#include "caf/abstract_actor.hpp"
#include "caf/detail/ringbuffer.hpp"
#include "caf/detail/simple_actor_clock.hpp"
#include "caf/detail/simple_actor_clock.hpp"
namespace
caf
{
namespace
caf
{
...
@@ -29,9 +33,15 @@ namespace detail {
...
@@ -29,9 +33,15 @@ namespace detail {
class
thread_safe_actor_clock
:
public
simple_actor_clock
{
class
thread_safe_actor_clock
:
public
simple_actor_clock
{
public:
public:
// -- constants --------------------------------------------------------------
static
constexpr
size_t
buffer_size
=
64
;
// -- member types -----------------------------------------------------------
using
super
=
simple_actor_clock
;
using
super
=
simple_actor_clock
;
thread_safe_actor_clock
();
// -- member functions -------------------------------------------------------
void
set_ordinary_timeout
(
time_point
t
,
abstract_actor
*
self
,
void
set_ordinary_timeout
(
time_point
t
,
abstract_actor
*
self
,
atom_value
type
,
uint64_t
id
)
override
;
atom_value
type
,
uint64_t
id
)
override
;
...
@@ -61,11 +71,14 @@ public:
...
@@ -61,11 +71,14 @@ public:
void
cancel_dispatch_loop
();
void
cancel_dispatch_loop
();
private:
private:
std
::
recursive_mutex
mx_
;
void
push
(
event
*
ptr
);
std
::
condition_variable_any
cv_
;
std
::
atomic
<
bool
>
done_
;
/// Receives timer events from other threads.
detail
::
ringbuffer
<
unique_event_ptr
,
buffer_size
>
queue_
;
/// Locally caches events for processing.
std
::
array
<
unique_event_ptr
,
buffer_size
>
events_
;
};
};
}
// namespace detail
}
// namespace detail
}
// namespace caf
}
// namespace caf
libcaf_core/src/simple_actor_clock.cpp
View file @
114dfc65
...
@@ -25,108 +25,40 @@
...
@@ -25,108 +25,40 @@
namespace
caf
{
namespace
caf
{
namespace
detail
{
namespace
detail
{
bool
simple_actor_clock
::
ordinary_predicate
::
simple_actor_clock
::
event
::~
event
()
{
operator
()(
const
secondary_map
::
value_type
&
x
)
const
noexcept
{
// nop
auto
ptr
=
get_if
<
ordinary_timeout
>
(
&
x
.
second
->
second
);
return
ptr
!=
nullptr
?
ptr
->
type
==
type
:
false
;
}
}
bool
simple_actor_clock
::
multi_predicate
::
void
simple_actor_clock
::
set_ordinary_timeout
(
time_point
t
,
operator
()(
const
secondary_map
::
value_type
&
x
)
const
noexcept
{
abstract_actor
*
self
,
auto
ptr
=
get_if
<
multi_timeout
>
(
&
x
.
second
->
second
);
return
ptr
!=
nullptr
?
ptr
->
type
==
type
:
false
;
}
bool
simple_actor_clock
::
request_predicate
::
operator
()(
const
secondary_map
::
value_type
&
x
)
const
noexcept
{
auto
ptr
=
get_if
<
request_timeout
>
(
&
x
.
second
->
second
);
return
ptr
!=
nullptr
?
ptr
->
id
==
id
:
false
;
}
void
simple_actor_clock
::
visitor
::
operator
()(
ordinary_timeout
&
x
)
{
CAF_ASSERT
(
x
.
self
!=
nullptr
);
x
.
self
->
get
()
->
eq_impl
(
make_message_id
(),
x
.
self
,
nullptr
,
timeout_msg
{
x
.
type
,
x
.
id
});
ordinary_predicate
pred
{
x
.
type
};
thisptr
->
drop_lookup
(
x
.
self
->
get
(),
pred
);
}
void
simple_actor_clock
::
visitor
::
operator
()(
multi_timeout
&
x
)
{
CAF_ASSERT
(
x
.
self
!=
nullptr
);
x
.
self
->
get
()
->
eq_impl
(
make_message_id
(),
x
.
self
,
nullptr
,
timeout_msg
{
x
.
type
,
x
.
id
});
multi_predicate
pred
{
x
.
type
};
thisptr
->
drop_lookup
(
x
.
self
->
get
(),
pred
);
}
void
simple_actor_clock
::
visitor
::
operator
()(
request_timeout
&
x
)
{
CAF_ASSERT
(
x
.
self
!=
nullptr
);
x
.
self
->
get
()
->
eq_impl
(
x
.
id
,
x
.
self
,
nullptr
,
sec
::
request_timeout
);
request_predicate
pred
{
x
.
id
};
thisptr
->
drop_lookup
(
x
.
self
->
get
(),
pred
);
}
void
simple_actor_clock
::
visitor
::
operator
()(
actor_msg
&
x
)
{
x
.
receiver
->
enqueue
(
std
::
move
(
x
.
content
),
nullptr
);
}
void
simple_actor_clock
::
visitor
::
operator
()(
group_msg
&
x
)
{
x
.
target
->
eq_impl
(
make_message_id
(),
std
::
move
(
x
.
sender
),
nullptr
,
std
::
move
(
x
.
content
));
}
void
simple_actor_clock
::
set_ordinary_timeout
(
time_point
t
,
abstract_actor
*
self
,
atom_value
type
,
uint64_t
id
)
{
atom_value
type
,
uint64_t
id
)
{
ordinary_predicate
pred
{
type
};
new_schedule_entry
<
ordinary_timeout
>
(
t
,
self
->
ctrl
(),
type
,
id
);
auto
i
=
lookup
(
self
,
pred
);
auto
sptr
=
actor_cast
<
strong_actor_ptr
>
(
self
);
ordinary_timeout
tmp
{
std
::
move
(
sptr
),
type
,
id
};
if
(
i
!=
actor_lookup_
.
end
())
{
schedule_
.
erase
(
i
->
second
);
i
->
second
=
schedule_
.
emplace
(
t
,
std
::
move
(
tmp
));
}
else
{
auto
j
=
schedule_
.
emplace
(
t
,
std
::
move
(
tmp
));
actor_lookup_
.
emplace
(
self
,
j
);
}
}
}
void
simple_actor_clock
::
set_multi_timeout
(
time_point
t
,
abstract_actor
*
self
,
void
simple_actor_clock
::
set_multi_timeout
(
time_point
t
,
abstract_actor
*
self
,
atom_value
type
,
uint64_t
id
)
{
atom_value
type
,
uint64_t
id
)
{
auto
sptr
=
actor_cast
<
strong_actor_ptr
>
(
self
);
new_schedule_entry
<
multi_timeout
>
(
t
,
self
->
ctrl
(),
type
,
id
);
multi_timeout
tmp
{
std
::
move
(
sptr
),
type
,
id
};
auto
j
=
schedule_
.
emplace
(
t
,
std
::
move
(
tmp
));
actor_lookup_
.
emplace
(
self
,
j
);
}
}
void
simple_actor_clock
::
set_request_timeout
(
time_point
t
,
abstract_actor
*
self
,
void
simple_actor_clock
::
set_request_timeout
(
time_point
t
,
abstract_actor
*
self
,
message_id
id
)
{
message_id
id
)
{
request_predicate
pred
{
id
};
new_schedule_entry
<
request_timeout
>
(
t
,
self
->
ctrl
(),
id
);
auto
i
=
lookup
(
self
,
pred
);
auto
sptr
=
actor_cast
<
strong_actor_ptr
>
(
self
);
request_timeout
tmp
{
std
::
move
(
sptr
),
id
};
if
(
i
!=
actor_lookup_
.
end
())
{
schedule_
.
erase
(
i
->
second
);
i
->
second
=
schedule_
.
emplace
(
t
,
std
::
move
(
tmp
));
}
else
{
auto
j
=
schedule_
.
emplace
(
t
,
std
::
move
(
tmp
));
actor_lookup_
.
emplace
(
self
,
j
);
}
}
}
void
simple_actor_clock
::
cancel_ordinary_timeout
(
abstract_actor
*
self
,
void
simple_actor_clock
::
cancel_ordinary_timeout
(
abstract_actor
*
self
,
atom_value
type
)
{
atom_value
type
)
{
ordinary_
predicate
pred
{
type
};
ordinary_
timeout_cancellation
tmp
{
self
->
id
(),
type
};
cancel
(
self
,
pred
);
handle
(
tmp
);
}
}
void
simple_actor_clock
::
cancel_request_timeout
(
abstract_actor
*
self
,
void
simple_actor_clock
::
cancel_request_timeout
(
abstract_actor
*
self
,
message_id
id
)
{
message_id
id
)
{
request_
predicate
pred
{
id
};
request_
timeout_cancellation
tmp
{
self
->
id
(),
id
};
cancel
(
self
,
pred
);
handle
(
tmp
);
}
}
void
simple_actor_clock
::
cancel_timeouts
(
abstract_actor
*
self
)
{
void
simple_actor_clock
::
cancel_timeouts
(
abstract_actor
*
self
)
{
auto
range
=
actor_lookup_
.
equal_range
(
self
);
auto
range
=
actor_lookup_
.
equal_range
(
self
->
id
()
);
if
(
range
.
first
==
range
.
second
)
if
(
range
.
first
==
range
.
second
)
return
;
return
;
for
(
auto
i
=
range
.
first
;
i
!=
range
.
second
;
++
i
)
for
(
auto
i
=
range
.
first
;
i
!=
range
.
second
;
++
i
)
...
@@ -137,14 +69,14 @@ void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
...
@@ -137,14 +69,14 @@ void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
void
simple_actor_clock
::
schedule_message
(
time_point
t
,
void
simple_actor_clock
::
schedule_message
(
time_point
t
,
strong_actor_ptr
receiver
,
strong_actor_ptr
receiver
,
mailbox_element_ptr
content
)
{
mailbox_element_ptr
content
)
{
schedule_
.
emplace
(
t
,
actor_msg
{
std
::
move
(
receiver
),
std
::
move
(
content
)}
);
new_schedule_entry
<
actor_msg
>
(
t
,
std
::
move
(
receiver
),
std
::
move
(
content
)
);
}
}
void
simple_actor_clock
::
schedule_message
(
time_point
t
,
group
target
,
void
simple_actor_clock
::
schedule_message
(
time_point
t
,
group
target
,
strong_actor_ptr
sender
,
strong_actor_ptr
sender
,
message
content
)
{
message
content
)
{
schedule_
.
emplace
(
new_schedule_entry
<
group_msg
>
(
t
,
std
::
move
(
target
),
std
::
move
(
sender
),
t
,
group_msg
{
std
::
move
(
target
),
std
::
move
(
sender
),
std
::
move
(
content
)}
);
std
::
move
(
content
)
);
}
}
void
simple_actor_clock
::
cancel_all
()
{
void
simple_actor_clock
::
cancel_all
()
{
...
@@ -152,5 +84,119 @@ void simple_actor_clock::cancel_all() {
...
@@ -152,5 +84,119 @@ void simple_actor_clock::cancel_all() {
schedule_
.
clear
();
schedule_
.
clear
();
}
}
void
simple_actor_clock
::
ship
(
delayed_event
&
x
)
{
switch
(
x
.
subtype
)
{
case
ordinary_timeout_type
:
{
auto
&
dref
=
static_cast
<
ordinary_timeout
&>
(
x
);
auto
&
self
=
dref
.
self
;
self
->
get
()
->
eq_impl
(
make_message_id
(),
self
,
nullptr
,
timeout_msg
{
dref
.
type
,
dref
.
id
});
break
;
}
case
multi_timeout_type
:
{
auto
&
dref
=
static_cast
<
multi_timeout
&>
(
x
);
auto
&
self
=
dref
.
self
;
self
->
get
()
->
eq_impl
(
make_message_id
(),
self
,
nullptr
,
timeout_msg
{
dref
.
type
,
dref
.
id
});
break
;
}
case
request_timeout_type
:
{
auto
&
dref
=
static_cast
<
request_timeout
&>
(
x
);
auto
&
self
=
dref
.
self
;
self
->
get
()
->
eq_impl
(
dref
.
id
,
self
,
nullptr
,
sec
::
request_timeout
);
break
;
}
case
actor_msg_type
:
{
auto
&
dref
=
static_cast
<
actor_msg
&>
(
x
);
dref
.
receiver
->
enqueue
(
std
::
move
(
dref
.
content
),
nullptr
);
break
;
}
case
group_msg_type
:
{
auto
&
dref
=
static_cast
<
group_msg
&>
(
x
);
dref
.
target
->
eq_impl
(
make_message_id
(),
std
::
move
(
dref
.
sender
),
nullptr
,
std
::
move
(
dref
.
content
));
break
;
}
default:
break
;
}
}
void
simple_actor_clock
::
handle
(
const
ordinary_timeout_cancellation
&
x
)
{
auto
pred
=
[
&
](
const
actor_lookup_map
::
value_type
&
kvp
)
{
auto
&
y
=
*
kvp
.
second
->
second
;
return
y
.
subtype
==
ordinary_timeout_type
&&
x
.
type
==
static_cast
<
const
ordinary_timeout
&>
(
y
).
type
;
};
cancel
(
x
.
aid
,
pred
);
}
void
simple_actor_clock
::
handle
(
const
multi_timeout_cancellation
&
x
)
{
auto
pred
=
[
&
](
const
actor_lookup_map
::
value_type
&
kvp
)
{
auto
&
y
=
*
kvp
.
second
->
second
;
if
(
y
.
subtype
!=
multi_timeout_type
)
return
false
;
auto
&
dref
=
static_cast
<
const
multi_timeout
&>
(
y
);
return
x
.
type
==
dref
.
type
&&
x
.
id
==
dref
.
id
;
};
cancel
(
x
.
aid
,
pred
);
}
void
simple_actor_clock
::
handle
(
const
request_timeout_cancellation
&
x
)
{
auto
pred
=
[
&
](
const
actor_lookup_map
::
value_type
&
kvp
)
{
auto
&
y
=
*
kvp
.
second
->
second
;
return
y
.
subtype
==
request_timeout_type
&&
x
.
id
==
static_cast
<
const
request_timeout
&>
(
y
).
id
;
};
cancel
(
x
.
aid
,
pred
);
}
void
simple_actor_clock
::
handle
(
const
timeouts_cancellation
&
x
)
{
auto
range
=
actor_lookup_
.
equal_range
(
x
.
aid
);
if
(
range
.
first
==
range
.
second
)
return
;
for
(
auto
i
=
range
.
first
;
i
!=
range
.
second
;
++
i
)
schedule_
.
erase
(
i
->
second
);
actor_lookup_
.
erase
(
range
.
first
,
range
.
second
);
}
size_t
simple_actor_clock
::
trigger_expired_timeouts
()
{
size_t
result
=
0
;
auto
t
=
now
();
auto
i
=
schedule_
.
begin
();
auto
e
=
schedule_
.
end
();
while
(
i
!=
e
&&
i
->
first
<=
t
)
{
auto
ptr
=
std
::
move
(
i
->
second
);
auto
backlink
=
ptr
->
backlink
;
if
(
backlink
!=
actor_lookup_
.
end
())
actor_lookup_
.
erase
(
backlink
);
i
=
schedule_
.
erase
(
i
);
ship
(
*
ptr
);
++
result
;
}
return
result
;
}
void
simple_actor_clock
::
add_schedule_entry
(
time_point
t
,
std
::
unique_ptr
<
ordinary_timeout
>
x
)
{
auto
aid
=
x
->
self
->
id
();
auto
type
=
x
->
type
;
auto
pred
=
[
&
](
const
actor_lookup_map
::
value_type
&
kvp
)
{
auto
&
y
=
*
kvp
.
second
->
second
;
return
y
.
subtype
==
ordinary_timeout_type
&&
static_cast
<
const
ordinary_timeout
&>
(
y
).
type
==
type
;
};
auto
i
=
lookup
(
aid
,
pred
);
if
(
i
!=
actor_lookup_
.
end
())
{
schedule_
.
erase
(
i
->
second
);
i
->
second
=
schedule_
.
emplace
(
t
,
std
::
move
(
x
));
}
else
{
auto
j
=
schedule_
.
emplace
(
t
,
std
::
move
(
x
));
i
=
actor_lookup_
.
emplace
(
aid
,
j
);
}
i
->
second
->
second
->
backlink
=
i
;
}
}
// namespace detail
}
// namespace detail
}
// namespace caf
}
// namespace caf
libcaf_core/src/test_actor_clock.cpp
View file @
114dfc65
...
@@ -46,13 +46,16 @@ bool test_actor_clock::trigger_timeout() {
...
@@ -46,13 +46,16 @@ bool test_actor_clock::trigger_timeout() {
CAF_LOG_TRACE
(
CAF_ARG2
(
"schedule.size"
,
schedule_
.
size
()));
CAF_LOG_TRACE
(
CAF_ARG2
(
"schedule.size"
,
schedule_
.
size
()));
if
(
schedule_
.
empty
())
if
(
schedule_
.
empty
())
return
false
;
return
false
;
visitor
f
{
this
};
auto
i
=
schedule_
.
begin
();
auto
i
=
schedule_
.
begin
();
auto
tout
=
i
->
first
;
auto
tout
=
i
->
first
;
if
(
tout
>
current_time
)
if
(
tout
>
current_time
)
current_time
=
tout
;
current_time
=
tout
;
visit
(
f
,
i
->
second
);
auto
ptr
=
std
::
move
(
i
->
second
);
schedule_
.
erase
(
i
);
schedule_
.
erase
(
i
);
auto
backlink
=
ptr
->
backlink
;
if
(
backlink
!=
actor_lookup_
.
end
())
actor_lookup_
.
erase
(
backlink
);
ship
(
*
ptr
);
return
true
;
return
true
;
}
}
...
@@ -60,28 +63,9 @@ size_t test_actor_clock::trigger_timeouts() {
...
@@ -60,28 +63,9 @@ size_t test_actor_clock::trigger_timeouts() {
CAF_LOG_TRACE
(
CAF_ARG2
(
"schedule.size"
,
schedule_
.
size
()));
CAF_LOG_TRACE
(
CAF_ARG2
(
"schedule.size"
,
schedule_
.
size
()));
if
(
schedule_
.
empty
())
if
(
schedule_
.
empty
())
return
0u
;
return
0u
;
visitor
f
{
this
};
auto
result
=
schedule_
.
size
();
for
(
auto
&
kvp
:
schedule_
)
{
auto
tout
=
kvp
.
first
;
if
(
tout
>
current_time
)
current_time
=
tout
;
visit
(
f
,
kvp
.
second
);
}
schedule_
.
clear
();
return
result
;
}
size_t
test_actor_clock
::
trigger_expired_timeouts
()
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"schedule.size"
,
schedule_
.
size
()));
visitor
f
{
this
};
size_t
result
=
0
;
size_t
result
=
0
;
auto
i
=
schedule_
.
begin
();
while
(
trigger_timeout
())
while
(
i
!=
schedule_
.
end
()
&&
i
->
first
<=
current_time
)
{
++
result
;
++
result
;
visit
(
f
,
i
->
second
);
i
=
schedule_
.
erase
(
i
);
}
return
result
;
return
result
;
}
}
...
...
libcaf_core/src/thread_safe_actor_clock.cpp
View file @
114dfc65
...
@@ -18,133 +18,148 @@
...
@@ -18,133 +18,148 @@
#include "caf/detail/thread_safe_actor_clock.hpp"
#include "caf/detail/thread_safe_actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
namespace
caf
{
namespace
caf
{
namespace
detail
{
namespace
detail
{
namespace
{
using
guard_type
=
std
::
unique_lock
<
std
::
recursive_mutex
>
;
}
// namespace <anonymous>
thread_safe_actor_clock
::
thread_safe_actor_clock
()
:
done_
(
false
)
{
// nop
}
void
thread_safe_actor_clock
::
set_ordinary_timeout
(
time_point
t
,
void
thread_safe_actor_clock
::
set_ordinary_timeout
(
time_point
t
,
abstract_actor
*
self
,
abstract_actor
*
self
,
atom_value
type
,
atom_value
type
,
uint64_t
id
)
{
uint64_t
id
)
{
guard_type
guard
{
mx_
};
push
(
new
ordinary_timeout
(
t
,
self
->
ctrl
(),
type
,
id
));
if
(
!
done_
)
{
super
::
set_ordinary_timeout
(
t
,
self
,
type
,
id
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
set_request_timeout
(
time_point
t
,
void
thread_safe_actor_clock
::
set_request_timeout
(
time_point
t
,
abstract_actor
*
self
,
abstract_actor
*
self
,
message_id
id
)
{
message_id
id
)
{
guard_type
guard
{
mx_
};
push
(
new
request_timeout
(
t
,
self
->
ctrl
(),
id
));
if
(
!
done_
)
{
super
::
set_request_timeout
(
t
,
self
,
id
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
set_multi_timeout
(
time_point
t
,
abstract_actor
*
self
,
void
thread_safe_actor_clock
::
set_multi_timeout
(
time_point
t
,
abstract_actor
*
self
,
atom_value
type
,
uint64_t
id
)
{
atom_value
type
,
uint64_t
id
)
{
guard_type
guard
{
mx_
};
push
(
new
multi_timeout
(
t
,
self
->
ctrl
(),
type
,
id
));
if
(
!
done_
)
{
super
::
set_multi_timeout
(
t
,
self
,
type
,
id
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
cancel_ordinary_timeout
(
abstract_actor
*
self
,
void
thread_safe_actor_clock
::
cancel_ordinary_timeout
(
abstract_actor
*
self
,
atom_value
type
)
{
atom_value
type
)
{
guard_type
guard
{
mx_
};
push
(
new
ordinary_timeout_cancellation
(
self
->
id
(),
type
));
if
(
!
done_
)
{
super
::
cancel_ordinary_timeout
(
self
,
type
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
cancel_request_timeout
(
abstract_actor
*
self
,
void
thread_safe_actor_clock
::
cancel_request_timeout
(
abstract_actor
*
self
,
message_id
id
)
{
message_id
id
)
{
guard_type
guard
{
mx_
};
push
(
new
request_timeout_cancellation
(
self
->
id
(),
id
));
if
(
!
done_
)
{
super
::
cancel_request_timeout
(
self
,
id
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
cancel_timeouts
(
abstract_actor
*
self
)
{
void
thread_safe_actor_clock
::
cancel_timeouts
(
abstract_actor
*
self
)
{
guard_type
guard
{
mx_
};
push
(
new
timeouts_cancellation
(
self
->
id
()));
if
(
!
done_
)
{
super
::
cancel_timeouts
(
self
);
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
schedule_message
(
time_point
t
,
void
thread_safe_actor_clock
::
schedule_message
(
time_point
t
,
strong_actor_ptr
receiver
,
strong_actor_ptr
receiver
,
mailbox_element_ptr
content
)
{
mailbox_element_ptr
content
)
{
guard_type
guard
{
mx_
};
push
(
new
actor_msg
(
t
,
std
::
move
(
receiver
),
std
::
move
(
content
)));
if
(
!
done_
)
{
super
::
schedule_message
(
t
,
std
::
move
(
receiver
),
std
::
move
(
content
));
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
schedule_message
(
time_point
t
,
group
target
,
void
thread_safe_actor_clock
::
schedule_message
(
time_point
t
,
group
target
,
strong_actor_ptr
sender
,
strong_actor_ptr
sender
,
message
content
)
{
message
content
)
{
guard_type
guard
{
mx_
};
push
(
new
group_msg
(
t
,
std
::
move
(
target
),
std
::
move
(
sender
),
if
(
!
done_
)
{
std
::
move
(
content
)));
super
::
schedule_message
(
t
,
std
::
move
(
target
),
std
::
move
(
sender
),
std
::
move
(
content
));
cv_
.
notify_all
();
}
}
}
void
thread_safe_actor_clock
::
cancel_all
()
{
void
thread_safe_actor_clock
::
cancel_all
()
{
guard_type
guard
{
mx_
};
push
(
new
drop_all
);
super
::
cancel_all
();
cv_
.
notify_all
();
}
}
void
thread_safe_actor_clock
::
run_dispatch_loop
()
{
void
thread_safe_actor_clock
::
run_dispatch_loop
()
{
visitor
f
{
this
};
for
(;;)
{
guard_type
guard
{
mx_
};
// Wait until queue is non-empty.
while
(
done_
==
false
)
{
// Wait for non-empty schedule.
// Note: The thread calling run_dispatch_loop() is guaranteed not to lock
// the mutex recursively. Otherwise, cv_.wait() or cv_.wait_until()
// would be unsafe, because wait operations call unlock() only once.
if
(
schedule_
.
empty
())
{
if
(
schedule_
.
empty
())
{
cv_
.
wait
(
guard
);
queue_
.
wait_nonempty
(
);
}
else
{
}
else
{
auto
tout
=
schedule_
.
begin
()
->
first
;
auto
t
=
schedule_
.
begin
()
->
second
->
due
;
cv_
.
wait_until
(
guard
,
tout
);
if
(
!
queue_
.
wait_nonempty
(
t
))
{
// Handle timeout by shipping timed-out events and starting anew.
trigger_expired_timeouts
();
continue
;
}
}
// Double-check whether schedule is non-empty and execute it.
if
(
!
schedule_
.
empty
())
{
auto
t
=
now
();
auto
i
=
schedule_
.
begin
();
while
(
i
!=
schedule_
.
end
()
&&
i
->
first
<=
t
)
{
visit
(
f
,
i
->
second
);
i
=
schedule_
.
erase
(
i
);
}
}
// Push all elements from the queue to the events buffer.
auto
i
=
events_
.
begin
();
auto
e
=
queue_
.
get_all
(
i
);
for
(;
i
!=
e
;
++
i
)
{
auto
&
x
=
*
i
;
CAF_ASSERT
(
x
!=
nullptr
);
switch
(
x
->
subtype
)
{
case
ordinary_timeout_cancellation_type
:
{
handle
(
static_cast
<
ordinary_timeout_cancellation
&>
(
*
x
));
break
;
}
}
case
request_timeout_cancellation_type
:
{
handle
(
static_cast
<
request_timeout_cancellation
&>
(
*
x
));
break
;
}
}
case
timeouts_cancellation_type
:
{
handle
(
static_cast
<
timeouts_cancellation
&>
(
*
x
));
break
;
}
case
drop_all_type
:
{
schedule_
.
clear
();
actor_lookup_
.
clear
();
break
;
}
case
shutdown_type
:
{
schedule_
.
clear
();
schedule_
.
clear
();
actor_lookup_
.
clear
();
// Call it a day.
return
;
}
case
ordinary_timeout_type
:
{
auto
dptr
=
static_cast
<
ordinary_timeout
*>
(
x
.
release
());
add_schedule_entry
(
std
::
unique_ptr
<
ordinary_timeout
>
{
dptr
});
break
;
}
case
multi_timeout_type
:
{
auto
dptr
=
static_cast
<
multi_timeout
*>
(
x
.
release
());
add_schedule_entry
(
std
::
unique_ptr
<
multi_timeout
>
{
dptr
});
break
;
}
case
request_timeout_type
:
{
auto
dptr
=
static_cast
<
request_timeout
*>
(
x
.
release
());
add_schedule_entry
(
std
::
unique_ptr
<
request_timeout
>
{
dptr
});
break
;
}
case
actor_msg_type
:
{
auto
dptr
=
static_cast
<
actor_msg
*>
(
x
.
release
());
add_schedule_entry
(
std
::
unique_ptr
<
actor_msg
>
{
dptr
});
break
;
}
case
group_msg_type
:
{
auto
dptr
=
static_cast
<
group_msg
*>
(
x
.
release
());
add_schedule_entry
(
std
::
unique_ptr
<
group_msg
>
{
dptr
});
break
;
}
default:
{
CAF_LOG_ERROR
(
"unexpected event type"
);
break
;
}
}
x
.
reset
();
}
}
}
}
void
thread_safe_actor_clock
::
cancel_dispatch_loop
()
{
void
thread_safe_actor_clock
::
cancel_dispatch_loop
()
{
guard_type
guard
{
mx_
};
push
(
new
shutdown
);
done_
=
true
;
}
cv_
.
notify_all
();
void
thread_safe_actor_clock
::
push
(
event
*
ptr
)
{
queue_
.
push_back
(
unique_event_ptr
{
ptr
});
}
}
}
// namespace detail
}
// namespace detail
...
...
libcaf_core/test/ringbuffer.cpp
View file @
114dfc65
...
@@ -82,6 +82,40 @@ CAF_TEST(push_back) {
...
@@ -82,6 +82,40 @@ CAF_TEST(push_back) {
CAF_CHECK_EQUAL
(
buf
.
front
(),
0
);
CAF_CHECK_EQUAL
(
buf
.
front
(),
0
);
}
}
CAF_TEST
(
get
all
)
{
using
array_type
=
std
::
array
<
int
,
buf_size
>
;
using
vector_type
=
std
::
vector
<
int
>
;
array_type
tmp
;
auto
fetch_all
=
[
&
]
{
auto
i
=
tmp
.
begin
();
auto
e
=
buf
.
get_all
(
i
);
return
vector_type
(
i
,
e
);
};
CAF_MESSAGE
(
"add five element"
);
for
(
int
i
=
0
;
i
<
5
;
++
i
)
buf
.
push_back
(
std
::
move
(
i
));
CAF_CHECK_EQUAL
(
buf
.
empty
(),
false
);
CAF_CHECK_EQUAL
(
buf
.
full
(),
false
);
CAF_CHECK_EQUAL
(
buf
.
size
(),
5u
);
CAF_CHECK_EQUAL
(
buf
.
front
(),
0
);
CAF_MESSAGE
(
"drain elements"
);
CAF_CHECK_EQUAL
(
fetch_all
(),
vector_type
({
0
,
1
,
2
,
3
,
4
}));
CAF_CHECK_EQUAL
(
buf
.
empty
(),
true
);
CAF_CHECK_EQUAL
(
buf
.
full
(),
false
);
CAF_CHECK_EQUAL
(
buf
.
size
(),
0u
);
CAF_MESSAGE
(
"add 60 elements (wraps around)"
);
vector_type
expected
;
for
(
int
i
=
0
;
i
<
60
;
++
i
)
{
expected
.
push_back
(
i
);
buf
.
push_back
(
std
::
move
(
i
));
}
CAF_CHECK_EQUAL
(
buf
.
size
(),
60u
);
CAF_CHECK_EQUAL
(
fetch_all
(),
expected
);
CAF_CHECK_EQUAL
(
buf
.
empty
(),
true
);
CAF_CHECK_EQUAL
(
buf
.
full
(),
false
);
CAF_CHECK_EQUAL
(
buf
.
size
(),
0u
);
}
CAF_TEST
(
concurrent
access
)
{
CAF_TEST
(
concurrent
access
)
{
std
::
vector
<
std
::
thread
>
producers
;
std
::
vector
<
std
::
thread
>
producers
;
producers
.
emplace_back
(
producer
,
std
::
ref
(
buf
),
0
,
100
);
producers
.
emplace_back
(
producer
,
std
::
ref
(
buf
),
0
,
100
);
...
...
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