View difference between Paste ID: 0kuzm3Uf and

-
+// lump/list.H
 
 #ifndef INCLUDED_LUMP_LIST_H
 #define INCLUDED_LUMP_LIST_H
 
 #include <cstddef>
 #include <bitset>
 
 #include <boost/interprocess/offset_ptr.hpp>
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/tuple/tuple.hpp>
 #include <boost/tuple/tuple_comparison.hpp>
 
 #include <sys/time.h>
 
 // The __sync functions are GCC atomic builtins. They're documented here: http://gcc.gnu.org/onlinedocs/gcc-4.1.2/gcc/Atomic-Builtins.html
 // method_one() and method_two() have commented out condition waits. You can try uncommenting them, they only make latency worse. Same for the condition
 // code in method_three(), it's only left in since that's how I was testing last.
 
 #define FULL_MEMORY_BARRIER __sync_synchronize()
 
 #ifndef MEMORY_BARRIER_DEBUG // If you set this and things start working, you have a memory barrier bug
 #define RELEASE_MEMORY_BARRIER_IMPL2(line) int release_barrier##line; __sync_lock_release(&release_barrier##line)
 #define RELEASE_MEMORY_BARRIER_IMPL(line) RELEASE_MEMORY_BARRIER_IMPL2(line)
 #define RELEASE_MEMORY_BARRIER RELEASE_MEMORY_BARRIER_IMPL(__LINE__)
 #else
 #define RELEASE_MEMORY_BARRIER FULL_MEMORY_BARRIER
 #endif
 
 #ifndef MEMORY_BARRIER_DEBUG
 #define ACQUIRE_MEMORY_BARRIER_IMPL2(line) int acquire_barrier##line; __sync_lock_test_and_set(&acquire_barrier##line, 1)
 #define ACQUIRE_MEMORY_BARRIER_IMPL(line) ACQUIRE_MEMORY_BARRIER_IMPL2(line)
 #define ACQUIRE_MEMORY_BARRIER ACQUIRE_MEMORY_BARRIER_IMPL(__LINE__)
 #else
 #define ACQUIRE_MEMORY_BARRIER FULL_MEMORY_BARRIER
 #endif
 
 // Meant to represent an offset like offset_ptr, but doesn't
 // do anything automagically. Nice if you want to have pointers
 // from one segment into another segment.
 template<class T>
 class diff_ptr
 {
 public:
 	inline diff_ptr() {}
 
 	inline diff_ptr(diff_ptr const& other)
 		: offset_(other.offset_)
 	{}
 
 	inline diff_ptr(boost::interprocess::offset_ptr<T> const& init,
 					boost::interprocess::managed_shared_memory const& segment)
 	{
 		set(init.get(), segment);
 	}
 
 	inline diff_ptr(T* init,
 					boost::interprocess::managed_shared_memory const& segment)
 	{
 		set(init, segment);
 	}
 
 	inline void set(T* init,
 					boost::interprocess::managed_shared_memory const& segment)
 	{
 		set_offset((char*)(init) - (char*)(segment.get_address()));
 	}
 
 	inline void set(T volatile* init,
 					boost::interprocess::managed_shared_memory const& segment) volatile
 	{
 		set_offset((char*)(init) - (char*)(segment.get_address()));
 	}
 
 	inline void set_offset(ptrdiff_t offset) volatile
 	{
 		offset_ = offset;
 	}
 
 	inline T* get(boost::interprocess::managed_shared_memory const& segment) const volatile
 	{
 		return reinterpret_cast<T*>((char*)(segment.get_address()) + offset_);
 	}
 
 private:
 	std::ptrdiff_t offset_;
 };
 
 enum NodeState {
 	HEAD,
 	NOT_FILLED_YET,
 	FILLED
 };
 
 struct Node {
 	volatile NodeState state_;
 	volatile diff_ptr<Node> next_;
 };
 
 struct UpdateID : public Node {
 	::uint32_t list_id_;
 };
 
 #define DUMMY_DATA_SIZE 512
 
 struct Data : public Node {
 	volatile long tv_sec_;
 	volatile long tv_usec_;
 
 	char dummy_data_[DUMMY_DATA_SIZE];
 };
 
 struct Channel {
 	diff_ptr<Node> head_;
 	volatile diff_ptr<Node> tail_;
 	volatile std::size_t size_;
 	::uint32_t id_;
 };
 
 #endif
 
 
 // lump/child/main.C
 
 #include <iostream>
 #include <vector>
 #include <string>
 #include <set>
 
 #include <boost/thread.hpp>
 #include <boost/tokenizer.hpp>
 
 #include <boost/interprocess/sync/scoped_lock.hpp>
 #include <boost/interprocess/sync/named_mutex.hpp>
 #include <boost/interprocess/sync/named_condition.hpp>
 #include <boost/interprocess/allocators/allocator.hpp>
 #include <boost/interprocess/containers/vector.hpp>
 #include <boost/interprocess/managed_shared_memory.hpp>
 
 #include <lump/list.H>
 
 using namespace boost;
 using namespace boost::interprocess;
 using namespace std;
 
 managed_shared_memory segment(open_only, "SharedMemoryArea");
 
 typedef boost::interprocess::allocator< diff_ptr<Channel>, managed_shared_memory::segment_manager> ShmemAllocator;
 typedef boost::interprocess::vector< diff_ptr<Channel>, ShmemAllocator > ChannelList;
 
 ChannelList& channel_list = *(segment.find<ChannelList>("channel_list").first);
 Channel& update_channel = *(segment.find<Channel>("update_channel").first);
 
 named_mutex update_mutex(open_only, "update_mutex");
 named_condition update_condition(open_only, "update_condition");
 
 void log_latency(long send_tv_sec, long send_tv_usec)
 {
 	timeval arrive_timestamp;
 	gettimeofday(&arrive_timestamp, NULL);
 
 	if(send_tv_sec == 0)
 		std::cerr << "Send timestamp was 0... something is wrong." << std::endl; // never prints
 	if(arrive_timestamp.tv_sec == 0)
 		std::cerr << "Arrive timestamp was 0... something is wrong." << std::endl; // never prints
 
 	hrtime_t send_time = hrtime_t(send_tv_sec) * hrtime_t(1e9) + hrtime_t(send_tv_usec) * hrtime_t(1e3);
 	hrtime_t arrive_time = hrtime_t(arrive_timestamp.tv_sec) * hrtime_t(1e9) + hrtime_t(arrive_timestamp.tv_usec) * hrtime_t(1e3);
 	hrtime_t latency = arrive_time - send_time;
 
 	std::cout << "Latency: " << double(latency) * 1.0e-6 << "ms " << double(latency) * 1.0e-3 << "us " << latency << "ns " << std::endl;
 }
 
 void method_one()
 {
 	std::vector<Data*> channel_cursors;
 	for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
 	{
 		Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
 
 		if(current_item->state_ == HEAD)
 			current_item = static_cast<Data*>(current_item->next_.get(segment));
 
 		channel_cursors.push_back(current_item);
 	}
 
 	while(true)
 	{
 		for(std::size_t i = 0; i < channel_list.size(); ++i)
 		{
 			// {
 			// 	scoped_lock<named_mutex> update_lock(update_mutex);
 			// 	update_condition.wait(update_lock);
 			// }
 
 			Data* current_item = channel_cursors[i];
 
 			ACQUIRE_MEMORY_BARRIER;
 			if(current_item->state_ == NOT_FILLED_YET) {
 				continue;
 			}
 
 			log_latency(current_item->tv_sec_, current_item->tv_usec_);
 
 			channel_cursors[i] = static_cast<Data*>(current_item->next_.get(segment));
 		}
 	}
 }
 
 void method_two()
 {
 	std::vector<Data*> channel_cursors;
 	for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
 	{
 		Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
 
 		if(current_item->state_ == HEAD)
 			current_item = static_cast<Data*>(current_item->next_.get(segment));
 
 		channel_cursors.push_back(current_item);
 	}
 
 	UpdateID* update_cursor = static_cast<UpdateID*>(update_channel.tail_.get(segment));
 
 	while(true)
 	{
 		// {
 		// 	scoped_lock<named_mutex> update_lock(update_mutex);
 		// 	update_condition.wait(update_lock);
 		// }
 
 		if(update_cursor->state_ == NOT_FILLED_YET) {
 			continue;
 		}
 
 		::uint32_t update_id = update_cursor->list_id_;
 
 		Data* current_item = channel_cursors[update_id];
 
 		if(current_item->state_ == NOT_FILLED_YET) {
 			std::cerr << "This should never print." << std::endl; // it doesn't
 			continue;
 		}
 
 		log_latency(current_item->tv_sec_, current_item->tv_usec_);
 
 		channel_cursors[update_id] = static_cast<Data*>(current_item->next_.get(segment));
 		update_cursor = static_cast<UpdateID*>(update_cursor->next_.get(segment));
 	}
 }
 
 void method_three()
 {
 	std::vector<Data*> channel_cursors;
 	for(ChannelList::iterator i = channel_list.begin(); i != channel_list.end(); ++i)
 	{
 		Data* current_item = static_cast<Data*>(i->get(segment)->tail_.get(segment));
 
 		if(current_item->state_ == HEAD)
 			current_item = static_cast<Data*>(current_item->next_.get(segment));
 
 		channel_cursors.push_back(current_item);
 	}
 
 	UpdateID* update_cursor = static_cast<UpdateID*>(update_channel.tail_.get(segment));
 
 	while(true)
 	{
 		bool found_an_update = false;
 		for(std::size_t i = 0; i < channel_list.size(); ++i)
 		{
 			std::size_t idx = i;
 
 			ACQUIRE_MEMORY_BARRIER;
 			if(update_cursor->state_ != NOT_FILLED_YET) {
 				//std::cerr << "Found via update" << std::endl;
 				i--;
 				idx = update_cursor->list_id_;
 				update_cursor = static_cast<UpdateID*>(update_cursor->next_.get(segment));
 			}
 
 			Data* current_item = channel_cursors[idx];
 
 			ACQUIRE_MEMORY_BARRIER;
 			if(current_item->state_ == NOT_FILLED_YET) {
 				continue;
 			}
 
 			found_an_update = true;
 
 			log_latency(current_item->tv_sec_, current_item->tv_usec_);
 			channel_cursors[idx] = static_cast<Data*>(current_item->next_.get(segment));
 		}
 
 		if(!found_an_update)
 		{
 			std::cerr << "Sleeping" << std::endl;
 			scoped_lock<named_mutex> update_lock(update_mutex);
 			update_condition.wait(update_lock);
 		}
 	}
 }
 
 int main(int argc, char** argv)
 {
 	using ::int32_t;
 	using ::uint32_t;
 
 	if(argc == 1)
 	{
 		std::cout << "Using 1st method" << std::endl;
 		method_one();
 	}
 	else if(argc == 2)
 	{
 		std::cout << "Using 2nd method" << std::endl;
 		method_two();
 	}
 	else if(argc == 3)
 	{
 		std::cout << "Using 3rd method" << std::endl;
 		method_three();
 	}
 
 	return 0;
 }
 
 
 // lump/server/main.C
 
 #include <vector>
 #include <map>
 #include <cstring>
 #include <sstream>
 #include <iostream>
 
 #include <boost/interprocess/sync/scoped_lock.hpp>
 #include <boost/interprocess/sync/named_mutex.hpp>
 #include <boost/interprocess/sync/named_condition.hpp>
 #include <boost/interprocess/allocators/allocator.hpp>
 #include <boost/interprocess/containers/vector.hpp>
 #include <boost/interprocess/managed_shared_memory.hpp>
 
 #include <lump/list.H>
 
 using namespace boost;
 using namespace boost::interprocess;
 using namespace std;
 
 struct shm_remove
 {
 	shm_remove()
 	{
 		shared_memory_object::remove("SharedMemoryArea");
 		named_mutex::remove("update_mutex");
 		named_condition::remove("update_condition");
 	}
 
 	~shm_remove()
 	{
 		shared_memory_object::remove("SharedMemoryArea");
 		named_mutex::remove("update_mutex");
 		named_condition::remove("update_condition");
 	}
 } remover;
 
 static int CACHE_SIZE = 512 * 1024 * 1024;
 managed_shared_memory segment(create_only, "SharedMemoryArea", CACHE_SIZE);
 
 typedef boost::interprocess::allocator< diff_ptr<Channel>, managed_shared_memory::segment_manager> ShmemAllocator;
 typedef boost::interprocess::vector< diff_ptr<Channel>, ShmemAllocator > ChannelList;
 ShmemAllocator alloc_instance(segment.get_segment_manager());
 
 ChannelList& channel_list = *(segment.construct<ChannelList>("channel_list")(alloc_instance));
 Channel& update_channel = *(segment.construct<Channel>("update_channel")());
 
 named_mutex update_mutex(create_only, "update_mutex");
 named_condition update_condition(create_only, "update_condition");
 
 timeval receive_timestamp;
 
 static int NEXT_ID_TO_USE = 0;
 static int CHANNEL_TO_USE = 0;
 
 template<class T>
 void init_channel(Channel& channel)
 {
 	T* new_head = static_cast<T*>(segment.allocate(sizeof(T)));
 	new_head->state_ = HEAD;
 
 	T* new_empty = static_cast<T*>(segment.allocate(sizeof(T)));
 	new_empty->state_ = NOT_FILLED_YET;
 	new_head->next_.set(new_empty, segment);
 
 	channel.id_ = NEXT_ID_TO_USE++;
 
 	RELEASE_MEMORY_BARRIER;
 	channel.head_.set(new_head, segment);
 	channel.tail_.set(new_head, segment);
 }
 
 void update(Channel& channel, char *new_data)
 {
 	gettimeofday(&receive_timestamp, NULL);
 
 	Data* new_empty_node = static_cast<Data*>(segment.allocate(sizeof(Data)));
 	new_empty_node->state_ = NOT_FILLED_YET;
 
 	Data* node_to_fill = static_cast<Data*>(channel.tail_.get(segment)->next_.get(segment));
 
 	node_to_fill->next_.set(new_empty_node, segment);
 
 	node_to_fill->tv_sec_ = receive_timestamp.tv_sec;
 	node_to_fill->tv_usec_ = receive_timestamp.tv_usec;
 
 	strncpy(node_to_fill->dummy_data_, new_data, DUMMY_DATA_SIZE);
 
 	__sync_fetch_and_add(&(channel.size_), 1);
 
 	//
 	// Separate out UpdateID work
 	//
 
 	UpdateID* new_empty_update = static_cast<UpdateID*>(segment.allocate(sizeof(UpdateID)));
 	new_empty_update->state_ = NOT_FILLED_YET;
 
 	UpdateID* update_node_to_fill = static_cast<UpdateID*>(update_channel.tail_.get(segment)->next_.get(segment));
 
 	update_node_to_fill->next_.set(new_empty_update, segment);
 
 	update_node_to_fill->list_id_ = channel.id_;
 
 	__sync_fetch_and_add(&(update_channel.size_), 1);
 
 	RELEASE_MEMORY_BARRIER;
 	node_to_fill->state_ = FILLED;
 	RELEASE_MEMORY_BARRIER;
 	channel.tail_.set(node_to_fill, segment);
 
 	RELEASE_MEMORY_BARRIER;
 	update_node_to_fill->state_ = FILLED;
 	RELEASE_MEMORY_BARRIER;
 	update_channel.tail_.set(update_node_to_fill, segment);
 
 	//scoped_lock<named_mutex> update_lock(update_mutex);
 	update_condition.notify_all();
 }
 
 int main(int argc, char** argv)
 {
 	for(int i = 0; i < 250000; ++i)
 	{
 		Channel* data_channel = static_cast<Channel*>(segment.allocate(sizeof(Channel)));
 		init_channel<Data>(*data_channel);
 		diff_ptr<Channel> ptr(data_channel, segment);
 		channel_list.push_back(ptr);
 	}
 
 	init_channel<UpdateID>(update_channel);
 
 	std::cout << "Cache Size: " << CACHE_SIZE << std::endl;
 
 	srand ( time(NULL) );
 	while(true)
 	{
 		char dummy_data_to_transmit[DUMMY_DATA_SIZE];
 		for(int i = 0; i < DUMMY_DATA_SIZE - 1; ++i)
 		{
 			dummy_data_to_transmit[i] = rand() % (90 - 65) + 65;
 		}
 		dummy_data_to_transmit[DUMMY_DATA_SIZE - 1] = '\0';
 
 		//std::cout << "Update: " << dummy_data_to_transmit << std::endl;
 		Channel& channel = *(channel_list[rand() % channel_list.size()].get(segment));
 		update(channel, &(dummy_data_to_transmit[0]));
 
 		timespec interval;
 		interval.tv_sec = 0;
 		//interval.tv_sec = rand() % 2;
 		//interval.tv_nsec = rand() % 5000;
 		interval.tv_nsec = rand() % 5000000;
 		nanosleep(&interval, NULL);
 	}
 
 	return 0;
 }