BroadcastTimeClient.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_BROADCASTTIMECLIENT_HXX_
#define _OPENLCB_BROADCASTTIMECLIENT_HXX_
 
#include "openlcb/BroadcastTime.hxx"
#include "openlcb/BroadcastTimeAlarm.hxx"
 
namespace openlcb
{
 
class BroadcastTimeClient : public BroadcastTime
{
public:
    BroadcastTimeClient(Node *node, NodeID clock_id,
                        bool configure_agent = false)
        : BroadcastTime(node, clock_id)
        , configureAgent_(configure_agent)
        , immediateUpdate_(false)
        , immediatePending_(false)
        , sleeping_(false)
        , waiting_(false)
        , rolloverPending_(false)
        , rolloverPendingDate_(false)
        , rolloverPendingYear_(false)
        , serverDetected_(false)
    {
        EventRegistry::instance()->register_handler(
            EventRegistryEntry(this, eventBase_), 16);
    }
 
    ~BroadcastTimeClient()
    {
        EventRegistry::instance()->unregister_handler(this);
    }
 
    bool is_server_detected() override
    {
        return serverDetected_;
    }
 
    bool is_server_self() override
    {
        return false;
    }
 
    void handle_identify_global(const EventRegistryEntry &entry,
                                EventReport *event,
                                BarrierNotifiable *done) override
    {
        AutoNotify an(done);
        if (event->dst_node && event->dst_node != node_)
        {
            // not for us
            return;
        }
 
        if (event->dst_node == node_)
        {
            // This is directed at us. We were likely re-initialized, so we
            // need to query for a new clock server and possibly sync.
            query();
        }
 
        if (is_terminated())
        {
            start_flow(STATE(initialize));
        }
 
        event->event_write_helper<1>()->WriteAsync(
            node_, Defs::MTI_CONSUMER_IDENTIFIED_RANGE, WriteHelper::global(),
            eventid_to_buffer(EncodeRange(eventBase_, 0x1 << 16)),
            done->new_child());
        if (configureAgent_)
        {
            // we can configure our complementary time server
            event->event_write_helper<2>()->WriteAsync(
                node_, Defs::MTI_PRODUCER_IDENTIFIED_RANGE,
                WriteHelper::global(),
                eventid_to_buffer(
                    EncodeRange(entry.event |
                                    BroadcastTimeDefs::EVENT_SET_SUFFIX_MASK,
                                0x1 << 15)),
                done->new_child());
        }
        else
        {
            // we cannot configure our complementary time server
            event->event_write_helper<2>()->WriteAsync(
                node_, Defs::MTI_PRODUCER_IDENTIFIED_UNKNOWN,
                WriteHelper::global(),
                eventid_to_buffer(entry.event |
                                  BroadcastTimeDefs::QUERY_EVENT_SUFFIX),
                done->new_child());
        }
    }
 
    void handle_identify_consumer(const EventRegistryEntry &entry,
                                  EventReport *event,
                                  BarrierNotifiable *done) override
    {
        AutoNotify an(done);
        if (event->event < (eventBase_ | 0x5000))
        {
            event->event_write_helper<1>()->WriteAsync(
                node_, Defs::MTI_CONSUMER_IDENTIFIED_UNKNOWN,
                WriteHelper::global(), eventid_to_buffer(event->event),
                done->new_child());
        }
    }
 
    void handle_identify_producer(const EventRegistryEntry &entry,
                                  EventReport *event,
                                  BarrierNotifiable *done) override
    {
        AutoNotify an(done);
        if ((configureAgent_ &&
             (entry.event & BroadcastTimeDefs::EVENT_SET_SUFFIX_MASK)) ||
            event->event == (entry.event |
                             BroadcastTimeDefs::QUERY_EVENT_SUFFIX))
        {
            // if we cannot configure our complementary time server,
            // we only produce one event.
            event->event_write_helper<1>()->WriteAsync(
                node_, Defs::MTI_PRODUCER_IDENTIFIED_UNKNOWN,
                WriteHelper::global(),
                eventid_to_buffer(entry.event |
                                  BroadcastTimeDefs::QUERY_EVENT_SUFFIX),
                done->new_child());
        }
    }
 
    void handle_producer_identified(const EventRegistryEntry &entry,
                                    EventReport *event,
                                    BarrierNotifiable *done) override
    {
        done->notify();
 
        if (event->state == EventState::VALID)
        {
            // Look for a Report Date Event ID.
            if ((event->event & 0x000000000000F000ULL) == 0x2000ULL)
            {
                // We can only get here if there is a time server detected.
                serverDetected_ = true;
            }
 
            // We only care about valid event state.
            // Producer identified only happens when a clock synchronization
            // is taking place.  This voids previous date rollover events.
            rolloverPending_ = false;
            handle_updates(event, false);
        }
 
    }
 
    void handle_event_report(const EventRegistryEntry &entry,
                             EventReport *event,
                             BarrierNotifiable *done) override
    {
        done->notify();
 
        handle_updates(event, true);
    }
 
private:
    void handle_updates(EventReport *event, bool report);
 
    void start_stop_logic(bool started)
    {
        bool notify = false;
        time_t val = 0;
        {
            AtomicHolder h(this);
            if (started_ != started)
            {
                if (!started)
                {
                    seconds_ = time();
                }
                timestamp_ = OSTime::get_monotonic();
                if (!started)
                {
                    ::gmtime_r(&seconds_, &tm_);
                }
                started_ = started;
                notify = true;
                val = seconds_;
            }
            // release AtomicHolder
        }
        if (notify)
        {
            service_callbacks(val, val);
        }
    }
 
    Action initialize()
    {
        rolloverPending_ = false;
        rolloverPendingDate_ = false;
        rolloverPendingYear_ = false;
 
        waiting();
        return wait_and_call(STATE(client_update));
    }
 
    Action client_update()
    {
        if (immediateUpdate_ || rolloverPending_)
        {
            return call_immediately(STATE(client_update_commit));
        }
        else
        {
            // because the set time sometimes comes in bursts, we will wait
            // a little while for them to finish coming in.  This avoids
            // extra redundant processing.
            sleeping();
            return sleep_and_call(&timer_, MSEC_TO_NSEC(100),
                                  STATE(client_update_maybe));
        }
    }
 
    Action client_update_maybe()
    {
        if (timer_.is_triggered())
        {
            return call_immediately(STATE(client_update));
        }
        else
        {
            sleeping_ = false;
            return call_immediately(STATE(client_update_commit));
        }
    }
 
    Action client_update_commit()
    {
        if (rolloverPendingDate_ || rolloverPendingYear_)
        {
            // initiate timeout, in case the time server doesn't complete the
            // sequence successfully.
            sleeping();
            return sleep_and_call(&timer_, SEC_TO_NSEC(6),
                                  STATE(rollover_pending));
        }
 
 
        {
            bool notify = false;
            time_t old_seconds = 0;
            time_t new_seconds = 0;
            {
                AtomicHolder h(this);
                old_seconds = time();
                if (rate_ != rateRequested_ ||
                    (immediateUpdate_ && immediatePending_))
                {
                    // rate changed or an immediate update was pending.
                    notify = true;
                }
 
                // we are about to commit an updated time, reset the flags
                immediateUpdate_ = false;
                immediatePending_ = false;
 
                rate_ = rateRequested_;
                seconds_ = ::mktime(&tm_);
                timestamp_ = OSTime::get_monotonic();
                new_seconds = seconds_;
                if (rolloverPending_)
                {
                    // roll forward/back the 3 second delay for the year/date
                    // report
                    seconds_ += rate_ >= 0 ? 3 : -3;
                    rolloverPending_ = false;
                }
 
                if (notify == false)
                {
                    if (std::abs(compare_realtime(seconds_, old_seconds)) > 3)
                    {
                        // We hadd a drift of more than 3 real seconds.  3 real
                        // seconds is chosen because it is within the magnitude
                        // of normal network jitter.
                        notify = true;
                    }
                }
                // release AtomicHolder
            }
            if (notify)
            {
                service_callbacks(old_seconds, new_seconds);
            }
        }
 
        waiting();
        return wait_and_call(STATE(client_update));
    }
 
    Action rollover_pending()
    {
        if (timer_.is_triggered())
        {
            return call_immediately(STATE(client_update_commit));
        }
        else
        {
            sleeping_ = false;
            // A date rollover sequence failed.  Query the clock in order to
            // resync.  In the meantime, continue using our internal time.
            return call_immediately(STATE(initialize));
        }
    }
 
    void waiting()
    {
        waiting_ = true;
        sleeping_ = false;
    }
 
    void sleeping()
    {
        waiting_ = false;
        sleeping_ = true;
    }
 
    void wakeup()
    {
        if (waiting_)
        {
            notify();
            waiting_ = false;
        }
        if (sleeping_)
        {
            timer_.trigger();
            sleeping_ = false;
        }
    }
 
    uint16_t configureAgent_   : 1; 
    uint16_t immediateUpdate_  : 1; 
    uint16_t immediatePending_ : 1; 
    uint16_t sleeping_         : 1; 
    uint16_t waiting_          : 1; 
    uint16_t rolloverPending_  : 1; 
    uint16_t rolloverPendingDate_ : 1; 
    uint16_t rolloverPendingYear_ : 1; 
    uint16_t serverDetected_      : 1; 
 
 
    DISALLOW_COPY_AND_ASSIGN(BroadcastTimeClient);
};
 
} // namespace openlcb
 
#endif // _OPENLCB_BROADCASTTIMECLIENT_HXX_