GridConnectHub.cxx

Go to the documentation of this file.

 
//#define LOGLEVEL VERBOSE
 
#include "openmrn_features.h"
 
#include "utils/GridConnectHub.hxx"
 
#include "executor/StateFlow.hxx"
#include "can_frame.h"
#include "nmranet_config.h"
#include "utils/Buffer.hxx"
#include "utils/BufferPort.hxx"
#include "utils/HubDevice.hxx"
#if OPENMRN_FEATURE_EXECUTOR_SELECT
#include "utils/HubDeviceSelect.hxx"
#endif
#include "utils/Hub.hxx"
#include "utils/GcStreamParser.hxx"
#include "utils/gc_format.h"
 
class GCAdapter : public GCAdapterBase
{
public:
    GCAdapter(HubFlow *gc_side, CanHubFlow *can_side, bool double_bytes)
        : parser_(can_side->service(), can_side, &formatter_)
        , formatter_(can_side->service(), gc_side, &parser_, double_bytes)
    {
        gc_side->register_port(&parser_);
        can_side->register_port(&formatter_);
        isRegistered_ = 1;
    }
 
    GCAdapter(HubFlow *gc_side_read, HubFlow *gc_side_write,
        CanHubFlow *can_side, bool double_bytes)
        : parser_(can_side->service(), can_side, &formatter_)
        , formatter_(can_side->service(), gc_side_write, &parser_, double_bytes)
    {
        gc_side_read->register_port(&parser_);
        can_side->register_port(&formatter_);
        isRegistered_ = 1;
    }
 
    virtual ~GCAdapter()
    {
        unregister();
    }
 
    void unregister()
    {
        if (isRegistered_)
        {
            parser_.destination()->unregister_port(&formatter_);
            formatter_.destination()->unregister_port(&parser_);
            isRegistered_ = 0;
        }
    }
 
    bool shutdown() OVERRIDE
    {
        unregister();
        return formatter_.shutdown() && parser_.shutdown_ready() &&
            formatter_.is_waiting();
    }
 
    class BinaryToGCMember : public CanHubPort
    {
    public:
        BinaryToGCMember(Service *service, HubFlow *destination,
            HubPort *skip_member, int double_bytes)
            : CanHubPort(service)
            , delayPort_(service, destination, config_gridconnect_buffer_size(),
                  USEC_TO_NSEC(config_gridconnect_buffer_delay_usec()))
            , destination_(destination)
            , skipMember_(skip_member)
            , double_bytes_(double_bytes)
        {
            const int cnt = config_gridconnect_bridge_max_outgoing_packets();
            if (cnt > 1)
            {
                ownedPool_.reset(
                    new LimitedPool(sizeof(Buffer<CanHubData>), cnt));
                pool_ = ownedPool_.get();
            }
            else
            {
                pool_ = mainBufferPool;
            }
        }
 
        HubFlow *destination()
        {
            return destination_;
        }
 
        bool shutdown()
        {
            const int cnt = config_gridconnect_bridge_max_outgoing_packets();
            bool state_delay = delayPort_.shutdown();
            bool state_pool = true;
            if (ownedPool_)
            {
                state_pool = (int(ownedPool_->free_items()) == cnt);
            }
            return state_delay && state_pool;
        }
 
        Pool *pool() override
        {
            return pool_;
        }
 
        Action entry() override
        {
            LOG(VERBOSE, "can packet arrived: %" PRIx32,
                GET_CAN_FRAME_ID_EFF(*message()->data()));
            char *end =
                gc_format_generate(message()->data(), dbuf_, double_bytes_);
            size_t size = (end - dbuf_);
            if (size)
            {
                Buffer<HubData> *target_buffer = nullptr;
                mainBufferPool->alloc(&target_buffer);
                target_buffer->data()->skipMember_ = skipMember_;
                target_buffer->data()->resize(size);
                memcpy((char *)target_buffer->data()->data(), dbuf_, size);
                target_buffer->set_done(bn_.reset(this));
                delayPort_.send(target_buffer, 0);
                release();
                return wait_and_call(STATE(buffer_accepted));
            }
            else
            {
                LOG(INFO, "gc generate failed.");
            }
            return release_and_exit();
        }
 
        Action buffer_accepted()
        {
            return exit();
        }
 
    private:
        BufferPort delayPort_;
        std::unique_ptr<LimitedPool> ownedPool_;
        Pool *pool_;
        char dbuf_[56];
        HubFlow *destination_;
        HubPort *skipMember_;
        int double_bytes_;
        BarrierNotifiable bn_;
    };
 
    class GCToBinaryMember : public HubPort
    {
    public:
        GCToBinaryMember(
            Service *service, CanHubFlow *destination, CanHubPort *skip_member)
            : HubPort(service)
            , destination_(destination)
            , skipMember_(skip_member)
        {
            int max_frames_to_parse =
                config_gridconnect_bridge_max_incoming_packets();
            if (max_frames_to_parse > 1) {
                frameAllocator_.reset(new LimitedPool(
                    sizeof(CanHubFlow::buffer_type), max_frames_to_parse));
            }
        }
 
        bool shutdown_ready()
        {
            int max_frames_to_parse =
                config_gridconnect_bridge_max_incoming_packets();
            if (max_frames_to_parse > 1)
            {
                if (frameAllocator_->free_items() < (size_t)max_frames_to_parse)
                {
                    return false;
                }
            }
            return is_waiting();
        }
 
        CanHubFlow *destination()
        {
            return destination_;
        }
 
        Action entry() override
        {
            inBuf_ = message()->data()->data();
            inBufSize_ = message()->data()->size();
            return call_immediately(STATE(parse_more_data));
        }
 
        Action parse_more_data()
        {
            while (inBufSize_--)
            {
                char c = *inBuf_++;
                if (streamSegmenter_.consume_byte(c))
                {
                    // End of frame. Allocate an output buffer and parse the
                    // frame.
                    return allocate_and_call(destination_, STATE(parse_to_output_frame), frameAllocator_.get());
                }
            }
            // Will notify the caller.
            return release_and_exit();
        }
 
        Action parse_to_output_frame()
        {
            auto* b = get_allocation_result(destination_);
            if (streamSegmenter_.parse_frame_to_output(b->data()))
            {
                b->data()->skipMember_ = skipMember_;
                destination_->send(b);
            }
            else
            {
                // Releases the buffer.
                b->unref();
            }
            return call_immediately(STATE(parse_more_data));
        }
 
    private:
        GcStreamParser streamSegmenter_;
        
        const char *inBuf_;
        size_t inBufSize_;
 
        // Allocator to get the frame from. If NULL, the target's default
        // buffer pool will be used.
        std::unique_ptr<LimitedPool> frameAllocator_;
        
        // ==== static data ====
 
        CanHubFlow *destination_;
        CanHubPortInterface *skipMember_;
    };
 
private:
    GCToBinaryMember parser_;
    BinaryToGCMember formatter_;
    unsigned isRegistered_ : 1;
};
 
GCAdapterBase *GCAdapterBase::CreateGridConnectAdapter(HubFlow *gc_side,
                                                       CanHubFlow *can_side,
                                                       bool double_bytes)
{
    return new GCAdapter(gc_side, can_side, double_bytes);
}
 
GCAdapterBase *GCAdapterBase::CreateGridConnectAdapter(HubFlow *gc_side_read,
                                                       HubFlow *gc_side_write,
                                                       CanHubFlow *can_side,
                                                       bool double_bytes)
{
    return new GCAdapter(gc_side_read, gc_side_write, can_side, double_bytes);
}
 
struct GcPacketPrinter::Impl : public CanHubPortInterface
{
    Impl(CanHubFlow *can_hub, bool timestamped)
        : canHub_(can_hub)
        , timestamped_(timestamped)
    {
        canHub_->register_port(this);
    }
 
    ~Impl()
    {
        canHub_->unregister_port(this);
    }
 
    void send(Buffer<CanHubData> *message, unsigned priority) OVERRIDE
    {
        AutoReleaseBuffer<CanHubData> b(message);
        char str[40];
        char* p = gc_format_generate(message->data(), str, false);
        *p = 0;
        if (timestamped_)
        {
#if defined(__linux__) || defined(__MACH__)
            struct timeval tv;
            gettimeofday(&tv, nullptr);
            struct tm t;
            localtime_r(&tv.tv_sec, &t);
            fprintf(stderr, "%04d-%02d-%02d %02d:%02d:%02d:%06ld [%p] ",
                t.tm_year + 1900, t.tm_mon + 1, t.tm_mday, t.tm_hour, t.tm_min,
                t.tm_sec, (long)tv.tv_usec, message->data()->skipMember_);
#endif
        }
        fprintf(stderr, "%s", str);
        if (config_gc_generate_newlines() != 1)
        {
            fprintf(stderr, "\n");
        }
    }
 
    CanHubFlow* canHub_;
    bool timestamped_;
};
 
GcPacketPrinter::GcPacketPrinter(CanHubFlow *can_hub, bool timestamped) : impl_(new Impl(can_hub, timestamped))
{
}
 
GcPacketPrinter::~GcPacketPrinter()
{
}
 
struct GcHubPort : public Executable
{
    GcHubPort(CanHubFlow *can_hub, int fd, Notifiable *on_exit, bool use_select)
        : gcHub_(can_hub->service())
        , bridge_(
              GCAdapterBase::CreateGridConnectAdapter(&gcHub_, can_hub, false))
        , onExit_(on_exit)
    {
        LOG(VERBOSE, "gchub port %p", (Executable *)this);
        if (use_select) {
#ifndef OPENMRN_FEATURE_EXECUTOR_SELECT
            DIE("select is not supported");
#else
            gcWrite_.reset(new HubDeviceSelect<HubFlow>(&gcHub_, fd, this));
#endif
        } else {
            gcWrite_.reset(new FdHubPort<HubFlow>(&gcHub_, fd, this));
        }
    }
    virtual ~GcHubPort()
    {
    }
 
    HubFlow gcHub_;
    std::unique_ptr<GCAdapterBase> bridge_;
    std::unique_ptr<FdHubPortInterface> gcWrite_;
    Notifiable* onExit_;
 
    void notify() OVERRIDE
    {
        /* We would like to delete *this but we cannot do that in this
         * callback, because we don't know what executor we are running
         * on. Deleting on the write executor would cause a deadlock for
         * example. */
        gcHub_.service()->executor()->add(this);
    }
 
    void run() OVERRIDE
    {
        if (!bridge_->shutdown() || !gcHub_.is_waiting())
        {
            // Yield.
            gcHub_.service()->executor()->add(this);
            return;
        }
        LOG(INFO, "GCHubPort: Shutting down gridconnect port %d. (%p)",
            gcWrite_->fd(), bridge_.get());
        if (onExit_) {
            onExit_->notify();
            onExit_ = nullptr;
        }
        /* We get this call when something is wrong with the FDs and we need to
         * close the connection. It is guaranteed that by the time we got this
         * call the device is unregistered from the char bridge, and the
         * service thread is ready to be stopped. */
        delete this;
    }
};
 
void create_gc_port_for_can_hub(
    CanHubFlow *can_hub, int fd, Notifiable *on_exit, bool use_select)
{
    new GcHubPort(can_hub, fd, on_exit, use_select);
}