cluckd.cpp

Go to the documentation of this file.

// Copyright (c) 2016-2025  Made to Order Software Corp.  All Rights Reserved
//
// https://snapwebsites.org/project/cluck
// contact@m2osw.com
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 
// self
//
#include    "cluckd.h"
 
 
 
// cluck
//
#include    <cluck/exception.h>
#include    <cluck/names.h>
#include    <cluck/version.h>
 
 
// communicator
//
#include    <communicator/flags.h>
#include    <communicator/names.h>
 
 
// as2js
//
#include    <as2js/json.h>
 
 
// eventdispatcher
//
#include    <eventdispatcher/names.h>
 
 
// snapdev
//
#include    <snapdev/gethostname.h>
#include    <snapdev/hexadecimal_string.h>
#include    <snapdev/stringize.h>
#include    <snapdev/tokenize_string.h>
#include    <snapdev/to_string_literal.h>
 
 
// snaplogger
//
#include    <snaplogger/logger.h>
#include    <snaplogger/options.h>
#include    <snaplogger/severity.h>
 
 
// advgetopt
//
#include    <advgetopt/advgetopt.h>
#include    <advgetopt/exception.h>
 
 
// C++
//
#include    <algorithm>
#include    <iostream>
#include    <sstream>
 
 
// openssl
//
#include    <openssl/rand.h>
 
 
// last include
//
#include    <snapdev/poison.h>
 
 
 
namespace cluck_daemon
{
 
 
namespace
{
 
 
 
constexpr std::string_view  g_default_candidate_priority =
    snapdev::integer_to_string_literal<computer::PRIORITY_DEFAULT>.data();
 
 
advgetopt::option const g_options[] =
{
    advgetopt::define_option(
          advgetopt::Name("candidate-priority")
        , advgetopt::ShortName('p')
        , advgetopt::Flags(advgetopt::all_flags<
                      advgetopt::GETOPT_FLAG_REQUIRED
                    , advgetopt::GETOPT_FLAG_GROUP_OPTIONS>())
        , advgetopt::Help("Define the priority of this candidate (1 to 14) to gain a leader position or \"off\".")
        , advgetopt::DefaultValue(g_default_candidate_priority.data())
    ),
    advgetopt::define_option(
          advgetopt::Name("server-name")
        , advgetopt::ShortName('n')
        , advgetopt::Flags(advgetopt::all_flags<
                      advgetopt::GETOPT_FLAG_DYNAMIC_CONFIGURATION
                    , advgetopt::GETOPT_FLAG_REQUIRED
                    , advgetopt::GETOPT_FLAG_GROUP_OPTIONS>())
        , advgetopt::Help("Set the name of this server instance.")
    ),
    advgetopt::end_options()
};
 
 
advgetopt::group_description const g_group_descriptions[] =
{
    advgetopt::define_group(
          advgetopt::GroupNumber(advgetopt::GETOPT_FLAG_GROUP_COMMANDS)
        , advgetopt::GroupName("command")
        , advgetopt::GroupDescription("Commands:")
    ),
    advgetopt::define_group(
          advgetopt::GroupNumber(advgetopt::GETOPT_FLAG_GROUP_OPTIONS)
        , advgetopt::GroupName("option")
        , advgetopt::GroupDescription("Options:")
    ),
    advgetopt::end_groups()
};
 
 
constexpr char const * const g_configuration_files[] =
{
    "/etc/cluck/cluckd.conf",
    nullptr
};
 
 
advgetopt::options_environment const g_options_environment =
{
    .f_project_name = "cluckd",
    .f_group_name = "cluck",
    .f_options = g_options,
    .f_environment_variable_name = "CLUCKD_OPTIONS",
    .f_configuration_files = g_configuration_files,
    .f_environment_flags = advgetopt::GETOPT_ENVIRONMENT_FLAG_SYSTEM_PARAMETERS
                         | advgetopt::GETOPT_ENVIRONMENT_FLAG_PROCESS_SYSTEM_PARAMETERS,
    .f_help_header = "Usage: %p [-<opt>]\n"
                     "where -<opt> is one or more of:",
    .f_help_footer = "%c",
    .f_version = CLUCK_VERSION_STRING,
    .f_license = "GNU GPL v3",
    .f_copyright = "Copyright (c) 2013-"
                   SNAPDEV_STRINGIZE(UTC_BUILD_YEAR)
                   " by Made to Order Software Corporation -- All Rights Reserved",
    .f_groups = g_group_descriptions,
};
 
 
 
}
// no name namespace
 
 
 
 
 
 
 
 
 
 
 
 
 
cluckd::cluckd(int argc, char * argv[])
    : f_opts(g_options_environment)
{
    snaplogger::add_logger_options(f_opts);
 
    // before we can parse command line arguments, we must create the
    // fluid settings & communicator client objects which happen to
    // dynamically add command line options to f_opts
    //
    f_messenger = std::make_shared<messenger>(this, f_opts);
 
    f_opts.finish_parsing(argc, argv);
    if(!snaplogger::process_logger_options(f_opts, "/etc/cluck/logger"))
    {
        throw advgetopt::getopt_exit("logger options generated an error.", 1);
    }
 
    // determine this server name
    //
    // TODO: if the name of the server is changed, we should reboot, but
    //       to the minimum we need to restart cluckd (among other daemons)
    //       remember that snapmanager.cgi gives you that option;
    //       we CANNOT use fluid-settings for this one since each computer
    //       must have a different name
    //
    if(f_opts.is_defined("server-name"))
    {
        f_server_name = f_opts.get_string("server-name");
    }
    if(f_server_name.empty())
    {
        f_server_name = snapdev::gethostname();
    }
 
    f_start_time = snapdev::now();
}
 
 
cluckd::~cluckd()
{
}
 
 
void cluckd::add_connections()
{
    f_communicator = ed::communicator::instance();
 
    // capture Ctrl-C (SIGINT) to get a clean exit by default
    //
    f_interrupt = std::make_shared<interrupt>(this);
    f_communicator->add_connection(f_interrupt);
 
    // timer so we can timeout locks
    //
    f_timer = std::make_shared<timer>(this);
    f_communicator->add_connection(f_timer);
 
    // add the messenger used to communicate with the communicator daemon
    // and other services as required
    //
    f_communicator->add_connection(f_messenger);
 
    // the following call actually connects the messenger to the
    // communicator daemon
    //
    f_messenger->finish_parsing();
}
 
 
void cluckd::set_my_ip_address(addr::addr const & a)
{
    f_my_ip_address = a;
}
 
 
void cluckd::run()
{
    SNAP_LOG_INFO
        << "--------------------------------- cluckd started."
        << SNAP_LOG_SEND;
 
    // now run our listening loop
    //
    f_communicator->run();
}
 
 
 
 
 
int cluckd::get_computer_count() const
{
    return f_computers.size();
}
 
 
std::string const & cluckd::get_server_name() const
{
    return f_server_name;
}
 
 
bool cluckd::is_daemon_ready() const
{
    // we are not yet properly registered
    //
    if(f_messenger == nullptr || !f_messenger->is_ready())
    {
        return false;
    }
 
    // we need a stable clock otherwise we cannot guarantee that the time
    // between servers is going to be sufficiently close
    //
    if(!f_stable_clock)
    {
        return false;
    }
 
    // without at least one leader we are definitely not ready
    //
    if(f_leaders.empty())
    {
        SNAP_LOG_TRACE
            << "not considered ready: no leaders."
            << SNAP_LOG_SEND;
        return false;
    }
 
    // enough leaders for that cluster?
    //
    // we consider that having at least 2 leaders is valid because locks
    // will still work, an election should be happening when we lose a
    // leader fixing that temporary state
    //
    // the test below allows for the case where we have a single computer
    // too (i.e. "one neighbor")
    //
    // notice how not having received the CLUSTERUP would be taken in
    // account here since f_neighbors_count will still be 0 in that case
    // (however, the previous empty() test already takes that in account)
    //
    if(f_leaders.size() == 1
    && f_neighbors_count != 1)
    {
        SNAP_LOG_TRACE
            << "not considered ready: not enough leaders for this cluster."
            << SNAP_LOG_SEND;
        return false;
    }
 
    // the election_status() function verifies that the quorum is
    // attained, but it can change if the cluster grows or shrinks
    // so we have to check here again as the lock system becomes
    // "unready" when the quorum is lost; see that other function
    // for additional info
 
    // this one probably looks complicated...
    //
    // if our quorum is 1 or 2 then we need a number of computers
    // equal to the total number of computers (i.e. a CLUSTER_COMPLETE
    // status which we verify here)
    //
    if(f_neighbors_quorum < 3
    && f_computers.size() < f_neighbors_count)
    {
        SNAP_LOG_TRACE
            << "not considered ready: quorum changed, re-election expected soon (number of neighbors: "
            << f_neighbors_count
            << ", neighbors quorum: "
            << f_neighbors_quorum
            << ", number of computers: "
            << f_computers.size()
            << "."
            << SNAP_LOG_SEND;
        return false;
    }
 
    // the neighbors count & quorum can change over time so
    // we have to verify that the number of computers is
    // still acceptable here
    //
    if(f_computers.size() < f_neighbors_quorum)
    {
        SNAP_LOG_TRACE
            << "not considered ready: quorum lost, re-election expected soon."
            << SNAP_LOG_SEND;
        return false;
    }
 
    // are all leaders connected to us?
    //
    std::size_t ready(0);
    computer::pointer_t last_leader;
    for(auto const & l : f_leaders)
    {
        if(l->get_connected())
        {
            ++ready;
        }
        else
        {
            // attempt resending a LOCK_STARTED because it could be that it
            // did not work quite right and the cluck daemons are not
            // going to ever talk with each others otherwise
            //
            // we also make sure we do not send the message too many times,
            // it should be resolved in five seconds or less...
            //
            time_t const now(time(nullptr));
            if(now >= f_pace_lockstarted)
            {
                // pause for 5 to 6 seconds in case this happens a lot
                //
                f_pace_lockstarted = now + 5;
 
                // only send it to that specific cluck daemon
                //
                ed::message temporary_message;
                temporary_message.set_sent_from_server(l->get_name());
                temporary_message.set_sent_from_service(cluck::g_name_cluck_service_name);
                const_cast<cluckd *>(this)->send_lock_started(&temporary_message);
            }
 
            last_leader = l;
        }
    }
 
    // we only need to have 2 leaders to have a functional system with
    // 2 or 3 leaders total
    //
    if(ready >= 2
    || ready == f_leaders.size())
    {
        return true;
    }
 
    // we're still not ready
    //
    SNAP_LOG_TRACE
        << "not considered ready: no direct connection with leader: \""
        << last_leader->get_name()
        << "\"."
        << SNAP_LOG_SEND;
 
    return false;
}
 
 
computer::pointer_t cluckd::is_leader(std::string id) const
{
    if(id.empty())
    {
        id = f_my_id;
    }
 
    auto const l(std::find_if(
          f_leaders.begin()
        , f_leaders.end()
        , [id](auto const & c){
              return c->get_id() == id;
          }));
    if(l != f_leaders.end())
    {
        return *l;
    }
 
    return computer::pointer_t();
}
 
 
computer::pointer_t cluckd::get_leader_a() const
{
#ifdef _DEBUG
    if(is_leader() == nullptr)
    {
        throw cluck::logic_error("cluckd::get_leader_a(): only a leader can call this function.");
    }
#endif
 
    switch(f_leaders.size())
    {
    case 0: // LCOV_EXCL_LINE -- because of the debug above, this cannot happen here
    default:
        throw cluck::logic_error("cluckd::get_leader_a(): call this function only when leaders were elected."); // LCOV_EXCL_LINE
 
    case 1:
        return computer::pointer_t();
 
    case 2:
    case 3:
        return f_leaders[f_leaders[0]->is_self() ? 1 : 0];
 
    }
}
 
 
computer::pointer_t cluckd::get_leader_b() const
{
#ifdef _DEBUG
    if(is_leader() == nullptr)
    {
        throw cluck::logic_error("cluckd::get_leader_b(): only a leader can call this function.");
    }
#endif
 
    switch(f_leaders.size())
    {
    case 0: // LCOV_EXCL_LINE -- because of the debug above, this cannot happen here
    default:
        throw cluck::unexpected_case("cluckd::get_leader_b(): call this function only when leaders were elected."); // LCOV_EXCL_LINE
 
    case 1:
    case 2: // we have a leader A but no leader B when we have only 2 leaders
        return computer::pointer_t();
 
    case 3:
        return f_leaders[f_leaders[2]->is_self() ? 1 : 2];
 
    }
}
 
 
 
void cluckd::msg_info(ed::message & msg)
{
    std::stringstream ss;
 
    as2js::position p;
    p.set_filename("cluckd.cpp");
    p.set_function("msg_info");
 
    as2js::json::json_value::object_t obj;
    as2js::json::json_value::pointer_t result(std::make_shared<as2js::json::json_value>(p, obj));
 
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, is_daemon_ready()));
        result->set_member("daemon_ready", value);
    }
 
    if(!f_my_id.empty())
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, f_my_id));
        result->set_member("id", value);
    }
 
    {
        addr::addr zero;
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p,
            f_my_ip_address == zero
                ? "<not assigned>"
                : f_my_ip_address.to_ipv4or6_string(addr::STRING_IP_ADDRESS | addr::STRING_IP_BRACKET_ADDRESS)));
        result->set_member("ip", value);
    }
 
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, as2js::integer(f_neighbors_count)));
        result->set_member("neighbors_count", value);
    }
 
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, as2js::integer(f_neighbors_quorum)));
        result->set_member("neighbors_quorum", value);
    }
 
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, as2js::integer(f_leaders.size())));
        result->set_member("leaders_count", value);
    }
 
    if(!f_message_cache.empty())
    {
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, as2js::integer(f_message_cache.size())));
        result->set_member("cache_size", value);
    }
 
    {
        as2js::json::json_value::array_t computers;
        as2js::json::json_value::pointer_t list(std::make_shared<as2js::json::json_value>(p, computers));
 
        for(auto const & c : f_computers)
        {
            as2js::json::json_value::object_t computer;
            as2js::json::json_value::pointer_t item(std::make_shared<as2js::json::json_value>(p, computer));
 
            {
                as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, c.second->get_name()));
                item->set_member("name", value);
            }
 
            {
                as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, c.second->get_id()));
                item->set_member("id", value);
            }
 
            {
                as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, c.second->get_ip_address().to_ipv4or6_string(addr::STRING_IP_ADDRESS | addr::STRING_IP_BRACKET_ADDRESS)));
                item->set_member("ip", value);
            }
 
            {
                as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, c.second->get_connected()));
                item->set_member("connected", value);
            }
 
            {
                auto const it(std::find_if(
                          f_leaders.begin()
                        , f_leaders.end()
                        , [&c](auto const & l)
                        {
                            return c.second == l;
                        }));
                std::string leader;
                if(it != f_leaders.end())
                {
                    as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, as2js::integer(it - f_leaders.begin())));
                    item->set_member("leader", value);
                }
            }
 
            list->set_item(list->get_array().size(), item);
        }
        result->set_member("computers", list);
    }
 
    if(msg.has_parameter(cluck::g_name_cluck_param_mode)
    && msg.get_parameter(cluck::g_name_cluck_param_mode) == cluck::g_name_cluck_value_debug)
    {
        // TODO: create a serialization to JSON instead of a specialized string
        //
        as2js::json::json_value::pointer_t value(std::make_shared<as2js::json::json_value>(p, serialized_tickets()));
        result->set_member("tickets", value);
    }
 
    ed::message reply_message;
    reply_message.set_command(cluck::g_name_cluck_cmd_cluckd_status);
    reply_message.reply_to(msg);
    reply_message.add_parameter(communicator::g_name_communicator_param_status, result->to_string());
    f_messenger->send_message(reply_message);
}
 
 
 
std::string cluckd::ticket_list() const
{
    std::stringstream list;
    for(auto const & obj_ticket : f_tickets)
    {
        for(auto const & key_ticket : obj_ticket.second)
        {
            list
                << "ticket_id: "
                << key_ticket.second->get_ticket_number()
                << "  object name: \""
                << key_ticket.second->get_object_name()
                << "\"  key: "
                << key_ticket.second->get_entering_key();
 
            cluck::timeout_t const lock_timeout(key_ticket.second->get_lock_timeout_date());
            if(lock_timeout != cluck::timeout_t())
            {
                list
                    << "  timeout "
                    << lock_timeout.to_string();
            }
            else
            {
                cluck::timeout_t const obtention_timeout(key_ticket.second->get_obtention_timeout());
                list
                    << "  obtention "
                    << obtention_timeout.to_string();
            }
 
            list << '\n';
        }
    }
 
    return list.str();
}
 
 
void cluckd::election_status()
{
    // we already have election results?
    //
    if(!f_leaders.empty())
    {
        // the results may have been "tempered" with (i.e. one of
        // the leaders was lost)
        //
        if(f_leaders.size() == 3
        || (f_neighbors_count < 3 && f_leaders.size() == f_neighbors_count))
        {
            // status is fine
            //
            return;
        }
    }
 
    // we do not yet know our IP address, we cannot support an election just yet
    //
    if(f_my_ip_address.is_default())
    {
        // this should not be possible since we expect to receive the
        // REGISTER's reply before any other message
        //
        return; // LCOV_EXCL_LINE
    }
 
    // neighbors count is 0 until we receive a very first CLUSTER_UP
    // (note that it does not go back to zero on CLUSTER_DOWN, however,
    // the quorum as checked in the next if() is never going to be
    // reached if the cluster is down)
    //
    if(f_neighbors_count == 0)
    {
        return;
    }
 
    // this one probably looks complicated...
    //
    // if our quorum is 1 or 2 then we need a number of computers
    // equal to the total number of computers (i.e. a CLUSTER_COMPLETE
    // status which we compute here)
    //
    if(f_neighbors_quorum < 3
    && f_computers.size() < f_neighbors_count)
    {
        return;
    }
 
    // since the neighbors count & quorum never go back to zero (on a
    // CLUSTER_DOWN) we have to verify that the number of computers is
    // acceptable here
    //
    // Note: further we will not count computers marked disabled, which
    //       is done below when sorting by ID, however, that does not
    //       prevent the quorum to be attained, even with disabled
    //       computers
    //
    if(f_computers.size() < f_neighbors_quorum)
    {
        return;
    }
 
    // to proceed with an election we must have the smallest IP address
    // (it is not absolutely required, but that way we avoid many
    // consensus problems, in effect we have one "temporary-leader" that ends
    // up telling us who the final three leaders are)
    //
    // TODO: verify that this works properly in a non-complete cluster
    //
    for(auto & c : f_computers)
    {
        // Note: the test fails when we compare to ourselves so we do not
        //       need any special case
        //
        if(c.second->get_ip_address() < f_my_ip_address)
        {
            return;
        }
    }
 
    // to select the leaders sort them by identifier and take the first
    // three (i.e. lower priority, random, IP, pid.)
    //
    int off(0);
    computer::map_t sort_by_id;
    for(auto const & c : f_computers)
    {
        // ignore nodes with a priority of 15 (i.e. OFF)
        //
        if(c.second->get_priority() != computer::PRIORITY_OFF)
        {
            sort_by_id[c.second->get_id()] = c.second;
        }
        else
        {
            ++off;
        }
    }
 
//for(auto const & s : sort_by_id)
//{
//SNAP_LOG_WARNING << "--- sort by ID: " << s.first << SNAP_LOG_SEND;
//}
 
    bool too_many_computers_off(false);
    if(f_computers.size() <= 3)
    {
        if(off != 0
        && f_computers.size() >= f_neighbors_count)
        {
            SNAP_LOG_FATAL
                << "you cannot have any cluck computer turned OFF when you"
                   " have three or less computers total in your cluster."
                   " The elections cannot be completed in these"
                   " conditions."
                << SNAP_LOG_SEND;
            too_many_computers_off = true;
        }
    }
    else if(f_computers.size() - off < 3
         && f_computers.size() >= f_neighbors_count)
    {
        SNAP_LOG_FATAL
            << "you have a total of "
            << f_computers.size()
            << " computers in your cluster. You turned off "
            << off
            << " of them, which means less than three are left"
               " as candidates for leadership which is not enough."
               " You can have a maximum of "
            << f_computers.size() - 3
            << " that are turned off on this cluster."
            << SNAP_LOG_SEND;
        too_many_computers_off = true;
    }
    if(too_many_computers_off)
    {
        // only generate the flag once we reach the CLUSTER_COMPLETE status
        // (we cannot be sure that the `off` variable is valid until then)
        //
        if(f_computers.size() >= f_neighbors_count)
        {
            // this is something that breaks the entire system so someone
            // needs to fix it and thus it has a really high priority
            //
            communicator::flag::pointer_t flag(COMMUNICATOR_FLAG_UP(
                          "cluckd"
                        , "election"
                        , "instances-off"
                        , "the cluck daemon detected that too many of the"
                          " daemons have their priority set to OFF;"
                          " you must turn some of these back ON."
                    ));
            flag->set_priority(99);
            flag->add_tag("settings");
            flag->set_manual_down(true);
            flag->save();
        }
        return;
    }
 
    // an election works as soon as we have at least 2 leaders
    // or we reached the total number of computers
    //
    if(sort_by_id.size() < 2
    && sort_by_id.size() != f_computers.size())
    {
        return;
    }
 
//std::cerr << f_communicator_port << " is conducting an election:\n";
//for(auto s : sort_by_id)
//{
//std::cerr << "  " << s.second->get_name() << "    " << s.first << "\n";
//}
 
    // the first three are the new leaders
    //
    ed::message lock_leaders_message;
    lock_leaders_message.set_command(cluck::g_name_cluck_cmd_lock_leaders);
    lock_leaders_message.set_service(communicator::g_name_communicator_server_any);
    f_leaders.clear();
    f_election_date = snapdev::now();
    lock_leaders_message.add_parameter(cluck::g_name_cluck_param_election_date, f_election_date);
    auto leader(sort_by_id.begin());
    std::size_t const max(std::min(static_cast<computer::map_t::size_type>(3), sort_by_id.size()));
    for(std::size_t idx(0); idx < max; ++idx, ++leader)
    {
        lock_leaders_message.add_parameter(
              cluck::g_name_cluck_param_leader + std::to_string(idx)
            , leader->second->get_id());
        f_leaders.push_back(leader->second);
    }
    f_messenger->send_message(lock_leaders_message);
 
#if 1
SNAP_LOG_WARNING
<< "election status = add leader(s)... "
<< f_computers.size()
<< " computers and "
<< f_leaders.size()
<< " leaders."
<< SNAP_LOG_SEND;
#endif
 
    // we need to synchronize from this cluckd daemon if it is one of the
    // leaders otherwise we won't get the LOCK_STARTED which would also
    // call that function
    //
    synchronize_leaders();
}
 
 
void cluckd::check_lock_status()
{
    bool const lock_status(is_daemon_ready());
    if(f_lock_status == lock_status)
    {
        return;
    }
    f_lock_status = lock_status;
 
    ed::message status_message;
    status_message.set_command(f_lock_status
                    ? cluck::g_name_cluck_cmd_lock_ready
                    : cluck::g_name_cluck_cmd_no_lock);
    status_message.set_service(communicator::g_name_communicator_server_me);
    status_message.add_parameter(communicator::g_name_communicator_param_cache, communicator::g_name_communicator_value_no);
    f_messenger->send_message(status_message);
 
    if(lock_status
    && !f_message_cache.empty())
    {
        // we still have a cache of locks that can now be processed
        //
        // note:
        // although msg_lock() could re-add some of those messages
        // in the f_message_cache vector, it should not since it
        // calls the same is_daemon_ready() function which we know returns
        // true and therefore no cache is required
        //
        message_cache::list_t cache;
        cache.swap(f_message_cache);
        for(auto & mc : cache)
        {
            msg_lock(mc.f_message);
        }
    }
}
 
 
void cluckd::send_lock_started(ed::message const * msg)
{
    // tell other cluck daemon instances that are already listening that
    // we are ready; this way we can calculate the number of computers
    // available in our network and use that to calculate the QUORUM
    //
    ed::message lock_started_message;
    lock_started_message.set_command(cluck::g_name_cluck_cmd_lock_started);
    if(msg == nullptr)
    {
        lock_started_message.set_service(communicator::g_name_communicator_service_public_broadcast);
 
        // unfortunately, the following does NOT work as expected...
        // (i.e. the following ends up sending the message to ourselves only
        // and does not forward to any remote communicators).
        //
        //lock_started_message.set_server(communicator::g_name_communicator_server_any);
        //lock_started_message.set_service(cluck::g_name_cluck_service_name);
    }
    else
    {
        lock_started_message.reply_to(*msg);
    }
 
    // our info: server name and id
    //
    lock_started_message.add_parameter(communicator::g_name_communicator_param_server_name, f_server_name);
    lock_started_message.add_parameter(cluck::g_name_cluck_param_lock_id, f_my_id);
    lock_started_message.add_parameter(cluck::g_name_cluck_param_start_time, f_start_time);
 
    // include the leaders if present
    //
    if(!f_leaders.empty())
    {
        lock_started_message.add_parameter(cluck::g_name_cluck_param_election_date, f_election_date);
        for(size_t idx(0); idx < f_leaders.size(); ++idx)
        {
            lock_started_message.add_parameter(
                  cluck::g_name_cluck_param_leader + std::to_string(idx)
                , f_leaders[idx]->get_id());
        }
    }
 
    f_messenger->send_message(lock_started_message);
}
 
 
void cluckd::stop(bool quitting)
{
    if(f_messenger != nullptr)
    {
        f_messenger->unregister_fluid_settings(quitting);
        f_communicator->remove_connection(f_messenger);
        f_messenger.reset();
    }
 
    if(f_communicator != nullptr)
    {
        f_communicator->remove_connection(f_interrupt);
        f_interrupt.reset();
 
        f_communicator->remove_connection(f_timer);
        f_timer.reset();
    }
}
 
 
void cluckd::activate_first_lock(std::string const & object_name)
{
    auto ticket(find_first_lock(object_name));
 
    if(ticket != nullptr)
    {
        // there is what we think is the first ticket
        // that should be actived now; we need to share
        // with the other 2 leaders to make sure of that
        //
        ticket->activate_lock();
    }
}
 
 
ticket::pointer_t cluckd::find_first_lock(std::string const & object_name)
{
    ticket::pointer_t first_ticket;
    auto const obj_ticket(f_tickets.find(object_name));
 
    if(obj_ticket != f_tickets.end())
    {
        // loop through making sure that we activate a ticket only
        // if the obtention date was not already reached; if that
        // date was reached before we had the time to activate the
        // lock, then the client should have abandonned the lock
        // request anyway...
        //
        // (this is already done in the cleanup(), but a couple of
        // other functions may call the activate_first_lock()
        // function!)
        //
        for(auto key_ticket(obj_ticket->second.begin()); key_ticket != obj_ticket->second.end(); )
        {
            if(key_ticket->second->timed_out())
            {
                // that ticket timed out, send an UNLOCKING, UNLOCKED,
                // or LOCK_FAILED message and get rid of it
                //
                key_ticket->second->lock_failed("timed out while searching for first lock");
                if(key_ticket->second->timed_out())
                {
                    // still timed out, remove it
                    //
                    key_ticket = obj_ticket->second.erase(key_ticket);
                }
            }
            else
            {
                if(first_ticket == nullptr)
                {
                    first_ticket = key_ticket->second;
                }
                ++key_ticket;
            }
        }
 
        if(obj_ticket->second.empty())
        {
            // it is empty now, get rid of that set of tickets
            //
            f_tickets.erase(obj_ticket);
        }
    }
 
    return first_ticket;
}
 
 
void cluckd::synchronize_leaders()
{
    // there is nothing to do if we are by ourselves because we cannot
    // gain any type of concensus unless we are expected to be the only
    // one in which case there is no synchronization requirements anyway
    //
    if(f_leaders.size() <= 1)
    {
        return;
    }
 
    // only leaders can synchronize each others
    // (other cluck daemons do not have any tickets to synchronize)
    //
    if(is_leader() == nullptr)
    {
        return;
    }
 
    // determine whether we are leader #0 or not, if zero, then we
    // call msg_lock() directly, otherwise we do a f_messenger->send_message()
    //
    // TODO: review the logic here, I do not think that leader0 has anything
    //       to do with lock requests and tickets; instead, I think that the
    //       sharing that needs to happen is between the old and the new
    //       leaders (i.e. when we lose a leader and assign another computer
    //       as a new leader to compensate, that new leader needs to get
    //       all the info which is what the LOCK_TICKETS message is about)
    //
    bool const leader0(f_leaders[0]->get_id() == f_my_id);
 
    // a vector of messages for which we have to call msg_lock()
    //
    ed::message::vector_t local_locks;
 
    // if entering a ticket is definitely not locked, although it
    // could be ready (one step away from being locked!) we still
    // restart the whole process with the new leaders if such
    // exist
    //
    // Note: of course we restart the process only if the owner
    //       was that one leader that disappeared, not if the
    //       ticket is owned by a remaining leader
    //
    for(auto obj_entering(f_entering_tickets.begin()); obj_entering != f_entering_tickets.end(); ++obj_entering)
    {
        for(auto key_entering(obj_entering->second.begin()); key_entering != obj_entering->second.end(); )
        {
            std::string const owner_name(key_entering->second->get_owner());
            auto key_leader(std::find_if(
                      f_leaders.begin()
                    , f_leaders.end()
                    , [&owner_name](auto const & l)
                    {
                        return l->get_name() == owner_name;
                    }));
            if(key_leader == f_leaders.end())
            {
                // give new ownership to leader[0]
                //
                ed::message lock_message;
                lock_message.set_command(cluck::g_name_cluck_cmd_lock);
                lock_message.set_server(f_leaders[0]->get_name());
                lock_message.set_service(cluck::g_name_cluck_service_name);
                lock_message.set_sent_from_server(key_entering->second->get_server_name());
                lock_message.set_sent_from_service(key_entering->second->get_service_name());
                lock_message.add_parameter(cluck::g_name_cluck_param_object_name, key_entering->second->get_object_name());
                lock_message.add_parameter(cluck::g_name_cluck_param_tag, key_entering->second->get_tag());
                lock_message.add_parameter(cluck::g_name_cluck_param_pid, key_entering->second->get_client_pid());
                lock_message.add_parameter(cluck::g_name_cluck_param_timeout, key_entering->second->get_obtention_timeout());
                lock_message.add_parameter(cluck::g_name_cluck_param_duration, key_entering->second->get_lock_duration());
                lock_message.add_parameter(cluck::g_name_cluck_param_unlock_duration, key_entering->second->get_unlock_duration());
                if(leader0)
                {
                    // we are leader #0 so directly call msg_lock()
                    //
                    // first we remove the entry otherwise we get a duplicate
                    // error since we try to re-add the same ticket
                    //
                    key_entering = obj_entering->second.erase(key_entering);
                    local_locks.push_back(lock_message);
                }
                else
                {
                    // we are not leader #0, so send the message to it
                    //
                    ++key_entering;
                    lock_message.add_parameter(cluck::g_name_cluck_param_serial, key_entering->second->get_serial());
                    f_messenger->send_message(lock_message);
                }
            }
            else
            {
                ++key_entering;
            }
        }
    }
 
    // a ticket may still be unlocked in which case we want to
    // restart the lock process as if still entering
    //
    // if locked, a ticket is assigned leader0 as its new owner so
    // further work on that ticket works as expected
    //
    std::string serialized;
    for(auto obj_ticket(f_tickets.begin()); obj_ticket != f_tickets.end(); ++obj_ticket)
    {
        for(auto key_ticket(obj_ticket->second.begin()); key_ticket != obj_ticket->second.end(); )
        {
            std::string const owner_name(key_ticket->second->get_owner());
            auto key_leader(std::find_if(
                      f_leaders.begin()
                    , f_leaders.end()
                    , [&owner_name](auto const & l)
                    {
                        return l->get_name() == owner_name;
                    }));
            if(key_ticket->second->is_locked())
            {
                // if ticket was locked by the leader that disappeared, we
                // transfer ownership to leader #0
                //
                if(key_leader == f_leaders.end())
                {
                    key_ticket->second->set_owner(f_leaders[0]->get_name());
                }
 
                // and send that ticket to the other leaders to make sure
                // they all agree on its current state
                //
                serialized += key_ticket->second->serialize();
                serialized += '\n';
 
                ++key_ticket;
            }
            else
            {
                // it was not locked yet, restart the LOCK process from
                // the very beginning
                //
                if(key_leader == f_leaders.end())
                {
                    // give new ownership to leader[0]
                    //
                    ed::message lock_message;
                    lock_message.set_command(cluck::g_name_cluck_cmd_lock);
                    lock_message.set_server(f_leaders[0]->get_name());
                    lock_message.set_service(cluck::g_name_cluck_service_name);
                    lock_message.set_sent_from_server(key_ticket->second->get_server_name());
                    lock_message.set_sent_from_service(key_ticket->second->get_service_name());
                    lock_message.add_parameter(cluck::g_name_cluck_param_object_name, key_ticket->second->get_object_name());
                    lock_message.add_parameter(cluck::g_name_cluck_param_tag, key_ticket->second->get_tag());
                    lock_message.add_parameter(cluck::g_name_cluck_param_pid, key_ticket->second->get_client_pid());
                    lock_message.add_parameter(cluck::g_name_cluck_param_timeout, key_ticket->second->get_obtention_timeout());
                    lock_message.add_parameter(cluck::g_name_cluck_param_duration, key_ticket->second->get_lock_duration());
                    lock_message.add_parameter(cluck::g_name_cluck_param_unlock_duration, key_ticket->second->get_unlock_duration());
                    if(leader0)
                    {
                        // we are leader #0 so directly call msg_lock()
                        //
                        key_ticket = obj_ticket->second.erase(key_ticket);
                        local_locks.push_back(lock_message);
                    }
                    else
                    {
                        // we are not leader #0, so send the message to it
                        //
                        ++key_ticket;
                        lock_message.add_parameter(cluck::g_name_cluck_param_serial, key_ticket->second->get_serial());
                        f_messenger->send_message(lock_message);
                    }
                }
                else
                {
                    ++key_ticket;
                }
            }
        }
    }
 
    // we send those after the loops above because the msg_lock() is
    // not unlikely to change the f_entering_tickets map and looping
    // through it when another function is going to modify it is not
    // wise
    //
    for(auto lm : local_locks)
    {
        msg_lock(lm);
    }
 
    // send LOCK_TICkETS if there is serialized ticket data
    //
    if(!serialized.empty())
    {
        ed::message lock_tickets_message;
        lock_tickets_message.set_command(cluck::g_name_cluck_cmd_lock_tickets);
        lock_tickets_message.set_service(cluck::g_name_cluck_service_name);
        lock_tickets_message.add_parameter(cluck::g_name_cluck_param_tickets, serialized);
 
        auto const la(get_leader_a());
        if(la != nullptr)
        {
            lock_tickets_message.set_server(la->get_name());
            f_messenger->send_message(lock_tickets_message);
 
            auto const lb(get_leader_b());
            if(lb != nullptr)
            {
                lock_tickets_message.set_server(lb->get_name());
                f_messenger->send_message(lock_tickets_message);
            }
        }
    }
}
 
 
void cluckd::forward_message_to_leader(ed::message & msg)
{
    // for safety, we do not call this function if the daemon is not
    // considered ready meaning it has at least one leader
    //
    if(f_leaders.empty())
    {
        return; // LCOV_EXCL_LINE
    }
 
    // we are not a leader, we work as a proxy by forwarding the
    // message to a leader, we add our trail so the LOCKED and
    // other messages can be proxied back
    //
    // Note: using the get_sent_from_server() means that we may not
    //       even see the returned message, it may be proxied to another
    //       server directly or through another route
    //
    msg.set_service(cluck::g_name_cluck_service_name);
    msg.add_parameter(cluck::g_name_cluck_param_lock_proxy_server_name, msg.get_sent_from_server());
    msg.add_parameter(cluck::g_name_cluck_param_lock_proxy_service_name, msg.get_sent_from_service());
 
    f_next_leader = (f_next_leader + 1) % f_leaders.size();
    msg.set_server(f_leaders[f_next_leader]->get_name());
 
    f_messenger->send_message(msg);
}
 
 
void cluckd::cleanup()
{
    cluck::timeout_t next_timeout(snapdev::timespec_ex::max());
 
    // when we receive LOCK requests before we have leaders elected, they
    // get added to our cache, so do some cache clean up when not empty
    //
    cluck::timeout_t const now(snapdev::now());
    for(auto c(f_message_cache.begin()); c != f_message_cache.end(); )
    {
        if(c->f_timeout <= now)
        {
            std::string object_name;
            ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
            pid_t client_pid(0);
            cluck::timeout_t timeout;
            if(!get_parameters(c->f_message, &object_name, &tag, &client_pid, &timeout, nullptr, nullptr))
            {
                // we should never cache messages that are invalid
                //
                throw cluck::logic_error("cluck::cleanup() of LOCK message failed get_parameters()."); // LCOV_EXCL_LINE
            }
 
            SNAP_LOG_WARNING
                << "Lock on \""
                << object_name
                << "\" / \""
                << client_pid
                << "\" / \""
                << tag
                << "\" timed out before leaders were known."
                << SNAP_LOG_SEND;
 
            std::string const server_name(c->f_message.has_parameter("lock_proxy_server_name")
                                        ? c->f_message.get_parameter("lock_proxy_server_name")
                                        : c->f_message.get_sent_from_server());
            std::string const service_name(c->f_message.has_parameter("lock_proxy_service_name")
                                         ? c->f_message.get_parameter("lock_proxy_service_name")
                                         : c->f_message.get_sent_from_service());
            std::string const entering_key(server_name + '/' + std::to_string(client_pid));
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.set_service(service_name);
            lock_failed_message.set_server(server_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_timedout);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "cleanup() found a timed out lock");
#endif
            f_messenger->send_message(lock_failed_message);
 
            c = f_message_cache.erase(c);
        }
        else
        {
            if(c->f_timeout < next_timeout)
            {
                next_timeout = c->f_timeout;
            }
            ++c;
        }
    }
 
    // remove any f_ticket that timed out
    //
    for(auto obj_ticket(f_tickets.begin()); obj_ticket != f_tickets.end(); )
    {
        bool try_activate(false);
        for(auto key_ticket(obj_ticket->second.begin()); key_ticket != obj_ticket->second.end(); )
        {
            bool move_next(true);
            if(key_ticket->second->timed_out())
            {
                key_ticket->second->lock_failed("ticket: timed out while cleaning up");
                if(key_ticket->second->timed_out())
                {
                    // still timed out, remove it
                    //
                    key_ticket = obj_ticket->second.erase(key_ticket);
                    try_activate = true;
                    move_next = false;
                }
            }
            if(move_next)
            {
                if(key_ticket->second->get_current_timeout_date() < next_timeout)
                {
                    next_timeout = key_ticket->second->get_current_timeout_date();
                }
                ++key_ticket;
            }
        }
 
        if(obj_ticket->second.empty())
        {
            obj_ticket = f_tickets.erase(obj_ticket);
        }
        else
        {
            if(try_activate)
            {
                // something was erased, a new ticket may be first
                //
                activate_first_lock(obj_ticket->first);
            }
 
            ++obj_ticket;
        }
    }
 
    // remove any f_entering_tickets that timed out
    //
    for(auto obj_entering(f_entering_tickets.begin()); obj_entering != f_entering_tickets.end(); )
    {
        for(auto key_entering(obj_entering->second.begin()); key_entering != obj_entering->second.end(); )
        {
            if(key_entering->second->timed_out())
            {
                key_entering->second->lock_failed("entering ticket: timed out while cleanup");
                if(key_entering->second->timed_out())
                {
                    // still timed out, remove it
                    //
                    key_entering = obj_entering->second.erase(key_entering);
                }
            }
            else
            {
                if(key_entering->second->get_current_timeout_date() < next_timeout)
                {
                    next_timeout = key_entering->second->get_current_timeout_date();
                }
                ++key_entering;
            }
        }
 
        if(obj_entering->second.empty())
        {
            obj_entering = f_entering_tickets.erase(obj_entering);
        }
        else
        {
            ++obj_entering;
        }
    }
 
    // got a new timeout?
    //
    if(next_timeout != snapdev::timespec_ex::max())
    {
        // we add one second to avoid looping like crazy
        // if we timeout just around the "wrong" time
        //
        f_timer->set_timeout_date(next_timeout + cluck::timeout_t(1, 0));
    }
    else
    {
        f_timer->set_timeout_date(-1);
    }
}
 
 
ticket::ticket_id_t cluckd::get_last_ticket(std::string const & object_name)
{
    ticket::ticket_id_t last_ticket(ticket::NO_TICKET);
 
    // Note: There is no need to check the f_entering_tickets list
    //       since that one does not yet have any ticket number assigned
    //       and thus the maximum there would return 0 every time
    //
    auto obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        // note:
        // the std::max_element() algorithm would require many more
        // get_ticket_number() when our loop uses one per ticket
        //
        for(auto key_ticket : obj_ticket->second)
        {
            ticket::ticket_id_t const ticket_number(key_ticket.second->get_ticket_number());
            if(ticket_number > last_ticket)
            {
                last_ticket = ticket_number;
            }
        }
    }
 
    return last_ticket;
}
 
 
void cluckd::set_ticket(
      std::string const & object_name
    , std::string const & key
    , ticket::pointer_t ticket)
{
    f_tickets[object_name][key] = ticket;
}
 
 
ticket::key_map_t const cluckd::get_entering_tickets(std::string const & object_name)
{
    auto const it(f_entering_tickets.find(object_name));
    if(it == f_entering_tickets.end())
    {
        // LCOV_EXCL_START
        //
        // I could not get this covered and I do not think it can happen,
        // also the caller function does not verify that the returned map
        // is valid so it is safer this way
        //
        throw std::logic_error(
              "could not find entering ticket with object name: \""
            + object_name
            + "\".");
        // LCOV_EXCL_STOP
    }
 
    return it->second;
}
 
 
void cluckd::lock_exiting(ed::message & msg)
{
    msg_lock_exiting(msg);
}
 
 
 
 
 
 
std::string cluckd::serialized_tickets()
{
    std::stringstream result;
 
    for(auto const & obj_ticket : f_tickets)
    {
        for(auto const & key_ticket : obj_ticket.second)
        {
            result
                << key_ticket.second->serialize()
                << '\n';
        }
    }
 
    return result.str();
}
 
 
bool cluckd::get_parameters(
      ed::message const & msg
    , std::string * object_name
    , ed::dispatcher_match::tag_t * tag
    , pid_t * client_pid
    , cluck::timeout_t * timeout
    , std::string * key
    , std::string * source)
{
    // get the "object name" (what we are locking)
    // in Snap, the object name is often a URI plus the action we are performing
    //
    if(object_name != nullptr)
    {
        *object_name = msg.get_parameter(cluck::g_name_cluck_param_object_name);
    }
 
    // the same application may want to hold multiple locks simultaneously
    // and this is made possible by using a tag (a 16 bits number)
    //
    if(tag != nullptr)
    {
        *tag = msg.get_integer_parameter(cluck::g_name_cluck_param_tag);
    }
 
    // get the pid (process identifier) of the process that is
    // requesting the lock; this is important to be able to distinguish
    // multiple processes on the same computer requesting a lock
    //
    if(client_pid != nullptr)
    {
        *client_pid = msg.get_integer_parameter(cluck::g_name_cluck_param_pid);
        if(*client_pid < 1)
        {
            // invalid pid
            //
            SNAP_LOG_NOISY_ERROR
                << "cluckd::get_parameters(): invalid pid specified for a lock ("
                << std::to_string(*client_pid)
                << "); it must be a positive decimal number."
                << SNAP_LOG_SEND;
            return false;
        }
    }
 
    // get the time limit we will wait up to before we decide we
    // cannot obtain that lock
    //
    if(timeout != nullptr)
    {
        if(msg.has_parameter(cluck::g_name_cluck_param_timeout))
        {
            // this timeout may already be out of date in which case
            // the lock immediately fails
            //
            *timeout = msg.get_timespec_parameter(cluck::g_name_cluck_param_timeout);
        }
        else
        {
            *timeout = snapdev::now() + cluck::get_lock_obtention_timeout();
        }
    }
 
    // get the key of a ticket or entering object
    //
    if(key != nullptr)
    {
        *key = msg.get_parameter(cluck::g_name_cluck_param_key);
    }
 
    // get the source of a ticket (i.e. <server> '/' <service>)
    //
    if(source != nullptr)
    {
        *source = msg.get_parameter(cluck::g_name_cluck_param_source);
    }
 
    return true;
}
 
 
void cluckd::msg_absolutely(ed::message & msg)
{
    // we may receive the ABSOLUTELY message from anywhere so don't expect
    // the "serial" parameter to be defined
    //
    if(!msg.has_parameter(ed::g_name_ed_param_serial))
    {
        return;
    }
 
    std::string const serial(msg.get_parameter(ed::g_name_ed_param_serial));
    std::vector<std::string> segments;
    snapdev::tokenize_string(segments, serial, "/");
 
    if(segments[0] == "cluckd")
    {
        // check serial as defined in msg_lock()
        //
        if(segments.size() != 4)
        {
            SNAP_LOG_WARNING
                << "ABSOLUTELY reply has an invalid cluckd serial parameters \""
                << serial
                << "\" was expected to have exactly 4 segments."
                << SNAP_LOG_SEND;
 
            ed::message invalid;
            invalid.set_command(ed::g_name_ed_cmd_invalid);
            invalid.reply_to(msg);
            invalid.add_parameter(
                  ed::g_name_ed_param_command
                , msg.get_command());
            invalid.add_parameter(
                  ed::g_name_ed_param_message
                , "invalid number of segments in \""
                + serial
                + "\".");
            f_messenger->send_message(invalid);
 
            return;
        }
 
        // notice how the split() re-split the entering key
        //
        std::string const object_name(segments[1]);
        std::string const server_name(segments[2]);
        std::string const client_pid(segments[3]);
 
        auto entering_ticket(f_entering_tickets.find(object_name));
        if(entering_ticket != f_entering_tickets.end())
        {
            std::string const entering_key(server_name + '/' + client_pid);
            auto key_ticket(entering_ticket->second.find(entering_key));
            if(key_ticket != entering_ticket->second.end())
            {
                // remove the alive timeout
                //
                key_ticket->second->set_alive_timeout(cluck::timeout_t());
 
                // got it! start the bakery algorithm
                //
                key_ticket->second->entering();
            }
        }
    }
 
    // ignore other messages
}
 
 
void cluckd::msg_activate_lock(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    std::string first_key("no-key");
 
    auto ticket(find_first_lock(object_name));
    if(ticket != nullptr)
    {
        // found a lock
        //
        first_key = ticket->get_ticket_key();
 
        if(key == first_key)
        {
            // we can mark this ticket as activated
            //
            ticket->lock_activated();
        }
    }
 
    // always reply, if we could not find the key, then we returned 'no-key'
    // as the key parameter
    //
    ed::message lock_activated_message;
    lock_activated_message.set_command(cluck::g_name_cluck_cmd_lock_activated);
    lock_activated_message.reply_to(msg);
    lock_activated_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
    lock_activated_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
    lock_activated_message.add_parameter(cluck::g_name_cluck_param_key, key);
    lock_activated_message.add_parameter(cluck::g_name_cluck_param_other_key, first_key);
    f_messenger->send_message(lock_activated_message);
 
    // the list of tickets is not unlikely changed so we need to make
    // a call to cleanup to make sure the timer is reset appropriately
    //
    cleanup();
}
 
 
void cluckd::msg_add_ticket(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    cluck::timeout_t timeout;
    if(!get_parameters(msg, &object_name, &tag, nullptr, &timeout, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // make sure the ticket is unique
    //
    auto const obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        auto const key_ticket(obj_ticket->second.find(key));
        if(key_ticket != obj_ticket->second.end())
        {
            SNAP_LOG_ERROR
                << "an existing ticket has the same object name \""
                << object_name
                << "\" ("
                << tag
                << ") and key \""
                << key
                << "\"."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_duplicate);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "ADD_TICKET called with an existing ticket name and key");
#endif
            f_messenger->send_message(lock_failed_message);
 
            communicator::flag::pointer_t flag(COMMUNICATOR_FLAG_UP(
                          "cluckd"
                        , "ticket"
                        , "invalid-algorithm"
                        , "msg_add_ticket() received a second call to add the"
                          " same ticket. This either means there is a bug in our"
                          " algorithm or there is a hacker sending us messages"
                          " trying to create invalid tickets."
                    ));
            flag->set_priority(25);
            flag->add_tag("bug");
            flag->set_manual_down(true);
            flag->save();
 
            return;
        }
    }
 
    // the client_pid parameter is part of the key (3rd segment)
    //
    std::vector<std::string> segments;
    snapdev::tokenize_string(segments, key, "/");
    if(segments.size() != 3)
    {
        SNAP_LOG_ERROR
            << "Expected exactly 3 segments in \""
            << key
            << "\" to add a ticket."
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "ADD_TICKET called with an invalid key (expected three segments)");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    // TODO: we probably want to look at using a function which returns false
    //       instead of having to do a try/catch
    //
    bool ok(true);
    std::uint32_t number(0);
    try
    {
        number = snapdev::hex_to_int<std::uint32_t>(segments[0]);
    }
    catch(snapdev::hexadecimal_string_exception const &)
    {
        ok = false;
    }
    catch(snapdev::hexadecimal_string_out_of_range const &)
    {
        ok = false;
    }
    if(!ok)
    {
        SNAP_LOG_ERROR
            << "somehow ticket number \""
            << segments[0]
            << "\" is not a valid hexadecimal number."
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "ADD_TICKET called with an invalid key (first segment is not a valid hexadecimal number)");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    // by now all the leaders should already have
    // an entering ticket for that one ticket
    //
    auto const obj_entering_ticket(f_entering_tickets.find(object_name));
    if(obj_entering_ticket == f_entering_tickets.end())
    {
        SNAP_LOG_ERROR
            << "Expected entering ticket object for \""
            << object_name
            << "\" not found when adding a ticket."
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "ADD_TICKET could not find an entering ticket with the specified object_name");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    // the key we need to search is not the new ticket key but the
    // entering key, build it from the segments
    //
    std::string const entering_key(segments[1] + '/' + segments[2]);
    auto const key_entering_ticket(obj_entering_ticket->second.find(entering_key));
    if(key_entering_ticket == obj_entering_ticket->second.end())
    {
        SNAP_LOG_ERROR
            << "Expected entering ticket key for \""
            << object_name
            << "\" not found when adding a ticket."
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "ADD_TICKET could not find the very entering ticket with the specified key");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    // make it an official ticket now
    //
    // this should happen on all cluck daemon other than the one that
    // first received the LOCK message
    //
    set_ticket(object_name, key, key_entering_ticket->second);
 
    // WARNING: the set_ticket_number() function has the same side
    //          effects as the add_ticket() function without the
    //          f_messenger->send_message() call
    //
    f_tickets[object_name][key]->set_ticket_number(number);
 
    ed::message ticket_added_message;
    ticket_added_message.set_command(cluck::g_name_cluck_cmd_ticket_added);
    ticket_added_message.reply_to(msg);
    ticket_added_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
    ticket_added_message.add_parameter(cluck::g_name_cluck_param_key, key);
    ticket_added_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
    f_messenger->send_message(ticket_added_message);
}
 
 
void cluckd::msg_clock_stable(ed::message & msg)
{
    if(!f_stable_clock)
    {
        f_stable_clock = msg.get_parameter(communicator::g_name_communicator_param_clock_resolution)
                                        == communicator::g_name_communicator_value_verified;
 
        check_lock_status();
    }
}
 
 
void cluckd::msg_cluster_down(ed::message & msg)
{
    snapdev::NOT_USED(msg);
 
    SNAP_LOG_INFO
        << "cluster is down, canceling existing locks and we have to"
           " refuse any further lock requests for a while."
        << SNAP_LOG_SEND;
 
    // in this case, we cannot safely keep the leaders
    //
    f_leaders.clear();
 
    // in case services listen to the NO_LOCK, let them know it's gone
    //
    check_lock_status();
 
    // we do not call the lock_gone() because the HANGUP will be sent
    // if required so we do not have to do that twice
}
 
 
void cluckd::msg_cluster_up(ed::message & msg)
{
    f_neighbors_count = msg.get_integer_parameter("neighbors_count");
    f_neighbors_quorum = f_neighbors_count / 2 + 1;
 
    computer::priority_t priority(computer::PRIORITY_OFF);
    std::string candidate_priority(f_opts.get_string("candidate-priority"));
    if(candidate_priority != "off")
    {
        priority = f_opts.get_long("candidate-priority"
                                , 0
                                , computer::PRIORITY_USER_MIN
                                , computer::PRIORITY_MAX);
    }
 
#ifdef _DEBUG
    // the READY message is expected to happen first and setup this parameter
    //
    if(f_my_ip_address.is_default())
    {
        throw cluck::logic_error("cluckd::msg_cluster_up(): somehow f_my_ip_address is still the default in msg_cluster_up()."); // LCOV_EXCL_LINE
    }
#endif
 
    // add ourselves to the list of computers; mark us connected; get our ID
    //
    f_computers[f_server_name] = std::make_shared<computer>(f_server_name, priority, f_my_ip_address);
    f_computers[f_server_name]->set_start_time(f_start_time);
    f_computers[f_server_name]->set_connected(true);
    f_my_id = f_computers[f_server_name]->get_id();
 
    SNAP_LOG_INFO
        << "cluster is up with "
        << f_neighbors_count
        << " neighbors, attempt an election"
           " then check for leaders by sending a LOCK_STARTED message."
        << SNAP_LOG_SEND;
 
    election_status();
    send_lock_started(nullptr);
    check_lock_status();
}
 
 
void cluckd::msg_drop_ticket(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    std::vector<std::string> segments;
    snapdev::tokenize_string(segments, key, "/");
 
    // drop the regular ticket
    //
    // if we have only 2 segments, then there is no corresponding ticket
    // since tickets are added only once we have a ticket_id
    //
    std::string entering_key;
    if(segments.size() == 3)
    {
        auto obj_ticket(f_tickets.find(object_name));
        if(obj_ticket != f_tickets.end())
        {
            auto key_ticket(obj_ticket->second.find(key));
            if(key_ticket != obj_ticket->second.end())
            {
                obj_ticket->second.erase(key_ticket);
            }
 
            if(obj_ticket->second.empty())
            {
                f_tickets.erase(obj_ticket);
            }
 
            // one ticket was erased, another may be first now
            //
            activate_first_lock(object_name);
        }
 
        // we received the ticket_id in the message, so
        // we have to regenerate the entering_key without
        // the ticket_id (which is the first element)
        //
        entering_key = segments[1] + '/' + segments[2];
    }
    else
    {
        // we received the entering_key in the message, use as is
        //
        entering_key = key;
    }
 
    // drop the entering ticket
    //
    auto obj_entering_ticket(f_entering_tickets.find(object_name));
    if(obj_entering_ticket != f_entering_tickets.end())
    {
        auto key_entering_ticket(obj_entering_ticket->second.find(entering_key));
        if(key_entering_ticket != obj_entering_ticket->second.end())
        {
            obj_entering_ticket->second.erase(key_entering_ticket);
        }
 
        if(obj_entering_ticket->second.empty())
        {
            f_entering_tickets.erase(obj_entering_ticket);
        }
    }
 
    // the list of tickets is not unlikely changed so we need to make
    // a call to cleanup to make sure the timer is reset appropriately
    //
    cleanup();
}
 
 
void cluckd::msg_get_max_ticket(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // remove any f_tickets that timed out by now because these should
    // not be taken in account in the max. computation
    //
    cleanup();
 
    ticket::ticket_id_t const last_ticket(get_last_ticket(object_name));
 
    ed::message reply;
    reply.set_command(cluck::g_name_cluck_cmd_max_ticket);
    reply.reply_to(msg);
    reply.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
    reply.add_parameter(cluck::g_name_cluck_param_key, key);
    reply.add_parameter(cluck::g_name_cluck_param_tag, tag);
    reply.add_parameter(cluck::g_name_cluck_param_ticket_id, last_ticket);
    f_messenger->send_message(reply);
}
 
 
void cluckd::msg_list_tickets(ed::message & msg)
{
    ed::message list_message;
    list_message.set_command(cluck::g_name_cluck_cmd_ticket_list);
    list_message.reply_to(msg);
    list_message.add_parameter(cluck::g_name_cluck_param_list, ticket_list());
    f_messenger->send_message(list_message);
}
 
 
void cluckd::msg_lock(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    pid_t client_pid(0);
    cluck::timeout_t timeout;
    if(!get_parameters(msg, &object_name, &tag, &client_pid, &timeout, nullptr, nullptr))
    {
        return;
    }
 
    // do some cleanup as well
    //
    cleanup();
 
    // if we are a leader, create an entering key
    //
    std::string const server_name(msg.has_parameter(cluck::g_name_cluck_param_lock_proxy_server_name)
                                ? msg.get_parameter(cluck::g_name_cluck_param_lock_proxy_server_name)
                                : msg.get_sent_from_server());
 
    std::string const service_name(msg.has_parameter(cluck::g_name_cluck_param_lock_proxy_service_name)
                                ? msg.get_parameter(cluck::g_name_cluck_param_lock_proxy_service_name)
                                : msg.get_sent_from_service());
 
    std::string const entering_key(server_name + '/' + std::to_string(client_pid));
 
    if(timeout <= snapdev::now())
    {
        SNAP_LOG_WARNING
            << "lock on \""
            << object_name
            << "\" ("
            << tag
            << ")/ \""
            << client_pid
            << "\" timed out before we could start the locking process."
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_timedout);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK timeout date is already in the past");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    cluck::timeout_t const duration(msg.get_timespec_parameter(cluck::g_name_cluck_param_duration));
    if(duration < cluck::CLUCK_MINIMUM_TIMEOUT)
    {
        // duration too small
        //
        SNAP_LOG_ERROR
            << duration
            << " is an invalid duration, the minimum accepted is "
            << cluck::CLUCK_MINIMUM_TIMEOUT
            << '.'
            << SNAP_LOG_SEND;
 
        ed::message lock_failed_message;
        lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
        lock_failed_message.reply_to(msg);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
        lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK called with a duration that is too small");
#endif
        f_messenger->send_message(lock_failed_message);
 
        return;
    }
 
    cluck::timeout_t unlock_duration(cluck::CLUCK_DEFAULT_TIMEOUT);
    if(msg.has_parameter(cluck::g_name_cluck_param_unlock_duration))
    {
        unlock_duration = msg.get_timespec_parameter(cluck::g_name_cluck_param_unlock_duration);
        if(unlock_duration < cluck::CLUCK_UNLOCK_MINIMUM_TIMEOUT)
        {
            // invalid duration, minimum is cluck::CLUCK_UNLOCK_MINIMUM_TIMEOUT
            //
            SNAP_LOG_ERROR
                << unlock_duration
                << " is an invalid unlock duration, the minimum accepted is "
                << cluck::CLUCK_UNLOCK_MINIMUM_TIMEOUT
                << '.'
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK called with an unlock duration that is too small");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
    }
 
    if(!is_daemon_ready())
    {
        SNAP_LOG_TRACE
            << "caching LOCK message for \""
            << object_name
            << "\" ("
            << tag
            << ") as the cluck system is not yet considered ready."
            << SNAP_LOG_SEND;
 
        f_message_cache.emplace_back(timeout, msg);
 
        // make sure the cache gets cleaned up if the message times out
        //
        std::int64_t const timeout_date(f_timer->get_timeout_date());
        if(timeout_date == -1
        || cluck::timeout_t(timeout_date / 1'000'000, timeout_date % 1'000'000) > timeout)
        {
            f_timer->set_timeout_date(timeout);
        }
        return;
    }
 
    if(is_leader() == nullptr)
    {
        // we are not a leader, we need to forward the message to one
        // of the leaders instead
        //
        forward_message_to_leader(msg);
        return;
    }
 
    // make sure this is a new ticket
    //
    auto entering_ticket(f_entering_tickets.find(object_name));
    if(entering_ticket != f_entering_tickets.end())
    {
        auto key_ticket(entering_ticket->second.find(entering_key));
        if(key_ticket != entering_ticket->second.end())
        {
            // if this is a re-LOCK, then it may be a legitimate duplicate
            // in which case we do not want to generate a LOCK_FAILED error
            //
            if(msg.has_parameter(cluck::g_name_cluck_param_serial))
            {
                ticket::serial_t const serial(msg.get_integer_parameter(cluck::g_name_cluck_param_serial));
                if(key_ticket->second->get_serial() == serial)
                {
                    // legitimate double request from leaders
                    // (this happens when a leader dies and we have to restart
                    // a lock negotiation)
                    //
                    return;
                }
            }
 
            // the object already exists... do not allow duplicates
            //
            SNAP_LOG_ERROR
                << "an entering ticket has the same object name \""
                << object_name
                << "\" ("
                << tag
                << ") and entering key \""
                << entering_key
                << "\"."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_duplicate);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK called with the same entering ticket object_name and entering_key");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
        if(entering_ticket->second.size() >= cluck::CLUCK_MAXIMUM_ENTERING_LOCKS)
        {
            // this is a failure in the algorithm (unfortunately), if you
            // send LOCK commands without much pause, the number of entering
            // ticket can grow forever; the following is a way to avoid
            // that situation by preventing such inconsiderate growth.
            //
            SNAP_LOG_ERROR
                << "too many entering tickets for object name \""
                << object_name
                << "\"."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_overflow);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK called too quickly that the number of entering tickets overflowed");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
    }
 
    // make sure there is not a ticket with the same name already defined
    //
    // (this is is really important so we can actually properly UNLOCK an
    // existing lock since we use the same search and if two entries were
    // to be the same we could not know which to unlock; there are a few
    // other places where such a search is used actually...)
    //
    auto obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        auto key_ticket(std::find_if(
                  obj_ticket->second.begin()
                , obj_ticket->second.end()
                , [&entering_key](auto const & t)
                {
                    return t.second->get_entering_key() == entering_key;
                }));
        if(key_ticket != obj_ticket->second.end())
        {
            // there is already a ticket with this object name/entering key
            //
            SNAP_LOG_ERROR
                << "a ticket has the same object name \""
                << object_name
                << "\" ("
                << tag
                << ") and entering key \""
                << entering_key
                << "\"."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, entering_key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_duplicate);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK called with the same ticket object_name and entering_key");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
    }
 
    ticket::pointer_t ticket(std::make_shared<ticket>(
                                  this
                                , f_messenger
                                , object_name
                                , tag
                                , entering_key
                                , timeout
                                , duration
                                , server_name
                                , service_name));
 
    f_entering_tickets[object_name][entering_key] = ticket;
 
    // finish up ticket initialization
    //
    ticket->set_unlock_duration(unlock_duration);
 
    // generate a serial number for that ticket
    //
    f_ticket_serial = (f_ticket_serial + 1) & 0x00FFFFFF;   // 0 is a valid serial number (-1 is not)
    if(f_leaders[0]->get_id() != f_my_id)
    {
        if(f_leaders.size() >= 2
        && f_leaders[1]->get_id() != f_my_id)
        {
            f_ticket_serial |= 1 << 24;
        }
        else if(f_leaders.size() >= 3
             && f_leaders[2]->get_id() != f_my_id)
        {
            f_ticket_serial |= 2 << 24;
        }
    }
    ticket->set_serial(f_ticket_serial);
 
    if(msg.has_parameter(cluck::g_name_cluck_param_serial))
    {
        // if we have a "serial" number in that message, we lost a leader
        // and when that happens we are not unlikely to have lost the
        // client that requested the LOCK, send an ALIVE message to make
        // sure that the client still exists before entering the ticket
        //
        // TODO: we may want to make this 5s a parameter that we can change
        //
        ticket->set_alive_timeout(snapdev::now() + cluck::timeout_t(5, 0));
 
        ed::message alive_message;
        alive_message.set_command(ed::g_name_ed_cmd_alive);
        alive_message.set_server(server_name);
        alive_message.set_service(service_name);
        alive_message.add_parameter(ed::g_name_ed_param_serial, "cluckd/" + object_name + '/' + entering_key);
        alive_message.add_parameter(ed::g_name_ed_param_timestamp, snapdev::now());
        f_messenger->send_message(alive_message);
    }
    else
    {
        // act on the new ticket
        //
        ticket->entering();
    }
 
    // the list of tickets changed, make sure we update the timeout timer
    //
    cleanup();
}
 
 
void cluckd::msg_lock_activated(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    std::string const & other_key(msg.get_parameter(cluck::g_name_cluck_param_other_key));
    if(other_key == key)
    {
        auto obj_ticket(f_tickets.find(object_name));
        if(obj_ticket != f_tickets.end())
        {
            auto key_ticket(obj_ticket->second.find(key));
            if(key_ticket != obj_ticket->second.end())
            {
                // that key is still here!
                // time to activate
                //
                key_ticket->second->lock_activated();
            }
        }
    }
}
 
 
void cluckd::msg_lock_entered(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    auto const obj_entering_ticket(f_entering_tickets.find(object_name));
    if(obj_entering_ticket != f_entering_tickets.end())
    {
        auto const key_entering_ticket(obj_entering_ticket->second.find(key));
        if(key_entering_ticket != obj_entering_ticket->second.end())
        {
            key_entering_ticket->second->entered();
        }
    }
}
 
 
void cluckd::msg_lock_entering(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    cluck::timeout_t timeout;
    std::string key;
    std::string source;
    if(!get_parameters(msg, &object_name, &tag, nullptr, &timeout, &key, &source))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // lock still in the future?
    //
    if(timeout <= snapdev::now())
    {
        SNAP_LOG_DEBUG
            << "received LOCK_ENTERING for \""
            << object_name
            << "\" that already timed out."
            << SNAP_LOG_SEND;
        return;
    }
 
    // do we have enough leaders?
    //
    if(!is_daemon_ready())
    {
        SNAP_LOG_DEBUG
            << "received LOCK_ENTERING while we are thinking we are not ready."
            << SNAP_LOG_SEND;
        return;
    }
 
    // the entering is just a flag (i.e. entering[i] = true)
    // in our case the existance of a ticket is enough to know
    // that we entered
    //
    bool allocate(true);
    auto const obj_ticket(f_entering_tickets.find(object_name));
    if(obj_ticket != f_entering_tickets.end())
    {
        auto const key_ticket(obj_ticket->second.find(key));
        allocate = key_ticket == obj_ticket->second.end();
    }
    if(allocate)
    {
        // ticket does not exist, so create it now
        // (note: ticket should only exist on originator)
        //
        cluck::timeout_t const duration(msg.get_timespec_parameter(cluck::g_name_cluck_param_duration));
        if(duration < cluck::CLUCK_MINIMUM_TIMEOUT)
        {
            // invalid duration
            //
            SNAP_LOG_ERROR
                << duration
                << " is an invalid duration, the minimum accepted is "
                << cluck::CLUCK_MINIMUM_TIMEOUT
                << "."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK_ENTERING called with a duration which is too small");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
 
        cluck::timeout_t unlock_duration(cluck::CLUCK_DEFAULT_TIMEOUT);
        if(msg.has_parameter(cluck::g_name_cluck_param_unlock_duration))
        {
            unlock_duration = msg.get_timespec_parameter(cluck::g_name_cluck_param_unlock_duration);
            if(unlock_duration != cluck::CLUCK_DEFAULT_TIMEOUT
            && unlock_duration < cluck::CLUCK_UNLOCK_MINIMUM_TIMEOUT)
            {
                // invalid duration, minimum is 60
                //
                SNAP_LOG_ERROR
                    << unlock_duration
                    << " is an invalid unlock duration, the minimum accepted is "
                    << cluck::CLUCK_UNLOCK_MINIMUM_TIMEOUT
                    << "."
                    << SNAP_LOG_SEND;
 
                ed::message lock_failed_message;
                lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
                lock_failed_message.reply_to(msg);
                lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
                lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
                lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
                lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
                lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK_ENTERING called with an unlock duration which is too small or set to the default value");
#endif
                f_messenger->send_message(lock_failed_message);
 
                return;
            }
        }
 
        // we have to know where this message comes from
        //
        std::vector<std::string> source_segments;
        if(snapdev::tokenize_string(source_segments, source, "/") != 2)
        {
            SNAP_LOG_ERROR
                << "Invalid number of parameters in source parameter \""
                << source
                << "\" (found "
                << source_segments.size()
                << ", expected 2)."
                << SNAP_LOG_SEND;
 
            ed::message lock_failed_message;
            lock_failed_message.set_command(cluck::g_name_cluck_cmd_lock_failed);
            lock_failed_message.reply_to(msg);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_tag, tag);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_key, key);
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_error, cluck::g_name_cluck_value_invalid);
#ifndef CLUCKD_OPTIMIZATIONS
            lock_failed_message.add_parameter(cluck::g_name_cluck_param_description, "LOCK_ENTERING called with an invalid source parameter");
#endif
            f_messenger->send_message(lock_failed_message);
 
            return;
        }
 
        ticket::pointer_t ticket(std::make_shared<ticket>(
                                  this
                                , f_messenger
                                , object_name
                                , tag
                                , key
                                , timeout
                                , duration
                                , source_segments[0]
                                , source_segments[1]));
 
        f_entering_tickets[object_name][key] = ticket;
 
        // finish up on ticket initialization
        //
        ticket->set_owner(msg.get_sent_from_server());
        ticket->set_unlock_duration(unlock_duration);
        ticket->set_serial(msg.get_integer_parameter(cluck::g_name_cluck_param_serial));
    }
 
    ed::message reply;
    reply.set_command(cluck::g_name_cluck_cmd_lock_entered);
    reply.reply_to(msg);
    reply.add_parameter(cluck::g_name_cluck_param_object_name, object_name);
    reply.add_parameter(cluck::g_name_cluck_param_tag, tag);
    reply.add_parameter(cluck::g_name_cluck_param_key, key);
    f_messenger->send_message(reply);
 
    cleanup();
}
 
 
void cluckd::msg_lock_exiting(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // when exiting we just remove the entry with that key
    //
    auto const obj_entering(f_entering_tickets.find(object_name));
    if(obj_entering != f_entering_tickets.end())
    {
        auto const key_entering(obj_entering->second.find(key));
        if(key_entering != obj_entering->second.end())
        {
            obj_entering->second.erase(key_entering);
 
            // we also want to remove it from the ticket f_entering
            // map if it is there (older ones are there!)
            //
            bool run_activation(false);
            auto const obj_ticket(f_tickets.find(object_name));
            if(obj_ticket != f_tickets.end())
            {
                for(auto const & key_ticket : obj_ticket->second)
                {
                    key_ticket.second->remove_entering(key);
                    run_activation = true;
                }
            }
            if(run_activation)
            {
                // hmm... looking at the code closer, it seems that this is
                // still very much necessary
                //
                // --------------------------------------------------
                //
                // try to activate the lock right now since it could
                // very well be the only ticket and that is exactly
                // when it is viewed as active!
                //
                // Note: this is from my old version, if I am correct
                //       it cannot happen anymore because (1) this is
                //       not the owner so the activation would not
                //       take anyway and (2) the ticket is not going
                //       to be marked as being ready at this point
                //       (that happens later)
                //
                //       XXX we probably should remove this statement
                //           and the run_activation flag which would
                //           then be useless
                //
                activate_first_lock(object_name);
            }
 
            if(obj_entering->second.empty())
            {
                f_entering_tickets.erase(obj_entering);
            }
        }
        else
        {
            SNAP_LOG_WARNING
                << "entering lock \""
                << object_name
                << "\" with key \""
                << key
                << "\" in LOCK_EXITING specified lock not found."
                << SNAP_LOG_SEND;
        }
    }
    else
    {
        SNAP_LOG_WARNING
            << "LOCK_EXITING specified lock \""
            << object_name
            << "\" not found."
            << SNAP_LOG_SEND;
    }
 
    // the list of tickets is not unlikely changed so we need to make
    // a call to cleanup to make sure the timer is reset appropriately
    //
    cleanup();
}
 
 
void cluckd::msg_lock_failed(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    std::string errmsg("get LOCK_FAILED: ");
    errmsg += msg.get_parameter("error");
 
    std::string forward_server;
    std::string forward_service;
 
    // remove f_entering_tickets entries if we find matches there
    //
    auto obj_entering(f_entering_tickets.find(object_name));
    if(obj_entering != f_entering_tickets.end())
    {
        auto key_entering(obj_entering->second.find(key));
        if(key_entering != obj_entering->second.end())
        {
            forward_server = key_entering->second->get_server_name();
            forward_service = key_entering->second->get_service_name();
 
            obj_entering->second.erase(key_entering);
 
            errmsg += " -- happened while entering";
        }
 
        if(obj_entering->second.empty())
        {
            f_entering_tickets.erase(obj_entering);
        }
    }
 
    // remove any f_tickets entries if we find matches there
    //
    auto obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        bool try_activate(false);
        auto key_ticket(obj_ticket->second.find(key));
        if(key_ticket == obj_ticket->second.end())
        {
            key_ticket = std::find_if(
                      obj_ticket->second.begin()
                    , obj_ticket->second.end()
                    , [&key](auto const & t)
                    {
                        return t.second->get_entering_key() == key;
                    });
        }
        if(key_ticket != obj_ticket->second.end())
        {
            // Note: if we already found it in the f_entering_tickets then
            //       the server and service names are going to be exactly
            //       the same so there is no need to test that here
            //
            forward_server = key_ticket->second->get_server_name();
            forward_service = key_ticket->second->get_service_name();
 
            obj_ticket->second.erase(key_ticket);
            try_activate = true;
 
            errmsg += " -- happened when locked"; // TBD: are we really always locked in this case?
        }
 
        if(obj_ticket->second.empty())
        {
            f_tickets.erase(obj_ticket);
        }
        else if(try_activate)
        {
            // something was erased, a new ticket may be first
            //
            activate_first_lock(obj_ticket->first);
        }
    }
 
    if(!forward_server.empty()
    && !forward_service.empty())
    {
        // we deleted an entry, forward the message to the service
        // that requested that lock
        //
        msg.set_server(forward_server);
        msg.set_service(forward_service);
        f_messenger->send_message(msg);
 
        errmsg += " -> ";
        errmsg += forward_server;
        errmsg += ":";
        errmsg += forward_service;
    }
 
    SNAP_LOG_IMPORTANT
        << errmsg
        << '.'
        << SNAP_LOG_SEND;
 
    // the list of tickets is not unlikely changed so we need to make
    // a call to cleanup to make sure the timer is reset appropriately
    //
    cleanup();
}
 
 
void cluckd::msg_lock_leaders(ed::message & msg)
{
    f_election_date = msg.get_timespec_parameter(cluck::g_name_cluck_param_election_date);
 
    // save the new leaders in our own list
    //
    f_leaders.clear();
    for(int idx(0); idx < 3; ++idx)
    {
        std::string const param_name(cluck::g_name_cluck_param_leader + std::to_string(idx));
        if(msg.has_parameter(param_name))
        {
            computer::pointer_t leader(std::make_shared<computer>());
            std::string const lockid(msg.get_parameter(param_name));
            if(leader->set_id(lockid))
            {
                computer::map_t::iterator exists(f_computers.find(leader->get_name()));
                if(exists != f_computers.end())
                {
                    // it already exists, use our existing instance
                    //
                    f_leaders.push_back(exists->second);
                }
                else
                {
                    // we do not yet know of that computer, even though
                    // it is a leader! (i.e. we are not yet aware that
                    // somehow we are connected to it)
                    //
                    leader->set_connected(false);
                    f_computers[leader->get_name()] = leader;
 
                    f_leaders.push_back(leader);
                }
            }
        }
    }
 
    if(!f_leaders.empty())
    {
        synchronize_leaders();
 
        // set the round-robin position to a random value
        //
        // note: I know the result is likely skewed, c will be set to
        // a number between 0 and 255 and modulo 3 means that you get
        // one extra zero (255 % 3 == 0); however, there are 85 times
        // 3 in 255 so it probably won't be noticeable.
        //
        std::uint8_t c;
        RAND_bytes(reinterpret_cast<unsigned char *>(&c), sizeof(c));
        f_next_leader = c % f_leaders.size();
    }
 
    // the is_daemon_ready() function depends on having f_leaders defined
    // and when that happens we may need to empty our cache
    //
    check_lock_status();
}
 
 
void cluckd::msg_lock_started(ed::message & msg)
{
    // I do not think we would ever message ourselves, but in case it happens
    // the rest of the function does not support that case
    //
    std::string const server_name(msg.get_parameter(communicator::g_name_communicator_param_server_name));
    if(server_name == f_server_name)
    {
        return;
    }
 
    cluck::timeout_t const start_time(msg.get_timespec_parameter(cluck::g_name_cluck_param_start_time));
 
    computer::map_t::iterator it(f_computers.find(server_name));
    bool new_computer(it == f_computers.end());
    if(new_computer)
    {
        // create a computer instance so we know it exists
        //
        computer::pointer_t computer(std::make_shared<computer>());
 
        // fill the fields from the "lock_id" parameter
        //
        if(!computer->set_id(msg.get_parameter(cluck::g_name_cluck_param_lock_id)))
        {
            // this is not a valid identifier, ignore altogether
            //
            return;
        }
        computer->set_start_time(start_time);
 
#ifdef _DEBUG
        // LCOV_EXCL_START
        if(computer->get_name() != server_name)
        {
            throw cluck::logic_error("cluckd::msg_lock_started(): server_name ("
                + server_name
                + ") does not match the new computer name ("
                + computer->get_name()
                + ").");
        }
        // LCOV_EXCL_STOP
#endif
        f_computers[computer->get_name()] = computer;
    }
    else
    {
        if(!it->second->get_connected())
        {
            // we heard of this computer (because it is/was a leader) but
            // we had not yet received a LOCK_STARTED message from it; so here
            // we consider it a new computer and will reply to the LOCK_STARTED
            //
            new_computer = true;
            it->second->set_connected(true);
        }
 
        if(it->second->get_start_time() != start_time)
        {
            // when the start time changes that means cluckd
            // restarted which can happen without communicatord
            // restarting so here we would not know about the feat
            // without this parameter and in this case it is very
            // much the same as a new computer so send it a
            // LOCK_STARTED message back
            //
            new_computer = true;
            it->second->set_start_time(start_time);
        }
    }
 
    // keep the newest election results
    //
    computer::pointer_t old_leader(is_leader());
    if(msg.has_parameter(cluck::g_name_cluck_param_election_date))
    {
        snapdev::timespec_ex const election_date(msg.get_timespec_parameter(cluck::g_name_cluck_param_election_date));
        if(election_date > f_election_date)
        {
            f_election_date = election_date;
            f_leaders.clear();
        }
    }
 
    if(f_leaders.empty())
    {
        for(int idx(0); idx < 3; ++idx)
        {
            std::string const param_name(cluck::g_name_cluck_param_leader + std::to_string(idx));
            if(msg.has_parameter(param_name))
            {
                computer::pointer_t leader(std::make_shared<computer>());
                std::string const lockid(msg.get_parameter(param_name));
                if(leader->set_id(lockid))
                {
                    computer::map_t::iterator exists(f_computers.find(leader->get_name()));
                    if(exists != f_computers.end())
                    {
                        // it already exists, use our existing instance
                        //
                        f_leaders.push_back(exists->second);
                    }
                    else
                    {
                        // we do not yet know of that computer, even though
                        // it is a leader! (i.e. we are not yet aware that
                        // somehow we are connected to it)
                        //
                        leader->set_connected(false);
                        f_computers[leader->get_name()] = leader;
 
                        f_leaders.push_back(leader);
                    }
                }
            }
        }
    }
 
    election_status();
 
    // this can have an effect on the lock statuses
    //
    check_lock_status();
 
    if(new_computer
    || old_leader != is_leader())
    {
        // send a reply if that was a new computer
        //
        send_lock_started(&msg);
    }
}
 
 
void cluckd::msg_lock_status(ed::message & msg)
{
    ed::message status_message;
    status_message.set_command(is_daemon_ready()
            ? cluck::g_name_cluck_cmd_lock_ready
            : cluck::g_name_cluck_cmd_no_lock);
 
    status_message.reply_to(msg);
    status_message.add_parameter(communicator::g_name_communicator_param_cache, communicator::g_name_communicator_value_no);
    f_messenger->send_message(status_message);
}
 
 
void cluckd::msg_lock_tickets(ed::message & msg)
{
    std::string const tickets(msg.get_parameter(cluck::g_name_cluck_param_tickets));
 
    // we have one ticket per line, so we first split per line and then
    // work on lines one at a time
    //
    bool added_tickets(false);
    std::list<std::string> lines;
    snapdev::tokenize_string(lines, tickets, "\n", true);
    for(auto const & l : lines)
    {
        ticket::pointer_t t;
        std::list<std::string> vars;
        snapdev::tokenize_string(vars, tickets, "|", true);
        auto object_name_value(std::find_if(
                  vars.begin()
                , vars.end()
                , [](std::string const & vv)
                {
                    return vv.starts_with("object_name=");
                }));
        if(object_name_value != vars.end())
        {
            auto entering_key_value(std::find_if(
                      vars.begin()
                    , vars.end()
                    , [](std::string const & vv)
                    {
                        return vv.starts_with("entering_key=");
                    }));
            if(entering_key_value != vars.end())
            {
                // extract the values which start after the '=' sign
                //
                std::string const object_name(object_name_value->substr(12));
                std::string const entering_key(entering_key_value->substr(13));
 
                auto entering_ticket(f_entering_tickets.find(object_name));
                if(entering_ticket != f_entering_tickets.end())
                {
                    auto key_ticket(entering_ticket->second.find(entering_key));
                    if(key_ticket != entering_ticket->second.end())
                    {
                        t = key_ticket->second;
                    }
                }
                if(t == nullptr)
                {
                    auto obj_ticket(f_tickets.find(object_name));
                    if(obj_ticket != f_tickets.end())
                    {
                        auto key_ticket(std::find_if(
                                  obj_ticket->second.begin()
                                , obj_ticket->second.end()
                                , [&entering_key](auto const & o)
                                {
                                    return o.second->get_entering_key() == entering_key;
                                }));
                        if(key_ticket != obj_ticket->second.end())
                        {
                            t = key_ticket->second;
                        }
                    }
                }
 
                // ticket exists? if not create a new one
                //
                bool const new_ticket(t == nullptr);
                if(new_ticket)
                {
                    // create a new ticket, some of the parameters are there just
                    // because they are required; they will be replaced by the
                    // unserialize call below...
                    //
                    t = std::make_shared<ticket>(
                                  this
                                , f_messenger
                                , object_name
                                , ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG
                                , entering_key
                                , cluck::CLUCK_DEFAULT_TIMEOUT + snapdev::now()
                                , cluck::CLUCK_DEFAULT_TIMEOUT
                                , f_server_name
                                , cluck::g_name_cluck_service_name);
                }
 
                t->unserialize(l);
                added_tickets = true;
 
                // do a couple of additional sanity tests to
                // make sure that we want to keep new tickets
                //
                // first make sure it is marked as "locked"
                //
                // second check that the owner is a leader that
                // exists (the sender uses a LOCK message for
                // locks that are not yet locked or require
                // a new owner)
                //
                if(new_ticket
                && t->is_locked())
                {
                    auto li(std::find_if(
                              f_leaders.begin()
                            , f_leaders.end()
                            , [&t](auto const & c)
                            {
                                return t->get_owner() == c->get_name();
                            }));
                    if(li != f_leaders.end())
                    {
                        f_tickets[object_name][t->get_ticket_key()] = t;
                    }
                }
            }
        }
    }
 
    // if we updated some tickets, we need to make sure our timer is setup
    // appropriately
    //
    if(added_tickets)
    {
        cleanup();
    }
}
 
 
void cluckd::msg_max_ticket(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // the MAX_TICKET is an answer that has to go in a still un-added ticket
    //
    auto const obj_entering_ticket(f_entering_tickets.find(object_name));
    if(obj_entering_ticket != f_entering_tickets.end())
    {
        auto const key_entering_ticket(obj_entering_ticket->second.find(key));
        if(key_entering_ticket != obj_entering_ticket->second.end())
        {
            key_entering_ticket->second->max_ticket(msg.get_integer_parameter(cluck::g_name_cluck_param_ticket_id));
        }
    }
}
 
 
void cluckd::msg_server_gone(ed::message & msg)
{
    // if the server is not defined, ignore that message
    // (already tested by the dispatcher)
    //
    if(!msg.has_parameter(communicator::g_name_communicator_param_server_name))
    {
        return; // LCOV_EXCL_LINE
    }
 
    // is it us?
    //
    std::string const server_name(msg.get_parameter(communicator::g_name_communicator_param_server_name));
    if(server_name.empty()
    || server_name == f_server_name)
    {
        // we never want to remove ourselves
        //
        return;
    }
 
    // is "server_name" known?
    //
    auto it(f_computers.find(server_name));
    if(it == f_computers.end())
    {
        // no computer found, nothing else to do here
        //
        return;
    }
    computer::pointer_t c(it->second);
 
    // got it, remove it
    //
    f_computers.erase(it);
SNAP_LOG_WARNING << "removed \"" << server_name << "\"" << SNAP_LOG_SEND;
 
    // is that computer a leader?
    //
    auto li(std::find(
              f_leaders.begin()
            , f_leaders.end()
            , c));
    if(li != f_leaders.end())
    {
        f_leaders.erase(li);
 
        if(f_messenger != nullptr)
        {
            // elect another computer in case the one we just erased was a leader
            //
            // (of course, no elections occur unless we are the computer with the
            // smallest IP address)
            //
            election_status();
 
            // if too many leaders were dropped, we may go back to the NO_LOCK status
            //
            // we only send a NO_LOCK if the election could not re-assign another
            // computer to replace the missing leader(s)
            //
            check_lock_status();
        }
    }
}
 
 
void cluckd::msg_status(ed::message & msg)
{
    // check the service name, it has to be one that means it is a remote
    // connection with another communicator daemon
    //
    std::string const service(msg.get_parameter(communicator::g_name_communicator_param_service));
 
    if(service.starts_with(communicator::g_name_communicator_connection_remote_communicator_in)   // remote host connected to us
    || service.starts_with(communicator::g_name_communicator_connection_remote_communicator_out)) // we connected to remote host
    {
        // check what the status is now: "up" or "down"
        //
        std::string const status(msg.get_parameter(communicator::g_name_communicator_param_status));
        if(status == communicator::g_name_communicator_value_up)
        {
            // we already broadcast a LOCK_STARTED from CLUSTER_UP
            // and that's enough
            //
            ;
        }
        else
        {
            // host is down, remove from our list of hosts
            //
            msg_server_gone(msg);
        }
    }
}
 
 
void cluckd::msg_ticket_added(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    auto const obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        auto const key_ticket(obj_ticket->second.find(key));
        if(key_ticket != obj_ticket->second.end())
        {
            // this ticket exists on this system
            //
            auto const obj_entering_ticket(f_entering_tickets.find(object_name));
            if(obj_entering_ticket == f_entering_tickets.end())
            {
                // this happens all the time because the entering ticket
                // gets removed on the first TICKET_ADDED we receive so
                // on the second one we get here...
                //
                SNAP_LOG_TRACE
                    << "called with object \""
                    << object_name
                    << "\" not present in f_entering_ticket (key: \""
                    << key
                    << "\")."
                    << SNAP_LOG_SEND;
                return;
            }
            key_ticket->second->ticket_added(obj_entering_ticket->second);
        }
        else
        {
            SNAP_LOG_WARNING
                << "found object \""
                << object_name
                << "\" but could not find a corresponding ticket with key \""
                << key
                << "\"..."
                << SNAP_LOG_SEND;
        }
    }
    else
    {
        SNAP_LOG_WARNING
            << "object \""
            << object_name
            << "\" not found."
            << SNAP_LOG_SEND;
    }
}
 
 
void cluckd::msg_ticket_ready(ed::message & msg)
{
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    std::string key;
    if(!get_parameters(msg, &object_name, &tag, nullptr, nullptr, &key, nullptr))
    {
        return; // LCOV_EXCL_LINE
    }
 
    auto obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        auto key_ticket(obj_ticket->second.find(key));
        if(key_ticket != obj_ticket->second.end())
        {
            // we can mark this ticket as activated
            //
            key_ticket->second->set_ready();
        }
    }
}
 
 
void cluckd::msg_unlock(ed::message & msg)
{
    if(!is_daemon_ready())
    {
        SNAP_LOG_ERROR
            << "received an UNLOCK when cluckd is not ready to receive lock related messages."
            << SNAP_LOG_SEND;
        return;
    }
 
    if(is_leader() == nullptr)
    {
        // we are not a leader, we need to forward to a leader to handle
        // the message properly
        //
        forward_message_to_leader(msg);
        return;
    }
 
    std::string object_name;
    ed::dispatcher_match::tag_t tag(ed::dispatcher_match::DISPATCHER_MATCH_NO_TAG);
    pid_t client_pid(0);
    if(!get_parameters(msg, &object_name, &tag, &client_pid, nullptr, nullptr, nullptr))
    {
        return;
    }
 
    // if the ticket still exists, send the UNLOCKED and then erase it
    //
    auto obj_ticket(f_tickets.find(object_name));
    if(obj_ticket != f_tickets.end())
    {
        std::string const server_name(msg.has_parameter("lock_proxy_server_name")
                                    ? msg.get_parameter("lock_proxy_server_name")
                                    : msg.get_sent_from_server());
 
        //std::string const service_name(msg.has_parameter("lock_proxy_service_name")
        //                            ? msg.get_parameter("lock_proxy_service_name")
        //                            : msg.get_sent_from_service());
 
        std::string const entering_key(server_name + '/' + std::to_string(client_pid));
        auto key_ticket(std::find_if(
                  obj_ticket->second.begin()
                , obj_ticket->second.end()
                , [&entering_key](auto const & t)
                {
                    return t.second->get_entering_key() == entering_key;
                }));
        if(key_ticket != obj_ticket->second.end())
        {
            // this function will send a DROPTICKET to the other leaders
            // and the UNLOCKED to the source (unless we already sent the
            // UNLOCKED which gets sent at most once.)
            //
            key_ticket->second->drop_ticket();
 
            obj_ticket->second.erase(key_ticket);
            if(obj_ticket->second.empty())
            {
                // we are done with this one!
                //
                f_tickets.erase(obj_ticket);
            }
 
            // TBD: the clean up calls this function if it removes a lock
            //      that timed out but not otherwise; maybe we could consider
            //      doing it slightly differently where we pass a flag to
            //      the clean up function to force the call either way
            //
            activate_first_lock(object_name);
        }
        else
        {
            SNAP_LOG_MAJOR
                << "UNLOCK could not find key \""
                << entering_key
                << "\" in object \""
                << object_name
                << "\"."
                << SNAP_LOG_SEND;
        }
    }
    else
    {
        SNAP_LOG_WARNING
            << "UNLOCK could not find object \""
            << object_name
            << "\"."
            << SNAP_LOG_SEND;
    }
 
    // reset the timeout with the other locks
    //
    cleanup();
}
 
 
 
} // namespace cluck_daemon
// vim: ts=4 sw=4 et