Commit 2b2c3c51 authored by Dominik Charousset's avatar Dominik Charousset

removed obsolete files

parent 2d92106e
#!/bin/bash
NUM_THREADS=8
XUNIT=1000000
# $1 = list name
# $2 = num messages
function exec_queue_test()
{
echo "./queue_test $2 $NUM_THREADS $1"
# run test five times
_1=$(./queue_test $2 $NUM_THREADS $1 "\$TIME")
_2=$(./queue_test $2 $NUM_THREADS $1 "\$TIME")
_3=$(./queue_test $2 $NUM_THREADS $1 "\$TIME")
_4=$(./queue_test $2 $NUM_THREADS $1 "\$TIME")
_5=$(./queue_test $2 $NUM_THREADS $1 "\$TIME")
# get average runtime
MID=$(echo "scale=50;($_1+$_2+$_3+$_4+$_5)/5" | bc)
# get max. runtime
MAX=$(echo "define max (x, y) { if (x < y) return (y); return (x); }; max(max(max(max($_1,$_2),$_3),$_4),$_5)" | bc)
# get min. runtime
MIN=$(echo "define min (x, y) { if (x < y) return (x); return (y); }; min(min(min(min($_1,$_2),$_3),$_4),$_5)" | bc)
# error
ERR=$(echo "scale=50; $MAX-$MIN" | bc)
# X value for gnuplot
XVAL=$(echo "scale=4; $2/$XUNIT" | bc)
# print result to file
echo "$XVAL $MID $ERR" >> "$1.dat"
}
# $1 = list name
# $2 = minimum num messages
# $3 = maximum num messages
# $4 = step
function test_run()
{
#rm -f "$list.dat"
i=$2
while test "$i" -le "$3" ; do
exec_queue_test $1 $i
XVAL=$(echo "scale=4; $i/$XUNIT" | bc)
i=$[$i+$4]
done
}
#for list in intrusive_sutter_list cached_stack lockfree_list ; do
for list in michael_scott_lockfree ; do
echo "benchmarking $list ..."
rm -f "$list.dat"
test_run $list 1000000 10000000 9000000
test_run $list 20000000 110000000 10000000
done
echo "done"
gnuplot -e "FILE='i7_${NUM_THREADS}threads.png';TITLE='2.66 GHz i7 (dual core) with $NUM_THREADS producer threads'" gnuplot_file
CXX = /usr/bin/g++-4.6
CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -O2
#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./
LIBS = -pthread
FLAGS = -DCACHE_LINE_SIZE=64
EXECUTABLE = queue_test
HEADERS = blocking_cached_stack2.hpp blocking_cached_stack.hpp blocking_sutter_list.hpp cached_stack.hpp defines.hpp intrusive_sutter_list.hpp lockfree_list.hpp sutter_list.hpp
SOURCES = main.cpp
OBJECTS = $(SOURCES:.cpp=.o)
%.o : %.cpp $(HEADERS)
$(CXX) $(CXXFLAGS) $(INCLUDES) $(FLAGS) -c $< -o $@
$(EXECUTABLE) : $(OBJECTS) $(HEADERS)
$(CXX) $(LIBS) $(OBJECTS) -o $(EXECUTABLE)
all : $(EXECUTABLE)
clean:
rm -f $(OBJECTS) $(EXECUTABLE)
#ifndef BLOCKING_CACHED_STACK_HPP
#define BLOCKING_CACHED_STACK_HPP
#include <thread>
#include <atomic>
#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
std::mutex m_mtx;
std::condition_variable m_cv;
typedef std::unique_lock<std::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
#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
std::mutex m_mtx;
std::condition_variable m_cv;
typedef std::unique_lock<std::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
#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
std::mutex m_mtx;
std::condition_variable m_cv;
typedef std::unique_lock<std::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)) {
std::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
#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) {
std::this_thread::yield();
result = try_pop();
}
return result;
}
};
#endif // CACHED_STACK_HPP
#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
echo 2 workers
./queue_test 10000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 20000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 30000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 40000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 50000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 70000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 80000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 90000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 100000000 2 cached_stack "MSG_IN_MILLION TIME"
./queue_test 110000000 2 cached_stack "MSG_IN_MILLION TIME"
echo
echo 4 workers
./queue_test 10000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 20000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 30000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 40000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 50000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 70000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 80000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 90000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 100000000 4 cached_stack "MSG_IN_MILLION TIME"
./queue_test 110000000 4 cached_stack "MSG_IN_MILLION TIME"
echo
echo 8 workers
./queue_test 10000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 20000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 30000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 40000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 50000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 70000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 80000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 90000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 100000000 8 cached_stack "MSG_IN_MILLION TIME"
./queue_test 110000000 8 cached_stack "MSG_IN_MILLION TIME"
echo
echo 16 workers
./queue_test 10000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 20000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 30000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 40000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 50000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 70000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 80000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 90000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 100000000 16 cached_stack "MSG_IN_MILLION TIME"
./queue_test 110000000 16 cached_stack "MSG_IN_MILLION TIME"
#ifndef INTRUSIVE_SUTTER_LIST_HPP
#define INTRUSIVE_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 intrusive_sutter_list {
public:
struct node {
node(T val = T()) : value(val), next(0) { }
T value;
std::atomic<node*> next;
char pad[CACHE_LINE_SIZE - sizeof(T)- sizeof(std::atomic<node*>)];
};
private:
// 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:
intrusive_sutter_list() {
m_first = m_last = new node;
m_producer_lock = false;
}
~intrusive_sutter_list() {
while (m_first) {
node* tmp = m_first;
m_first = tmp->next;
delete tmp;
}
}
// takes ownership of what
void push(node* tmp) {
// acquire exclusivity
while (m_producer_lock.exchange(true)) {
std::this_thread::yield();
}
// publish & swing last forward
m_last->next = tmp;
m_last = tmp;
// release exclusivity
m_producer_lock = false;
}
// returns nullptr on failure
bool try_pop(T& result) {
// no critical section; only one consumer allowed
node* first = m_first;
node* next = m_first->next;
if (next) {
// queue is not empty
result = next->value; // take it out of the node
// swing first forward
m_first = next;
// delete old dummy
delete first;
// done
return true;
}
return false;
}
// polls the queue until an element was dequeued
T pop() {
T result;
while (!try_pop(result)) {
std::this_thread::yield();
}
return result;
}
};
#endif // INTRUSIVE_SUTTER_LIST_HPP
#ifndef LOCKFREE_LIST_HPP
#define LOCKFREE_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 lockfree_list {
public:
struct node {
node(T val = T()) : value(val), next(0) { }
T value;
std::atomic<node*> next;
char pad[CACHE_LINE_SIZE - sizeof(T)- sizeof(std::atomic<node*>)];
};
private:
// one consumer at a time
node* m_first;
char m_pad1[CACHE_LINE_SIZE - sizeof(node*)];
// shared among producers
std::atomic<node*> m_last;
public:
lockfree_list() {
m_first = m_last = new node;
}
~lockfree_list() {
while (m_first) {
node* tmp = m_first;
m_first = tmp->next;
delete tmp;
}
}
// takes ownership of what
void push(node* tmp) {
// m_last becomes our predecessor
node* predecessor = m_last.load();
// swing last forward
while (!m_last.compare_exchange_weak(predecessor, tmp)) { }
// link pieces together
predecessor->next = tmp;
}
// returns nullptr on failure
bool try_pop(T& result) {
// no critical section; only one consumer allowed
node* first = m_first;
node* next = m_first->next;
if (next) {
// queue is not empty
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 true;
}
// queue was empty
return false;
}
// polls the queue until an element was dequeued
T pop() {
T result;
while (!try_pop(result)) {
std::this_thread::yield();
}
return result;
}
};
#endif // LOCKFREE_LIST_HPP
#include <thread>
#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 "lockfree_list.hpp"
#include "blocking_sutter_list.hpp"
#include "intrusive_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
<< " - intrusive_sutter_list" << endl
<< " - blocking_sutter_list" << endl
<< " - cached_stack" << endl
<< " - blocking_cached_stack" << endl
<< " - blocking_cached_stack2" << endl
<< " - lockfree_list" << endl
<< endl
<< " possible format string variables: " << endl
<< " - MESSAGES" << endl
<< " - MSG_IN_MILLION" << endl
<< " - PRODUCERS" << endl
<< " - TIME" << endl
<< endl
<< "example: ./queue_test 10000 10 cached_stack \"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<std::thread*> producer_threads(num_producers);
for (size_t i = 0; i < num_producers; ++i) {
producer_threads[i] = new std::thread(producer<Queue, Allocator>,
std::ref(list),
std::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 == "intrusive_sutter_list") {
typedef intrusive_sutter_list<size_t> isl;
elapsed_time = run_test<isl>(
num_messages,
num_producers,
[] (size_t value) -> isl::node* {
return new isl::node(value);
},
[] (const size_t& value, size_t& storage) {
storage = value;
}
);
}
else if (list_name == "lockfree_list") {
typedef lockfree_list<size_t> isl;
elapsed_time = run_test<isl>(
num_messages,
num_producers,
[] (size_t value) -> isl::node* {
return new isl::node(value);
},
[] (const size_t& value, size_t& storage) {
storage = 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] }, { "MSG_IN_MILLION", boost::lexical_cast<std::string>(num_messages / 1000000.0) }, { "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;
}
#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)) {
std::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) {
std::this_thread::yield();
result = try_pop();
}
return result;
}
};
#endif // SUTTER_LIST_HPP
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment