<commit_before>/** \copyright
 * Copyright (c) 2013, <NAME>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *  - Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 *
 *  - Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * \file AsyncIfCan.cxx
 *
 * Asynchronous NMRAnet interface.
 *
 * @author <NAME>
 * @date 3 Dec 2013
 */

#include "nmranet/AsyncIfCan.hxx"

#include "nmranet/AsyncAliasAllocator.hxx"
#include "nmranet/AsyncIfImpl.hxx"
#include "nmranet/AsyncIfCanImpl.hxx"
#include "nmranet/NMRAnetIfCan.hxx"
//#include "nmranet/NMRAnetWriteFlow.hxx"
#include "nmranet_can.h"

namespace NMRAnet
{

size_t g_alias_use_conflicts = 0;

DynamicPool *CanFrameWriteFlow::pool()
{
    return ifCan_->device()->pool();
}

void CanFrameWriteFlow::send(Buffer<CanHubData> *message, unsigned priority)
{
    LOG(INFO, "outgoing message %" PRIx32 ".",
        GET_CAN_FRAME_ID_EFF(message->data()->frame()));
    message->data()->skipMember_ = ifCan_->hub_port();
    ifCan_->device()->send(message, priority);
}

DynamicPool *CanFrameReadFlow::pool()
{
    return ifCan_->dispatcher()->pool();
}

void CanFrameReadFlow::send(Buffer<CanHubData> *message, unsigned priority)
{
    const struct can_frame &frame = message->data()->frame();
    if (IS_CAN_FRAME_ERR(frame) || IS_CAN_FRAME_RTR(frame) ||
        !IS_CAN_FRAME_EFF(frame))
    {
        // Ignores these frames.
        message->unref();
        return;
    }

    // We typecast the incoming buffer to a different buffer type that should be
    // the subset of the data.
    Buffer<CanMessageData> *incoming_buffer;

    // Checks that it fits.
    HASSERT(sizeof(*incoming_buffer) <= sizeof(*message));
    // Does the cast.
    incoming_buffer = static_cast<Buffer<CanMessageData> *>(
        static_cast<BufferBase *>(message));
    // Checks that the frame is still in the same place (by pointer).
    HASSERT(incoming_buffer->data()->mutable_frame() ==
            message->data()->mutable_frame());

    /** @TODO(balazs.racz): Figure out what priority the new message should be
     * at. */
    ifCan_->frame_dispatcher()->send(incoming_buffer, priority);
}

/** Specifies how long to wait for a response to an alias mapping enquiry
 * message when trying to send an addressed message to a destination. The final
 * timeout will be twice this time, because after the first timeout a global
 * message for verify node id request will also be sent out.
 *
 * This value is writable for unittesting purposes. We might consider moving it
 * to flash with a weak definition instead. */
extern long long ADDRESSED_MESSAGE_LOOKUP_TIMEOUT_NSEC;
long long ADDRESSED_MESSAGE_LOOKUP_TIMEOUT_NSEC = SEC_TO_NSEC(1);

namespace
{

/* This write flow inherits all the business logic from the parent, just
 * maintains a separate allocation queue. This allows global messages to go out
 * even if addressed messages are waiting for destination address
 * resolution. */
class GlobalCanMessageWriteFlow : public CanMessageWriteFlow
{
public:
    GlobalCanMessageWriteFlow(AsyncIfCan *if_can) : CanMessageWriteFlow(if_can)
    {
    }

protected:
    virtual Action entry()
    {
        return call_immediately(STATE(send_to_hardware));
    }

    virtual Action send_finished()
    {
        return call_immediately(STATE(global_entry));
    }
};

} // namespace

#if 0

/** This class listens for incoming CAN messages, and if it sees a local alias
 * conflict, then takes the appropriate action:
 *
 * . if the conflict happened in alias check, it responds with an AMD frame.
 *
 * . if the conflict is with an allocated alias, kicks it out of the local
 * cache, forcing an alias reallocation for that node.
 *
 * . if the conflict is with a reserved but unused alias, kicks it out of the
 * cache. This condition will be detected when a new node tries using that
 * alias.
 *
 * NOTE: this handler could be a control flow, but since the flow is very
 * simple, it is implemented on first principles. The resulting object will
 * take much less RAM. */
class AliasConflictHandler : public AllocationResult,
                             public IncomingFrameHandler
{
public:
    AliasConflictHandler(AsyncIfCan *if_can) : ifCan_(if_can)
    {
        lock_.TypedRelease(this);
        ifCan_->frame_dispatcher()->register_handler(0, ~((1 << 30) - 1), this);
    }

    ~AliasConflictHandler()
    {
        ifCan_->frame_dispatcher()->unregister_handler(0, ~((1 << 30) - 1),
                                                       this);
    }

    virtual AllocatorBase *get_allocator()
    {
        return &lock_;
    }

    /// Handler callback for incoming messages.
    virtual void handle_message(struct can_frame *f, Notifiable *done)
    {
        uint32_t id = GET_CAN_FRAME_ID_EFF(*f);
        done->notify(); // We don't need the frame anymore.
        f = nullptr;
        done = nullptr;
        if (IfCan::get_priority(id) != IfCan::NORMAL_PRIORITY)
        {
            // Probably not an OpenLCB frame.
            lock_.TypedRelease(this);
            return;
        }
        NodeAlias alias = IfCan::get_src(id);
        // If the caller comes with alias 000, we ignore that.
        NodeID node = alias ? ifCan_->local_aliases()->lookup(alias) : 0;
        if (!node)
        {
            // This is not a local alias of ours.
            lock_.TypedRelease(this);
            return;
        }
        if (IfCan::is_cid_frame(id))
        {
            // This is a CID frame. We own the alias, let them know.
            alias_ = alias;
            ifCan_->write_allocator()->AllocateEntry(this);
            // We keep the lock.
            return;
        }
        /* Removes the alias from the local alias cache.  If it was assigned to
         * a node, the node will grab a new alias next time it tries to send
         * something.  If it was reserved but not used, then whoever tries to
         * use it will realize due to the RESERVED_ALIAS_NODE_ID guard missing
         * from the cache.  We do not aggressively start looking for a new
         * alias in place of the missing one. If we want to do that, we would
         * have to find the entry in the list of
         * ifCan_->alias_allocator()->reserved_alias_allocator() and put it
         * into the empty_alias_allocator() instead to be picked up by the
         * alias allocator flow.  */
        g_alias_use_conflicts++;
        ifCan_->local_aliases()->remove(alias);
        lock_.TypedRelease(this);
    }

    virtual void AllocationCallback(QueueMember *entry)
    {
        frameWriteFlow_ = ifCan_->write_allocator()->cast_result(entry);
        ifCan_->frame_dispatcher()->executor()->Add(this);
    }

    virtual void Run()
    {
        // We are sending an RID frame.
        struct can_frame *f = frameWriteFlow_->mutable_frame();
        IfCan::control_init(*f, alias_, IfCan::RID_FRAME, 0);
        frameWriteFlow_->Send(nullptr);
        lock_.TypedRelease(this);
    }

private:
    /// Lock for ourselves.
    TypedAllocator<IncomingFrameHandler> lock_;
    /// Parent interface.
    AsyncIfCan *ifCan_;
    /// Alias being checked.
    unsigned alias_ : 12;
    /// The write flow we received from the allocator.
    CanFrameWriteFlow *frameWriteFlow_;
};

/** This class listens for incoming CAN frames of regular unaddressed global
 * OpenLCB messages, then translates it in a generic way into a message,
 * computing its MTI. The resulting message is then passed to the generic If
 * for dispatching. */<commit_msg>Updates the code for the alias conflict flow.<commit_after>/** \copyright
 * Copyright (c) 2013, <NAME>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *  - Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 *
 *  - Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * \file AsyncIfCan.cxx
 *
 * Asynchronous NMRAnet interface.
 *
 * @author <NAME>
 * @date 3 Dec 2013
 */

#include "nmranet/AsyncIfCan.hxx"

#include "nmranet/AsyncAliasAllocator.hxx"
#include "nmranet/AsyncIfImpl.hxx"
#include "nmranet/AsyncIfCanImpl.hxx"
#include "nmranet/NMRAnetIfCan.hxx"
//#include "nmranet/NMRAnetWriteFlow.hxx"
#include "nmranet_can.h"

namespace NMRAnet
{

size_t g_alias_use_conflicts = 0;

DynamicPool *CanFrameWriteFlow::pool()
{
    return ifCan_->device()->pool();
}

void CanFrameWriteFlow::send(Buffer<CanHubData> *message, unsigned priority)
{
    LOG(INFO, "outgoing message %" PRIx32 ".",
        GET_CAN_FRAME_ID_EFF(message->data()->frame()));
    message->data()->skipMember_ = ifCan_->hub_port();
    ifCan_->device()->send(message, priority);
}

DynamicPool *CanFrameReadFlow::pool()
{
    return ifCan_->dispatcher()->pool();
}

void CanFrameReadFlow::send(Buffer<CanHubData> *message, unsigned priority)
{
    const struct can_frame &frame = message->data()->frame();
    if (IS_CAN_FRAME_ERR(frame) || IS_CAN_FRAME_RTR(frame) ||
        !IS_CAN_FRAME_EFF(frame))
    {
        // Ignores these frames.
        message->unref();
        return;
    }

    // We typecast the incoming buffer to a different buffer type that should be
    // the subset of the data.
    Buffer<CanMessageData> *incoming_buffer;

    // Checks that it fits.
    HASSERT(sizeof(*incoming_buffer) <= sizeof(*message));
    // Does the cast.
    incoming_buffer = static_cast<Buffer<CanMessageData> *>(
        static_cast<BufferBase *>(message));
    // Checks that the frame is still in the same place (by pointer).
    HASSERT(incoming_buffer->data()->mutable_frame() ==
            message->data()->mutable_frame());

    /** @TODO(balazs.racz): Figure out what priority the new message should be
     * at. */
    ifCan_->frame_dispatcher()->send(incoming_buffer, priority);
}

/** Specifies how long to wait for a response to an alias mapping enquiry
 * message when trying to send an addressed message to a destination. The final
 * timeout will be twice this time, because after the first timeout a global
 * message for verify node id request will also be sent out.
 *
 * This value is writable for unittesting purposes. We might consider moving it
 * to flash with a weak definition instead. */
extern long long ADDRESSED_MESSAGE_LOOKUP_TIMEOUT_NSEC;
long long ADDRESSED_MESSAGE_LOOKUP_TIMEOUT_NSEC = SEC_TO_NSEC(1);

namespace
{

/* This write flow inherits all the business logic from the parent, just
 * maintains a separate allocation queue. This allows global messages to go out
 * even if addressed messages are waiting for destination address
 * resolution. */
class GlobalCanMessageWriteFlow : public CanMessageWriteFlow
{
public:
    GlobalCanMessageWriteFlow(AsyncIfCan *if_can) : CanMessageWriteFlow(if_can)
    {
    }

protected:
    virtual Action entry()
    {
        return call_immediately(STATE(send_to_hardware));
    }

    virtual Action send_finished()
    {
        return call_immediately(STATE(global_entry));
    }
};

} // namespace

/** This class listens for incoming CAN messages, and if it sees a local alias
 * conflict, then takes the appropriate action:
 *
 * . if the conflict happened in alias check, it responds with an AMD frame.
 *
 * . if the conflict is with an allocated alias, kicks it out of the local
 * cache, forcing an alias reallocation for that node.
 *
 * . if the conflict is with a reserved but unused alias, kicks it out of the
 * cache. This condition will be detected when a new node tries using that
 * alias.
 */
class AliasConflictHandler : public StateFlow<Buffer<CanMessageData>, QList<1>>
{
public:
    AliasConflictHandler(AsyncIfCan *service)
        : StateFlow<Buffer<CanMessageData>, QList<1>>(service)
    {
        if_can()->frame_dispatcher()->register_handler(this, 0,
                                                       ~((1 << 30) - 1));
    }

    ~AliasConflictHandler()
    {
        if_can()->frame_dispatcher()->unregister_handler(this, 0,
                                                         ~((1 << 30) - 1));
    }

    AsyncIfCan *if_can()
    {
        return static_cast<AsyncIfCan *>(service());
    }

    /// Handler callback for incoming messages.
    virtual Action entry()
    {
        uint32_t id = GET_CAN_FRAME_ID_EFF(*message()->data());
        release();
        if (IfCan::get_priority(id) != IfCan::NORMAL_PRIORITY)
        {
            // Probably not an OpenLCB frame.
            return exit();
        }
        NodeAlias alias = IfCan::get_src(id);
        // If the caller comes with alias 000, we ignore that.
        NodeID node = alias ? if_can()->local_aliases()->lookup(alias) : 0;
        if (!node)
        {
            // This is not a local alias of ours.
            return exit();
        }
        if (IfCan::is_cid_frame(id))
        {
            // This is a CID frame. We own the alias, let them know.
            alias_ = alias;
            return allocate_and_call(if_can()->frame_write_flow(),
                                     STATE(send_reserved_alias));
        }
        /* Removes the alias from the local alias cache.  If it was assigned to
         * a node, the node will grab a new alias next time it tries to send
         * something.  If it was reserved but not used, then whoever tries to
         * use it will realize due to the RESERVED_ALIAS_NODE_ID guard missing
         * from the cache.  We do not aggressively start looking for a new
         * alias in place of the missing one. If we want to do that, we would
         * have to find the entry in the list of
         * ifCan_->alias_allocator()->reserved_alias_allocator() and put it
         * into the empty_alias_allocator() instead to be picked up by the
         * alias allocator flow.  */
        g_alias_use_conflicts++;
        if_can()->local_aliases()->remove(alias);
        return exit();
    }

    Action send_reserved_alias()
    {
        auto *b = get_allocation_result(if_can()->frame_write_flow());
        // We are sending an RID frame.
        struct can_frame *f = b->data();
        IfCan::control_init(*f, alias_, IfCan::RID_FRAME, 0);
        if_can()->frame_write_flow()->send(b);
        return exit();
    }

private:
    /// Alias being checked.
    unsigned alias_ : 12;
};

#if 0
/** This class listens for incoming CAN frames of regular unaddressed global
 * OpenLCB messages, then translates it in a generic way into a message,
 * computing its MTI. The resulting message is then passed to the generic If
 * for dispatching. */
{'fraction_non_alphanumeric': 0.09151304764184821, 'fraction_numerical': 0.004069906631553747, 'mean_word_length': 2.9688836104513063, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 103, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2730760', 'n_tokens_mistral': 4784, 'n_tokens_neox': 4602, 'n_words': 1700}