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
874b8980
Commit
874b8980
authored
Apr 08, 2011
by
neverlord
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
benchmarking
parent
594858b8
Changes
11
Show whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
886 additions
and
0 deletions
+886
-0
.gitignore
.gitignore
+3
-0
Makefile
Makefile
+2
-0
cppa.files
cppa.files
+7
-0
queue_performances/Makefile
queue_performances/Makefile
+30
-0
queue_performances/blocking_cached_stack.hpp
queue_performances/blocking_cached_stack.hpp
+126
-0
queue_performances/blocking_cached_stack2.hpp
queue_performances/blocking_cached_stack2.hpp
+140
-0
queue_performances/blocking_sutter_list.hpp
queue_performances/blocking_sutter_list.hpp
+114
-0
queue_performances/cached_stack.hpp
queue_performances/cached_stack.hpp
+106
-0
queue_performances/defines.hpp
queue_performances/defines.hpp
+9
-0
queue_performances/main.cpp
queue_performances/main.cpp
+242
-0
queue_performances/sutter_list.hpp
queue_performances/sutter_list.hpp
+107
-0
No files found.
.gitignore
View file @
874b8980
...
@@ -2,3 +2,6 @@
...
@@ -2,3 +2,6 @@
*.o
*.o
*.dylib
*.dylib
test
test
*.png
*.dat
queue_test
Makefile
View file @
874b8980
all
:
all
:
make
-f
libcppa.Makefile
make
-f
libcppa.Makefile
make
-C
unit_testing
make
-C
unit_testing
make
-C
queue_performances
clean
:
clean
:
make
-f
libcppa.Makefile clean
make
-f
libcppa.Makefile clean
make
-C
unit_testing clean
make
-C
unit_testing clean
make
-C
queue_performances clean
cppa.files
View file @
874b8980
...
@@ -98,3 +98,10 @@ cppa/scheduling_hint.hpp
...
@@ -98,3 +98,10 @@ cppa/scheduling_hint.hpp
src/actor_behavior.cpp
src/actor_behavior.cpp
cppa/group.hpp
cppa/group.hpp
src/group.cpp
src/group.cpp
queue_performances/main.cpp
queue_performances/sutter_list.hpp
queue_performances/defines.hpp
queue_performances/cached_stack.hpp
queue_performances/blocking_cached_stack.hpp
queue_performances/blocking_cached_stack2.hpp
queue_performances/blocking_sutter_list.hpp
queue_performances/Makefile
0 → 100644
View file @
874b8980
include
../Makefile.rules
#CXX = /opt/local/bin/g++-mp-4.5
#CXX = /opt/local/bin/g++-mp-4.6
#CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -g -O0 -I/opt/local/include/
#CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -O2 -I/opt/local/include/
#LIBS = -L/opt/local/lib -lboost_thread-mt -L../ -lcppa
INCLUDES
=
-I
./
-I
../
FLAGS
=
-DCACHE_LINE_SIZE
=
64
EXECUTABLE
=
../queue_test
HEADERS
=
sutter_list.hpp
SOURCES
=
main.cpp
OBJECTS
=
$(SOURCES:.cpp=.o)
%.o
:
%.cpp $(HEADERS) $(HEADERS)
$(CXX)
$(CXXFLAGS)
$(INCLUDES)
$(FLAGS)
-c
$<
-o
$@
$(EXECUTABLE)
:
$(OBJECTS) $(HEADERS)
$(CXX)
$(LIBS)
-L
../
-lcppa
$(OBJECTS)
-o
$(EXECUTABLE)
all
:
$(EXECUTABLE)
clean
:
rm
-f
$(OBJECTS)
$(EXECUTABLE)
queue_performances/blocking_cached_stack.hpp
0 → 100644
View file @
874b8980
#ifndef BLOCKING_CACHED_STACK_HPP
#define BLOCKING_CACHED_STACK_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "defines.hpp"
// This class is intrusive.
template
<
typename
T
>
class
blocking_cached_stack
{
// singly linked list, serves as cache
T
*
m_head
;
char
m_pad1
[
CACHE_LINE_SIZE
-
sizeof
(
T
*
)];
// modified by consumers
std
::
atomic
<
T
*>
m_stack
;
char
m_pad2
[
CACHE_LINE_SIZE
-
sizeof
(
std
::
atomic
<
T
*>
)];
// locked on enqueue/dequeue operations to/from an empty list
boost
::
mutex
m_mtx
;
boost
::
condition_variable
m_cv
;
typedef
boost
::
unique_lock
<
boost
::
mutex
>
lock_type
;
// read all elements of m_stack, convert them to FIFO order and store
// them in m_head
// precondition: m_head == nullptr
bool
consume_stack
()
{
T
*
e
=
m_stack
.
load
();
while
(
e
)
{
if
(
m_stack
.
compare_exchange_weak
(
e
,
0
))
{
// m_stack is now empty (m_stack == nullptr)
while
(
e
)
{
T
*
next
=
e
->
next
;
// enqueue to m_head
e
->
next
=
m_head
;
m_head
=
e
;
// next iteration
e
=
next
;
}
return
true
;
}
}
// nothing to consume
return
false
;
}
void
wait_for_data
()
{
if
(
!
m_head
&&
!
(
m_stack
.
load
()))
{
lock_type
lock
(
m_mtx
);
while
(
!
(
m_stack
.
load
()))
m_cv
.
wait
(
lock
);
}
}
public:
blocking_cached_stack
()
:
m_head
(
0
)
{
m_stack
=
0
;
}
~
blocking_cached_stack
()
{
do
{
while
(
m_head
)
{
T
*
next
=
m_head
->
next
;
delete
m_head
;
m_head
=
next
;
}
}
// repeat if m_stack is not empty
while
(
consume_stack
());
}
void
push
(
T
*
what
)
{
T
*
e
=
m_stack
.
load
();
for
(;;)
{
what
->
next
=
e
;
if
(
!
e
)
{
lock_type
lock
(
m_mtx
);
if
(
m_stack
.
compare_exchange_weak
(
e
,
what
))
{
m_cv
.
notify_one
();
return
;
}
}
// compare_exchange_weak stores the
// new value to e if the operation fails
else
if
(
m_stack
.
compare_exchange_weak
(
e
,
what
))
return
;
}
}
T
*
try_pop
()
{
if
(
m_head
||
consume_stack
())
{
T
*
result
=
m_head
;
m_head
=
m_head
->
next
;
return
result
;
}
return
0
;
}
T
*
pop
()
{
wait_for_data
();
return
try_pop
();
}
};
#endif // BLOCKING_CACHED_STACK_HPP
queue_performances/blocking_cached_stack2.hpp
0 → 100644
View file @
874b8980
#ifndef BLOCKING_CACHED_STACK2_HPP
#define BLOCKING_CACHED_STACK2_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "defines.hpp"
// This class is intrusive.
template
<
typename
T
>
class
blocking_cached_stack2
{
// singly linked list, serves as cache
T
*
m_head
;
char
m_pad1
[
CACHE_LINE_SIZE
-
sizeof
(
T
*
)];
// modified by consumers
std
::
atomic
<
T
*>
m_stack
;
char
m_pad2
[
CACHE_LINE_SIZE
-
sizeof
(
std
::
atomic
<
T
*>
)];
T
*
m_dummy
;
char
m_pad3
[
CACHE_LINE_SIZE
-
sizeof
(
T
)];
// locked on enqueue/dequeue operations to/from an empty list
boost
::
mutex
m_mtx
;
boost
::
condition_variable
m_cv
;
typedef
boost
::
unique_lock
<
boost
::
mutex
>
lock_type
;
// read all elements of m_stack, convert them to FIFO order and store
// them in m_head
// precondition: m_head == nullptr
void
consume_stack
()
{
T
*
e
=
m_stack
.
load
();
while
(
e
)
{
// enqueue dummy instead of nullptr to reduce
// lock operations
if
(
m_stack
.
compare_exchange_weak
(
e
,
m_dummy
))
{
// m_stack is now empty (m_stack == m_dummy)
// m_dummy marks always the end of the stack
while
(
e
&&
e
!=
m_dummy
)
{
T
*
next
=
e
->
next
;
// enqueue to m_head
e
->
next
=
m_head
;
m_head
=
e
;
// next iteration
e
=
next
;
}
return
;
}
}
// nothing to consume
}
void
wait_for_data
()
{
if
(
!
m_head
)
{
T
*
e
=
m_stack
.
load
();
while
(
e
==
m_dummy
)
{
if
(
m_stack
.
compare_exchange_weak
(
e
,
0
))
e
=
0
;
}
if
(
!
e
)
{
lock_type
lock
(
m_mtx
);
while
(
!
(
m_stack
.
load
()))
m_cv
.
wait
(
lock
);
}
consume_stack
();
}
}
void
delete_head
()
{
while
(
m_head
)
{
T
*
next
=
m_head
->
next
;
delete
m_head
;
m_head
=
next
;
}
}
public:
blocking_cached_stack2
()
:
m_head
(
0
)
{
m_stack
=
0
;
m_dummy
=
new
T
;
}
~
blocking_cached_stack2
()
{
delete_head
();
T
*
e
=
m_stack
.
load
();
if
(
e
&&
e
!=
m_dummy
)
{
consume_stack
();
delete_head
();
}
delete
m_dummy
;
}
void
push
(
T
*
what
)
{
T
*
e
=
m_stack
.
load
();
for
(;;)
{
what
->
next
=
e
;
if
(
!
e
)
{
lock_type
lock
(
m_mtx
);
if
(
m_stack
.
compare_exchange_weak
(
e
,
what
))
{
m_cv
.
notify_one
();
return
;
}
}
// compare_exchange_weak stores the
// new value to e if the operation fails
else
if
(
m_stack
.
compare_exchange_weak
(
e
,
what
))
return
;
}
}
T
*
pop
()
{
wait_for_data
();
T
*
result
=
m_head
;
m_head
=
m_head
->
next
;
return
result
;
}
};
#endif // BLOCKING_CACHED_STACK2_HPP
queue_performances/blocking_sutter_list.hpp
0 → 100644
View file @
874b8980
#ifndef BLOCKING_SUTTER_LIST_HPP
#define BLOCKING_SUTTER_LIST_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "defines.hpp"
// This implementation is a single-consumer version
// of an example implementation of Herb Sutter:
// http://drdobbs.com/cpp/211601363.
// T is any type
template
<
typename
T
>
class
blocking_sutter_list
{
struct
node
{
node
(
T
*
val
=
0
)
:
value
(
val
),
next
(
0
)
{
}
T
*
value
;
std
::
atomic
<
node
*>
next
;
char
pad
[
CACHE_LINE_SIZE
-
sizeof
(
T
*
)
-
sizeof
(
std
::
atomic
<
node
*>
)];
};
// one consumer at a time
node
*
m_first
;
char
m_pad1
[
CACHE_LINE_SIZE
-
sizeof
(
node
*
)];
// for one producers at a time
node
*
m_last
;
char
m_pad2
[
CACHE_LINE_SIZE
-
sizeof
(
node
*
)];
// shared among producers
std
::
atomic
<
bool
>
m_producer_lock
;
char
m_pad3
[
CACHE_LINE_SIZE
-
sizeof
(
std
::
atomic
<
bool
>
)];
// locked on enqueue/dequeue operations to/from an empty list
boost
::
mutex
m_mtx
;
boost
::
condition_variable
m_cv
;
typedef
boost
::
unique_lock
<
boost
::
mutex
>
lock_type
;
public:
blocking_sutter_list
()
{
m_first
=
m_last
=
new
node
;
m_producer_lock
=
false
;
}
~
blocking_sutter_list
()
{
while
(
m_first
)
{
node
*
tmp
=
m_first
;
m_first
=
tmp
->
next
;
delete
tmp
;
}
}
// takes ownership of what
void
push
(
T
*
what
)
{
bool
consumer_might_sleep
=
0
;
node
*
tmp
=
new
node
(
what
);
// acquire exclusivity
while
(
m_producer_lock
.
exchange
(
true
))
{
boost
::
this_thread
::
yield
();
}
// do we have to wakeup a sleeping consumer?
// this is a sufficient condition because m_last->value is 0
// if and only if m_head == m_tail
consumer_might_sleep
=
(
m_last
->
value
==
0
);
// publish & swing last forward
m_last
->
next
=
tmp
;
m_last
=
tmp
;
// release exclusivity
m_producer_lock
=
false
;
// wakeup consumer if needed
if
(
consumer_might_sleep
)
{
lock_type
lock
(
m_mtx
);
m_cv
.
notify_one
();
}
}
// polls the queue until an element was dequeued
T
*
pop
()
{
node
*
first
=
m_first
;
node
*
next
=
m_first
->
next
;
if
(
!
next
)
{
lock_type
lock
(
m_mtx
);
while
(
!
(
next
=
m_first
->
next
))
{
m_cv
.
wait
(
lock
);
}
}
T
*
result
=
next
->
value
;
// take it out
next
->
value
=
0
;
// of the node
// swing first forward
m_first
=
next
;
// delete old dummy
delete
first
;
// done
return
result
;
}
};
#endif // BLOCKING_SUTTER_LIST_HPP
queue_performances/cached_stack.hpp
0 → 100644
View file @
874b8980
#ifndef CACHED_STACK_HPP
#define CACHED_STACK_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "defines.hpp"
// This class is intrusive.
template
<
typename
T
>
class
cached_stack
{
// singly linked list, serves as cache
T
*
m_head
;
char
m_pad1
[
CACHE_LINE_SIZE
-
sizeof
(
T
*
)];
// modified by consumers
std
::
atomic
<
T
*>
m_stack
;
// read all elements of m_stack, convert them to FIFO order and store
// them in m_head
// precondition: m_head == nullptr
bool
consume_stack
()
{
T
*
e
=
m_stack
.
load
();
while
(
e
)
{
if
(
m_stack
.
compare_exchange_weak
(
e
,
0
))
{
// m_stack is now empty (m_stack == nullptr)
while
(
e
)
{
T
*
next
=
e
->
next
;
// enqueue to m_head
e
->
next
=
m_head
;
m_head
=
e
;
// next iteration
e
=
next
;
}
return
true
;
}
}
// nothing to consume
return
false
;
}
public:
cached_stack
()
:
m_head
(
0
)
{
m_stack
=
0
;
}
~
cached_stack
()
{
do
{
while
(
m_head
)
{
T
*
next
=
m_head
->
next
;
delete
m_head
;
m_head
=
next
;
}
}
// repeat if m_stack is not empty
while
(
consume_stack
());
}
void
push
(
T
*
what
)
{
T
*
e
=
m_stack
.
load
();
for
(;;)
{
what
->
next
=
e
;
// compare_exchange_weak stores the
// new value to e if the operation fails
if
(
m_stack
.
compare_exchange_weak
(
e
,
what
))
return
;
}
}
T
*
try_pop
()
{
if
(
m_head
||
consume_stack
())
{
T
*
result
=
m_head
;
m_head
=
m_head
->
next
;
return
result
;
}
return
0
;
}
T
*
pop
()
{
T
*
result
=
try_pop
();
while
(
!
result
)
{
boost
::
this_thread
::
yield
();
result
=
try_pop
();
}
return
result
;
}
};
#endif // CACHED_STACK_HPP
queue_performances/defines.hpp
0 → 100644
View file @
874b8980
#ifndef DEFINES_HPP
#define DEFINES_HPP
#ifndef CACHE_LINE_SIZE
// the cache line size of an i7 under Mac OS X
#define CACHE_LINE_SIZE 64
#endif
#endif // DEFINES_HPP
queue_performances/main.cpp
0 → 100644
View file @
874b8980
#include <vector>
#include <cstddef>
#include <sstream>
#include <iostream>
#include <stdexcept>
#include <boost/thread.hpp>
#include <boost/progress.hpp>
#include <boost/lexical_cast.hpp>
#include "sutter_list.hpp"
#include "cached_stack.hpp"
#include "blocking_sutter_list.hpp"
#include "blocking_cached_stack.hpp"
#include "blocking_cached_stack2.hpp"
using
std
::
cout
;
using
std
::
cerr
;
using
std
::
endl
;
namespace
{
//const size_t num_messages = 1000000;
//const size_t num_producers = 10;
//const size_t num_messages_per_producer = num_messages / num_producers;
}
// namespace <anonymous>
template
<
typename
Queue
,
typename
Allocator
>
void
producer
(
Queue
&
q
,
Allocator
&
a
,
size_t
begin
,
size_t
end
)
{
for
(
;
begin
!=
end
;
++
begin
)
{
q
.
push
(
a
(
begin
));
}
}
template
<
typename
Queue
,
typename
Processor
>
void
consumer
(
Queue
&
q
,
Processor
&
p
,
size_t
num_messages
)
{
// vector<bool> scales better (in memory) than bool[num_messages]
std
::
vector
<
bool
>
received
(
num_messages
);
for
(
size_t
i
=
0
;
i
<
num_messages
;
++
i
)
received
[
i
]
=
false
;
for
(
size_t
i
=
0
;
i
<
num_messages
;
++
i
)
{
size_t
value
;
p
(
q
.
pop
(),
value
);
if
(
value
>=
num_messages
)
{
throw
std
::
runtime_error
(
"value out of bounds"
);
}
else
if
(
received
[
value
])
{
std
::
ostringstream
oss
;
oss
<<
"ERROR: received element nr. "
<<
value
<<
" two times"
;
throw
std
::
runtime_error
(
oss
.
str
());
}
received
[
value
]
=
true
;
}
// done
}
void
usage
()
{
cout
<<
"usage:"
<<
endl
<<
"queue_test [messages] [producer threads] "
<<
"[list impl.] {format string}"
<<
endl
<<
" available implementations:"
<<
endl
<<
" - sutter_list"
<<
endl
<<
" - blocking_sutter_list"
<<
endl
<<
" - cached_stack"
<<
endl
<<
" - blocking_cached_stack"
<<
endl
<<
" - blocking_cached_stack2"
<<
endl
<<
endl
<<
" possible format string variables: "
<<
endl
<<
" - $MESSAGES"
<<
endl
<<
" - $PRODUCERS"
<<
endl
<<
" - $TIME"
<<
endl
<<
endl
<<
"example: ./queue_test 10000 10 cached_list
\"
$MESSAGES $TIME
\"
"
<<
endl
;
}
template
<
typename
Queue
,
typename
Allocator
,
typename
Processor
>
double
run_test
(
size_t
num_messages
,
size_t
num_producers
,
Allocator
element_allocator
,
Processor
element_processor
)
{
size_t
num_messages_per_producer
=
num_messages
/
num_producers
;
// measurement
boost
::
timer
t0
;
// locals
Queue
list
;
std
::
vector
<
boost
::
thread
*>
producer_threads
(
num_producers
);
for
(
size_t
i
=
0
;
i
<
num_producers
;
++
i
)
{
producer_threads
[
i
]
=
new
boost
::
thread
(
producer
<
Queue
,
Allocator
>
,
boost
::
ref
(
list
),
boost
::
ref
(
element_allocator
),
i
*
num_messages_per_producer
,
(
i
+
1
)
*
num_messages_per_producer
);
}
// run consumer in main thread
consumer
(
list
,
element_processor
,
num_messages
);
// print result
return
t0
.
elapsed
();
}
struct
cs_element
{
size_t
value
;
std
::
atomic
<
cs_element
*>
next
;
cs_element
(
size_t
val
=
0
)
:
value
(
val
),
next
(
0
)
{
}
};
int
main
(
int
argc
,
char
**
argv
)
{
if
(
argc
<
4
||
argc
>
5
)
{
usage
();
return
-
1
;
}
size_t
num_messages
=
boost
::
lexical_cast
<
size_t
>
(
argv
[
1
]);
size_t
num_producers
=
boost
::
lexical_cast
<
size_t
>
(
argv
[
2
]);
if
(
num_messages
==
0
||
num_producers
==
0
)
{
cerr
<<
"invalid arguments"
<<
endl
;
return
-
2
;
}
if
((
num_messages
%
num_producers
)
!=
0
)
{
cerr
<<
"(num_messages % num_producers) != 0"
<<
endl
;
return
-
3
;
}
std
::
string
format_string
;
if
(
argc
==
5
)
{
format_string
=
argv
[
4
];
}
else
{
format_string
=
"$MESSAGES $TIME"
;
}
std
::
string
list_name
=
argv
[
3
];
double
elapsed_time
;
if
(
list_name
==
"sutter_list"
)
{
elapsed_time
=
run_test
<
sutter_list
<
size_t
>>
(
num_messages
,
num_producers
,
[]
(
size_t
value
)
->
size_t
*
{
return
new
size_t
(
value
);
},
[]
(
size_t
*
value
,
size_t
&
storage
)
{
storage
=
*
value
;
delete
value
;
}
);
}
else
if
(
list_name
==
"blocking_sutter_list"
)
{
elapsed_time
=
run_test
<
blocking_sutter_list
<
size_t
>>
(
num_messages
,
num_producers
,
[]
(
size_t
value
)
->
size_t
*
{
return
new
size_t
(
value
);
},
[]
(
size_t
*
value
,
size_t
&
storage
)
{
storage
=
*
value
;
delete
value
;
}
);
}
else
if
(
list_name
==
"cached_stack"
)
{
elapsed_time
=
run_test
<
cached_stack
<
cs_element
>>
(
num_messages
,
num_producers
,
[]
(
size_t
value
)
->
cs_element
*
{
return
new
cs_element
(
value
);
},
[]
(
cs_element
*
e
,
size_t
&
storage
)
{
storage
=
e
->
value
;
delete
e
;
}
);
}
else
if
(
list_name
==
"blocking_cached_stack"
)
{
elapsed_time
=
run_test
<
blocking_cached_stack
<
cs_element
>>
(
num_messages
,
num_producers
,
[]
(
size_t
value
)
->
cs_element
*
{
return
new
cs_element
(
value
);
},
[]
(
cs_element
*
e
,
size_t
&
storage
)
{
storage
=
e
->
value
;
delete
e
;
}
);
}
else
if
(
list_name
==
"blocking_cached_stack2"
)
{
elapsed_time
=
run_test
<
blocking_cached_stack2
<
cs_element
>>
(
num_messages
,
num_producers
,
[]
(
size_t
value
)
->
cs_element
*
{
return
new
cs_element
(
value
);
},
[]
(
cs_element
*
e
,
size_t
&
storage
)
{
storage
=
e
->
value
;
delete
e
;
}
);
}
else
{
cerr
<<
"unknown list"
<<
endl
;
usage
();
return
-
4
;
}
// build output message
std
::
vector
<
std
::
pair
<
std
::
string
,
std
::
string
>>
replacements
=
{
{
"$MESSAGES"
,
argv
[
1
]
},
{
"$PRODUCERS"
,
argv
[
2
]
},
{
"$TIME"
,
boost
::
lexical_cast
<
std
::
string
>
(
elapsed_time
)
}
};
for
(
auto
i
=
replacements
.
begin
();
i
!=
replacements
.
end
();
++
i
)
{
const
std
::
string
&
needle
=
i
->
first
;
const
std
::
string
&
value
=
i
->
second
;
std
::
string
::
size_type
pos
=
format_string
.
find
(
needle
);
if
(
pos
!=
std
::
string
::
npos
)
{
format_string
.
replace
(
pos
,
pos
+
needle
.
size
(),
value
);
}
}
cout
<<
format_string
<<
endl
;
// done
return
0
;
}
queue_performances/sutter_list.hpp
0 → 100644
View file @
874b8980
#ifndef SUTTER_LIST_HPP
#define SUTTER_LIST_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "defines.hpp"
// This implementation is a single-consumer version
// of an example implementation of Herb Sutter:
// http://drdobbs.com/cpp/211601363.
// T is any type
template
<
typename
T
>
class
sutter_list
{
struct
node
{
node
(
T
*
val
=
0
)
:
value
(
val
),
next
(
0
)
{
}
T
*
value
;
std
::
atomic
<
node
*>
next
;
char
pad
[
CACHE_LINE_SIZE
-
sizeof
(
T
*
)
-
sizeof
(
std
::
atomic
<
node
*>
)];
};
// one consumer at a time
node
*
m_first
;
char
m_pad1
[
CACHE_LINE_SIZE
-
sizeof
(
node
*
)];
// for one producers at a time
node
*
m_last
;
char
m_pad2
[
CACHE_LINE_SIZE
-
sizeof
(
node
*
)];
// shared among producers
std
::
atomic
<
bool
>
m_producer_lock
;
public:
sutter_list
()
{
m_first
=
m_last
=
new
node
;
m_producer_lock
=
false
;
}
~
sutter_list
()
{
while
(
m_first
)
{
node
*
tmp
=
m_first
;
m_first
=
tmp
->
next
;
delete
tmp
;
}
}
// takes ownership of what
void
push
(
T
*
what
)
{
node
*
tmp
=
new
node
(
what
);
// acquire exclusivity
while
(
m_producer_lock
.
exchange
(
true
))
{
boost
::
this_thread
::
yield
();
}
// publish & swing last forward
m_last
->
next
=
tmp
;
m_last
=
tmp
;
// release exclusivity
m_producer_lock
=
false
;
}
// returns nullptr on failure
T
*
try_pop
()
{
// no critical section; only one consumer allowed
node
*
first
=
m_first
;
node
*
next
=
m_first
->
next
;
if
(
next
)
{
// queue is not empty
T
*
result
=
next
->
value
;
// take it out
next
->
value
=
0
;
// of the node
// swing first forward
m_first
=
next
;
// delete old dummy
delete
first
;
// done
return
result
;
}
// queue was empty
return
0
;
}
// polls the queue until an element was dequeued
T
*
pop
()
{
T
*
result
=
try_pop
();
while
(
!
result
)
{
boost
::
this_thread
::
yield
();
result
=
try_pop
();
}
return
result
;
}
};
#endif // SUTTER_LIST_HPP
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