C++11.cpp

// stupid comment
#include <string>
#include <iostream>
#include <sstream>
#include <memory>
#include <thread>
#include <future>
#include <ratio>
#include <sstream>
#include <boost/range/irange.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
#include "Probe.h"
#include "zhelpers.hpp"

using namespace std;
using namespace std::chrono;

struct X: public Traceable
{
    X(const std::string& s, double d)
        : name_(s), rate_(d) {}
    X()
        : rate_(0.0)  { }
    virtual ~X() {}

private:
    std::string name_ ;
    double rate_ ;
};
typedef std::shared_ptr<X> Xptr;

struct Y: public Traceable {
    virtual ~Y() {}
};
typedef std::shared_ptr<Y> Yptr;


struct Z: public Traceable {
    virtual ~Z() {}
};
typedef std::shared_ptr<Z> Zptr;

template <typename T>
string
oid (const shared_ptr<T>& ptr)
{
    ostringstream oss ;
    oss << hex << ptr.get();
    return oss.str();
}

string tid()
{
    ostringstream oss ;
    oss << this_thread::get_id();
    return oss.str();
}

void dummy4(Xptr br)
{
    Trace<X> t(br, "dummy4_"+tid());
    std::this_thread::sleep_for( milliseconds(400) );
}

void dummy3(Xptr br)
{
    Trace<X> t(br, "dummy3_"+tid());
    dummy4(br);
    std::this_thread::sleep_for( milliseconds(300) );
}

void dummy2(Xptr br)
{
    Trace<X> t(br, "dummy2_"+tid());
    std::this_thread::sleep_for( milliseconds(200) );
    dummy3(br);
}

void dummy1(Xptr br)
{
    Trace<X> t(br, "dummy1_"+tid());
    auto rc = thread([&] { dummy2(br); });
    std::this_thread::sleep_for( milliseconds(100) );
    rc.join();
}

void test(Xptr br)
{
    {
        {
            Trace<X> t1(br, "A1_"+tid());
            {
                Trace<X> t2(br, "A2_"+tid());
            }
        }
        Trace<X> t3(br, "A3_"+tid());
    }
    {
        Trace<X> t4(br, "A4_"+tid());
        {
            Trace<X> t (br, "A5_"+tid());
            {
                Trace<X> t(br, "A6_"+tid());
                {
                    Trace<X> t(br, "A7_"+tid());
                }
            }
        }
    }
}

void f2(Xptr x)
{
    Trace<X> t(x, "f2_"+tid());
    auto f2 = thread([&] { return dummy1(x); });
    std::this_thread::sleep_for( milliseconds(1000) );
    f2.join();
}

void f1(Xptr x)
{
    Trace<X> t(x, "f1_"+tid());
    f2(x);
    auto f1 = thread([&] { return dummy1(x); });
    std::this_thread::sleep_for( milliseconds(2000) );
    f1.join();
}

void g1(Xptr x)
{
    Trace<X> t(x, "g1_1"+tid());
    {
        Trace<X> t(x, "g1.2_"+tid());
    }
    std::this_thread::sleep_for( milliseconds(2000) );
}

zmq::context_t context(1);
zmq::socket_t publisher(context, ZMQ_PUB);

auto sink_fn = [] (const boost::uuids::uuid& uuid, const std::vector<Span>& vs)
{
    // TODO: do something useful.
    // use google protocol buffer i/o serialize
    std::ostringstream oss;
    using namespace std::chrono;
    using boost::property_tree::ptree;
    using boost::property_tree::write_json;

    for (const auto& s: vs)
    {
        ptree pt;
        pt.put("uuid", uuid);
        pt.put("pid", s.pid_);
        pt.put("id", s.id_);
        pt.put("tag", s.tag_);
        pt.put("t1", duration_cast<microseconds>(s.start_.time_since_epoch()).count());
        pt.put("t2", duration_cast<microseconds>(s.end_.time_since_epoch()).count());
        ostringstream buf;
        write_json(buf,pt,false);
        s_sendmore(publisher, "RTBkitTRACE");
        s_send(publisher, buf.str());
        std::cerr << buf.str() ;
    }
};

void clone_test(Xptr x)
{
    // auto x = Probe<X>::create (sink_fn, "x", 0.11111);
    Trace<X> t(x, "clone_test()_x"+tid());
    {
        auto y = Probe<Y>::clone (x, sink_fn);
        Trace<Y> t(y, "Y1-"+oid(y));
        {
            Trace <Y> t(y, "Y2-"+oid(y));
            {
                Trace <Y> t(y, "Y3-"+oid(y));
            }
        }
    }

}

void thread_test()
{
    auto x = Probe<X>::create(sink_fn, "j1", 0.11111);
    Trace<X> t1 (x, "thread_test()_x_"+tid());
    auto rc1 = thread ([=] { dummy1(x);});
    auto rc2 = thread ([=] { clone_test(x);});
    rc1.join();
    rc2.join();
}

int main()
{
    publisher.bind("tcp://*:5563");

//    for (auto i: boost::irange<int>(0,1))
//    {
//        auto x = Probe<X>::create(sink_fn, "X"+to_string(i), (i+1)*0.111111);
//        Trace<X> t(x, "x-"+oid(x));
//#if 1
//        thread thr1, thr2;
//        auto rc = thread ([=] {g1(x);});
//        {
//            Trace<X> t(x, "main_1_"+tid());
//            {
//                Trace<X> t(x, "main_2_"+tid());
//                {
//                    thr1 = thread([=] {f1(x);});
//                    {
//                        Trace <X> t   (x, "main_3_"+tid());
//                        thr2 = thread ([=] {test(x);});
//                    }
//                }
//            }
//        }
//        rc.join();
//        thr1.join();
//        thr2.join();
//        auto y = clone(x);
//        Trace<Y> t2(y, "y");
//#else
//        {
//             auto y = Probe<Y>::clone(x, sink_fn);
//             Trace<Y> ty1(y, "ty1_"+oid(y));
//             {
//               auto z = Probe<Z>::clone(x, sink_fn);
//            	 Trace<Z> tz(z, "tz1_"+oid(z));
//             }
//
//        }
//#endif
//    }

    auto x = Probe<X>::create(sink_fn, "foobar", 0.123);
    f1 (x);
    return 0;
}

gistfile1.txt

RTBkit Tracking. The command: c++ -std=c++11 -stdlib=libc++ -g -o C++11 ./C++11.cpp -lboost_thread-mt -lzmq&&./C++11 should produce something like: {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"5","id":"6","tag":"dummy4_0x105b76000","t1":"172721123568","t2":"172721523753"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"4","id":"5","tag":"dummy3_0x105b76000","t1":"172721123560","t2":"172721824160"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"3","id":"4","tag":"dummy2_0x105b76000","t1":"172720923441","t2":"172721824170"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"2","id":"3","tag":"dummy1_0x105af3000","t1":"172720923351","t2":"172721824302"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"1","id":"2","tag":"f2_0x7fff72166180","t1":"172720923232","t2":"172721923796"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"9","id":"10","tag":"dummy4_0x105b76000","t1":"172722124445","t2":"172722524785"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"8","id":"9","tag":"dummy3_0x105b76000","t1":"172722124439","t2":"172722825478"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"7","id":"8","tag":"dummy2_0x105b76000","t1":"172721924030","t2":"172722825484"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"6","id":"7","tag":"dummy1_0x105af3000","t1":"172721923934","t2":"172722825632"} {"uuid":"90b6acb4-c24e-429d-9537-4fb9b09e2899","pid":"0","id":"1","tag":"f1_0x7fff72166180","t1":"172720923216","t2":"172723924943"} 

Probe.h

#ifndef PROBE_H_
#define PROBE_H_

/*
 * Probe.h
 *
 *  Created on: Aug 27, 2013
 *      Author: jan
 *
 */
#include <string>
#include <stack>
#include <utility> // forward
#include <functional>
#include <atomic>
#include <chrono>
#include <iostream>
#include <memory>
#include <ctime>
#include <unordered_map>
#include <cassert>
#include <limits>

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/functional/hash.hpp>
#include <boost/uuid/uuid_io.hpp>

#if 1 //  __GNUC_MINOR__ <= 7
#define USE_BOOST_TSS
#endif

#ifdef USE_BOOST_TSS
#include <boost/thread/tss.hpp>
#endif

struct Span
{
    std::string              tag_ ;
    uint32_t                 id_, pid_;
    std::chrono::steady_clock::time_point start_, end_;
    Span(uint32_t id, uint32_t pid) : id_ (id), pid_(pid) {}
    Span() : Span(0,0) {}
    Span& operator=(const Span&) =delete;
};

typedef std::unordered_map<
boost::uuids::uuid,
      std::stack<Span>,
      boost::hash<boost::uuids::uuid>> ProbeStacks;

#ifndef USE_BOOST_TSS
static thread_local ProbeStacks PSTACKS;
#else
static boost::thread_specific_ptr<ProbeStacks> PSTACKS;
#endif

typedef std::function<void(const boost::uuids::uuid&,
                           const std::vector<Span>&)> SinkCb;
/*
 *  Probe context.
 *  Will be shared between related (cloned) probes
 */
struct Pctx {
    Pctx(SinkCb s)
        : sink_(s)
        , uuid_(boost::uuids::random_generator()())
        , span_id_ (ATOMIC_VAR_INIT(0)) {}
    SinkCb                    sink_ ;
    const boost::uuids::uuid  uuid_ ;
    std::atomic<uint32_t>     span_id_;
};

struct Traceable {
    virtual ~Traceable()                  {}
    virtual void push(const std::string&) {}
    virtual void pop()                    {}
    virtual bool probed () const          {
        return false;
    }
    virtual std::shared_ptr<Pctx> ctx ()  {
        return std::shared_ptr<Pctx>();
    }
};

/**
 * Mixin like.
 * Assumes T overrides all member functions
 * of Traceable.
 * TODO: compile check that.
 */
template <typename T, int N = 1>
class Probe: public T
{
public:
    typedef typename std::shared_ptr<T> ProbePtr;

    // perfectly forwarding constructors
    template <typename... Args>
    explicit Probe(SinkCb sink, Args&&... args)
        :  T(std::forward<Args>(args)...)
        , ctx_ (std::make_shared<Pctx>(sink))
        , state_ (Unlocked)
    {
    }

    template <typename... Args>
    explicit Probe(const std::shared_ptr<Pctx>& ctx,
                   SinkCb sink, Args&&... args)
        :  T(std::forward<Args>(args)...)
        ,  ctx_ (ctx)
        ,  state_ (Unlocked)
    {
    }


    template <typename... Args>
    static
    ProbePtr
    create(SinkCb sink, Args&&... args)
    {
        // static std::atomic<uint32_t> count = ATOMIC_VAR_INIT(0);
        static std::atomic<uint32_t> count (0);
        ProbePtr rc ;
        auto n = count++;
        if (n%N)
            rc.reset(new T(std::forward<Args>(args)...)) ;
        else
            rc.reset(new Probe<T>(sink, std::forward<Args>(args)...));
        return rc ;
    }


    /**
     * link a trace of type U, with a trace of type T
     */
    template <typename U, typename... Args>
    static
    ProbePtr
    clone(std::shared_ptr<U> u, SinkCb sink, Args&&... args)
    {
        ProbePtr rc ;
        if (u->probed())
            rc.reset(new Probe<T>(u->ctx(), sink, std::forward<Args>(args)...));
        else
            rc.reset(new T(std::forward<Args>(args)...)) ;
        return rc ;
    }

    virtual ~Probe()     {
        flush_spans();
    }

    virtual void push (const std::string& s) {
        // grab our context
        auto& ctx = *ctx_.get();
#ifdef USE_BOOST_TSS
        if (!PSTACKS.get())
            PSTACKS.reset (new ProbeStacks());
        auto& stk = (*PSTACKS.get())[ctx.uuid_];
#else
        auto& stk = PSTACKS[ctx.uuid_];
#endif
        // if first stack empty, set dummy stack entry
        // with pid = id=current_span_id (i.e., create a root)
        Span sp;
        sp.tag_   = s;
        if (stk.empty())
        {
            stk.emplace(Span(std::atomic_load(&ctx.span_id_), 0));
        }

        sp.id_    = ++ctx.span_id_;
        sp.pid_   = stk.top().id_;
        sp.start_ = std::chrono::steady_clock::now() ;
        stk.emplace (sp);
    }

    virtual void pop () {
        // grab our context
        auto& ctx = *ctx_.get();
#ifdef USE_BOOST_TSS
        if (!PSTACKS.get())
            PSTACKS.reset (new ProbeStacks());
        auto it = (*PSTACKS.get()).find(ctx.uuid_);
        assert (it != (*PSTACKS.get()).end());
#else
        auto it = PSTACKS.find(ctx.uuid_);
        assert (it != PSTACKS.end());
#endif
        // take ending time, save the span and pop it.
        it->second.top().end_ = std::chrono::steady_clock::now();
        {
            std::lock_guard<Probe<T>> l(*this);
            spans_.emplace_back (it->second.top());
        }
        it->second.pop();

        // if nothing left but root, cleanup
        if (1U==it->second.size())
        {
            assert (it->second.top().pid_ == 0);
#ifdef BOOST_USE_TSS
            (*PSTACKS.get()).erase(it);
#else
            PSTACKS->erase(it);
#endif
        }
    }

    void lock ()  {
        while (std::atomic_exchange(&state_, Locked) == Locked) ;
    }
    void unlock() {
        std::atomic_store(&state_, Unlocked);
    }


private:
    void flush_spans()        {
        auto& ctx = *ctx_.get();
        ctx.sink_(ctx.uuid_, spans_);
    }
    // Probe Context. Will be shared between
    // linked probes.

    std::shared_ptr<Pctx>     ctx_;
    enum LockState            {Locked, Unlocked};
    std::atomic<LockState>    state_;
    std::vector<Span>         spans_ ;
    std::shared_ptr<Pctx> ctx () override {
        return ctx_;
    }
    bool probed() const override {
        return true;
    }


};


template <typename T>
struct Trace
{
    Trace(std::shared_ptr<T> ptr, const std::string& tag)
        : ptr_(ptr) {
        ptr_->push (tag);
    }
    ~Trace() {
        ptr_->pop();
    }
    std::shared_ptr<T> ptr_ ;
};


#endif /* PROBE_H_ */

sinkTrace.py

"""
   do very useful thing with
   trace received.

"""
import zmq

def do_something_useful(trace):
    print trace

def main():
    """ main method """

    # Prepare our context and publisher
    context    = zmq.Context(1)
    subscriber = context.socket(zmq.SUB)
    subscriber.connect("tcp://localhost:5563")
    subscriber.setsockopt(zmq.SUBSCRIBE, "RTBkitTRACE")

    while True:
        # Read envelope with address
        [address, contents] = subscriber.recv_multipart()
        do_something_useful(contents)

    # We never get here but clean up anyhow
    subscriber.close()
    context.term()


if __name__ == "__main__":
    main()

zhelpers.hpp

/*  =========================================================================
    zhelpers.h - ZeroMQ helpers for example applications

    Copyright (c) 1991-2010 iMatix Corporation and contributors

    This is free software; you can redistribute it and/or modify it under
    the terms of the Lesser GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    This software is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    Lesser GNU General Public License for more details.

    You should have received a copy of the Lesser GNU General Public License
    along with this program. If not, see <http://www.gnu.org/licenses/>.
    =========================================================================
*/

// Olivier Chamoux <[email protected]>


#ifndef __ZHELPERS_HPP_INCLUDED__
#define __ZHELPERS_HPP_INCLUDED__

//  Include a bunch of headers that we will need in the examples

#include <zmq.hpp>

#include <iostream>
#include <iomanip>
#include <string>
#include <sstream>

#if (!defined(__WINDOWS__))
#include <sys/time.h>
#include <unistd.h>
#include <pthread.h>
#endif
#include <time.h>
#include <assert.h>
#include <stdlib.h>        // random()  RAND_MAX
#include <stdio.h>
#include <stdarg.h>
#include <signal.h>

//  Bring Windows MSVC up to C99 scratch
#if (defined (__WINDOWS__))
    typedef unsigned long ulong;
    typedef unsigned int  uint;
    typedef __int64 int64_t;
#endif

//  Provide random number from 0..(num-1)
#if (!defined(__WINDOWS__))
#define within(num) (int) ((float) (num) * random () / (RAND_MAX + 1.0))
#else
#define within(num) (int) ((float) (num) * rand () / (RAND_MAX + 1.0))
#endif

//  Receive 0MQ string from socket and convert into string
static std::string
s_recv (zmq::socket_t & socket) {

    zmq::message_t message;
    socket.recv(&message);

    return std::string(static_cast<char*>(message.data()), message.size());
}

//  Convert string to 0MQ string and send to socket
static bool
s_send (zmq::socket_t & socket, const std::string & string) {

    zmq::message_t message(string.size());
    memcpy (message.data(), string.data(), string.size());

    bool rc = socket.send (message);
    return (rc);
}

//  Sends string as 0MQ string, as multipart non-terminal
static bool
s_sendmore (zmq::socket_t & socket, const std::string & string) {

    zmq::message_t message(string.size());
    memcpy (message.data(), string.data(), string.size());

    bool rc = socket.send (message, ZMQ_SNDMORE);
    return (rc);
}

//  Receives all message parts from socket, prints neatly
//
static void
s_dump (zmq::socket_t & socket)
{
    std::cout << "----------------------------------------" << std::endl;

    while (1) {
        //  Process all parts of the message
        zmq::message_t message;
        socket.recv(&message);

        //  Dump the message as text or binary
        int size = message.size();
        std::string data(static_cast<char*>(message.data()), size);

        bool is_text = true;

        int char_nbr;
        unsigned char byte;
        for (char_nbr = 0; char_nbr < size; char_nbr++) {
            byte = data [char_nbr];
            if (byte < 32 || byte > 127)
                is_text = false;
        }
        std::cout << "[" << std::setfill('0') << std::setw(3) << size << "]";
        for (char_nbr = 0; char_nbr < size; char_nbr++) {
            if (is_text)
                std::cout << (char)data [char_nbr];
            else
                std::cout << std::setfill('0') << std::setw(2)
                   << std::hex << (unsigned int) data [char_nbr];
        }
        std::cout << std::endl;

        int more = 0;           //  Multipart detection
        size_t more_size = sizeof (more);
        socket.getsockopt (ZMQ_RCVMORE, &more, &more_size);
        if (!more)
            break;              //  Last message part
    }
}

//  Set simple random printable identity on socket
//
inline std::string
s_set_id (zmq::socket_t & socket)
{
    std::stringstream ss;
    ss << std::hex << std::uppercase
       << std::setw(4) << std::setfill('0') << within (0x10000) << "-"
       << std::setw(4) << std::setfill('0') << within (0x10000);
    socket.setsockopt(ZMQ_IDENTITY, ss.str().c_str(), ss.str().length());
    return ss.str();
}

//  Report 0MQ version number
//
static void
s_version (void)
{
    int major, minor, patch;
    zmq_version (&major, &minor, &patch);
    std::cout << "Current 0MQ version is " << major << "." << minor << "." << patch << std::endl;
}

static void
s_version_assert (int want_major, int want_minor)
{
    int major, minor, patch;
    zmq_version (&major, &minor, &patch);
    if (major < want_major
    || (major == want_major && minor < want_minor)) {
        std::cout << "Current 0MQ version is " << major << "." << minor << std::endl;
        std::cout << "Application needs at least " << want_major << "." << want_minor
              << " - cannot continue" << std::endl;
        exit (EXIT_FAILURE);
    }
}

//  Return current system clock as milliseconds
static int64_t
s_clock (void)
{
#if (defined (__WINDOWS__))
    SYSTEMTIME st;
    GetSystemTime (&st);
    return (int64_t) st.wSecond * 1000 + st.wMilliseconds;
#else
    struct timeval tv;
    gettimeofday (&tv, NULL);
    return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
#endif
}

//  Sleep for a number of milliseconds
static void
s_sleep (int msecs)
{
#if (defined (__WINDOWS__))
    Sleep (msecs);
#else
    struct timespec t;
    t.tv_sec = msecs / 1000;
    t.tv_nsec = (msecs % 1000) * 1000000;
    nanosleep (&t, NULL);
#endif
}

static void
s_console (const char *format, ...)
{
    time_t curtime = time (NULL);
    struct tm *loctime = localtime (&curtime);
    char *formatted = new char[20];
    strftime (formatted, 20, "%y-%m-%d %H:%M:%S ", loctime);
    printf ("%s", formatted);
    delete[] formatted;

    va_list argptr;
    va_start (argptr, format);
    vprintf (format, argptr);
    va_end (argptr);
    printf ("\n");
}

//  ---------------------------------------------------------------------
//  Signal handling
//
//  Call s_catch_signals() in your application at startup, and then exit
//  your main loop if s_interrupted is ever 1. Works especially well with
//  zmq_poll.

static int s_interrupted = 0;
static void s_signal_handler (int signal_value)
{
    s_interrupted = 1;
}

static void s_catch_signals ()
{
#if (!defined(__WINDOWS__))
    struct sigaction action;
    action.sa_handler = s_signal_handler;
    action.sa_flags = 0;
    sigemptyset (&action.sa_mask);
    sigaction (SIGINT, &action, NULL);
    sigaction (SIGTERM, &action, NULL);
#endif
}

#endif