FeedEngine.h

Go to the documentation of this file.

// Copyright Onix Solutions Limited [OnixS]. All rights reserved.
//
// This software owned by Onix Solutions Limited [OnixS] and is 
// protected by copyright law and international copyright treaties.
//
// Access to and use of the software is governed by the terms of the applicable 
// OnixS Software Services Agreement (the Agreement) and Customer end user license 
// agreements granting a non-assignable, non-transferable and non-exclusive license 
// to use the software for it's own data processing purposes under the terms defined 
// in the Agreement.
//
// Except as otherwise granted within the terms of the Agreement, copying or 
// reproduction of any part of this source code or associated reference material 
// to any other location for further reproduction or redistribution, and any 
// amendments to this copyright notice, are expressly prohibited.
//
// Any reproduction or redistribution for sale or hiring of the Software not in 
// accordance with the terms of the Agreement is a violation of copyright law.
//

#pragma once

#include <OnixS/CME/Streamlined/String.h>
#include <OnixS/CME/Streamlined/Integral.h>
#include <OnixS/CME/Streamlined/TinySet.h>
#include <OnixS/CME/Streamlined/Time.h>

ONIXS_CMESTREAMLINEDMDH_NAMESPACE_BEGIN

/// Zero-based index of CPU.
typedef Int32 CpuIndex;

/// Represents set of CPU indices.
ONIXS_CMESTREAMLINEDMDH_LTWT_CLASS ThreadAffinity
{
public:
    /// Alias for collection of CPU indices.
    typedef 
    TinySet<CpuIndex> 
    CpuIndices;

    /// Initializes empty set.
    ThreadAffinity()
    {
    }

    /// Initializes as copy of other set.
    ThreadAffinity(
        const ThreadAffinity& other)
        : indices_(other.indices_)
    {
    }

    /// Utilizes all the resources.
    ~ThreadAffinity()
    {
    }

    /// Read-only access to index collection.
    const CpuIndices& cpus() const
    {
        return indices_;
    }

    /// Collection of CPU indices.
    CpuIndices& cpus()
    {
        return indices_;
    }

private:
    CpuIndices indices_;
};

/// Serializes thread affinity into a string. 
ONIXS_CMESTREAMLINEDMDH_EXPORTED 
void 
toStr(
    std::string&, 
    const ThreadAffinity&);

/// Serializes thread affinity into a string. 
inline 
std::string 
toStr(
    const ThreadAffinity& affinity)
{
    std::string str;

    toStr(str, affinity);

    return str;
}

/// Collection of settings affecting Feed Engine behavior.
ONIXS_CMESTREAMLINEDMDH_LTWT_CLASS FeedEngineSettings
{
    ThreadAffinity threadAffinity_;
    UInt32 threadCount_;

    WatchService* watch_;

    UInt32 dataWaitTime_;
    UInt32 spinBeforeIdleTime_;

    UInt32 socketBufferSize_;

public:
    /// Initializes options with default values. 
    FeedEngineSettings()
        : threadAffinity_()
        , threadCount_(1)
        , watch_(
            &UtcWatch::service())
        , dataWaitTime_(10)
        , spinBeforeIdleTime_(1)
        , socketBufferSize_(
            8 * 1024 * 1024)
    {
    }

    /// Cleans everything up.
    ~FeedEngineSettings()
    {
    }

    /// Defines set of CPUs allowed for each working thread 
    /// to be executed on while processing market data by Handlers 
    /// binded to Feed Engine instance which is configured by given 
    /// settings.
    ///
    /// @note By default set is empty thus allowing threads 
    /// to be executed on any CPU available in the system. 
    const 
    ThreadAffinity& 
    threadAffinity() const
    {
        return threadAffinity_;
    }

    /// Defines set of CPUs allowed for each working thread 
    /// to be executed on while processing market data by Handlers 
    /// binded to Feed Engine instance which is configured by given 
    /// settings.
    ///
    /// @note By default set is empty thus allowing threads 
    /// to be executed on any CPU available in the system. 
    ThreadAffinity& threadAffinity()
    {
        return threadAffinity_;
    }

    /// Number of working threads to be used by feed engine.
    /// 
    /// @note Default value is '1'.
    UInt32 
    threadCount() const
    {
        return threadCount_;
    }

    /// Sets threadsCount. @see threadsCount.
    void 
    threadCount(
        UInt32 value)
    {
        threadCount_ = value;
    }

    /// Watch service to be used by Feed Engine.
    /// 
    /// Watch is used by Feed Engine to timestamp 
    /// packets received from the feeds.
    /// 
    /// @note By default, UTC watch service is used.
    WatchService& watch() const
    {
        return *watch_;
    }

    /// Watch service to be used by Feed Engine.
    /// 
    /// If no instance associated, UTC watch is used. 
    void 
    watch(
        WatchService& watch)
    {
        watch_ = &watch;
    }

    /// Defines amount of time Feed Engine spends on socket
    /// waiting for I/O while running master processing loop.
    ///
    /// Time is measured in milliseconds.
    ///
    /// @note Default value is '10'.
    ///
    /// @warning Given parameter significantly affects
    /// Handler's responsiveness and load onto CPU! 
    UInt32 
    dataWaitTime() const
    {
        return dataWaitTime_;
    }

    /// Sets dataWaitTime. @see dataWaitTime.
    void 
    dataWaitTime(
        UInt32 value)
    {
        dataWaitTime_ = value;
    }

    /// Defines amount of time Feed Engine keeps cycling before 
    /// going to sleep when no useful activity can be done. 
    ///
    /// Time is measured in milliseconds.
    ///
    /// @note Default value is '1'. 
    ///
    /// @warning Given parameter has direct influence onto CPU load! 
    UInt32 
    spinBeforeIdleTime() const
    {
        return spinBeforeIdleTime_;
    }

    /// Sets redundancySpinTime. @see redundancySpinTime.
    void 
    spinBeforeIdleTime(
        UInt32 value)
    {
        spinBeforeIdleTime_ = value;
    }

    /// Defines size of receiving buffer in bytes for sockets.
    ///
    /// @note Default value is 4 MiB.
    UInt32 
    socketBufferSize() const
    {
        return socketBufferSize_;
    }

    /// Sets udpSocketBufferSize. @see udpSocketBufferSize.
    void 
    socketBufferSize(
        UInt32 value)
    {
        socketBufferSize_ = value;
    }
};

/// Serializes feed engine settings into a string. 
ONIXS_CMESTREAMLINEDMDH_EXPORTED
void
toStr(
    std::string&,
    const FeedEngineSettings&);

/// Serializes feed engine settings into a string. 
inline 
std::string
toStr(
    const FeedEngineSettings& settings)
{
    std::string str;

    toStr(str, settings);

    return str;
}

/// Identifies reasons feed engine threads becomes idle. 
ONIXS_CMESTREAMLINEDMDH_LTWT_STRUCT FeedEngineThreadIdle
{
    enum Reason
    {
        /// Thread waited for incoming data using corresponding 
        /// I/O operations (like 'select') and exited waiting with 
        /// no data availability signs. 
        DataWaitTimeout,

        /// Thread entered idle state due to absence of any data and 
        /// while other threads are waiting for new incoming data. 
        Redundant
    };
};

//

ONIXS_CMESTREAMLINEDMDH_EXPORTED_CLASS_DECL(FeedEngine);

/// Listener for thread-related events.
/// 
/// Members of this classes are invoked to reflect 
/// various life-time events of threads spawned and 
/// used by the feed engine while processing market data. 
ONIXS_CMESTREAMLINEDMDH_EXPORTED_STRUCT FeedEngineListener
{
    /// Member invoked by feed engine when 
    /// a new processing thread is spawned. 
    /// 
    /// @note Invocation is done within newly 
    /// started thread. 
    virtual 
    void 
    onFeedEngineThreadBegin(
        const FeedEngine&)
    {
    }

    /// Member is invoked by feed engine when
    /// processing thread is about to end. 
    /// 
    /// @note Invocation is done within the 
    /// thread that is about to end. 
    virtual
    void
    onFeedEngineThreadEnd(
        const FeedEngine&)
    {
    }

    /// Is called when feed engine's thread is idle. 
    ///
    /// Thread becomes idle when either no data is received within 
    /// time interval defined by FeedEngineSettings::dataWaitTime 
    /// parameter or no pending data is available for processing. 
    /// In the first case, callback is invoked with 'DataWaitTimeout'
    /// reason. In the second case, thread is considered as redundant
    /// and thus callback is invoked with 'Redundant' reason. 
    /// After callback invocation threads may sleep in kernel to reduce 
    /// load onto CPU and racing between feed engine working threads. 
    ///
    /// Integer parameter-variable defines amount of time feed engine 
    /// suggest for thread to sleep in kernel after invoking given member.
    virtual 
    void
    onFeedEngineThreadIdle(
        const FeedEngine&,
        FeedEngineThreadIdle::Reason,
        UInt32&)
    {
    }

    /// Invoked if working thread experiences an issue 
    /// while processing its tasks. Once invoked, working 
    /// thread continues performing its tasks. 
    virtual
    void
    onFeedEngineThreadIssue(
        const FeedEngine&,
        const Char*)
    {
    }
};

// Workhorse for the public Feed Engine. 
ONIXS_CMESTREAMLINEDMDH_INTERNAL_CLASS_DECL
(
    MultithreadedFeedEngine
);

/// Manages processing machinery for market data received from feeds.
ONIXS_CMESTREAMLINEDMDH_EXPORTED_CLASS FeedEngine
{
    // Current workhorse is multi-threaded 
    // feed engine using on socket API. 
    friend 
    ONIXS_CMESTREAMLINEDMDH_INTERNAL_CLASS_DECL
    (
        MultithreadedFeedEngine
    );

    MultithreadedFeedEngine* engine_;

    // Copying is not supposed for given class. 

    FeedEngine(const FeedEngine&);
    FeedEngine& operator =(const FeedEngine&);

public:
    /// Initializes engine with given configuration.
    FeedEngine(
        const FeedEngineSettings&,
        FeedEngineListener* = ONIXS_CMESTREAMLINEDMDH_NULLPTR);

    /// Destructs given instance.
    ~FeedEngine();

    /// Settings used define behavior of given instance. 
    const 
    FeedEngineSettings& 
    settings() const;
};

ONIXS_CMESTREAMLINEDMDH_NAMESPACE_END