soa/service/socket_per_thread.h

/* socket_per_thread.h                                             -*- C++ -*-
   Jeremy Barnes, 14 April 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

   ZeroMQ Socket per thread.
*/

#ifndef __zmq__socket_per_thread_h__
#define __zmq__socket_per_thread_h__


#include "zmq.hpp"
#include <boost/thread/tss.hpp>
#include <boost/noncopyable.hpp>
#include <boost/thread/locks.hpp>
#include "jml/compiler/compiler.h"
#include "jml/arch/spinlock.h"
#include "jml/arch/exception.h"
#include <set>


namespace Datacratic {


/*****************************************************************************/
/* SOCKET PER THREAD                                                         */
/*****************************************************************************/

struct SocketPerThread : boost::noncopyable {

    SocketPerThread();

    SocketPerThread(zmq::context_t & context,
                    int type,
                    const std::string & uri,
                    bool allowForceClose = true);

    ~SocketPerThread();

    void init(zmq::context_t & context,
              int type,
              const std::string & uri,
              bool allowForceClose = true);

    void shutdown();

    zmq::context_t * context;   
    int type;                   
    std::string uri;            
    bool allowForceClose;       
    mutable int numOpen;        

    inline zmq::socket_t & operator () () const
    {
        if (state != READY)
            throw ML::Exception("socket not ready: %d", state);

        if (JML_UNLIKELY(!entries.get())) {
            initForThisThread();
        }

        return *entries->sock;
    }

    void initForThisThread() const;
    
    void cleanupThisThread();

    static void cleanupAllForThread();
    
private:
    enum {
        NOTINITIALIZED = 12321,
        READY = 349244,
        FINISHED = 293845
    };
    int state;

    struct Entry {
        zmq::socket_t * sock;
        SocketPerThread * owner;
    };

    std::set<Entry *> allThreads;

    typedef ML::Spinlock Lock;
    typedef boost::unique_lock<Lock> Guard;
    mutable Lock allThreadsLock;

    void removeThreadEntry(Entry * entry);

    void addThreadEntry(Entry * entry);

    static void onFreeEntry(Entry * entry);
    
    mutable boost::thread_specific_ptr<Entry> entries;

    static boost::thread_specific_ptr<std::set<SocketPerThread *> > allForThread;
};

} // namespace Datacratic



#endif /* __zmq__socket_per_thread_h__ */