UT/UT_NetMessage.h

Go to the documentation of this file.

/*
 * PROPRIETARY INFORMATION.  This software is proprietary to
 * Side Effects Software Inc., and is not to be reproduced,
 * transmitted, or disclosed in any way without written permission.
 *
 * Produced by:
 *      Jeff Lait
 *      Side Effects Software Inc
 *      123 Front Street West, Suite 1401
 *      Toronto, Ontario
 *      Canada   M5J 2M2
 *      416-504-9876
 *
 * NAME:        UT_NetMessage.h ( UT Library, C++)
 *
 * COMMENTS:
 *      Very similar to UT_NetStream, but rather than using threads
 *      to interleave packets successfully, it is designed
 *      around using ::select for the same purpose.
 *
 *      This allows the user to only have to worry about dealing with complete
 *      messages.
 *
 *      The key tools for end users is the UT_NetMessage and the 
 *      UT_NetExchange.
 */

#ifndef __UT_NetMessage__
#define __UT_NetMessage__

#include "UT_API.h"

#include "UT_NetSocket.h"
#include "UT_PtrArray.h"
#include "UT_Interrupt.h"

class UT_StopWatch;

///
/// UT_NetMessage bears similarity to a UT_NetPacket in that
/// it tries to abstract the necessitiy of retrying TCP/IP streams
/// until the message is completely received.  However, it is built
/// around the idea of having a datapump, pumpData, to handle this,
/// allowing one to interleave many messages in one thread using
/// select()
///
/// Write Protocol:
///     STATE_WRITE:
///             send 32bit length in network byte order
///             send myLength from myData
///     STATE_WAITACK:
///             receive 'j'
///             shutdown the connection
///
/// Read Protocol:
///     STATE_READ:
///             receive 32bit length in network byte order
///             set myLength and allocate myData
///             receive myLength into myData
///     STATE_SENDACK:
///             send 'j'
///     STATE_READFIN:
///             receive '' (ie, wait till safe to close)
///
class UT_API UT_NetMessage
{
public:
    UT_NetMessage();

    /// Frees the allocated data buffer.
    ~UT_NetMessage();

    /// What state we are in.
    /// Thanks to the vagaries of TCP/IP we need an application
    /// level ACK to verify the data made it across the wire
    /// before we do our shutdown.
    enum TransmitState
    {
        STATE_INVALID,

        STATE_READ,     /// Read message
        STATE_SENDACK,  /// Send ack
        STATE_READFIN,  /// Connection closed after ack sent.

        STATE_WRITE,    /// Send message
        STATE_WAITACK   /// Wait for connection ack.
    };

    /// Sends or receives data from the socket until the message
    /// is complete.  When it is complete or errored, returns true.  Returns
    /// false if you should put it back in the select loop.
    bool        pumpData();

    /// Trys to connect the socket, if already connected a no-op
    /// Returns false if still not connected, true if connected or error.
    /// failed is set to true if the connectionf ailed due to a 
    /// UT_CONNECT_FAILED as opposed to being delayed by a UT_WOULD_BLOCK
    bool        tryConnection(bool *failed=0);

    /// The socket we are attached to.
    UT_NetSocket *getSocket() const { return mySocket; }
    /// File descriptor of socket.
    int          getSocketFD() const { return mySocket->getSocket(); }

    /// Adds our file descriptor to the select set and update
    /// maxfd.
    void         addToFDSet(fd_set *set, int &maxfd) const;

    TransmitState       state() const { return myState; }
    bool         isErrored() const { return myErrorState; }
    void         flagErrored() { myErrorState = true; }

    const char  *data() const { return myData; }
    char        *data() { return myData; }
    int          length() const { return myLength; }

    /// Extracts data from the message, applying proper byteswaps.
    /// Offset is in bytes.  Does not need to be aligned.
    int64        extractInt64(int offset);
    int32        extractInt32(int offset);
    fpreal32     extractFloat32(int offset);
    fpreal64     extractFloat64(int offset);
    int16        extractInt16(int offset);
    int8         extractInt8(int offset);

    /// Inserts data into the message, applying proper byte swaps.
    /// Offset is in bytes.  Does not need to be aligned.
    void         overwriteInt64(int offset, int64 val);
    void         overwriteInt32(int offset, int32 val);
    void         overwriteFloat32(int offset, fpreal32 val);
    void         overwriteFloat64(int offset, fpreal64 val);
    void         overwriteInt16(int offset, int16 val);
    void         overwriteInt8(int offset, int8 val);

    /// Determine source/sink of this message.
    /// Because we shutdown the socket on completion of the
    /// protocol, we always gain ownership of the socket and
    /// will handle deleting it ourselves.
    void         setReadSocket(UT_NetSocket *socket);
    void         setWriteSocket(UT_NetSocket *socket);
    void         setWriteSocket(const char *addr, int port);

    /// The headersize, in bytes, is prepended as nulls to the message.
    /// The headersize *is* included in the myLength, the receiver
    /// doesn't know about the distinction between header and data!
    void         setWriteData(const char *data, int len) 
                 { setWriteDataWithHeader(0, data, len); }
    void         setWriteData(const char *data) 
                 { setWriteData(data, strlen(data)); }
    void         setWriteDataWithHeader(int headersize, const char *data, int len);
    void         setWriteDataWithHeader(int headersize, const char *data)
                 { setWriteDataWithHeader(headersize, data, strlen(data)); }

    /// Allocates a blank write buffer of the given size.
    /// It is assumed the caller will fill it out by writing to data()
    void         setWriteDataLength(int bufsize)
                 { setWriteDataWithHeader(bufsize, 0, 0); }

    static int64         totalBytesSent();
    static int64         totalBytesReceived();
    static void          clearByteCounters();

private:

    /// Tracking variables.
    static int64         ourTotalBytesSent;
    static int64         ourTotalBytesReceived;

    UT_NetSocket        *mySocket;
    int                  myLength;
    char                *myData;
    TransmitState        myState;
    bool                 myErrorState;

    // Internal state of where we are in terms of pumping...
    // -4 to -1 are for sending the length.
    int                  myDataPos;
    int                  myNetLength;   // htonl version of length
};

///
/// UT_NetMessagePump
/// 
/// Interleaves processing from possibly many different message
/// sources.
///
/// postMessage() can be used to queue messages to send after
/// their setWriteSocket and setWriteData has been invoked.
///
/// listenSocket() will invoke accept() and add to the completed()
/// list messages as they are done.
///
///
class UT_API UT_NetMessagePump
{
public:
                 UT_NetMessagePump();
    /// On destruction all messages on our internal lists will be destroyed.
    virtual     ~UT_NetMessagePump();

    /// Adds the message to our queue to process.
    /// We take ownership of the UT_NetMessage until it is removed
    /// from one of our lists.
    void         postMessage(UT_NetMessage *msg);

    /// List of all completed incoming messages. 
    ///
    /// To process a message from the completed list,
    /// UT_NetMessage *msg = pump.completed()(i);
    /// pump.completed()(i) = 0;
    /// ... use msg ...
    /// delete msg;
    UT_PtrArray<UT_NetMessage *> &completed() { return myCompleteList; }

    /// List of messages that have entered the error status.
    /// This includes failed outgoing messages as well as incomplete
    /// incoming messages.
    UT_PtrArray<UT_NetMessage *> &errored() { return myErrorList; }
    
    /// Starts listening to the given socket for messages
    /// We do not own this socket, caller should delete it (but not
    /// before stopping listening or deleting the pump!)
    /// Connections accepted from this socket will be read
    /// via UT_NetMessage's protocol.
    void         listenSocket(UT_NetSocket *socket);
    /// Stops listening from a given socket.
    void         stopListening(UT_NetSocket *socket);

    /// Processes active messages and listens for incoming messages.
    /// Returns true if there is the *potential* of more work - if
    /// listening to a socket this is always true!
    bool         pumpData(int timeoutms);

    /// The number of messages still pending.
    int          pendingMessages() const { return myWorkList.entries(); }
protected:
    UT_PtrArray<UT_NetMessage *>        myWorkList;
    UT_PtrArray<UT_NetMessage *>        myErrorList;
    UT_PtrArray<UT_NetMessage *>        myCompleteList;
    UT_PtrArray<UT_NetSocket *>         myServerList;
};

class ut_NetPeerData
{
public:
    ut_NetPeerData() {}
    virtual ~ut_NetPeerData() {}
    ut_NetPeerData(const ut_NetPeerData &src)
    {
        *this = src;
    }

    const ut_NetPeerData &operator=(const ut_NetPeerData &src)
    {
        myAddress.harden(src.myAddress);
        myPort = src.myPort;
        myPeer = src.myPeer;
        return *this;
    }

    UT_String           myAddress;
    int                 myPort;
    int                 myPeer;
};

///
/// UT_NetExchange
///
/// Uses a tracker to exchange data between peers.
/// The tracker handles peer discovery and synchronization.  A unique
/// jobname is used to allow multiple net exchanges to use the same
/// tracker.
///
/// EXTERNAL PROTOCOL:
///
/// UT_NetMessages accepted or returned from this have an 8 byte
/// header.  The fields are filled out for you on a sendData()
/// and can be queried to determine the message source when
/// pulling completed messages from the result list.
///
/// The first byte is the message type.  If it is 'p', it is
/// peer data.   Users of UT_NetExchange should only see these
/// messages as the other are internal tracker messages.
///
/// extractInt16(4): Destination peer #
/// extractInt16(6): Source peer #
///
/// Anything 8 and after is user data.
///
/// INTERNAL PROTOCOL:
///
///     From the tracker:
///
/// Message of type 'b' is a barrier value.  The 32 bit network
/// byte order barrier value starts at offset 1.
///
/// Message type of 'c' is a peer list.  A space delimited list
/// of address port peer triples is present, one for each peer.
///
/// Message type of 'd' is a done flag.  It informs that all
/// the peers have reported done to the tracker.
///
/// Message type of 'e' is an error.  One of the peers reported
/// an error to the tracker that got broadcast back.
///
///     To the tracker:
///
/// Messages to the tracker all have the format
/// command port peer npeer jobname
///
/// port is the listen port used by this peer.
/// peer is the peer number
/// npeer is the number of peers
/// jobname is the name of the current job
///
/// command is one of:
/// acquire: A peer is logging on to this jobname.
/// done:    A peer has all the data it cares about from the netexchange.
/// error:   A peer has encountered a fatal network error and wants a clean
///          shutdown.
/// barrierset: Establish a barrier, the current value is peer.
/// barrierwait: Will recevie a done message when the peer value is reached or
///          exceeded by the named barrier.
///
class UT_API UT_NetExchange
{
public:
                 UT_NetExchange(const char *trackeraddr, int trackerport,
                            int peer, int npeer, const char *jobname);
    virtual     ~UT_NetExchange();

    /// Posts the data to be sent to the specific destination
    /// machine.
    /// The data will be copied into an internal buffer.
    /// Note that destpeer could be this, in which case we'll just
    /// have effected a very slow copy.
    void         sendData(int destpeer, const char *data, int len);

    /// Gains ownership of the message.  The message must have
    /// an 8 byte header reserved for the net exchange protocol.
    /// The rest of the data in the net message is unaffected.
    void         sendData(int destpeer, UT_NetMessage *msg);

    /// Flags this peer as done.  You should still process the pump until
    /// interrupted or completed, however, as the other peers
    /// may not yet be done.   Likewise, you may still continue
    /// to get new requests from other peers.
    void         sendDone();

    /// Returns true until the tracker notifies us that all of the
    /// peers have invoked sendDone() or an error occurs.
    bool         pumpData(int timeoutms);

    /// Pumps the message pump until we have received the expected
    /// number of messages and the tracker reports us as done.  The
    /// messages in the completed array must be deleted by the caller.
    /// UT_Interrupt is used to interrupt this.
    /// true if everything finished properly, false if there was
    /// an interrupt or error.
    bool         receiveDataLoop(UT_PtrArray<UT_NetMessage *> &completed,
                                int expectedmessages, int timeoutms = 100);

    /// Pumps the message pump until we have received the expected
    /// number of data messages and the tracker reports us as done.  The
    /// messages in the completed array must be deleted by the caller.
    /// UT_Interrupt is used to interrupt this.
    /// true if everything finished properly, false if there was
    /// an interrupt or error.
    ///
    /// The callback processes any message with a 'r' as the 8th byte.
    /// Final messages (which do not get further processed) should have 
    /// 'd' as the 8th byte - they will
    /// be added to the completed list and count to the expectedmessages.
    /// It should be a function object taking two parameters,
    /// UT_NetExchange * and UT_NetMessage *.
    ///
    /// UT_Interrupt is used to interrupt this.
    /// true if everything finished properly, false if there was
    /// an interrupt or error.
    template <class CallbackType>
    bool        processDataLoop(UT_PtrArray<UT_NetMessage *> &complist,
                                int expectedmessages,
                                CallbackType callback,
                                int timeoutms = 100)
    {
        // Because we append to complist, allow for the case where
        // we are accumulating...
        int             goal = complist.entries() + expectedmessages;
        bool            done = false;
        int             i;
        UT_NetMessage   *msg;
        UT_Interrupt    *boss = UTgetInterrupt();

        while (1)
        {
            if (boss->opInterrupt())
                return false;

            if (!pumpData(timeoutms))
                break;

            for (i = 0; i < completed().entries(); i++)
            {
                msg = completed()(i);
                completed()(i) = 0;

                if (msg->extractInt8(8) == 'r')
                {
                    callback(this, msg);
                    delete msg;
                }
                else
                    complist.append(msg);
            }

            // See if we got our done package.
            if (!done && (complist.entries() >= goal))
            {
                sendDone();
                done = true;
            }
        }

        return !isErrored();
    }

    /// The net exchange goes into an error state when the connection
    /// the tracker sends an error message. 
    /// peer-to-peer errors are dealt with by sending the tracker
    /// an error message which is supposed to broadcast the error
    /// back to all the peers.  Error recovery can thus be effected
    /// provided it is the tracker that stays up.
    bool         isErrored() const { return myErrorFromTracker; }

    /// Same semantics as UT_NetMessagePump.  Stores all the messages
    /// that have successfully arrived.
    /// This first 8 bytes of theses messages is the header with
    /// which you can extract the source peer of the message.
    UT_PtrArray<UT_NetMessage *>        &completed() { return myCompleted; }

    /// Returns if we have received a peer list from the tracker.
    bool         gotPeerList() const { return myGotPeerList; }

protected:
    ut_NetPeerData              *findPeer(int peer);
    void                        processTrackerMessage(UT_NetMessage *msg);
    /// Assembles a message to the tracker.
    void                        sendTrackerMessage(const char *msg);


    UT_NetMessagePump           myPump;
    int                         myNPeer, myPeer;
    UT_NetSocket                *myToSelf;

    UT_String                   myTrackerAddr;
    int                         myTrackerPort;

    bool                        myGotPeerList;
    bool                        myError;
    bool                        myErrorFromTracker;
    UT_RefArray<ut_NetPeerData> myPeerList;

    UT_PtrArray<UT_NetMessage *>        myWaitingForPeers;
    UT_PtrArray<UT_NetMessage *>        myCompleted;

    UT_String                   myJobName;

    /// We track our elapsed time for printing status
    UT_StopWatch                *myTimer;
    int                          myTimerDelay, myHeartbeat;
    fpreal64                     myLastTimerVal;
};

///
/// UT_NetBarrier
///
/// Very similar to UT_NetExchange, using the same tracker.  However,
/// it is meant for creating producer/consumer queues where the data
/// is transmitted OOB (ie, on shared disk resource)  The producer
/// can call setValue() to indicate how far it has completed.  The consumer
/// can invoke waitValue() to idle until the setValue is equal or
/// greater than the waitValue.
///
/// These calls will all block until success, error, or interrupt.
///
class UT_API UT_NetBarrier
{
public:
                 UT_NetBarrier(const char *trackeraddr, int trackerport,
                          const char *jobname);
    virtual     ~UT_NetBarrier();

    /// Tells the tracker to update the barrier value to the given
    /// value.  Does not listen for a tracker response.
    void         setValue(int val);

    /// Requests the tracker to alert us when the given value is
    /// reached on the barrier.
    void         waitValue(int val);

    /// Returns current value of the barrier and then increments
    /// the barrier's value.
    /// Blocks until the tracker responds.
    int          incrementValue(int defaultval);

protected:
    /// Creates and posts a message for the tracker.
    void         sendTrackerMessage(const char *msg, int value);

    /// A simplified method for running the message pump until
    /// the system errors or is interrupted.
    void         blockUntilComplete();

    /// Returns true until the barrier has been resolved or error
    /// occurs
    bool         pumpData(int timeoutms);

    UT_NetMessagePump            myPump;
    UT_NetSocket                *myToSelf;
    bool                         myError;

    int                          myBarrierValue;

    UT_String                    myTrackerAddr;
    int                          myTrackerPort;
    UT_String                    myJobName;
};


#endif

---