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

#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/epoll.h>
#include <netinet/tcp.h>
#include <stdexcept>

#include "addr_util.h"
#include "messenger.h"
#ifdef WITH_RDMA
#include "msgr_rdma.h"
#endif

void osd_messenger_t::init()
{
#ifdef WITH_RDMACM
    if (use_rdmacm)
    {
        // RDMA-CM only requires the event channel. All the remaining work is done separately
        rdmacm_evch = rdma_create_event_channel();
        if (!rdmacm_evch)
        {
            // ENODEV means that the client doesn't have RDMA devices available
            if (errno != ENODEV || log_level > 0)
                fprintf(stderr, "Failed to initialize RDMA-CM event channel: %s (code %d)\n", strerror(errno), errno);
        }
        else
        {
            fcntl(rdmacm_evch->fd, F_SETFL, fcntl(rdmacm_evch->fd, F_GETFL, 0) | O_NONBLOCK);
            tfd->set_fd_handler(rdmacm_evch->fd, false, [this](int rdmacm_eventfd, int epoll_events)
            {
                handle_rdmacm_events();
            });
        }
    }
    else
#endif
#ifdef WITH_RDMA
    if (use_rdma)
    {
        rdma_contexts = msgr_rdma_context_t::create_all(
            osd_num && osd_cluster_network_masks.size() ? osd_cluster_network_masks : osd_network_masks,
            rdma_device != "" ? rdma_device.c_str() : NULL,
            rdma_port_num, rdma_gid_index, rdma_mtu, log_level
        );
        if (!rdma_contexts.size())
        {
            if (log_level > 0)
                fprintf(stderr, "[OSD %ju] Couldn't initialize RDMA, proceeding with TCP only\n", osd_num);
        }
        else
        {
            fprintf(stderr, "[OSD %ju] RDMA initialized successfully\n", osd_num);
            for (msgr_rdma_context_t* rdma_context: rdma_contexts)
            {
                fcntl(rdma_context->channel->fd, F_SETFL, fcntl(rdma_context->channel->fd, F_GETFL, 0) | O_NONBLOCK);
                tfd->set_fd_handler(rdma_context->channel->fd, false, [this, rdma_context](int notify_fd, int epoll_events)
                {
                    handle_rdma_events(rdma_context);
                });
                handle_rdma_events(rdma_context);
            }
        }
    }
#endif
    if (ringloop)
    {
        has_sendmsg_zc = ringloop->has_sendmsg_zc();
    }
    if (ringloop && iothread_count > 0)
    {
        init_iothreads();
    }
    keepalive_timer_id = tfd->set_timer(1000, true, [this](int)
    {
        std::vector<uint64_t> clients_to_stop;
        std::vector<osd_op_t*> ops_to_send;
        auto cl_it = clients.begin();
        while (cl_it != clients.end())
        {
            auto cl = cl_it->second;
            cl_it++;
            if (!cl->osd_num && !cl->in_osd_num || cl->peer_state != PEER_CONNECTED && cl->peer_state != PEER_RDMA)
            {
                // Do not run keepalive on regular clients
                continue;
            }
            if (cl->ping_time_remaining > 0)
            {
                cl->ping_time_remaining--;
                if (!cl->ping_time_remaining)
                {
                    // Ping timed out, stop the client
                    fprintf(stderr, "Ping timed out for OSD %ju (client %ju), disconnecting peer\n",
                        cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->client_id);
                    clients_to_stop.push_back(cl->client_id);
                }
            }
            else if (cl->idle_time_remaining > 0)
            {
                cl->idle_time_remaining--;
                if (!cl->idle_time_remaining)
                {
                    // Connection is idle for <osd_idle_time>, send ping
                    osd_op_t *op = new osd_op_t();
                    op->op_type = OSD_OP_OUT;
                    op->client_id = cl->client_id;
                    op->req = (osd_any_op_t){
                        .hdr = {
                            .magic = SECONDARY_OSD_OP_MAGIC,
                            .opcode = OSD_OP_PING,
                        },
                    };
                    op->callback = [this](osd_op_t *op)
                    {
                        auto cl_it = clients.find(op->client_id);
                        if (cl_it == clients.end())
                        {
                            // client is already dropped
                            delete op;
                            return;
                        }
                        auto cl = cl_it->second;
                        uint64_t fail_client_id = (op->reply.hdr.retval != 0 ? op->client_id : 0);
                        auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num;
                        cl->ping_time_remaining = 0;
                        delete op;
                        if (fail_client_id)
                        {
                            fprintf(stderr, "Ping failed for OSD %ju (client %ju), disconnecting peer\n", fail_osd_num, fail_client_id);
                            stop_client(fail_client_id);
                        }
                    };
                    cl->ping_time_remaining = osd_ping_timeout;
                    cl->idle_time_remaining = osd_idle_timeout;
                    ops_to_send.push_back(op);
                }
            }
            else
            {
                cl->idle_time_remaining = osd_idle_timeout;
            }
        }
        for (uint64_t client_id: clients_to_stop)
        {
            stop_client(client_id);
        }
        for (osd_op_t *op: ops_to_send)
        {
            outbox_push(op);
        }
    });
}

osd_messenger_t::~osd_messenger_t()
{
    if (keepalive_timer_id >= 0)
    {
        tfd->clear_timer(keepalive_timer_id);
        keepalive_timer_id = -1;
    }
    while (clients.size() > 0)
    {
        stop_client(clients.begin()->first, true);
    }
    destroy_iothreads();
#ifdef WITH_RDMA
    for (auto rdma_context: rdma_contexts)
    {
        tfd->set_fd_handler(rdma_context->channel->fd, false, NULL);
        delete rdma_context;
    }
    rdma_contexts.clear();
#endif
#ifdef WITH_RDMACM
    if (rdmacm_evch)
    {
        tfd->set_fd_handler(rdmacm_evch->fd, false, NULL);
        rdma_destroy_event_channel(rdmacm_evch);
        rdmacm_evch = NULL;
    }
#endif
}

void osd_messenger_t::parse_config(const json11::Json & config)
{
#ifdef WITH_RDMA
    if (!config["use_rdma"].is_null())
    {
        // RDMA is on by default in RDMA-enabled builds
        this->use_rdma = config["use_rdma"].bool_value() || config["use_rdma"].uint64_value() != 0;
    }
#ifdef WITH_RDMACM
    // Use RDMA CM? (required for iWARP and may be useful for IB)
    // FIXME: Only parse during start
    this->use_rdmacm = config["use_rdmacm"].bool_value() || config["use_rdmacm"].uint64_value() != 0;
    this->disable_tcp = this->use_rdmacm && (config["disable_tcp"].bool_value() || config["disable_tcp"].uint64_value() != 0);
#endif
    this->rdma_device = config["rdma_device"].string_value();
    this->rdma_port_num = (uint8_t)config["rdma_port_num"].uint64_value();
    if (!config["rdma_gid_index"].is_null())
        this->rdma_gid_index = (uint8_t)config["rdma_gid_index"].uint64_value();
    this->rdma_mtu = (uint32_t)config["rdma_mtu"].uint64_value();
    this->rdma_max_sge = config["rdma_max_sge"].uint64_value();
    if (!this->rdma_max_sge)
        this->rdma_max_sge = 128;
    this->rdma_max_send = config["rdma_max_send"].uint64_value();
    if (!this->rdma_max_send)
        this->rdma_max_send = 8;
    this->rdma_max_recv = config["rdma_max_recv"].uint64_value();
    if (!this->rdma_max_recv)
        this->rdma_max_recv = 16;
    this->rdma_max_msg = config["rdma_max_msg"].uint64_value();
    if (!this->rdma_max_msg || this->rdma_max_msg > 128*1024*1024)
        this->rdma_max_msg = 129*1024;
#endif
    if (!osd_num)
        this->iothread_count = (uint32_t)config["client_iothread_count"].uint64_value();
    else
        this->iothread_count = (uint32_t)config["osd_iothread_count"].uint64_value();
    this->receive_buffer_size = (uint32_t)config["tcp_header_buffer_size"].uint64_value();
    if (!this->receive_buffer_size || this->receive_buffer_size > 1024*1024*1024)
        this->receive_buffer_size = 65536;
    this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
        config["use_sync_send_recv"].uint64_value();
    this->min_zerocopy_send_size = config["min_zerocopy_send_size"].is_null()
        ? DEFAULT_MIN_ZEROCOPY_SEND_SIZE
        : (int)config["min_zerocopy_send_size"].int64_value();
    this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
    if (!this->peer_connect_interval)
        this->peer_connect_interval = 5;
    this->peer_connect_timeout = config["peer_connect_timeout"].uint64_value();
    if (!this->peer_connect_timeout)
        this->peer_connect_timeout = 5;
    this->osd_idle_timeout = config["osd_idle_timeout"].uint64_value();
    if (!this->osd_idle_timeout)
        this->osd_idle_timeout = 5;
    this->osd_ping_timeout = config["osd_ping_timeout"].uint64_value();
    if (!this->osd_ping_timeout)
        this->osd_ping_timeout = 5;
    this->log_level = config["log_level"].uint64_value();
    // OSD public & cluster networks
    this->osd_networks.clear();
    if (config["osd_network"].is_string())
        this->osd_networks.push_back(config["osd_network"].string_value());
    else
        for (auto v: config["osd_network"].array_items())
            this->osd_networks.push_back(v.string_value());
    this->osd_cluster_networks.clear();
    if (config["osd_cluster_network"].is_string())
        this->osd_cluster_networks.push_back(config["osd_cluster_network"].string_value());
    else
        for (auto v: config["osd_cluster_network"].array_items())
            this->osd_cluster_networks.push_back(v.string_value());
    if (this->osd_cluster_networks.size())
        for (auto & net: this->osd_cluster_networks)
            for (int i = this->osd_networks.size()-1; i >= 0; i--)
                if (this->osd_networks[i] == net)
                    this->osd_networks.erase(this->osd_networks.begin()+i, this->osd_networks.begin()+i+1);
    this->osd_network_masks.clear();
    for (auto & netstr: this->osd_networks)
        this->osd_network_masks.push_back(cidr_parse(netstr));
    this->osd_cluster_network_masks.clear();
    for (auto & netstr: this->osd_cluster_networks)
        this->osd_cluster_network_masks.push_back(cidr_parse(netstr));
    this->all_osd_networks.clear();
    this->all_osd_networks.insert(this->all_osd_networks.end(), this->osd_networks.begin(), this->osd_networks.end());
    this->all_osd_networks.insert(this->all_osd_networks.end(), this->osd_cluster_networks.begin(), this->osd_cluster_networks.end());
    this->all_osd_network_masks.clear();
    this->all_osd_network_masks.insert(this->all_osd_network_masks.end(), this->osd_network_masks.begin(), this->osd_network_masks.end());
    this->all_osd_network_masks.insert(this->all_osd_network_masks.end(), this->osd_cluster_network_masks.begin(), this->osd_cluster_network_masks.end());
    if (!this->osd_networks.size())
    {
        this->osd_networks = this->osd_cluster_networks;
        this->osd_network_masks = this->osd_cluster_network_masks;
    }
}

void osd_messenger_t::connect_peer(uint64_t peer_osd, json11::Json peer_state)
{
    if (wanted_peers[peer_osd].raw_address_list != peer_state["addresses"])
    {
        wanted_peers[peer_osd].raw_address_list = peer_state["addresses"];
        // We are an OSD -> try to select a cluster address
        // We are a client -> try to select a public address
        // OSD only has 1 address -> don't try anything, it's pointless
        // FIXME: Maybe support optional fallback from cluster to public network?
        auto & match_masks = (this->osd_num ? osd_cluster_network_masks : osd_network_masks);
        if (peer_state["addresses"].array_items().size() > 1 && match_masks.size())
        {
            json11::Json::array address_list;
            for (auto json_addr: peer_state["addresses"].array_items())
            {
                struct sockaddr_storage addr;
                auto ok = string_to_addr(json_addr.string_value(), false, 0, &addr);
                if (ok)
                {
                    bool matches = false;
                    for (auto & mask: match_masks)
                    {
                        if (cidr_sockaddr_match(addr, mask))
                        {
                            matches = true;
                            break;
                        }
                    }
                    if (matches)
                        address_list.push_back(json_addr);
                }
            }
            if (!address_list.size())
                address_list = peer_state["addresses"].array_items();
            wanted_peers[peer_osd].address_list = address_list;
        }
        else
            wanted_peers[peer_osd].address_list = peer_state["addresses"];
        wanted_peers[peer_osd].address_changed = true;
    }
#ifdef WITH_RDMACM
    wanted_peers[peer_osd].rdmacm_port = (int)peer_state["rdmacm_port"].int64_value();
#endif
    wanted_peers[peer_osd].port = (int)peer_state["port"].int64_value();
    try_connect_peer(peer_osd);
}

void osd_messenger_t::try_connect_peer(uint64_t peer_osd)
{
    auto wp_it = wanted_peers.find(peer_osd);
    if (wp_it == wanted_peers.end() || wp_it->second.connecting ||
        (time(NULL) - wp_it->second.last_connect_attempt) < peer_connect_interval)
    {
        return;
    }
    if (osd_peers.find(peer_osd) != osd_peers.end())
    {
        wanted_peers.erase(peer_osd);
        return;
    }
    auto & wp = wp_it->second;
    if (wp.address_index >= wp.address_list.array_items().size())
    {
        return;
    }
    wp.cur_addr = wp.address_list[wp.address_index].string_value();
    wp.cur_port = wp.port;
    wp.connecting = true;
#ifdef WITH_RDMACM
    if (use_rdmacm && wp.rdmacm_port)
        rdmacm_try_connect_peer(peer_osd, wp.cur_addr.c_str(), wp.rdmacm_port, wp.cur_port);
    else
#endif
        try_connect_peer_tcp(peer_osd, wp.cur_addr.c_str(), wp.cur_port);
}

void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_host, int peer_port)
{
    assert(peer_osd != this->osd_num);
#ifdef WITH_RDMACM
    if (disable_tcp)
    {
        on_connect_peer(peer_osd, -EINVAL, 0);
        return;
    }
#endif
    struct sockaddr_storage addr;
    if (!string_to_addr(peer_host, 0, peer_port, &addr))
    {
        on_connect_peer(peer_osd, -EINVAL, 0);
        return;
    }
    int peer_fd = socket(addr.ss_family, SOCK_STREAM, 0);
    if (peer_fd < 0)
    {
        on_connect_peer(peer_osd, -errno, 0);
        return;
    }
    fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
    int r = connect(peer_fd, (sockaddr*)&addr, sizeof(addr));
    if (r < 0 && errno != EINPROGRESS)
    {
        close(peer_fd);
        on_connect_peer(peer_osd, -errno, 0);
        return;
    }
    const uint64_t client_id = next_client_id++;
    osd_client_t *cl = new osd_client_t();
    if (log_level > 0)
    {
        fprintf(stderr, "Connecting to OSD %ju at %s:%d (client %ju, FD %d)\n", peer_osd, peer_host, peer_port, client_id, peer_fd);
    }
    cl->client_id = client_id;
    cl->peer_addr = addr;
    cl->peer_port = peer_port;
    cl->peer_fd = peer_fd;
    cl->peer_state = PEER_CONNECTING;
    cl->connect_timeout_id = -1;
    cl->osd_num = peer_osd;
    cl->in_buf = malloc_or_die(receive_buffer_size);
    clients[client_id] = cl;
    clients_by_fd[peer_fd] = cl;
    tfd->set_fd_handler(peer_fd, true, [this](int peer_fd, int epoll_events)
    {
        // Either OUT (connected) or HUP
        handle_connect_epoll(peer_fd);
    });
    if (peer_connect_timeout > 0)
    {
        cl->connect_timeout_id = tfd->set_timer(1000*peer_connect_timeout, false, [this, client_id](int timer_id)
        {
            osd_num_t peer_osd = clients.at(client_id)->osd_num;
            stop_client(client_id);
            on_connect_peer(peer_osd, -EPIPE, 0);
            return;
        });
    }
}

void osd_messenger_t::handle_connect_epoll(int peer_fd)
{
    auto cl = clients_by_fd.at(peer_fd);
    if (cl->connect_timeout_id >= 0)
    {
        tfd->clear_timer(cl->connect_timeout_id);
        cl->connect_timeout_id = -1;
    }
    osd_num_t peer_osd = cl->osd_num;
    int result = 0;
    socklen_t result_len = sizeof(result);
    if (getsockopt(peer_fd, SOL_SOCKET, SO_ERROR, &result, &result_len) < 0)
    {
        result = errno;
    }
    if (result != 0)
    {
        stop_client(cl->client_id);
        on_connect_peer(peer_osd, -result, 0);
        return;
    }
    int one = 1;
    setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
    cl->peer_state = PEER_CONNECTED;
    tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
    {
        handle_peer_epoll(peer_fd, epoll_events);
    });
    // Check OSD number
    check_peer_config(cl);
}

void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
{
    // Mark client as ready (i.e. some data is available)
    auto cl = clients_by_fd.at(peer_fd);
    if (epoll_events & EPOLLRDHUP)
    {
        // Stop client
        if (log_level > 0)
        {
            fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id);
        }
        stop_client(cl->client_id);
    }
    else if (epoll_events & EPOLLIN)
    {
        // Mark client as ready (i.e. some data is available)
        cl->read_ready++;
        if (cl->read_ready == 1)
        {
            read_ready_clients.push_back(cl->client_id);
            if (ringloop)
                ringloop->wakeup();
            else
                read_requests();
        }
    }
}

void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id)
{
    auto & wp = wanted_peers.at(peer_osd);
    wp.connecting = false;
    if (errcode < 0)
    {
        fprintf(stderr, "Failed to connect to peer OSD %ju address %s port %d: %s\n", peer_osd, wp.cur_addr.c_str(), wp.cur_port, strerror(-errcode));
        if (wp.address_changed)
        {
            wp.address_changed = false;
            wp.address_index = 0;
            try_connect_peer(peer_osd);
        }
        else if (wp.address_index < wp.address_list.array_items().size()-1)
        {
            // Try other addresses
            wp.address_index++;
            try_connect_peer(peer_osd);
        }
        else
        {
            // Retry again in <peer_connect_interval> seconds
            wp.last_connect_attempt = time(NULL);
            wp.address_index = 0;
            tfd->set_timer(1000*peer_connect_interval, false, [this, peer_osd](int)
            {
                try_connect_peer(peer_osd);
            });
        }
        return;
    }
    if (log_level > 0)
    {
        fprintf(stderr, "[OSD %ju] Connected with peer OSD %ju (client %ju)\n", osd_num, peer_osd, client_id);
    }
    wanted_peers.erase(peer_osd);
    repeer_pgs(peer_osd);
}

void osd_messenger_t::check_peer_config(osd_client_t *cl)
{
    osd_op_t *op = new osd_op_t();
    op->op_type = OSD_OP_OUT;
    op->client_id = cl->client_id;
    op->req = (osd_any_op_t){
        .show_conf = {
            .header = {
                .magic = SECONDARY_OSD_OP_MAGIC,
                .opcode = OSD_OP_SHOW_CONFIG,
            },
        },
    };
    json11::Json::object payload;
    if (osd_num)
    {
        // Inform that we're OSD <osd_num>
        payload["osd_num"] = osd_num;
    }
    payload["features"] = json11::Json::object{ { "check_sequencing", true } };
#ifdef WITH_RDMA
    if (!use_rdmacm && rdma_contexts.size())
    {
        // Choose the right context for the selected network
        msgr_rdma_context_t *selected_ctx = choose_rdma_context(cl);
        if (!selected_ctx)
        {
            if (log_level > 0)
                fprintf(stderr, "No RDMA context for OSD %ju connection (client %ju), using only TCP\n", cl->osd_num, cl->client_id);
        }
        else
        {
            cl->rdma_conn = msgr_rdma_connection_t::create(selected_ctx, rdma_max_send, rdma_max_recv, rdma_max_sge, rdma_max_msg);
            if (cl->rdma_conn)
            {
                payload["connect_rdma"] = cl->rdma_conn->addr.to_string();
                payload["rdma_max_msg"] = cl->rdma_conn->max_msg;
            }
        }
    }
#endif
    if (payload.size())
    {
        std::string payload_str = json11::Json(payload).dump();
        op->req.show_conf.json_len = payload_str.size();
        op->buf = malloc_or_die(payload_str.size());
        op->iov.push_back(op->buf, payload_str.size());
        memcpy(op->buf, payload_str.c_str(), payload_str.size());
    }
    op->callback = [this, cl](osd_op_t *op)
    {
        std::string json_err;
        json11::Json config;
        bool err = false;
        if (op->reply.hdr.retval < 0)
        {
            err = true;
            fprintf(stderr, "Failed to get config from OSD %ju (retval=%jd), disconnecting peer\n", cl->osd_num, op->reply.hdr.retval);
        }
        else
        {
            config = json11::Json::parse(std::string((char*)op->buf), json_err);
            if (json_err != "")
            {
                err = true;
                fprintf(stderr, "Failed to get config from OSD %ju: bad JSON: %s, disconnecting peer\n", cl->osd_num, json_err.c_str());
            }
            else if (config["osd_num"].uint64_value() != cl->osd_num)
            {
                err = true;
                fprintf(stderr, "Connected to OSD %ju instead of OSD %ju, peer state is outdated, disconnecting peer\n", config["osd_num"].uint64_value(), cl->osd_num);
            }
            else if (config["protocol_version"].uint64_value() != OSD_PROTOCOL_VERSION)
            {
                err = true;
                fprintf(
                    stderr, "OSD %ju protocol version is %ju, but only version %u is supported.\n"
                    " If you need to upgrade from 0.5.x please request it via the issue tracker.\n",
                    cl->osd_num, config["protocol_version"].uint64_value(), OSD_PROTOCOL_VERSION
                );
            }
            if (check_config_hook)
            {
                err = !check_config_hook(cl, config);
            }
        }
        if (err)
        {
            osd_num_t peer_osd = cl->osd_num;
            stop_client(op->client_id);
            on_connect_peer(peer_osd, -EINVAL, 0);
            delete op;
            return;
        }
#ifdef WITH_RDMA
        if (!use_rdmacm && cl->rdma_conn && config["rdma_address"].is_string())
        {
            msgr_rdma_address_t addr;
            if (!msgr_rdma_address_t::from_string(config["rdma_address"].string_value().c_str(), &addr) ||
                cl->rdma_conn->connect(&addr) != 0)
            {
                fprintf(
                    stderr, "Failed to connect to OSD %ju (address %s) using RDMA, switching back to TCP\n",
                    cl->osd_num, config["rdma_address"].string_value().c_str()
                );
                delete cl->rdma_conn;
                cl->rdma_conn = NULL;
            }
            else
            {
                uint64_t server_max_msg = config["rdma_max_msg"].uint64_value();
                if (cl->rdma_conn->max_msg > server_max_msg)
                {
                    cl->rdma_conn->max_msg = server_max_msg;
                }
                if (log_level > 0)
                {
                    fprintf(stderr, "Connected to OSD %ju using RDMA\n", cl->osd_num);
                }
                cl->peer_state = PEER_RDMA;
                // Add the initial receive request
                init_recv_rdma(cl);
            }
        }
#endif
        osd_peers[cl->osd_num] = cl;
        on_connect_peer(cl->osd_num, 0, cl->client_id);
        delete op;
    };
    outbox_push(op);
}

void osd_messenger_t::accept_connections(int listen_fd)
{
    // Accept new connections
    sockaddr_storage addr;
    socklen_t peer_addr_size = sizeof(addr);
    int peer_fd;
    while ((peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size)) >= 0)
    {
        assert(peer_fd != 0);
        const uint64_t client_id = next_client_id++;
        fprintf(stderr, "[OSD %ju] new client %ju (FD %d): connection from %s\n", this->osd_num, client_id, peer_fd,
            addr_to_string(addr).c_str());
        fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
        int one = 1;
        setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
        auto cl = new osd_client_t();
        cl->client_id = client_id;
        clients[cl->client_id] = cl;
        clients_by_fd[peer_fd] = cl;
        cl->is_incoming = true;
        cl->peer_addr = addr;
        cl->peer_addr = addr;
        cl->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port);
        cl->peer_fd = peer_fd;
        cl->peer_state = PEER_CONNECTED;
        cl->in_buf = malloc_or_die(receive_buffer_size);
        // Add FD to epoll
        tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
        {
            handle_peer_epoll(peer_fd, epoll_events);
        });
        // Try to accept next connection
        peer_addr_size = sizeof(addr);
    }
    if (peer_fd == -1 && errno != EAGAIN)
    {
        throw std::runtime_error(std::string("accept: ") + strerror(errno));
    }
}

#ifdef WITH_RDMA
msgr_rdma_context_t* osd_messenger_t::choose_rdma_context(osd_client_t *cl)
{
    // Choose the right context for the selected network
    msgr_rdma_context_t *selected_ctx = NULL;
    for (auto rdma_ctx: rdma_contexts)
    {
        if (!rdma_ctx->net_mask.family && !selected_ctx ||
            rdma_ctx->net_mask.family && cidr_sockaddr_match(cl->peer_addr, rdma_ctx->net_mask))
        {
            selected_ctx = rdma_ctx;
        }
    }
    return selected_ctx;
}

bool osd_messenger_t::is_rdma_enabled()
{
    return rdma_contexts.size() > 0;
}
#endif
#ifdef WITH_RDMACM
bool osd_messenger_t::is_use_rdmacm()
{
    return use_rdmacm;
}
#endif

json11::Json::object osd_messenger_t::read_config(const json11::Json & config)
{
    json11::Json::object file_config;
    const char *config_path = config["config_path"].string_value() != ""
        ? config["config_path"].string_value().c_str() : VITASTOR_CONFIG_PATH;
    int fd = open(config_path, O_RDONLY);
    if (fd < 0)
    {
        if (errno != ENOENT)
            fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
        return file_config;
    }
    struct stat st;
    if (fstat(fd, &st) != 0)
    {
        fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
        close(fd);
        return file_config;
    }
    std::string buf;
    buf.resize(st.st_size);
    int done = 0;
    while (done < st.st_size)
    {
        int r = read(fd, (uint8_t*)buf.data()+done, st.st_size-done);
        if (r < 0)
        {
            fprintf(stderr, "Error reading %s: %s\n", config_path, strerror(errno));
            close(fd);
            return file_config;
        }
        done += r;
    }
    close(fd);
    std::string json_err;
    file_config = json11::Json::parse(buf, json_err).object_items();
    if (json_err != "")
    {
        fprintf(stderr, "Invalid JSON in %s: %s\n", config_path, json_err.c_str());
    }
    return file_config;
}

static const char* cli_only_params[] = {
    // The list has to be sorted
    "bitmap_granularity",
    "block_size",
    "data_device",
    "data_offset",
    "data_size",
    "disable_data_fsync",
    "disable_device_lock",
    "disable_journal_fsync",
    "disable_meta_fsync",
    "disk_alignment",
    "flush_journal",
    "immediate_commit",
    "inmemory_journal",
    "inmemory_metadata",
    "journal_block_size",
    "journal_device",
    "journal_no_same_sector_overwrites",
    "journal_offset",
    "journal_sector_buffer_count",
    "journal_size",
    "meta_block_size",
    "meta_buf_size",
    "meta_device",
    "meta_offset",
    "osd_num",
    "readonly",
};

static const char **cli_only_end = cli_only_params + (sizeof(cli_only_params)/sizeof(cli_only_params[0]));

static const char* local_only_params[] = {
    // The list has to be sorted
    "config_path",
    "rdma_device",
    "rdma_gid_index",
    "rdma_max_msg",
    "rdma_max_recv",
    "rdma_max_send",
    "rdma_max_sge",
    "rdma_mtu",
    "rdma_port_num",
    "tcp_header_buffer_size",
    "use_rdma",
    "use_sync_send_recv",
    "min_zerocopy_send_size",
};

static const char **local_only_end = local_only_params + (sizeof(local_only_params)/sizeof(local_only_params[0]));

// Basically could be replaced by std::lower_bound()...
static int find_str_array(const char **start, const char **end, const std::string & s)
{
    int min = 0, max = end-start;
    while (max-min >= 2)
    {
        int mid = (min+max)/2;
        int r = strcmp(s.c_str(), start[mid]);
        if (r < 0)
            max = mid;
        else if (r > 0)
            min = mid;
        else
            return mid;
    }
    if (min < end-start && !strcmp(s.c_str(), start[min]))
        return min;
    return -1;
}

json11::Json::object osd_messenger_t::merge_configs(const json11::Json::object & cli_config,
    const json11::Json::object & file_config,
    const json11::Json::object & etcd_global_config,
    const json11::Json::object & etcd_osd_config)
{
    // Priority: most important -> less important:
    // etcd_osd_config -> cli_config -> etcd_global_config -> file_config
    json11::Json::object res = file_config;
    for (auto & kv: file_config)
    {
        int cli_only = find_str_array(cli_only_params, cli_only_end, kv.first);
        if (cli_only < 0)
        {
            res[kv.first] = kv.second;
        }
    }
    for (auto & kv: etcd_global_config)
    {
        int local_only = find_str_array(local_only_params, local_only_end, kv.first);
        if (local_only < 0)
        {
            res[kv.first] = kv.second;
        }
    }
    for (auto & kv: cli_config)
    {
        res[kv.first] = kv.second;
    }
    for (auto & kv: etcd_osd_config)
    {
        int local_only = find_str_array(local_only_params, local_only_end, kv.first);
        if (local_only < 0)
        {
            res[kv.first] = kv.second;
        }
    }
    return res;
}