#include <cassert>
#include <cstdio>
#include <iostream>
#include <vector>
#include <valarray>

#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <unistd.h>

#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>

const std::size_t message_size = /**/ 1*1024 /*/ 64*1024 /**/;
const int port = 9999;
const std::size_t message_iterations = /** 100 /*/ 100000 /**/;
const std::size_t test_iterations = 1 + 4 + 1;

namespace detail_test
{
struct timed_scope
{
    boost::xtime t0;
    boost::xtime t1;
    std::size_t n;

    std::vector<double> & results;

    inline timed_scope(std::vector<double> & r, std::size_t iterations = 1)
        : results(r), n(iterations)
        {
            boost::xtime_get(&t0,boost::TIME_UTC);
        }

    inline ~timed_scope()
        {
            boost::xtime_get(&t1,boost::TIME_UTC);
            double t = double(t1.sec)+double(t1.nsec)/double(1000000000);
            t -= double(t0.sec)+double(t0.nsec)/double(1000000000);
            std::cerr << "### TIME"
                      << ": total = " << t
                      << "; iterations = " << n
                      << "; iteration = " << t/double(n)
                      << "; iterations/second = " << double(n)/t
                      << '\n';
            results.push_back(double(n)/t);
        }
};

template <typename out>
out & result_summary(out & o, const std::vector<double> & results)
{
    assert (results.size() > 2);
    std::valarray<double> r(&results[1],results.size()-2);
    o << r.sum()/r.size();
    return o;
}

void sleep_for_secs(int n)
{
    boost::xtime t;
    boost::xtime_get(&t,boost::TIME_UTC);
    t.sec += n;
    boost::thread::sleep(t);
}

} // namespace detail_test

int open_socket ()
{
    int socket = ::socket (AF_INET, SOCK_DGRAM, 0);

    if (socket < 0)
        {
        perror ("socket");
        exit (EXIT_FAILURE);
        }
    
    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl (INADDR_ANY);
    addr.sin_port = htons (port);

    if (bind (socket, (struct sockaddr*) &addr, sizeof (addr)) != 0)
        {
        perror ("bind");
        exit (EXIT_FAILURE);
        }

    return socket;
}

struct async_server
{
    char io_buffer[message_size];
    int socket;
    std::auto_ptr<detail_test::timed_scope> timer;
    std::size_t message_count;
    std::size_t message_recount;
    std::size_t test_count;
    std::auto_ptr<boost::thread> runner;
    std::vector<double> results;
    volatile bool running;
    fd_set fds;

    async_server()
        : message_count(0)
        , message_recount(0)
        , test_count(0)
        , running (false)
        {
            socket = open_socket ();
            FD_ZERO (&fds);
            FD_SET (socket, &fds);
        }

    void handle_ready ()
        {
            int len = recvfrom (socket, (void*) io_buffer, message_size, 0,
                                0, 0);

            if (len <= 0)
                {
                perror ("recvfrom");
                exit (EXIT_FAILURE);
                }

            if (++message_count == this->timer->n)
                {
                ++test_count;
                this->clear_timer();
                message_count = 0;
                this->reset_timer(message_recount);
                }
        }

    void run ()
        {
            while (running)
                {
                fd_set read_fds = fds;
                struct timeval tv = { 1, 0 };

                int status = select (1 + socket, &read_fds, 0, 0, &tv);

                if (status < 0)
                    {
                    perror ("select");
                    exit (EXIT_FAILURE);
                    }
                else if (status > 0)
                    handle_ready ();
                }
        }

    
    void start()
        {
            this->running = true;
            this->runner.reset(new boost::thread(
                                   boost::bind(&async_server::run,
                                               this)));
        }

    void stop()
        {
            this->running = false;
            this->runner->join();
            close (socket);
            this->clear_timer();
        }

    void reset_timer(std::size_t i = 1)
        {
            this->message_recount = i;
            this->timer.reset(new detail_test::timed_scope(this->results,i));
        }

    void clear_timer()
        {
            this->timer.reset();
        }
};

struct sync_server
{
    char io_buffer[message_size];
    int socket;
    std::auto_ptr<detail_test::timed_scope> timer;
    std::size_t message_count;
    std::size_t message_recount;
    std::auto_ptr<boost::thread> runner;
    std::vector<double> results;
    volatile bool running;

    sync_server()
        : message_count(0)
        , message_recount(0)
        , running(false)
        {
            socket = open_socket ();
        }

    void handle_receive_from(size_t /*bytes_transferred*/)
        {
            if (++message_count == this->timer->n)
                {
                this->clear_timer();
                message_count = 0;
                this->reset_timer(message_recount);
                }
        }

    void start()
        {
            this->runner.reset(new boost::thread(
                                   boost::bind(&sync_server::run,this) ));
        }

    void run()
        {
            this->running = true;

            while (this->running)
                {
                int len = recvfrom (socket, (void*) io_buffer, message_size,
                                    0, 0, 0);

                if (len <= 0)
                    {
                    perror ("recvfrom");
                    exit (EXIT_FAILURE);
                    }
                else if (len == 1) // 1-byte packet = time to go
                    break;

                this->handle_receive_from (len);
                }

            this->running = false;
        }

    void stop()
        {
            this->running = false;
            close (socket);
            this->runner->join();
            this->clear_timer();
        }

    void reset_timer(std::size_t i = 1)
        {
            this->message_recount = i;
            this->timer.reset(new detail_test::timed_scope(this->results,i));
        }

    void clear_timer()
        {
            this->timer.reset();
        }
};

struct sync_client
{
    char io_buffer[message_size];
    int socket;
    struct sockaddr_in addr;

    sync_client()
        {
            memset (&addr, 0, sizeof (addr));
            addr.sin_family = AF_INET;
            addr.sin_port = htons (port);
            addr.sin_addr.s_addr = htonl (0x7f000001); // 127.0.0.1

            socket = ::socket (AF_INET, SOCK_DGRAM, 0);

            if (socket < 0)
                {
                perror ("socket");
                exit (EXIT_FAILURE);
                }
        }

    void send(size_t len)
        {
            int result = sendto (socket, io_buffer, len, 0,
                                 (sockaddr*) &addr, sizeof (addr));

            if (result < 0)
                {
                perror ("sendto");
                exit (EXIT_FAILURE);
                }
        }
};

int main()
{
    sync_client c0;

    {
    async_server s0;
    s0.start();
    detail_test::sleep_for_secs(2);
    std::cerr << "--- ASYNC...\n";
    s0.reset_timer(message_iterations);
    for (std::size_t m = 0; m < message_iterations*test_iterations; ++m)
        {
        c0.send(message_size);
        }
    s0.stop();
    detail_test::result_summary(
        std::cerr << "--  ...ASYNC: average iterations/second = ",
        s0.results) << "\n";
    }

    detail_test::sleep_for_secs(2);

    {
    sync_server s0;
    s0.start();
    detail_test::sleep_for_secs(2);
    std::cerr << "--- SYNC...\n";
    s0.reset_timer(message_iterations);
    for (std::size_t m = 0; m < message_iterations*test_iterations; ++m)
        {
        c0.send(message_size);
        }
    c0.send (1);
    s0.stop();
    detail_test::result_summary(
        std::cerr << "--  ...SYNC: average iterations/second = ",
        s0.results) << "\n";
    }
    return 1;
}