// Copyright (C) 2012  Davis E. King (davis@dlib.net)
// License: Boost Software License   See LICENSE.txt for the full license.


#include <dlib/bsp.h>
#include <dlib/threads.h>
#include <dlib/pipe.h>
#include <dlib/matrix.h>

#include "tester.h"

namespace  
{

    using namespace test;
    using namespace dlib;
    using namespace std;

    logger dlog("test.bsp");


    template <typename funct>
    struct callfunct_helper
    {
        callfunct_helper (
            funct f_,
            int port_,
            bool& error_occurred_
        ) :f(f_), port(port_), error_occurred(error_occurred_) {}

        funct f;
        int port;
        bool& error_occurred;

        void operator() (
        ) const
        {
            try
            {
                bsp_listen(port, f);
            }
            catch (exception& e)
            {
                dlog << LERROR << "error calling bsp_listen(): " << e.what();
                error_occurred = true;
            }
        }
    };

    template <typename funct>
    callfunct_helper<funct> callfunct(funct f, int port, bool& error_occurred)
    {
        return callfunct_helper<funct>(f,port,error_occurred);

    }

// ----------------------------------------------------------------------------------------

    template <typename funct>
    struct callfunct_helper_pn
    {
        callfunct_helper_pn (
            funct f_,
            int port_,
            bool& error_occurred_,
            dlib::pipe<unsigned short>& port_pipe_
        ) :f(f_), port(port_), error_occurred(error_occurred_), port_pipe(port_pipe_) {}

        funct f;
        int port;
        bool& error_occurred;
        dlib::pipe<unsigned short>& port_pipe;

        struct helper
        {
            helper (
                dlib::pipe<unsigned short>& port_pipe_
            ) : port_pipe(port_pipe_) {}

            dlib::pipe<unsigned short>& port_pipe;

            void operator() (unsigned short p) { port_pipe.enqueue(p); }
        };

        void operator() (
        ) const
        {
            try
            {
                bsp_listen_dynamic_port(port, helper(port_pipe), f);
            }
            catch (exception& e)
            {
                dlog << LERROR << "error calling bsp_listen_dynamic_port(): " << e.what();
                error_occurred = true;
            }
        }
    };

    template <typename funct>
    callfunct_helper_pn<funct> callfunct(funct f, int port, bool& error_occurred, dlib::pipe<unsigned short>& port_pipe)
    {
        return callfunct_helper_pn<funct>(f,port,error_occurred,port_pipe);
    }

// ----------------------------------------------------------------------------------------

    void sum_array_driver (
        bsp_context& obj,
        const std::vector<int>& v,
        int& result
    )
    {
        obj.broadcast(v);

        result = 0;
        int val;
        while(obj.try_receive(val))
            result += val;
    }

    void sum_array_other (
        bsp_context& obj
    )
    {
        std::vector<int> v;
        obj.receive(v);

        int sum = 0;
        for (unsigned long i = 0; i < v.size(); ++i)
            sum += v[i];

        obj.send(sum, 0);


    }


    void dotest1()
    {
        dlog << LINFO << "start dotest1()";
        print_spinner();
        bool error_occurred = false;
        {
            thread_function t1(callfunct(sum_array_other, 12345, error_occurred));
            thread_function t2(callfunct(sum_array_other, 12346, error_occurred));
            thread_function t3(callfunct(sum_array_other, 12347, error_occurred));
            std::vector<int> v;
            int true_value = 0;
            for (int i = 0; i < 10; ++i)
            {
                v.push_back(i);
                true_value += i;
            }

            // wait a little bit for the threads to start up
            dlib::sleep(200);

            try
            {
                int result;
                std::vector<network_address> hosts;
                hosts.push_back("127.0.0.1:12345");
                hosts.push_back("localhost:12346");
                hosts.push_back("127.0.0.1:12347");
                bsp_connect(hosts, sum_array_driver, dlib::ref(v), dlib::ref(result));

                dlog << LINFO << "result: "<< result;
                dlog << LINFO << "should be: "<< 3*true_value; 
                DLIB_TEST(result == 3*true_value);
            }
            catch (std::exception& e)
            {
                dlog << LERROR << "error during bsp_context: " << e.what();
                DLIB_TEST(false);
            }
        }
        DLIB_TEST(error_occurred == false);
    }

// ----------------------------------------------------------------------------------------

    template <unsigned long id>
    void test2_job(bsp_context& obj)
    {
        if (obj.node_id() == id)
            dlib::sleep(100);
    }

    template <unsigned long id>
    void dotest2()
    {
        dlog << LINFO << "start dotest2()";
        print_spinner();
        bool error_occurred = false;
        {
            thread_function t1(callfunct(test2_job<id>, 12345, error_occurred));
            thread_function t2(callfunct(test2_job<id>, 12346, error_occurred));
            thread_function t3(callfunct(test2_job<id>, 12347, error_occurred));

            // wait a little bit for the threads to start up
            dlib::sleep(200);

            try
            {
                std::vector<network_address> hosts;
                hosts.push_back("127.0.0.1:12345");
                hosts.push_back("127.0.0.1:12346");
                hosts.push_back("127.0.0.1:12347");
                bsp_connect(hosts, test2_job<id>);
            }
            catch (std::exception& e)
            {
                dlog << LERROR << "error during bsp_context: " << e.what();
                DLIB_TEST(false);
            }

        }
        DLIB_TEST(error_occurred == false);
    }

// ----------------------------------------------------------------------------------------

    void test3_job_driver(bsp_context& obj, int& result)
    {

        obj.broadcast(obj.node_id());

        int accum = 0;
        int temp = 0;
        while(obj.try_receive(temp))
            accum += temp;

        // send to node 1 so it can sum everything
        if (obj.node_id() != 1)
            obj.send(accum, 1);

        while(obj.try_receive(temp))
            accum += temp;

        // Now hop the accum values along the nodes until the value from node 1 gets to
        // node 0.
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);

        // this whole block is a noop since it doesn't end up doing anything.
        for (int k = 0; k < 100; ++k)
        {
            dlog << LINFO << "k: " << k;
            for (int i = 0; i < 4; ++i)
            {
                obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
                obj.receive(accum);
            }
        }


        dlog << LINFO << "TERMINATE";
        if (obj.node_id() == 0)
            result = accum;
    }


    void test3_job(bsp_context& obj)
    {
        int junk;
        test3_job_driver(obj, junk);
    }


    void dotest3()
    {
        dlog << LINFO << "start dotest3()";
        print_spinner();
        bool error_occurred = false;
        {
            dlib::pipe<unsigned short> ports(5);
            thread_function t1(callfunct(test3_job, 12345, error_occurred, ports));
            thread_function t2(callfunct(test3_job, 0, error_occurred, ports));
            thread_function t3(callfunct(test3_job, 12347, error_occurred, ports));


            try
            {
                std::vector<network_address> hosts;
                unsigned short port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                int result = 0;
                const int expected =  1+2+3 + 0+2+3 + 0+1+3 + 0+1+2;
                bsp_connect(hosts, test3_job_driver, dlib::ref(result));

                dlog << LINFO << "result: " << result;
                dlog << LINFO << "should be: " << expected;
                DLIB_TEST(result == expected);
            }
            catch (std::exception& e)
            {
                dlog << LERROR << "error during bsp_context: " << e.what();
                DLIB_TEST(false);
            }

        }
        DLIB_TEST(error_occurred == false);
    }

// ----------------------------------------------------------------------------------------

    void test4_job_driver(bsp_context& obj, int& result)
    {

        obj.broadcast(obj.node_id());

        int accum = 0;
        int temp = 0;
        while(obj.try_receive(temp))
            accum += temp;

        // send to node 1 so it can sum everything
        if (obj.node_id() != 1)
            obj.send(accum, 1);

        while(obj.try_receive(temp))
            accum += temp;

        // Now hop the accum values along the nodes until the value from node 1 gets to
        // node 0.
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);
        obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
        obj.receive(accum);

        // this whole block is a noop since it doesn't end up doing anything.
        for (int k = 0; k < 40; ++k)
        {
            dlog << LINFO << "k: " << k;
            for (int i = 0; i < 4; ++i)
            {
                obj.send(accum, (obj.node_id()+1)%obj.number_of_nodes());
                obj.receive(accum);

                obj.receive();
            }
        }


        dlog << LINFO << "TERMINATE";
        if (obj.node_id() == 0)
            result = accum;
    }


    void test4_job(bsp_context& obj)
    {
        int junk;
        test4_job_driver(obj, junk);
    }


    void dotest4()
    {
        dlog << LINFO << "start dotest4()";
        print_spinner();
        bool error_occurred = false;
        {
            dlib::pipe<unsigned short> ports(5);
            thread_function t1(callfunct(test4_job, 0, error_occurred, ports));
            thread_function t2(callfunct(test4_job, 0, error_occurred, ports));
            thread_function t3(callfunct(test4_job, 0, error_occurred, ports));


            try
            {
                std::vector<network_address> hosts;
                unsigned short port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                int result = 0;
                const int expected =  1+2+3 + 0+2+3 + 0+1+3 + 0+1+2;
                bsp_connect(hosts, test4_job_driver, dlib::ref(result));

                dlog << LINFO << "result: " << result;
                dlog << LINFO << "should be: " << expected;
                DLIB_TEST(result == expected);
            }
            catch (std::exception& e)
            {
                dlog << LERROR << "error during bsp_context: " << e.what();
                DLIB_TEST(false);
            }

        }
        DLIB_TEST(error_occurred == false);
    }

// ----------------------------------------------------------------------------------------

    void test5_job(
        bsp_context& ,
        int& val
    )
    {
        val = 25;
    }

    void dotest5()
    {
        dlog << LINFO << "start dotest5()";
        print_spinner();
        std::vector<network_address> hosts;
        int val = 0;
        bsp_connect(hosts, test5_job, dlib::ref(val));
        DLIB_TEST(val == 25);
    }

// ----------------------------------------------------------------------------------------

    double f ( double x)
    {
        return std::pow(x-2.0, 2.0);
    }


    void bsp_job_node_0 (
        bsp_context& context,
        double& min_value,
        double& optimal_x
    )
    {
        double left = -100;
        double right = 100;

        min_value = std::numeric_limits<double>::infinity();
        double interval_width = std::abs(right-left);

        // This is doing a BSP based grid search for the minimum of f().  Here we 
        // do 100 iterations where we keep shrinking the grid size.
        for (int i = 0; i < 100; ++i)
        {
            context.broadcast(left);
            context.broadcast(right);

            for (unsigned int k = 1; k < context.number_of_nodes(); ++k)
            {
                std::pair<double,double> val;
                context.receive(val);
                if (val.second < min_value)
                {
                    min_value = val.second;
                    optimal_x = val.first;
                }
            }

            interval_width *= 0.5;
            left  = optimal_x - interval_width/2;
            right = optimal_x + interval_width/2;
        }
    }


    void bsp_job_other_nodes (
        bsp_context& context
    )
    {
        double left, right;
        while (context.try_receive(left))
        {
            context.receive(right);

            const double l = (context.node_id()-1)/(context.number_of_nodes()-1.0);
            const double r = context.node_id()    /(context.number_of_nodes()-1.0);

            const double width = right-left;
            matrix<double> values_to_check = linspace(left +l*width, left + r*width, 100);

            double best_x;
            double best_val = std::numeric_limits<double>::infinity();
            for (long j = 0; j < values_to_check.size(); ++j)
            {
                double temp = f(values_to_check(j));
                if (temp < best_val)
                {
                    best_val = temp;
                    best_x = values_to_check(j);
                }
            }

            context.send(make_pair(best_x, best_val), 0);
        }
    }

    void dotest6()
    {
        dlog << LINFO << "start dotest6()";
        print_spinner();
        bool error_occurred = false;
        {
            dlib::pipe<unsigned short> ports(5);
            thread_function t1(callfunct(bsp_job_other_nodes, 0, error_occurred, ports));
            thread_function t2(callfunct(bsp_job_other_nodes, 0, error_occurred, ports));
            thread_function t3(callfunct(bsp_job_other_nodes, 0, error_occurred, ports));


            try
            {
                std::vector<network_address> hosts;
                unsigned short port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                ports.dequeue(port); hosts.push_back(network_address("127.0.0.1",port)); dlog << LINFO << "PORT: " << port;
                double min_value = 10, optimal_x = 0;
                bsp_connect(hosts, bsp_job_node_0, dlib::ref(min_value), dlib::ref(optimal_x));

                dlog << LINFO << "min_value: " << min_value;
                dlog << LINFO << "optimal_x: " << optimal_x;
                DLIB_TEST(std::abs(min_value - 0) < 1e-14);
                DLIB_TEST(std::abs(optimal_x - 2) < 1e-14);
            }
            catch (std::exception& e)
            {
                dlog << LERROR << "error during bsp_context: " << e.what();
                DLIB_TEST(false);
            }

        }
        DLIB_TEST(error_occurred == false);
    }
// ----------------------------------------------------------------------------------------

    class bsp_tester : public tester
    {

    public:
        bsp_tester (
        ) :
            tester ("test_bsp",
                    "Runs tests on the BSP components.")
        {}

        void perform_test (
        )
        {
            for (int i = 0; i < 3; ++i)
            {
                dotest1();
                dotest2<0>();
                dotest2<1>();
                dotest2<2>();
                dotest3();
                dotest4();
                dotest5();
                dotest6();
            }
        }
    } a;

}

