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
d1bb08aa
Commit
d1bb08aa
authored
Dec 06, 2017
by
Dominik Charousset
Committed by
Dominik Charousset
Feb 06, 2018
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add peek_all to all intrusive queues
parent
9092da76
Changes
12
Hide whitespace changes
Inline
Side-by-side
Showing
12 changed files
with
374 additions
and
97 deletions
+374
-97
libcaf_core/caf/detail/stringification_inspector.hpp
libcaf_core/caf/detail/stringification_inspector.hpp
+12
-0
libcaf_core/caf/detail/type_traits.hpp
libcaf_core/caf/detail/type_traits.hpp
+13
-0
libcaf_core/caf/intrusive/drr_cached_queue.hpp
libcaf_core/caf/intrusive/drr_cached_queue.hpp
+140
-37
libcaf_core/caf/intrusive/drr_queue.hpp
libcaf_core/caf/intrusive/drr_queue.hpp
+7
-0
libcaf_core/caf/intrusive/task_queue.hpp
libcaf_core/caf/intrusive/task_queue.hpp
+44
-33
libcaf_core/caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp
libcaf_core/caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp
+7
-0
libcaf_core/caf/intrusive/wdrr_fixed_multiplexed_queue.hpp
libcaf_core/caf/intrusive/wdrr_fixed_multiplexed_queue.hpp
+21
-0
libcaf_core/test/drr_cached_queue.cpp
libcaf_core/test/drr_cached_queue.cpp
+33
-5
libcaf_core/test/drr_queue.cpp
libcaf_core/test/drr_queue.cpp
+28
-0
libcaf_core/test/task_queue.cpp
libcaf_core/test/task_queue.cpp
+6
-22
libcaf_core/test/wdrr_dynamic_multiplexed_queue.cpp
libcaf_core/test/wdrr_dynamic_multiplexed_queue.cpp
+33
-0
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
+30
-0
No files found.
libcaf_core/caf/detail/stringification_inspector.hpp
View file @
d1bb08aa
...
@@ -145,6 +145,17 @@ public:
...
@@ -145,6 +145,17 @@ public:
result_
+=
']'
;
result_
+=
']'
;
}
}
template
<
class
T
>
enable_if_t
<
has_peek_all
<
T
>::
value
&&
!
is_iterable
<
T
>::
value
// pick begin()/end() over peek_all
&&
!
is_inspectable
<
stringification_inspector
,
T
>::
value
&&
!
has_to_string
<
T
>::
value
>
consume
(
T
&
xs
)
{
result_
+=
'['
;
xs
.
peek_all
(
*
this
);
result_
+=
']'
;
}
template
<
class
T
>
template
<
class
T
>
void
consume
(
T
*
xs
,
size_t
n
)
{
void
consume
(
T
*
xs
,
size_t
n
)
{
result_
+=
'('
;
result_
+=
'('
;
...
@@ -179,6 +190,7 @@ public:
...
@@ -179,6 +190,7 @@ public:
template
<
class
T
>
template
<
class
T
>
enable_if_t
<
enable_if_t
<
!
is_iterable
<
T
>::
value
!
is_iterable
<
T
>::
value
&&
!
has_peek_all
<
T
>::
value
&&
!
std
::
is_pointer
<
T
>::
value
&&
!
std
::
is_pointer
<
T
>::
value
&&
!
is_inspectable
<
stringification_inspector
,
T
>::
value
&&
!
is_inspectable
<
stringification_inspector
,
T
>::
value
&&
!
std
::
is_arithmetic
<
T
>::
value
&&
!
std
::
is_arithmetic
<
T
>::
value
...
...
libcaf_core/caf/detail/type_traits.hpp
View file @
d1bb08aa
...
@@ -545,6 +545,19 @@ public:
...
@@ -545,6 +545,19 @@ public:
static
constexpr
bool
value
=
false
;
static
constexpr
bool
value
=
false
;
};
};
template
<
class
T
>
class
has_peek_all
{
private:
template
<
class
U
>
static
int
fun
(
const
U
*
,
decltype
(
std
::
declval
<
U
&>
().
peek_all
(
unit
))
*
=
nullptr
);
static
char
fun
(
const
void
*
);
public:
static
constexpr
bool
value
=
sizeof
(
fun
(
static_cast
<
T
*>
(
nullptr
)))
>
1
;
};
CAF_HAS_MEMBER_TRAIT
(
size
);
CAF_HAS_MEMBER_TRAIT
(
size
);
CAF_HAS_MEMBER_TRAIT
(
data
);
CAF_HAS_MEMBER_TRAIT
(
data
);
...
...
libcaf_core/caf/intrusive/drr_cached_queue.hpp
View file @
d1bb08aa
...
@@ -22,6 +22,8 @@
...
@@ -22,6 +22,8 @@
#include <utility>
#include <utility>
#include "caf/config.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/intrusive/task_result.hpp"
...
@@ -31,77 +33,145 @@ namespace intrusive {
...
@@ -31,77 +33,145 @@ namespace intrusive {
/// A Deficit Round Robin queue with an internal cache for allowing skipping
/// A Deficit Round Robin queue with an internal cache for allowing skipping
/// consumers.
/// consumers.
template
<
class
Policy
>
template
<
class
Policy
>
class
drr_cached_queue
:
public
task_queue
<
Policy
>
{
class
drr_cached_queue
{
// Note that we do *not* inherit from
// task_queue<Policy>, because the cached queue can no
// longer offer iterator access.
public:
public:
// -- member types ----------------------------------------------------------
// -- member types ----------------------------------------------------------
using
policy_type
=
Policy
;
using
super
=
task_queue
<
Policy
>
;
using
deleter_type
=
typename
policy_type
::
deleter_type
;
using
typename
super
::
policy
_type
;
using
value_type
=
typename
policy_type
::
mapped
_type
;
using
typename
super
::
deleter
_type
;
using
node_type
=
typename
value_type
::
node
_type
;
using
typename
super
::
unique_pointer
;
using
node_pointer
=
node_type
*
;
using
typename
super
::
value_type
;
using
pointer
=
value_type
*
;
using
cache_type
=
task_queue
<
Policy
>
;
using
unique_pointer
=
typename
policy_type
::
unique_pointer
;
using
deficit_type
=
typename
policy_type
::
deficit_type
;
using
deficit_type
=
typename
policy_type
::
deficit_type
;
using
task_size_type
=
typename
policy_type
::
task_size_type
;
using
task_size_type
=
typename
policy_type
::
task_size_type
;
using
list_type
=
task_queue
<
policy_type
>
;
using
cache_type
=
task_queue
<
policy_type
>
;
// -- constructors, destructors, and assignment operators -------------------
// -- constructors, destructors, and assignment operators -------------------
drr_cached_queue
(
const
policy_type
&
p
)
:
super
(
p
),
deficit_
(
0
),
cache_
(
p
)
{
drr_cached_queue
(
policy_type
p
)
:
list_
(
p
),
deficit_
(
0
),
cache_
(
std
::
move
(
p
))
{
// nop
// nop
}
}
drr_cached_queue
(
drr_cached_queue
&&
other
)
drr_cached_queue
(
drr_cached_queue
&&
other
)
:
super
(
std
::
move
(
other
)),
:
list_
(
std
::
move
(
other
.
list_
)),
deficit_
(
0
)
{
deficit_
(
other
.
deficit_
),
cache_
(
std
::
move
(
other
.
cache_
))
{
// nop
// nop
}
}
drr_cached_queue
&
operator
=
(
drr_cached_queue
&&
other
)
{
drr_cached_queue
&
operator
=
(
drr_cached_queue
&&
other
)
{
super
::
operator
=
(
std
::
move
(
other
));
list_
=
std
::
move
(
other
.
list_
);
deficit_
=
other
.
deficit_
;
cache_
=
std
::
move
(
other
.
cache_
);
return
*
this
;
return
*
this
;
}
}
// -- observers -------------------------------------------------------------
// -- observers -------------------------------------------------------------
/// Returns the policy object.
policy_type
&
policy
()
noexcept
{
return
list_
.
policy
();
}
/// Returns the policy object.
const
policy_type
&
policy
()
const
noexcept
{
return
list_
.
policy
();
}
deficit_type
deficit
()
const
{
deficit_type
deficit
()
const
{
return
deficit_
;
return
deficit_
;
}
}
/// Returns the accumulated size of all stored tasks in the list, i.e., tasks
/// that are not in the cache.
task_size_type
total_task_size
()
const
{
return
list_
.
total_task_size
();
}
/// Returns whether the queue has no uncached tasks.
bool
empty
()
const
noexcept
{
return
total_task_size
()
==
0
;
}
/// Peeks at the first element of the cache or of the list in case the former
/// is empty.
pointer
peek
()
noexcept
{
auto
ptr
=
cache_
.
peek
();
return
ptr
==
nullptr
?
list_
.
peek
()
:
ptr
;
}
/// Applies `f` to each element in the queue, including cached elements.
template
<
class
F
>
void
peek_all
(
F
f
)
const
{
for
(
auto
i
=
cache_
.
begin
();
i
!=
cache_
.
end
();
++
i
)
f
(
*
promote
(
i
.
ptr
));
for
(
auto
i
=
list_
.
begin
();
i
!=
list_
.
end
();
++
i
)
f
(
*
promote
(
i
.
ptr
));
}
// -- modifiers -------------------------------------------------------------
// -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue.
void
clear
()
{
list_
.
clear
();
cache_
.
clear
();
}
void
inc_deficit
(
deficit_type
x
)
noexcept
{
void
inc_deficit
(
deficit_type
x
)
noexcept
{
if
(
!
super
::
empty
())
if
(
!
list_
.
empty
())
deficit_
+=
x
;
deficit_
+=
x
;
}
}
void
flush_cache
()
noexcept
{
void
flush_cache
()
noexcept
{
super
::
prepend
(
cache_
);
list_
.
prepend
(
cache_
);
}
/// @private
template
<
class
T
>
void
inc_total_task_size
(
T
&&
x
)
noexcept
{
list_
.
inc_total_task_size
(
std
::
forward
<
T
>
(
x
));
}
/// @private
template
<
class
T
>
void
dec_total_task_size
(
T
&&
x
)
noexcept
{
list_
.
dec_total_task_size
(
std
::
forward
<
T
>
(
x
));
}
}
/// Takes the first element out of the queue if the deficit allows it and
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
/// returns the element.
unique_pointer
take_front
()
noexcept
{
unique_pointer
take_front
()
noexcept
{
super
::
prepend
(
cache_
);
list_
.
prepend
(
cache_
);
unique_pointer
result
;
unique_pointer
result
;
if
(
!
super
::
empty
())
{
if
(
!
list_
.
empty
())
{
auto
ptr
=
promote
(
super
::
head_
.
next
);
auto
ptr
=
list_
.
front
(
);
auto
ts
=
super
::
policy_
.
task_size
(
*
ptr
);
auto
ts
=
policy
()
.
task_size
(
*
ptr
);
CAF_ASSERT
(
ts
>
0
);
CAF_ASSERT
(
ts
>
0
);
if
(
ts
<=
deficit_
)
{
if
(
ts
<=
deficit_
)
{
deficit_
-=
ts
;
deficit_
-=
ts
;
super
::
total_task_size_
-=
ts
;
dec_total_task_size
(
ts
)
;
super
::
head_
.
next
=
ptr
->
next
;
list_
.
before_begin
()
->
next
=
ptr
->
next
;
if
(
super
::
total_task_size_
==
0
)
{
if
(
total_task_size
()
==
0
)
{
CAF_ASSERT
(
super
::
head_
.
next
==
&
(
super
::
tail_
));
CAF_ASSERT
(
list_
.
begin
()
==
list_
.
end
(
));
deficit_
=
0
;
deficit_
=
0
;
super
::
tail_
.
next
=
&
(
super
::
head_
)
;
list_
.
end
()
->
next
=
list_
.
before_begin
().
ptr
;
}
}
result
.
reset
(
ptr
);
result
.
reset
(
ptr
);
}
}
...
@@ -116,19 +186,20 @@ public:
...
@@ -116,19 +186,20 @@ public:
bool
consume
(
F
&
f
)
noexcept
(
noexcept
(
f
(
std
::
declval
<
value_type
&>
())))
{
bool
consume
(
F
&
f
)
noexcept
(
noexcept
(
f
(
std
::
declval
<
value_type
&>
())))
{
long
consumed
=
0
;
long
consumed
=
0
;
if
(
!
cache_
.
empty
())
if
(
!
cache_
.
empty
())
super
::
prepend
(
cache_
);
list_
.
prepend
(
cache_
);
if
(
!
super
::
empty
())
{
if
(
!
list_
.
empty
())
{
auto
ptr
=
promote
(
super
::
head_
.
next
);
auto
ptr
=
list_
.
front
(
);
auto
ts
=
super
::
policy_
.
task_size
(
*
ptr
);
auto
ts
=
policy
()
.
task_size
(
*
ptr
);
while
(
ts
<=
deficit_
)
{
while
(
ts
<=
deficit_
)
{
CAF_ASSERT
(
ts
>
0
);
auto
next
=
ptr
->
next
;
auto
next
=
ptr
->
next
;
super
::
total_task_size_
-=
ts
;
dec_total_task_size
(
ts
)
;
// Make sure the queue is in a consistent state before calling `f` in
// Make sure the queue is in a consistent state before calling `f` in
// case `f` recursively calls consume.
// case `f` recursively calls consume.
super
::
head_
.
next
=
next
;
list_
.
before_begin
()
->
next
=
next
;
if
(
super
::
total_task_size_
==
0
)
{
if
(
total_task_size
()
==
0
)
{
CAF_ASSERT
(
super
::
head_
.
next
==
&
(
super
::
tail_
));
CAF_ASSERT
(
list_
.
begin
()
==
list_
.
end
(
));
super
::
tail_
.
next
=
&
(
super
::
head_
)
;
list_
.
end
()
->
next
=
list_
.
begin
().
ptr
;
}
}
// Always decrease the deficit_ counter, again because `f` is allowed
// Always decrease the deficit_ counter, again because `f` is allowed
// to call consume again.
// to call consume again.
...
@@ -138,7 +209,7 @@ public:
...
@@ -138,7 +209,7 @@ public:
// Fix deficit and push the unconsumed item to the cache.
// Fix deficit and push the unconsumed item to the cache.
deficit_
+=
ts
;
deficit_
+=
ts
;
cache_
.
push_back
(
ptr
);
cache_
.
push_back
(
ptr
);
if
(
super
::
empty
())
{
if
(
list_
.
empty
())
{
deficit_
=
0
;
deficit_
=
0
;
return
consumed
!=
0
;
return
consumed
!=
0
;
}
}
...
@@ -147,8 +218,8 @@ public:
...
@@ -147,8 +218,8 @@ public:
d
(
ptr
);
d
(
ptr
);
++
consumed
;
++
consumed
;
if
(
!
cache_
.
empty
())
if
(
!
cache_
.
empty
())
super
::
prepend
(
cache_
);
list_
.
prepend
(
cache_
);
if
(
super
::
empty
())
{
if
(
list_
.
empty
())
{
deficit_
=
0
;
deficit_
=
0
;
return
consumed
!=
0
;
return
consumed
!=
0
;
}
}
...
@@ -156,8 +227,8 @@ public:
...
@@ -156,8 +227,8 @@ public:
return
consumed
!=
0
;
return
consumed
!=
0
;
}
}
// Next iteration.
// Next iteration.
ptr
=
promote
(
super
::
head_
.
next
);
ptr
=
list_
.
front
(
);
ts
=
super
::
policy_
.
task_size
(
*
ptr
);
ts
=
policy
()
.
task_size
(
*
ptr
);
}
}
}
}
return
consumed
!=
0
;
return
consumed
!=
0
;
...
@@ -167,7 +238,7 @@ public:
...
@@ -167,7 +238,7 @@ public:
template
<
class
F
>
template
<
class
F
>
bool
new_round
(
deficit_type
quantum
,
F
&
consumer
)
bool
new_round
(
deficit_type
quantum
,
F
&
consumer
)
noexcept
(
noexcept
(
consumer
(
std
::
declval
<
value_type
&>
())))
{
noexcept
(
noexcept
(
consumer
(
std
::
declval
<
value_type
&>
())))
{
if
(
!
super
::
empty
())
{
if
(
!
list_
.
empty
())
{
deficit_
+=
quantum
;
deficit_
+=
quantum
;
return
consume
(
consumer
);
return
consume
(
consumer
);
}
}
...
@@ -178,8 +249,40 @@ public:
...
@@ -178,8 +249,40 @@ public:
return
cache_
;
return
cache_
;
}
}
// -- insertion --------------------------------------------------------------
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
pointer
ptr
)
noexcept
{
return
list_
.
push_back
(
ptr
);
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
unique_pointer
ptr
)
noexcept
{
return
push_back
(
ptr
.
release
());
}
/// Creates a new element from `xs...` and appends it.
template
<
class
...
Ts
>
bool
emplace_back
(
Ts
&&
...
xs
)
{
return
push_back
(
new
value_type
(
std
::
forward
<
Ts
>
(
xs
)...));
}
/// @private
void
lifo_append
(
node_pointer
ptr
)
{
list_
.
lifo_append
(
ptr
);
}
/// @private
void
stop_lifo_append
()
{
list_
.
stop_lifo_append
();
}
private:
private:
// -- member variables ------------------------------------------------------
// -- member variables ------------------------------------------------------
/// Stores current (unskipped) items.
list_type
list_
;
/// Stores the deficit on this queue.
/// Stores the deficit on this queue.
deficit_type
deficit_
=
0
;
deficit_type
deficit_
=
0
;
...
...
libcaf_core/caf/intrusive/drr_queue.hpp
View file @
d1bb08aa
...
@@ -64,6 +64,13 @@ public:
...
@@ -64,6 +64,13 @@ public:
return
deficit_
;
return
deficit_
;
}
}
/// Applies `f` to each element in the queue.
template
<
class
F
>
void
peek_all
(
F
f
)
const
{
for
(
auto
i
=
super
::
begin
();
i
!=
super
::
end
();
++
i
)
f
(
*
promote
(
i
.
ptr
));
}
// -- modifiers --------------------------------------------------------------
// -- modifiers --------------------------------------------------------------
void
inc_deficit
(
deficit_type
x
)
noexcept
{
void
inc_deficit
(
deficit_type
x
)
noexcept
{
...
...
libcaf_core/caf/intrusive/task_queue.hpp
View file @
d1bb08aa
...
@@ -22,6 +22,8 @@
...
@@ -22,6 +22,8 @@
#include <utility>
#include <utility>
#include "caf/config.hpp"
#include "caf/intrusive/forward_iterator.hpp"
#include "caf/intrusive/forward_iterator.hpp"
namespace
caf
{
namespace
caf
{
...
@@ -126,13 +128,6 @@ public:
...
@@ -126,13 +128,6 @@ public:
return
ptr
!=
&
tail_
?
promote
(
ptr
)
:
nullptr
;
return
ptr
!=
&
tail_
?
promote
(
ptr
)
:
nullptr
;
}
}
/// Applies `f` to each element in the queue.
template
<
class
F
>
void
peek_all
(
F
f
)
const
{
for
(
auto
i
=
begin
();
i
!=
end
();
++
i
)
f
(
*
promote
(
i
.
ptr
));
}
// -- modifiers -------------------------------------------------------------
// -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue.
/// Removes all elements from the queue.
...
@@ -141,31 +136,9 @@ public:
...
@@ -141,31 +136,9 @@ public:
init
();
init
();
}
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
pointer
ptr
)
noexcept
{
CAF_ASSERT
(
ptr
!=
nullptr
);
tail_
.
next
->
next
=
ptr
;
tail_
.
next
=
ptr
;
ptr
->
next
=
&
tail_
;
inc_total_task_size
(
*
ptr
);
return
true
;
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
unique_pointer
ptr
)
noexcept
{
return
push_back
(
ptr
.
release
());
}
/// Creates a new element from `xs...` and appends it.
template
<
class
...
Ts
>
bool
emplace_back
(
Ts
&&
...
xs
)
{
return
push_back
(
new
value_type
(
std
::
forward
<
Ts
>
(
xs
)...));
}
/// @private
/// @private
void
inc_total_task_size
(
task_size_type
x
)
noexcept
{
void
inc_total_task_size
(
task_size_type
x
)
noexcept
{
CAF_ASSERT
(
x
>
0
);
total_task_size_
+=
x
;
total_task_size_
+=
x
;
}
}
...
@@ -174,6 +147,17 @@ public:
...
@@ -174,6 +147,17 @@ public:
inc_total_task_size
(
policy_
.
task_size
(
x
));
inc_total_task_size
(
policy_
.
task_size
(
x
));
}
}
/// @private
void
dec_total_task_size
(
task_size_type
x
)
noexcept
{
CAF_ASSERT
(
x
>
0
);
total_task_size_
-=
x
;
}
/// @private
void
dec_total_task_size
(
const
value_type
&
x
)
noexcept
{
dec_total_task_size
(
policy_
.
task_size
(
x
));
}
// -- iterator access --------------------------------------------------------
// -- iterator access --------------------------------------------------------
/// Returns an iterator to the dummy before the first element.
/// Returns an iterator to the dummy before the first element.
...
@@ -216,6 +200,8 @@ public:
...
@@ -216,6 +200,8 @@ public:
return
&
tail_
;
return
&
tail_
;
}
}
// -- element access ---------------------------------------------------------
/// Returns a pointer to the first element.
/// Returns a pointer to the first element.
pointer
front
()
noexcept
{
pointer
front
()
noexcept
{
return
promote
(
head_
.
next
);
return
promote
(
head_
.
next
);
...
@@ -226,6 +212,31 @@ public:
...
@@ -226,6 +212,31 @@ public:
return
promote
(
tail_
.
next
);
return
promote
(
tail_
.
next
);
}
}
// -- insertion --------------------------------------------------------------
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
pointer
ptr
)
noexcept
{
CAF_ASSERT
(
ptr
!=
nullptr
);
tail_
.
next
->
next
=
ptr
;
tail_
.
next
=
ptr
;
ptr
->
next
=
&
tail_
;
inc_total_task_size
(
*
ptr
);
return
true
;
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool
push_back
(
unique_pointer
ptr
)
noexcept
{
return
push_back
(
ptr
.
release
());
}
/// Creates a new element from `xs...` and appends it.
template
<
class
...
Ts
>
bool
emplace_back
(
Ts
&&
...
xs
)
{
return
push_back
(
new
value_type
(
std
::
forward
<
Ts
>
(
xs
)...));
}
/// Transfers all element from `other` to the front of this queue.
/// Transfers all element from `other` to the front of this queue.
template
<
class
Container
>
template
<
class
Container
>
void
prepend
(
Container
&
other
)
{
void
prepend
(
Container
&
other
)
{
...
@@ -237,7 +248,7 @@ public:
...
@@ -237,7 +248,7 @@ public:
}
}
other
.
back
()
->
next
=
head_
.
next
;
other
.
back
()
->
next
=
head_
.
next
;
head_
.
next
=
other
.
front
();
head_
.
next
=
other
.
front
();
total_task_size_
+=
other
.
total_task_size
(
);
inc_total_task_size
(
other
.
total_task_size
()
);
other
.
init
();
other
.
init
();
}
}
...
@@ -253,7 +264,7 @@ public:
...
@@ -253,7 +264,7 @@ public:
back
()
->
next
=
other
.
front
();
back
()
->
next
=
other
.
front
();
other
.
back
()
->
next
=
&
tail_
;
other
.
back
()
->
next
=
&
tail_
;
tail_
.
next
=
other
.
back
();
tail_
.
next
=
other
.
back
();
total_task_size_
+=
other
.
total_task_size
(
);
inc_total_task_size
(
other
.
total_task_size
()
);
other
.
init
();
other
.
init
();
}
}
...
@@ -316,7 +327,7 @@ protected:
...
@@ -316,7 +327,7 @@ protected:
node_type
tail_
;
node_type
tail_
;
/// Stores the total size of all items in the queue.
/// Stores the total size of all items in the queue.
task_size_type
total_task_size_
=
0
;
task_size_type
total_task_size_
;
/// Used for LIFO -> FIFO insertion.
/// Used for LIFO -> FIFO insertion.
node_pointer
old_last_
;
node_pointer
old_last_
;
...
...
libcaf_core/caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp
View file @
d1bb08aa
...
@@ -132,6 +132,13 @@ public:
...
@@ -132,6 +132,13 @@ public:
return
nullptr
;
return
nullptr
;
}
}
/// Applies `f` to each element in the queue.
template
<
class
F
>
void
peek_all
(
F
f
)
const
{
for
(
auto
&
kvp
:
qs_
)
kvp
.
second
.
peek_all
(
f
);
}
/// Returns `true` if all queues are empty, `false` otherwise.
/// Returns `true` if all queues are empty, `false` otherwise.
bool
empty
()
const
noexcept
{
bool
empty
()
const
noexcept
{
return
total_task_size
()
==
0
;
return
total_task_size
()
==
0
;
...
...
libcaf_core/caf/intrusive/wdrr_fixed_multiplexed_queue.hpp
View file @
d1bb08aa
...
@@ -92,6 +92,12 @@ public:
...
@@ -92,6 +92,12 @@ public:
return
peek_recursion
<
0
>
();
return
peek_recursion
<
0
>
();
}
}
/// Applies `f` to each element in the queue.
template
<
class
F
>
void
peek_all
(
F
f
)
const
{
return
peek_all_recursion
<
0
>
(
f
);
}
/// Returns `true` if all queues are empty, `false` otherwise.
/// Returns `true` if all queues are empty, `false` otherwise.
bool
empty
()
const
noexcept
{
bool
empty
()
const
noexcept
{
return
total_task_size
()
==
0
;
return
total_task_size
()
==
0
;
...
@@ -126,6 +132,10 @@ public:
...
@@ -126,6 +132,10 @@ public:
private:
private:
// -- recursive helper functions ---------------------------------------------
// -- recursive helper functions ---------------------------------------------
// TODO: a lot of this code could be vastly simplified by using C++14 generic
// lambdas and simple utility to dispatch on the tuple index. Consider
// to revisite this code once we switch to C++14.
template
<
size_t
I
>
template
<
size_t
I
>
detail
::
enable_if_t
<
I
==
num_queues
,
bool
>
detail
::
enable_if_t
<
I
==
num_queues
,
bool
>
push_back_recursion
(
size_t
,
mapped_type
*
)
noexcept
{
push_back_recursion
(
size_t
,
mapped_type
*
)
noexcept
{
...
@@ -186,6 +196,17 @@ private:
...
@@ -186,6 +196,17 @@ private:
return
peek_recursion
<
I
+
1
>
();
return
peek_recursion
<
I
+
1
>
();
}
}
template
<
size_t
I
,
class
F
>
detail
::
enable_if_t
<
I
==
num_queues
>
peek_all_recursion
(
F
&
)
const
{
// nop
}
template
<
size_t
I
,
class
F
>
detail
::
enable_if_t
<
I
!=
num_queues
>
peek_all_recursion
(
F
&
f
)
const
{
std
::
get
<
I
>
(
qs_
).
peek_all
(
f
);
peek_all_recursion
<
I
+
1
>
(
f
);
}
template
<
size_t
I
>
template
<
size_t
I
>
detail
::
enable_if_t
<
I
==
num_queues
>
flush_cache_recursion
()
noexcept
{
detail
::
enable_if_t
<
I
==
num_queues
>
flush_cache_recursion
()
noexcept
{
// nop
// nop
...
...
libcaf_core/test/drr_cached_queue.cpp
View file @
d1bb08aa
...
@@ -64,12 +64,13 @@ struct fixture {
...
@@ -64,12 +64,13 @@ struct fixture {
inode_policy
policy
;
inode_policy
policy
;
queue_type
queue
{
policy
};
queue_type
queue
{
policy
};
void
fill
(
queue_type
&
)
{
template
<
class
Queue
>
void
fill
(
Queue
&
)
{
// nop
// nop
}
}
template
<
class
T
,
class
...
Ts
>
template
<
class
Queue
,
class
T
,
class
...
Ts
>
void
fill
(
queue_typ
e
&
q
,
T
x
,
Ts
...
xs
)
{
void
fill
(
Queu
e
&
q
,
T
x
,
Ts
...
xs
)
{
q
.
emplace_back
(
x
);
q
.
emplace_back
(
x
);
fill
(
q
,
xs
...);
fill
(
q
,
xs
...);
}
}
...
@@ -84,8 +85,6 @@ CAF_TEST(default_constructed) {
...
@@ -84,8 +85,6 @@ CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL
(
queue
.
deficit
(),
0
);
CAF_REQUIRE_EQUAL
(
queue
.
deficit
(),
0
);
CAF_REQUIRE_EQUAL
(
queue
.
total_task_size
(),
0
);
CAF_REQUIRE_EQUAL
(
queue
.
total_task_size
(),
0
);
CAF_REQUIRE_EQUAL
(
queue
.
peek
(),
nullptr
);
CAF_REQUIRE_EQUAL
(
queue
.
peek
(),
nullptr
);
CAF_REQUIRE_EQUAL
(
queue
.
begin
(),
queue
.
end
());
CAF_REQUIRE_EQUAL
(
queue
.
before_begin
()
->
next
,
queue
.
end
().
ptr
);
}
}
CAF_TEST
(
new_round
)
{
CAF_TEST
(
new_round
)
{
...
@@ -154,4 +153,33 @@ CAF_TEST(alternating_consumer) {
...
@@ -154,4 +153,33 @@ CAF_TEST(alternating_consumer) {
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
.
cache
()),
"[9]"
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
.
cache
()),
"[9]"
);
}
}
CAF_TEST
(
peek_all
)
{
auto
queue_to_string
=
[
&
]
{
std
::
string
str
;
auto
peek_fun
=
[
&
](
const
inode
&
x
)
{
if
(
!
str
.
empty
())
str
+=
", "
;
str
+=
std
::
to_string
(
x
.
value
);
};
queue
.
peek_all
(
peek_fun
);
return
str
;
};
CAF_CHECK_EQUAL
(
queue_to_string
(),
""
);
queue
.
emplace_back
(
2
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"2"
);
queue
.
cache
().
emplace_back
(
1
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2"
);
queue
.
emplace_back
(
3
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2, 3"
);
}
CAF_TEST
(
to_string
)
{
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[]"
);
fill
(
queue
,
3
,
4
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[3, 4]"
);
fill
(
queue
.
cache
(),
1
,
2
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[1, 2, 3, 4]"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/drr_queue.cpp
View file @
d1bb08aa
...
@@ -130,4 +130,32 @@ CAF_TEST(new_round) {
...
@@ -130,4 +130,32 @@ CAF_TEST(new_round) {
CAF_CHECK_EQUAL
(
queue
.
deficit
(),
0
);
CAF_CHECK_EQUAL
(
queue
.
deficit
(),
0
);
}
}
CAF_TEST
(
peek_all
)
{
auto
queue_to_string
=
[
&
]
{
std
::
string
str
;
auto
peek_fun
=
[
&
](
const
inode
&
x
)
{
if
(
!
str
.
empty
())
str
+=
", "
;
str
+=
std
::
to_string
(
x
.
value
);
};
queue
.
peek_all
(
peek_fun
);
return
str
;
};
CAF_CHECK_EQUAL
(
queue_to_string
(),
""
);
queue
.
emplace_back
(
1
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1"
);
queue
.
emplace_back
(
2
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2"
);
queue
.
emplace_back
(
3
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2, 3"
);
queue
.
emplace_back
(
4
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2, 3, 4"
);
}
CAF_TEST
(
to_string
)
{
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[]"
);
fill
(
queue
,
1
,
2
,
3
,
4
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[1, 2, 3, 4]"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/task_queue.cpp
View file @
d1bb08aa
...
@@ -143,28 +143,6 @@ CAF_TEST(peek) {
...
@@ -143,28 +143,6 @@ CAF_TEST(peek) {
CAF_CHECK_EQUAL
(
queue
.
peek
()
->
value
,
1
);
CAF_CHECK_EQUAL
(
queue
.
peek
()
->
value
,
1
);
}
}
CAF_TEST
(
peek_all
)
{
auto
queue_to_string
=
[
&
]
{
std
::
string
str
;
auto
peek_fun
=
[
&
](
const
inode
&
x
)
{
if
(
!
str
.
empty
())
str
+=
", "
;
str
+=
std
::
to_string
(
x
.
value
);
};
queue
.
peek_all
(
peek_fun
);
return
str
;
};
CAF_CHECK_EQUAL
(
queue_to_string
(),
""
);
queue
.
emplace_back
(
1
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1"
);
queue
.
emplace_back
(
2
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2"
);
queue
.
emplace_back
(
3
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2, 3"
);
queue
.
emplace_back
(
4
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2, 3, 4"
);
}
CAF_TEST
(
task_size
)
{
CAF_TEST
(
task_size
)
{
fill
(
queue
,
1
,
2
,
3
);
fill
(
queue
,
1
,
2
,
3
);
CAF_CHECK_EQUAL
(
queue
.
total_task_size
(),
6
);
CAF_CHECK_EQUAL
(
queue
.
total_task_size
(),
6
);
...
@@ -174,4 +152,10 @@ CAF_TEST(task_size) {
...
@@ -174,4 +152,10 @@ CAF_TEST(task_size) {
CAF_CHECK_EQUAL
(
queue
.
total_task_size
(),
0
);
CAF_CHECK_EQUAL
(
queue
.
total_task_size
(),
0
);
}
}
CAF_TEST
(
to_string
)
{
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[]"
);
fill
(
queue
,
1
,
2
,
3
,
4
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[1, 2, 3, 4]"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/wdrr_dynamic_multiplexed_queue.cpp
View file @
d1bb08aa
...
@@ -176,4 +176,37 @@ CAF_TEST(priorities) {
...
@@ -176,4 +176,37 @@ CAF_TEST(priorities) {
CAF_REQUIRE_EQUAL
(
queue
.
empty
(),
true
);
CAF_REQUIRE_EQUAL
(
queue
.
empty
(),
true
);
}
}
CAF_TEST
(
peek_all
)
{
auto
queue_to_string
=
[
&
]
{
std
::
string
str
;
auto
peek_fun
=
[
&
](
const
inode
&
x
)
{
if
(
!
str
.
empty
())
str
+=
", "
;
str
+=
std
::
to_string
(
x
.
value
);
};
queue
.
peek_all
(
peek_fun
);
return
str
;
};
make_queues
();
CAF_CHECK_EQUAL
(
queue_to_string
(),
""
);
queue
.
emplace_back
(
1
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1"
);
queue
.
emplace_back
(
2
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2"
);
queue
.
emplace_back
(
3
);
// Lists are iterated in order and 3 is stored in the first queue for
// `x mod 3 == 0` values.
CAF_CHECK_EQUAL
(
queue_to_string
(),
"3, 1, 2"
);
queue
.
emplace_back
(
4
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"3, 1, 4, 2"
);
}
CAF_TEST
(
to_string
)
{
make_queues
();
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[]"
);
fill
(
queue
,
1
,
2
,
3
,
4
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[3, 1, 4, 2]"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/wdrr_fixed_multiplexed_queue.cpp
View file @
d1bb08aa
...
@@ -175,4 +175,34 @@ CAF_TEST(priorities) {
...
@@ -175,4 +175,34 @@ CAF_TEST(priorities) {
CAF_REQUIRE_EQUAL
(
queue
.
empty
(),
true
);
CAF_REQUIRE_EQUAL
(
queue
.
empty
(),
true
);
}
}
CAF_TEST
(
peek_all
)
{
auto
queue_to_string
=
[
&
]
{
std
::
string
str
;
auto
peek_fun
=
[
&
](
const
inode
&
x
)
{
if
(
!
str
.
empty
())
str
+=
", "
;
str
+=
std
::
to_string
(
x
.
value
);
};
queue
.
peek_all
(
peek_fun
);
return
str
;
};
CAF_CHECK_EQUAL
(
queue_to_string
(),
""
);
queue
.
emplace_back
(
1
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1"
);
queue
.
emplace_back
(
2
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"1, 2"
);
queue
.
emplace_back
(
3
);
// Lists are iterated in order and 3 is stored in the first queue for
// `x mod 3 == 0` values.
CAF_CHECK_EQUAL
(
queue_to_string
(),
"3, 1, 2"
);
queue
.
emplace_back
(
4
);
CAF_CHECK_EQUAL
(
queue_to_string
(),
"3, 1, 4, 2"
);
}
CAF_TEST
(
to_string
)
{
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[]"
);
fill
(
queue
,
1
,
2
,
3
,
4
);
CAF_CHECK_EQUAL
(
deep_to_string
(
queue
),
"[3, 1, 4, 2]"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
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