Timed processing in a thread in a CLI app? 8ms needed

[Lehthanis]

I'm trying to write a simple CLI app that needs to send data to another machine every 8 milliseconds (125hz processing). I have my TCP/IP connection established, and I figured I could just create a processing loop, but how can I make the loop process at 8ms? Mind you, not the time that it takes to calculate the data it's sending plus 8ms...just 8ms. Kinda like a timer event that triggers every 8ms. Any thoughts or links?

[SamuelAdams]

If you would share your TCP/IP code with me I'd be happy to test that.

Just to make sure you just want to send every 8ms, regardless of how much time it takes to arrive.

You are aware that internet traffic can take longer than 8ms right ?

I'm pretty sure I have code for that already just need to see how to place it in your code.

[Lehthanis]

I'm using the PCSBSocket code found here: https://github.com/X-Plane/xptools/blob/master/src/Network/PCSBSocket.h

I already have a little experience using it for UDP so I figured it'd be an easy start for me.

I haven't even started forming a loop for this yet, I've started the initialization and handshake stuff. It will all be running on a lan with minimal other traffic, so timing should be fine. I'd love to see what you have either way.

here's my init in its early stages:

#include <iostream>
#include "EMCMessageDefinitions.h"
#include "PCSBSocket.h"

using namespace std;

PCSBSocket * mSocket;			// Socket
const char * inServerIP = "10.10.101.40";
unsigned short inPort = 10002;

MC_RUN_SETUP_STRUCT mcrunsetup;
EMC_RUN_SETUP_STRUCT emcrunsetup;

bool initSocket()
{
    mSocket = NULL;
    mSocket->StartupNetworking(0);
    unsigned long	ip = PCSBSocket::LookupAddress(inServerIP);
    if (ip == 0)
        return 0;

    mSocket = new PCSBSocket(0, false);

    bool isConnected = false;

    mSocket->Connect(ip, inPort);
    while (!isConnected)
    {
        isConnected = mSocket->GetStatus() == PCSBSocket::status_Connected;
    }

    return 1;
}

int main(int argc, char *argv[])
{
    // Initialize Socket Call

    if (initSocket()) {
        cout << "Socket Initialized\n";
    }
    else {
        cout << "Socket Not Initialized\n";
    }
    bool isConnected = mSocket->GetStatus() == PCSBSocket::status_Connected;
    if (isConnected)
    {
        cout << "Waiting for Run Setup\n";
        while (mSocket->ReadData(&mcrunsetup, sizeof(MC_RUN_SETUP_STRUCT)) == 0)
        {
         //nothing
        }

        cout << "Sending EMC Run Setup\n";
        mSocket->WriteData(&emcrunsetup, sizeof(EMC_RUN_SETUP_STRUCT));

        //Write code to wait for Run Button Signal
        
        //Loop here

    } else
    {
        cout << "Not Connected\n";
    }
    //Finished - Kill Socket Connection
    if (mSocket) mSocket->Disconnect();
    mSocket->ShutdownNetworking(0);
    cout << "Exiting\n";
}

[coder777]

This is exactly what the chrono library is designed for:

http://www.cplusplus.com/reference/chrono/

See time_point and duration.


The thread library makes it possible to wait for specified amount of time.

http://www.cplusplus.com/reference/thread/this_thread/

See this_thread/sleep_for.

But I don't think that 8 ms is a realistic time span for reading from and writing to the internet.

[Lehthanis]

The thing is, I don't want to "wait" really, there will be calculations and such taking place...I think I need some kind of callback or something that's triggered by an 8ms timer...

I need the event to be triggered every 8ms, and if every calculation takes less than 8ms to calculate, then even though there's an offset from the time its called, at least the sending will be consistently 8ms...I have no idea how to even start building this sort of loop. All of my C/C++ experience is in writing plugins for software that already has loops and framerates defined.

SamuelAdams, you were going to share some code?

Also, keep in mind, this will not be going out over the internet, it's for a local area network between two machines in the same room.

I'll read up on the chrono library...but if I can see some good examples, that'd be awesome.

[JLBorges]

#include <iostream>
#include <thread>
#include <chrono>
#include <functional>
#include <atomic>
#include <ctime>

// call the function periodically, once every 'period_in_milliseconds' milliseconds.
// repeat the call a maximum of 'ntimes' times or till 'stop_calls' is set to true
void periodic_call( int period_in_milliseconds, int ntimes, std::atomic<bool>& stop_calls, std::function< void() > function  )
{
    const auto period = std::chrono::milliseconds(period_in_milliseconds) ;

    for( int i = 0 ; i < ntimes && !stop_calls ; ++i )
    {
        auto start = std::chrono::steady_clock::now() ;
        function() ; // assumes that the function won't take more than 'period_in_milliseconds' milliseconds
        auto end = std::chrono::steady_clock::now() ;

        // set delta to the time (in milliseconds) that was used to execute the function
        const auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(end-start) ;

        // sleep till the wait time has elapsed
        std::this_thread::sleep_for( period - delta ) ;
    }
}

void foo( int, int )
{
    static int n = 0 ;
    auto t = std::time(nullptr) ;
    char now[128] ;
    std::strftime( now, sizeof(now), "foo called  at %H:%M:%S\n", std::localtime(&t) ) ;
    std::cout << ++n << ". " << now << std::flush ;
}

int main()
{
    // launch a thread to call foo(2,3) once every 2 seconds, max 500 times in all 
    // note: stop_calls is passed as a wrapped reference to std::atomic<bool>
    std::atomic<bool> stop_calls { false } ; 
    std::thread call_back_thread( periodic_call, 2000, 500, std::ref(stop_calls), std::bind( foo, 2, 3 )  ) ;
    
    std::this_thread::sleep_for( std::chrono::seconds(10) ) ; // sleep for 10 seconds
    stop_calls = true ; // stop the calls
    call_back_thread.join() ; // wait for the thread to finish
    std::cout << "done!\n" ;
}

http://coliru.stacked-crooked.com/a/6339d67ab36905d1

[Lehthanis]

Awesome! Thanks JLBorges

[Lehthanis]

What kind of conversion do I have to do to cout that delta variable? This code worked like a charm.

[JLBorges]

To print out the number of millisecond interval ticks:

// http://en.cppreference.com/w/cpp/chrono/duration/count
const auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(end-start) ;
std::cout << "   delta: " << delta.count() << " milliseconds\n" ;

To get a more precise estimate of the duration in milliseconds, use a duration with a floating point type as the representation and milliseconds as the period.

const std::chrono::duration< double, std::ratio<1,1000> > delta_fp(end-start) ;
std::cout << " delta_fp: " << delta_fp.count() << " milliseconds\n" ;

http://coliru.stacked-crooked.com/a/6284d8a0391b122d

[JLBorges]

If output streaming of objects of type std::chrono::duration is a frequent requirement, overloading the stream insertion operator would be convenient. For instance:

#include <iostream>
#include <chrono>
#include <string>

namespace utility
{
    namespace chrono
    {
        constexpr inline const char* units( std::chrono::nanoseconds ) noexcept { return "nanoseconds" ; }
        constexpr inline const char* units( std::chrono::microseconds )  noexcept { return "microseconds" ; }
        constexpr inline const char* units( std::chrono::milliseconds )  noexcept { return "milliseconds" ; }
        constexpr inline const char* units( std::chrono::seconds )  noexcept { return "seconds" ; }
        constexpr inline const char* units( std::chrono::minutes )  noexcept { return "minutes" ; }
        constexpr inline const char* units( std::chrono::hours )  noexcept { return "hours" ; }

        template < std::intmax_t A, std::intmax_t B > struct gcd
        { static constexpr std::intmax_t value = gcd<B,A%B>::value ; };

        template < std::intmax_t A > struct gcd<A,0>
        { static constexpr std::intmax_t value = A ; };

        template < typename R, std::intmax_t N, std::intmax_t D >
        std::string units( std::chrono::duration< R, std::ratio<N,D> > )
        {
            constexpr auto NN = N / gcd<N,D>::value ;
            constexpr auto DD = D / gcd<N,D>::value ;
            return "intervals of " + std::to_string( double(N) / D ) +
                   " (" + std::to_string(NN) + '/' + std::to_string(DD) + ") seconds each" ;
        }
    }
}

template < typename R, std::intmax_t N, std::intmax_t D >
std::ostream& operator<< ( std::ostream& stm, std::chrono::duration< R, std::ratio<N,D> > d )
{ return stm << d.count() << ' ' << utility::chrono::units( decltype(d){} ) ; }

int main()
{
    std::chrono::hours duration_hrs(10) ;
    std::cout << duration_hrs << '\n'
              << std::chrono::duration_cast<std::chrono::seconds>(duration_hrs) << '\n'
              << std::chrono::duration< double, std::ratio<36,60> >(duration_hrs) << '\n' ;
}

http://coliru.stacked-crooked.com/a/61c6417e69e2f10a
http://rextester.com/ZGA53533

For production code, use Hinnant's non-viral (MIT license) date and time library which extends the facilities in the standard C++ library <chrono> https://github.com/HowardHinnant/date

Output streaming of std::chrono::duration: https://howardhinnant.github.io/date/chrono_io.html

[Lehthanis]

Super helpful dude. I appreciate it so much! I don't think this will be much of a repeat function, I'm not sure how much I'll need to use...what I've got is working pretty good...Only thing is, I'm not 100% sure that the thread is stopping when I ask it to. I need to move stop_calls and call_back_thread up in scope so I can stop it from other functions. How do I instantiate call_back_thread globally? thread call_back_thread; doesn't work...does it need the full declaration/constructor?

    stop_calls = true ; // stop the calls
    call_back_thread.join() ; // wait for the thread to finish 

In your example, the lines above for join on call_back_thread...is that a blocking function? does it actually wait until the thread stops or will the thread continue to fire a few more times after this line runs?

[JLBorges]

> I need to move stop_calls and call_back_thread up in scope so I can stop it from other functions.

Is only one callback in the entire program, which can be stopped from other functions;
or are there many callbacks, each of which has to be individually started and stopped from different functions?


> the lines above for join on call_back_thread...is that a blocking function?

Yes.

> does it actually wait until the thread stops or will the thread continue to fire a few more times after this line runs?

It will wait until the thread had finished execution.

[Lehthanis]

Yes, only one callback in the program...yeah, I need it to be able to be stopped from multiple functions.

Thanks for the help!

[JLBorges]

The simplest modification to the earlier program would be: make the flag stop_calls an object declared at namespace scope (instead of a parameter passed by reference to the periodic_call function). Then it can be set to true from anywhere in the program.

#include <iostream>
#include <thread>
#include <chrono>
#include <functional>
#include <atomic>
#include <ctime>

extern std::atomic<bool> stop_calls ; // you may want to place this in a header file
                                      // (included by functions which want to stop the calls)

std::atomic<bool> stop_calls { false } ;

// call the function periodically, once every 'period_in_milliseconds' milliseconds.
// repeat the call a maximum of 'ntimes' times or till 'stop_calls' is set to true
void periodic_call( int period_in_milliseconds, int ntimes, std::function< void() > function  )
{
    const auto period = std::chrono::milliseconds(period_in_milliseconds) ;

    for( int i = 0 ; i < ntimes && !stop_calls ; ++i )
    {
        auto start = std::chrono::steady_clock::now() ;
        function() ; // assumes that the function won't take more than 'period_in_milliseconds' milliseconds
        auto end = std::chrono::steady_clock::now() ;

        // set delta to the time (in milliseconds) that was used to execute the function
        const std::chrono::duration< double, std::ratio<1,1000> > delta(end-start) ;

        // sleep till the wait time has elapsed
        if( !stop_calls ) std::this_thread::sleep_for(period-delta) ;
    }
}

void foo( int, int )
{
    static int n = 0 ;
    auto t = std::time(nullptr) ;
    char now[128] ;
    std::strftime( now, sizeof(now), "foo called  at %H:%M:%S\n", std::localtime(&t) ) ;
    std::cout << ++n << ". " << now << std::flush ;
}

void some_fun() ;

int main()
{
    // launch a thread to call foo(2,3) once every second, max 500 times in all
    std::thread call_back_thread( periodic_call, 1000, 500, std::bind( foo, 2, 3 )  ) ;

    some_fun() ;

    call_back_thread.join() ; // wait for the thread to finish
    std::cout << "done!\n" ;
}

void some_fun()
{
    for( int i = 0 ; i < 16 ; ++i )
    {
        std::cout << "waiting...\n" << std::flush ;
        std::this_thread::sleep_for( std::chrono::milliseconds(400) ) ;
    }

    std::cout << "\nstopping the calls\n" << std::flush ;
   stop_calls = true ; // this may be done from any function
}

http://coliru.stacked-crooked.com/a/a5c26127bc7dd376

[Lehthanis]

what about the call_back_thread.join() being callable externally though? Thanks much!

[JLBorges]

Pass the information about the callback thread to the function. For instance:

struct periodic_call_info
{
    std::thread& thread ;
    std::atomic<bool>& stop ;

    void stop_calls( bool wait = true )
    {
        stop = true ;
        if( wait && thread.joinable() ) thread.join() ;
    }
};

// ...

void some_fun( periodic_call_info info )
{
    for( int i = 0 ; i < 16 ; ++i )
    {
        std::cout << "waiting...\n" << std::flush ;
        std::this_thread::sleep_for( std::chrono::milliseconds(500) ) ;
    }

    std::cout << "\nstop calls and wait for thread to finish\n" << std::flush ;
    info.stop_calls() ; // wait == true
}

int main()
{
    std::atomic<bool> stop_calls { false } ;
    // launch a thread to call foo(2,3) once every 2 seconds, max 500 times in all
    std::thread call_back_thread( periodic_call, 2000, 500, std::ref(stop_calls), std::bind( foo, 2, 3 )  ) ;

    some_fun( { call_back_thread, stop_calls } ) ; 
    
    // ...
}

http://coliru.stacked-crooked.com/a/74ea2dc8b2b5aa7a