cluck.cpp

Go to the documentation of this file.

// Copyright (c) 2016-2025  Made to Order Software Corp.  All Rights Reserved
//
// 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    "cluck/cluck.h"
 
#include    "cluck/exception.h"
#include    "cluck/names.h"
 
 
// eventdispatcher
//
#include    <eventdispatcher/names.h>
 
 
// communicator
//
#include    <communicator/communicator_connection.h>
#include    <communicator/names.h>
 
 
// cppthread
//
#include    <cppthread/guard.h>
#include    <cppthread/mutex.h>
#include    <cppthread/thread.h>
 
 
// snapdev
//
#include    <snapdev/not_reached.h>
 
 
// last include
//
#include    <snapdev/poison.h>
 
 
 
namespace cluck
{
 
 
 
namespace
{
 
 
 
cppthread::mutex            g_mutex = cppthread::mutex();
cluck::serial_t             g_serial = cluck::serial_t();
timeout_t                   g_lock_obtention_timeout = CLUCK_LOCK_OBTENTION_DEFAULT_TIMEOUT;
timeout_t                   g_lock_duration_timeout = CLUCK_LOCK_DURATION_DEFAULT_TIMEOUT;
timeout_t                   g_unlock_timeout = CLUCK_UNLOCK_DEFAULT_TIMEOUT;
 
 
ed::match_t match_command_and_tag(ed::dispatcher_match const * m, ed::message & msg)
{
    if(m->f_expr != nullptr
    && m->f_expr == msg.get_command()
    && msg.has_parameter(g_name_cluck_param_tag)
    && msg.get_integer_parameter(g_name_cluck_param_tag) == m->f_tag)
    {
        return ed::match_t::MATCH_TRUE;
    }
 
    return ed::match_t::MATCH_FALSE;
}
 
 
cluck::serial_t get_next_serial()
{
    cppthread::guard lock(g_mutex);
    ++g_serial;
    if(g_serial == 0)
    {
        g_serial = 1; // LCOV_EXCL_LINE
    }
    return g_serial;
}
 
 
 
} // no name namespace
 
 
 
timeout_t get_lock_obtention_timeout()
{
    cppthread::guard lock(g_mutex);
    return g_lock_obtention_timeout;
}
 
 
void set_lock_obtention_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        timeout = CLUCK_LOCK_OBTENTION_DEFAULT_TIMEOUT;
    }
    else
    {
        timeout = std::clamp(timeout, CLUCK_MINIMUM_TIMEOUT, CLUCK_LOCK_OBTENTION_MAXIMUM_TIMEOUT);
    }
 
    cppthread::guard lock(g_mutex);
    g_lock_obtention_timeout = timeout;
}
 
 
timeout_t get_lock_duration_timeout()
{
    cppthread::guard lock(g_mutex);
    return g_lock_duration_timeout;
}
 
 
void set_lock_duration_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        timeout = CLUCK_LOCK_DURATION_DEFAULT_TIMEOUT;
    }
    else
    {
        timeout = std::clamp(timeout, CLUCK_MINIMUM_TIMEOUT, CLUCK_MAXIMUM_TIMEOUT);
    }
 
    cppthread::guard lock(g_mutex);
    g_lock_duration_timeout = timeout;
}
 
 
timeout_t get_unlock_timeout()
{
    cppthread::guard lock(g_mutex);
    return g_unlock_timeout;
}
 
 
void set_unlock_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        timeout = CLUCK_UNLOCK_DEFAULT_TIMEOUT;
    }
    else
    {
        timeout = std::clamp(timeout, CLUCK_UNLOCK_MINIMUM_TIMEOUT, CLUCK_MAXIMUM_TIMEOUT);
    }
 
    cppthread::guard lock(g_mutex);
    g_unlock_timeout = timeout;
}
 
 
cluck::cluck(
          std::string const & object_name
        , ed::connection_with_send_message::pointer_t messenger
        , ed::dispatcher::pointer_t dispatcher
        , mode_t mode)
    : timer(0)
    , f_object_name(object_name)
    , f_tag(ed::dispatcher_match::get_next_tag())
    , f_connection(messenger)
    , f_dispatcher(dispatcher)
    , f_mode(mode)
    , f_lock_obtention_timeout(CLUCK_DEFAULT_TIMEOUT)
    , f_lock_duration_timeout(CLUCK_DEFAULT_TIMEOUT)
    , f_unlock_timeout(CLUCK_DEFAULT_TIMEOUT)
{
    if(messenger == nullptr
    || dispatcher == nullptr)
    {
        throw invalid_parameter("messenger & dispatcher parameters must be defined in cluck::cluck() constructor.");
    }
 
    set_enable(false);
    set_name("cluck::" + object_name);
 
    f_connection->add_help_callback(std::bind(&cluck::help, this, std::placeholders::_1));
}
 
 
cluck::~cluck()
{
    f_dispatcher->remove_matches(f_tag);
}
 
 
cluck::callback_manager_t::callback_id_t cluck::add_lock_obtained_callback(callback_t func, callback_manager_t::priority_t priority)
{
    return f_lock_obtained_callbacks.add_callback(func, priority);
}
 
 
bool cluck::remove_lock_obtained_callback(callback_manager_t::callback_id_t id)
{
    return f_lock_obtained_callbacks.remove_callback(id);
}
 
 
cluck::callback_manager_t::callback_id_t cluck::add_lock_failed_callback(callback_t func, callback_manager_t::priority_t priority)
{
    return f_lock_failed_callbacks.add_callback(func, priority);
}
 
 
bool cluck::remove_lock_failed_callback(callback_manager_t::callback_id_t id)
{
    return f_lock_failed_callbacks.remove_callback(id);
}
 
 
cluck::callback_manager_t::callback_id_t cluck::add_finally_callback(callback_t func, callback_manager_t::priority_t priority)
{
    return f_finally_callbacks.add_callback(func, priority);
}
 
 
bool cluck::remove_finally_callback(callback_manager_t::callback_id_t id)
{
    return f_finally_callbacks.remove_callback(id);
}
 
 
timeout_t cluck::get_lock_obtention_timeout() const
{
    return f_lock_obtention_timeout;
}
 
 
void cluck::set_lock_obtention_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        f_lock_obtention_timeout = timeout;
    }
    else
    {
        f_lock_obtention_timeout = std::clamp(timeout, CLUCK_MINIMUM_TIMEOUT, CLUCK_LOCK_OBTENTION_MAXIMUM_TIMEOUT);
    }
}
 
 
timeout_t cluck::get_lock_duration_timeout() const
{
    return f_lock_duration_timeout;
}
 
 
void cluck::set_lock_duration_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        f_lock_duration_timeout = timeout;
    }
    else
    {
        f_lock_duration_timeout = std::clamp(timeout, CLUCK_MINIMUM_TIMEOUT, CLUCK_MAXIMUM_TIMEOUT);
    }
}
 
 
timeout_t cluck::get_unlock_timeout() const
{
    return f_unlock_timeout;
}
 
 
void cluck::set_unlock_timeout(timeout_t timeout)
{
    if(timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        f_unlock_timeout = timeout;
    }
    else
    {
        f_unlock_timeout = std::clamp(timeout, CLUCK_UNLOCK_MINIMUM_TIMEOUT, CLUCK_MAXIMUM_TIMEOUT);
    }
}
 
 
std::string const & cluck::get_object_name() const
{
    return f_object_name;
}
 
 
mode_t cluck::get_mode() const
{
    return f_mode;
}
 
 
type_t cluck::get_type() const
{
    return f_type;
}
 
 
void cluck::set_type(type_t type)
{
    if(is_busy())
    {
        throw busy("this cluck object is busy, you cannot change its type at the moment.");
    }
 
    f_type = type;
}
 
 
reason_t cluck::get_reason() const
{
    return f_reason;
}
 
 
void cluck::set_reason(reason_t reason)
{
    f_reason = reason;
}
 
 
bool cluck::help(advgetopt::string_set_t & commands)
{
    commands.insert(g_name_cluck_cmd_locked);
    commands.insert(g_name_cluck_cmd_lock_failed);
    commands.insert(g_name_cluck_cmd_unlocked);
    commands.insert(g_name_cluck_cmd_unlocking);
 
    return true;
}
 
 
bool cluck::lock()
{
    // is lock still busy?
    //
    if(is_busy())
    {
        return false;
    }
 
    timeout_t obtention_timeout_date(snapdev::now());
    if(f_lock_obtention_timeout == CLUCK_DEFAULT_TIMEOUT)
    {
        // use global timeout by default
        //
        obtention_timeout_date += ::cluck::get_lock_obtention_timeout();
    }
    else
    {
        obtention_timeout_date += f_lock_obtention_timeout;
    }
 
    f_serial = get_next_serial();
 
    // send the LOCK message
    //
    ed::message lock_message;
    lock_message.set_command(g_name_cluck_cmd_lock);
    lock_message.set_service(g_name_cluck_service_name);
    lock_message.add_parameter(g_name_cluck_param_object_name, f_object_name);
    lock_message.add_parameter(g_name_cluck_param_tag, static_cast<int>(f_tag));
    lock_message.add_parameter(g_name_cluck_param_pid, cppthread::gettid());
    lock_message.add_parameter(ed::g_name_ed_param_serial, f_serial);
    lock_message.add_parameter(g_name_cluck_param_timeout, obtention_timeout_date);
    communicator::request_failure(lock_message);
    if(f_lock_duration_timeout != CLUCK_DEFAULT_TIMEOUT)
    {
        lock_message.add_parameter(g_name_cluck_param_duration, f_lock_duration_timeout);
    }
    if(f_unlock_timeout != CLUCK_DEFAULT_TIMEOUT)
    {
        lock_message.add_parameter(g_name_cluck_param_unlock_duration, f_unlock_timeout);
    }
    if(f_type != type_t::CLUCK_TYPE_READ_WRITE)
    {
        lock_message.add_parameter(g_name_cluck_param_type, static_cast<int>(f_type));
    }
    if(!f_connection->send_message(lock_message))
    {
        // LCOV_EXCL_START
        f_state = state_t::CLUCK_STATE_FAILED;
        set_reason(reason_t::CLUCK_REASON_TRANSMISSION_ERROR);
        snapdev::NOT_REACHED_IN_TEST();
        return false;
        // LCOV_EXCL_STOP
    }
 
    set_timeout_date(obtention_timeout_date);
    set_enable(true);
 
    set_reason(reason_t::CLUCK_REASON_NONE);
    f_state = state_t::CLUCK_STATE_LOCKING;
 
    // start listening to our messages
    //
    ed::dispatcher_match locked(ed::define_match(
              ed::Expression(g_name_cluck_cmd_locked)
            , ed::Callback(std::bind(&cluck::msg_locked, this, std::placeholders::_1))
            , ed::MatchFunc(&match_command_and_tag)
            , ed::Tag(f_tag)));
    f_dispatcher->add_match(locked);
 
    ed::dispatcher_match lock_failed(ed::define_match(
              ed::Expression(g_name_cluck_cmd_lock_failed)
            , ed::Callback(std::bind(&cluck::msg_lock_failed, this, std::placeholders::_1))
            , ed::MatchFunc(&match_command_and_tag)
            , ed::Tag(f_tag)));
    f_dispatcher->add_match(lock_failed);
 
    ed::dispatcher_match unlocked(ed::define_match(
              ed::Expression(g_name_cluck_cmd_unlocked)
            , ed::Callback(std::bind(&cluck::msg_unlocked, this, std::placeholders::_1))
            , ed::MatchFunc(&match_command_and_tag)
            , ed::Tag(f_tag)));
    f_dispatcher->add_match(unlocked);
 
    ed::dispatcher_match unlocking(ed::define_match(
              ed::Expression(g_name_cluck_cmd_unlocking)
            , ed::Callback(std::bind(&cluck::msg_unlocking, this, std::placeholders::_1))
            , ed::MatchFunc(&match_command_and_tag)
            , ed::Tag(f_tag)));
    f_dispatcher->add_match(unlocking);
 
    ed::dispatcher_match transmission_report(ed::define_match(
              ed::Expression(communicator::g_name_communicator_cmd_transmission_report)
            , ed::Callback(std::bind(&cluck::msg_transmission_report, this, std::placeholders::_1))
            , ed::MatchFunc(&ed::one_to_one_callback_match)
            , ed::Tag(f_tag)
            , ed::Priority(ed::dispatcher_match::DISPATCHER_MATCH_CALLBACK_PRIORITY)));
    f_dispatcher->add_match(transmission_report);
 
    // we just added new commands (at least the first time) which we need to
    // share with the communicator deamon (otherwise it won't forward them
    // to us)
    //
    f_connection->send_commands();
 
    return true;
}
 
 
void cluck::unlock()
{
    if(f_state != state_t::CLUCK_STATE_LOCKED
    && f_state != state_t::CLUCK_STATE_LOCKING)
    {
        SNAP_LOG_NOTICE
            << "this cluck object is not currently locked."
            << SNAP_LOG_SEND;
        return;
    }
 
    f_lock_timeout_date = timeout_t();
 
    // explicitly send the UNLOCK message and then make sure to unregister
    // from the communicator; note that we do not wait for a reply to the
    // UNLOCK message, since to us it does not matter much as long as the
    // message was sent...
    //
    ed::message unlock_message;
    unlock_message.set_command(g_name_cluck_cmd_unlock);
    unlock_message.set_service(g_name_cluck_service_name);
    unlock_message.add_parameter(g_name_cluck_param_object_name, f_object_name);
    unlock_message.add_parameter(g_name_cluck_param_tag, static_cast<int>(f_tag));
    unlock_message.add_parameter(g_name_cluck_param_pid, gettid());
    unlock_message.add_parameter(ed::g_name_ed_param_serial, f_serial);
    if(!f_connection->send_message(unlock_message))
    {
        // LCOV_EXCL_START
        f_state = state_t::CLUCK_STATE_FAILED;
        set_reason(reason_t::CLUCK_REASON_TRANSMISSION_ERROR);
        lock_failed();
        finally();
        snapdev::NOT_REACHED_IN_TEST();
        return;
        // LCOV_EXCL_STOP
    }
 
    // give the UNLOCK 5 seconds to happen, if it does not happen, we'll
    // set the state to "failed" and still call the finally() callbacks
    //
    timeout_t unlock_timeout_date(snapdev::now());
    unlock_timeout_date += timeout_t(5, 0); // TODO: use correct timeout instead of hard coded 5s
    set_timeout_date(unlock_timeout_date);
    set_enable(true);
 
    f_state = state_t::CLUCK_STATE_UNLOCKING;
}
 
 
timeout_t cluck::get_timeout_date() const
{
    return f_lock_timeout_date;
}
 
 
bool cluck::is_locked() const
{
    return f_state == state_t::CLUCK_STATE_LOCKED
        && f_lock_timeout_date > snapdev::now();
}
 
 
bool cluck::is_busy() const
{
    return f_state != state_t::CLUCK_STATE_IDLE;
}
 
 
bool cluck::is_cluck_msg(ed::message & msg) const
{
    // the tag must match our tag -- this allows your process to support
    // more than one lock (as long as each one has a different object name)
    //
    if(msg.get_integer_parameter(g_name_cluck_param_tag) != f_tag)
    {
        // IMPORTANT NOTE: this tag is checked in match_command_and_tag()
        //                 before our msg_...() callbacks get called so this
        //                 error should never happen
        //
        throw logic_error("tag mismatch in is_cluck_msg()."); // LCOV_EXCL_LINE
    }
 
    if(msg.get_parameter(g_name_cluck_param_object_name) != f_object_name)
    {
        // somehow we received a message with the wrong object name
        //
        ed::message invalid;
        invalid.user_data(msg.user_data<void>());
        invalid.reply_to(msg);
        invalid.set_command(ed::g_name_ed_cmd_invalid);
        invalid.add_parameter(ed::g_name_ed_param_command, msg.get_command());
        invalid.add_parameter(
              ed::g_name_ed_param_message
            , "the \"object_name\" parameter does not match this cluck object. Got \""
              + msg.get_parameter("object_name")
              + "\", expected \""
              + f_object_name
              + "\".");
        f_connection->send_message(invalid);
        return false;
    }
 
    return true;
}
 
 
void cluck::process_timeout()
{
    set_enable(false);
 
    // the LOCK event and the lock duration can time out
    //
    switch(f_state)
    {
    // LCOV_EXCL_START
    case state_t::CLUCK_STATE_IDLE:
        SNAP_LOG_DEBUG
            << "process_timeout() called with state set to CLUCK_STATE_IDLE."
            << SNAP_LOG_SEND;
        snapdev::NOT_REACHED_IN_TEST();
        break;
    // LCOV_EXCL_STOP
 
    case state_t::CLUCK_STATE_LOCKING:
        // lock never obtained
        //
        set_reason(reason_t::CLUCK_REASON_LOCAL_TIMEOUT);
        lock_failed();
        finally();
        break;
 
    case state_t::CLUCK_STATE_LOCKED:
        // we are out of time, unlock now
        //
        set_reason(reason_t::CLUCK_REASON_LOCAL_TIMEOUT);
        unlock();
        break;
 
    case state_t::CLUCK_STATE_UNLOCKING:
        // the UNLOCK was never acknowledged
        //
        set_reason(reason_t::CLUCK_REASON_LOCAL_TIMEOUT);
        lock_failed();
        finally();
        break;
 
    // LCOV_EXCL_START
    case state_t::CLUCK_STATE_FAILED:
        SNAP_LOG_DEBUG
            << "process_timeout() called with state set to CLUCK_STATE_FAILED."
            << SNAP_LOG_SEND;
        snapdev::NOT_REACHED_IN_TEST();
        break;
    // LCOV_EXCL_STOP
 
    }
}
 
 
void cluck::lock_obtained()
{
    f_lock_obtained_callbacks.call(this);
 
    if(f_mode == mode_t::CLUCK_MODE_SIMPLE)
    {
        // in this case the user is done and we can just release the
        // lock automatically; otherwise the user is responsible for
        // releasing the lock once done
        //
        unlock();
    }
}
 
 
void cluck::lock_failed()
{
    if(f_state != state_t::CLUCK_STATE_FAILED)
    {
        f_state = state_t::CLUCK_STATE_FAILED;
 
        // disable our timer, we don't need to time out if the lock failed
        // since we're done in this case
        //
        set_enable(false);
 
        f_lock_failed_callbacks.call(this);
    }
}
 
 
void cluck::finally()
{
    f_state = state_t::CLUCK_STATE_IDLE;
    f_dispatcher->remove_matches(f_tag);
    f_finally_callbacks.call(this);
}
 
 
void cluck::msg_locked(ed::message & msg)
{
    if(!is_cluck_msg(msg))
    {
        set_reason(reason_t::CLUCK_REASON_INVALID);
        lock_failed();
        finally();
        return;
    }
 
    f_state = state_t::CLUCK_STATE_LOCKED;
    f_lock_timeout_date = msg.get_timespec_parameter(g_name_cluck_param_timeout_date);
    f_unlocked_timeout_date = msg.get_timespec_parameter(g_name_cluck_param_unlocked_date);
 
    // setup our timer so it times out on that date
    //
    set_timeout_date(f_lock_timeout_date);
    set_enable(true);
 
    lock_obtained();
}
 
 
void cluck::msg_lock_failed(ed::message & msg)
{
    if(!is_cluck_msg(msg))
    {
        set_reason(reason_t::CLUCK_REASON_INVALID);
    }
    else
    {
        std::string const error(msg.get_parameter(g_name_cluck_param_error));
        if(error == g_name_cluck_value_timedout)
        {
            set_reason(reason_t::CLUCK_REASON_REMOTE_TIMEOUT);
        }
        else
        {
            // this may be a programmer error that need fixing
            //
            SNAP_LOG_WARNING
                << "communicatord did not like our LOCK message: "
                << error
                << '.'
                << SNAP_LOG_SEND;
 
            set_reason(reason_t::CLUCK_REASON_INVALID);
        }
    }
 
    lock_failed();
    finally();
}
 
 
void cluck::msg_transmission_report(ed::message & msg)
{
    std::string const status(msg.get_parameter(communicator::g_name_communicator_param_status));
    if(msg.has_parameter(communicator::g_name_communicator_param_command)
    && msg.get_parameter(communicator::g_name_communicator_param_command) == g_name_cluck_cmd_lock
    && status == communicator::g_name_communicator_value_failed)
    {
        SNAP_LOG_RECOVERABLE_ERROR
            << "the transmission of our \""
            << msg.get_parameter(communicator::g_name_communicator_param_command)
            << "\" message failed to travel to a cluckd service."
            << SNAP_LOG_SEND;
 
        // TODO: this message is global, so we fail all the
        //       currently valid locks in this very process
        //       (i.e. all the cluck objects have this function called
        //       on a TRANSITION_REPORT message); this is only an issue
        //       if you use threads and more than one has a lock at a
        //       time
        //
        set_reason(reason_t::CLUCK_REASON_TRANSMISSION_ERROR);
        lock_failed();
        finally();
    }
}
 
 
void cluck::msg_unlocked(ed::message & msg)
{
    if(!is_cluck_msg(msg))
    {
        set_reason(reason_t::CLUCK_REASON_INVALID);
        lock_failed();
    }
    else
    {
        set_enable(false);
        if(snapdev::now() >= f_unlocked_timeout_date)
        {
            // we took too long and received the unlocked after the lock was
            // over (instead of snapdev::now() called here, we may want to
            // call it right after the callback returned)
            //
            set_reason(reason_t::CLUCK_REASON_REMOTE_TIMEOUT);
            lock_failed();
        }
        else
        {
            set_reason(reason_t::CLUCK_REASON_NONE);
        }
    }
    finally();
}
 
 
 
void cluck::msg_unlocking(ed::message & msg)
{
    if(!is_cluck_msg(msg))
    {
        set_reason(reason_t::CLUCK_REASON_INVALID);
        lock_failed();
        finally();
        return;
    }
 
    set_reason(reason_t::CLUCK_REASON_REMOTE_TIMEOUT);
    if(snapdev::now() >= f_unlocked_timeout_date)
    {
        // it looks like we are beyond unlocking this lock
        //
        set_enable(false);
        lock_failed();
        finally();
    }
    else
    {
        unlock();
    }
}
 
 
 
} // namespace cluck
// vim: ts=4 sw=4 et