/*

 * Copyright (c) 2001-2020 Mellanox Technologies, Ltd. All rights reserved.

 *

 * This software is available to you under a choice of one of two

 * licenses.  You may choose to be licensed under the terms of the GNU

 * General Public License (GPL) Version 2, available from the file

 * COPYING in the main directory of this source tree, or the

 * BSD license below:

 *

 *     Redistribution and use in source and binary forms, with or

 *     without modification, are permitted provided that the following

 *     conditions are met:

 *

 *      - Redistributions of source code must retain the above

 *        copyright notice, this list of conditions and the following

 *        disclaimer.

 *

 *      - Redistributions in binary form must reproduce the above

 *        copyright notice, this list of conditions and the following

 *        disclaimer in the documentation and/or other materials

 *        provided with the distribution.

 *

 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND

 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS

 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN

 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN

 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE

 * SOFTWARE.

 */

#include "client.h"

#include "options.h"

#include <sys/socket.h>

#include <sys/epoll.h>

#include <netdb.h>

#include <cstring>

#include <stdexcept>

#include <boost/foreach.hpp>

#include <boost/make_shared.hpp>

#include <boost/thread.hpp>

#include <boost/format.hpp>

#include <boost/ref.hpp>

#include <algorithm>

#include <iostream>

client::client(const options& opts) {

    for (unsigned id = 0; id < opts.num_threads(); ++id) {

        m_connections.push_back(boost::make_shared<connection>(id,

                                                               opts.server(),

                                                               opts.port(),

                                                               opts.packet_rate()));

    }

}

void client::run() {

    boost::thread_group tg;

    BOOST_FOREACH(const connection_ptr& conn, m_connections) {

        tg.create_thread(boost::ref(*conn.get()));

    }

    tg.join_all();

}

client::connection::connection(unsigned id, const std::string& server,

                               unsigned port, size_t packet_rate) :

        m_id(id),

        m_packet_rate(packet_rate),

        m_psn(0)

{

    struct hostent *he = gethostbyname(server.c_str());

    if (!he) {

        throw std::runtime_error(std::string("failed to resolve ") + server);

    }

    m_dest_addr.sin_family = he->h_addrtype;

    m_dest_addr.sin_port = htons(port);

    if (he->h_length != sizeof(m_dest_addr.sin_addr)) {

        throw std::runtime_error("invalid address length");

    }

    memcpy(&m_dest_addr.sin_addr, he->h_addr_list[0], he->h_length);

    memset(m_dest_addr.sin_zero, 0, sizeof(m_dest_addr.sin_zero));

    m_sockfd = socket(AF_INET, SOCK_DGRAM, 0);

    if (m_sockfd < 0) {

        throw std::runtime_error("failed to create socket");

    }

    m_epfd = epoll_create(1);

    if (m_epfd < 0) {

        throw std::runtime_error("failed to create epfd");

    }

    // Add the socket to the main epoll set

    struct epoll_event evt;

    evt.events   = EPOLLIN|EPOLLOUT;

    evt.data.fd  = m_sockfd;

    int ret = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_sockfd, &evt);

    if (ret < 0) {

        throw std::runtime_error("failed to add socket fd to epoll set");

    }

}

client::connection::~connection() {

    close(m_epfd);

    close(m_sockfd);

}

std::string client::connection::destination() const {

    char buf[256] = {0};

    inet_ntop(m_dest_addr.sin_family, &m_dest_addr.sin_addr, buf, sizeof(buf) - 1);

    return (boost::format("%s:%d") % buf % ntohs(m_dest_addr.sin_port)).str();

}

void client::connection::operator()() {

    const size_t maxevents = 2;

    struct epoll_event events[maxevents];

    unsigned next_worker = 0;

    m_start_time = vtime::current();

    m_recv_count = 0;

    m_send_count = 0;

    m_rtt_sum = 0;

    size_t sent_prev = 0;

    size_t recv_prev = 0;

    vtime::time_t time_prev = m_start_time;

    vtime::time_t prev_rtt_sum = 0;

    vtime::time_t packet_interval = vtime::time_from_sec(1.0 / m_packet_rate);

    vtime::time_t last_send_time = m_start_time;

    size_t print_rate = std::min(200000ul, m_packet_rate);

    std::cout << "connection " << m_id << ": sending to " << destination() << std::endl;

    do {

        int nevents = epoll_wait(m_epfd, events, maxevents, -1);

        if (nevents < 0) {

            throw std::runtime_error("epoll_wait failed");

        }

        vtime:time_t current_time = vtime::current();

        for (int i = 0; i < nevents; ++i) {

            if (events[i].data.fd == m_sockfd) {

                if (events[i].events & EPOLLIN) {

                    int nrecvd = recvfrom(m_sockfd, &m_recvbuf, sizeof(m_recvbuf),

                                          0, NULL, NULL);

                    if (nrecvd != sizeof(m_sendbuf)) {

                        throw std::runtime_error("recvfrom failed");

                    }

                    m_rtt_sum += (current_time - m_recvbuf.send_time);

                    ++m_recv_count;

                }

                if (events[i].events & EPOLLOUT) {

                    if (current_time >= last_send_time + packet_interval) {

                        // TODO maintain packet rate

                        m_sendbuf.psn = m_psn++;

                        m_sendbuf.send_time = current_time;

                        int nsent = sendto(m_sockfd, &m_sendbuf, sizeof(m_sendbuf), 0,

                                           reinterpret_cast<struct sockaddr*>(&m_dest_addr),

                                           sizeof(m_dest_addr));

                        if (nsent != sizeof(m_sendbuf)) {

                            throw std::runtime_error("sendto failed");

                        }

                        ++m_send_count;

                        /*last_send_time += packet_interval;*/

                        last_send_time =

                                        (current_time + last_send_time + packet_interval) / 2;

                    }

                }

            }

        }

        if (m_send_count - sent_prev >= print_rate) {

            double rtt = (m_recv_count) > 0 ?

                            vtime::time_to_sec(

                                            (m_rtt_sum - prev_rtt_sum) * 1000000.0 /

                                            (m_recv_count - recv_prev)) :

                            0;

            double packet_rate = (m_send_count - sent_prev) /

                            vtime::time_to_sec(current_time - time_prev);

            double recv_ratio = (m_recv_count - recv_prev) /

                            static_cast<double>(m_send_count - sent_prev);

            printf("sent: %Zu rate: %7.2f recvd: %Zu (%5.2f%%) rtt: %5.2f\n",

                   m_send_count, packet_rate, m_recv_count, recv_ratio * 100.0,

                   rtt);

            sent_prev = m_send_count;

            recv_prev = m_recv_count;

            time_prev = current_time;

            prev_rtt_sum = m_rtt_sum;

        }

    } while (1);

}