// 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;
}