// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)

#include <unistd.h>
#include <assert.h>

#include "addr_util.h"
#include "messenger.h"

void osd_messenger_t::cancel_osd_ops(osd_client_t *cl)
{
    std::vector<osd_op_t*> cancel_ops;
    cancel_ops.resize(cl->sent_ops.size());
    int i = 0;
    for (auto p: cl->sent_ops)
    {
        cancel_ops[i++] = p.second;
    }
    cl->sent_ops.clear();
    cl->outbox.clear();
    for (auto op: cancel_ops)
    {
        cancel_op(op);
    }
}

void osd_messenger_t::cancel_op(osd_op_t *op)
{
    if (op->op_type == OSD_OP_OUT)
    {
        op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
        op->reply.hdr.id = op->req.hdr.id;
        op->reply.hdr.opcode = op->req.hdr.opcode;
        op->reply.hdr.retval = -EPIPE;
        // Copy lambda to be unaffected by `delete op`
        std::function<void(osd_op_t*)>(op->callback)(op);
    }
    else
    {
        // This function is only called in stop_client(), so it's fine to destroy the operation
        delete op;
    }
}

void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
{
    assert(peer_fd != 0);
    auto it = clients.find(peer_fd);
    if (it == clients.end())
    {
        return;
    }
    osd_client_t *cl = it->second;
    if (cl->peer_state == PEER_CONNECTING && !force || cl->peer_state == PEER_STOPPED)
    {
        return;
    }
    if (log_level > 0)
    {
        if (cl->osd_num)
        {
            fprintf(stderr, "[OSD %lu] Stopping client %d (OSD %speer %lu)\n",
                osd_num, peer_fd, cl->meta_connection_fd >= 0 ? " data" : "", cl->osd_num);
        }
        else
        {
            fprintf(stderr, "[OSD %lu] Stopping client %d (regular client)\n", osd_num, peer_fd);
        }
    }
    // First set state to STOPPED so another stop_client() call doesn't try to free it again
    cl->refs++;
    cl->peer_state = PEER_STOPPED;
    if (cl->osd_num && cl->meta_connection_fd < 0)
    {
        // ...and forget OSD peer
        osd_peer_fds.erase(cl->osd_num);
    }
#ifndef __MOCK__
    // Then remove FD from the eventloop so we don't accidentally read something
    tfd->set_fd_handler(peer_fd, false, NULL);
    if (cl->connect_timeout_id >= 0)
    {
        tfd->clear_timer(cl->connect_timeout_id);
        cl->connect_timeout_id = -1;
    }
    for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
    {
        if (*rit == peer_fd)
        {
            read_ready_clients.erase(rit);
            break;
        }
    }
    for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
    {
        if (*wit == peer_fd)
        {
            write_ready_clients.erase(wit);
            break;
        }
    }
#endif
    if (cl->osd_num)
    {
        if (cl->meta_connection_fd < 0)
        {
            // Then repeer PGs because cancel_op() callbacks can try to perform
            // some actions and we need correct PG states to not do something silly
            repeer_pgs(cl->osd_num);
        }
        else
        {
            // FIXME Try to re-establish data connection
            // Only when the connection is outbound, but here it's always outbound
        }
    }
    // Then cancel all operations
    if (cl->read_op)
    {
        if (!cl->read_op->callback)
        {
            delete cl->read_op;
        }
        cl->read_op = NULL;
    }
    if (cl->osd_num)
    {
        // Cancel outbound operations
        cancel_osd_ops(cl);
    }
#ifndef __MOCK__
    // And close the FD only when everything is done
    // ...because peer_fd number can get reused after close()
    close(peer_fd);
#ifdef WITH_RDMA
    if (cl->rdma_conn)
    {
        delete cl->rdma_conn;
    }
#endif
    clients_by_addr.erase(addr_to_string(cl->peer_addr));
#endif
    // Find the item again because it can be invalidated at this point
    it = clients.find(peer_fd);
    if (it != clients.end())
    {
        clients.erase(it);
    }
    // Break metadata/data connection pair
    if (cl->data_connection_fd >= 0)
    {
        // No sense to keep data connection when metadata connection is stopped
        auto dc_it = clients.find(cl->data_connection_fd);
        cl->data_connection_fd = -1;
        if (dc_it != clients.end() && dc_it->second->meta_connection_fd == cl->peer_fd)
        {
            stop_client(dc_it->second->peer_fd);
        }
    }
    break_data_client_pair(cl);
    // Refcount and delete
    cl->refs--;
    if (cl->refs <= 0 || force_delete)
    {
        delete cl;
    }
}

void osd_messenger_t::break_data_client_pair(osd_client_t *cl)
{
    if (cl->meta_connection_fd >= 0)
    {
        auto dc_it = clients.find(cl->meta_connection_fd);
        if (dc_it != clients.end() && dc_it->second->data_connection_fd == cl->peer_fd)
            dc_it->second->data_connection_fd = -1;
        cl->meta_connection_fd = -1;
    }
    if (cl->data_connection_fd >= 0)
    {
        auto dc_it = clients.find(cl->data_connection_fd);
        if (dc_it != clients.end() && dc_it->second->meta_connection_fd == cl->peer_fd)
            dc_it->second->meta_connection_fd = -1;
        cl->data_connection_fd = -1;
    }
}