libcore/plugins/tcpsst/ASIOSocketWrapper.hpp

Go to the documentation of this file.

/*  Sirikata Network Utilities
 *  ASIOSocketWrapper.hpp
 *
 *  Copyright (c) 2009, Daniel Reiter Horn
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are
 *  met:
 *  * Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *  * Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *  * Neither the name of Sirikata nor the names of its contributors may
 *    be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
#include <sirikata/core/util/UUID.hpp>
#include <sirikata/core/queue/SizedThreadSafeQueue.hpp>
#include <sirikata/core/util/Time.hpp>
#include <sirikata/core/util/EWA.hpp>
#include <sirikata/core/network/IOStrand.hpp>

#define SEND_LATENCY_EWA_ALPHA .10f

namespace Sirikata { namespace Network {
class ASIOSocketWrapper;
class ASIOReadBuffer;

void triggerMultiplexedConnectionError(MultiplexedSocket*,ASIOSocketWrapper*,const boost::system::error_code &error);
void ASIOLogBuffer(void * pointerkey, const char extension[16], const uint8* buffer, size_t buffersize);
class ASIOSocketWrapper {
    IOStrand* mStrand;
    TCPSocket*mSocket;

    ASIOReadBuffer *mReadBuffer;
    AtomicValue<uint32> mSendingStatus;

    struct TimestampedChunk {
        TimestampedChunk()
         : chunk(NULL), time(Time::null())
        {}

        TimestampedChunk(Chunk* _c)
         : chunk(_c), time(Time::local())
        {}

        uint32 size() const {
            return chunk->size();
        }

        Duration sinceCreation() const {
            return Time::local() - time;
        }

        Chunk* chunk;
        Time time;
    };

    SizedThreadSafeQueue<TimestampedChunk>mSendQueue;
    enum {
        ASYNCHRONOUS_SEND_FLAG=(1<<29),
        QUEUE_CHECK_FLAG=(1<<30),

    };
    EWA<Duration> mAverageSendLatency;

    std::vector<Stream::StreamID> mPausedSendStreams;
    std::deque<TimestampedChunk> mToSend;
    std::tr1::weak_ptr<MultiplexedSocket>mParent;
    std::tr1::shared_ptr<MultiplexedSocket>mOutstandingDataParent;
    void finishedSendingChunk(const TimestampedChunk& tc);


    typedef boost::system::error_code ErrorCode;
    std::tr1::function<void(const ErrorCode &error, std::size_t bytes_sent)>mSendManyDequeItems;
    void finishAsyncSend(const MultiplexedSocketPtr&parentMultiSocket);

    void sendManyDequeItems(const std::tr1::weak_ptr<MultiplexedSocket>&parentMultiSocket, const ErrorCode &error, std::size_t bytes_sent);

    void sendToWire(const MultiplexedSocketPtr&parentMultiSocket, TimestampedChunk toSend);

    void sendToWire(const MultiplexedSocketPtr&parentMultiSocket, std::deque<TimestampedChunk>&const_toSend);

    void retryQueuedSend(const MultiplexedSocketPtr&parentMultiSocket, uint32 current_status);

public:

    ASIOSocketWrapper(TCPSocket* socket,uint32 queuedBufferSize,uint32 sendBufferSize, const MultiplexedSocketPtr&parent)
     : mSocket(socket),
       mReadBuffer(NULL),
       mSendingStatus(0),
       mSendQueue(SizedResourceMonitor(queuedBufferSize)),
       mAverageSendLatency(SEND_LATENCY_EWA_ALPHA),
       mParent(parent)
    {
        //mPacketLogger.reserve(268435456);
        bindFunctions(parent);
    }

    ASIOSocketWrapper(const ASIOSocketWrapper& socket)
     : mSocket(socket.mSocket),
       mReadBuffer(NULL),
       mSendingStatus(0),
       mSendQueue(socket.getResourceMonitor()),
       mAverageSendLatency(SEND_LATENCY_EWA_ALPHA)
    {
        MultiplexedSocketPtr parent(socket.mParent.lock());
        mParent=parent;
        bindFunctions(parent);
    }

    ASIOSocketWrapper(uint32 queuedBufferSize,uint32 sendBufferSize,const MultiplexedSocketPtr&parent)
     : mSocket(NULL),
       mReadBuffer(NULL),
       mSendingStatus(0),
       mSendQueue(SizedResourceMonitor(queuedBufferSize)),
       mAverageSendLatency(SEND_LATENCY_EWA_ALPHA),
       mParent(parent)
    {
        bindFunctions(parent);
    }
    void bindFunctions(const MultiplexedSocketPtr&parent);
    ~ASIOSocketWrapper() {
        if (mToSend.size()!=0) {
            SILOG(tcpsst,error,"Outstanding data left on socket that is being deleted. mOutstandingDataParent is "<<(size_t)mOutstandingDataParent.get());
        }
        assert(mToSend.size()==0);
    }

    ASIOSocketWrapper&operator=(const ASIOSocketWrapper& socket){
        mSocket=socket.mSocket;
        return *this;
    }
    const SizedResourceMonitor&getResourceMonitor()const{return mSendQueue.getResourceMonitor();}

    void clearReadBuffer(){
        mReadBuffer=NULL;
    }
    ASIOReadBuffer*getReadBuffer()const{
        return mReadBuffer;
    }
    ASIOReadBuffer* setReadBuffer(ASIOReadBuffer*arb) {
        return mReadBuffer=arb;
    }
    TCPSocket&getSocket() {return *mSocket;}

    const TCPSocket&getSocket()const {return *mSocket;}

    void shutdownAndClose();

    void createSocket(unsigned int kernelSendBufferSize, unsigned int kernelReceiveBufferSize);

    void destroySocket();
    bool rawSend(const MultiplexedSocketPtr&parentMultiSocket, Chunk * chunk, bool force);
    bool canSend(size_t dataSize)const;
    static Chunk*constructControlPacket(const MultiplexedSocketPtr&parentMultiSocket, TCPStream::TCPStreamControlCodes code,const Stream::StreamID&sid);
    static Chunk* constructPing(const MultiplexedSocketPtr&parentMultiSocket, MemoryReference data, bool isPong);
    void sendControlPacket(const MultiplexedSocketPtr&parentMultiSocket, TCPStream::TCPStreamControlCodes code,const Stream::StreamID&sid) {
        rawSend(parentMultiSocket,constructControlPacket(parentMultiSocket, code,sid),true);
    }
    void sendProtocolHeader(const MultiplexedSocketPtr&parentMultiSocket, const Address& address, const UUID&value, unsigned int numConnections);
    void sendServerProtocolHeader(const MultiplexedSocketPtr& thus, const std::string&origin, const std::string&host, const std::string&port, const std::string&resource_name, const std::string&subprotocol, bool doSendSubProtocol, const std::string& response);

    void ioReactorThreadPauseStream(const MultiplexedSocketPtr&parentMultiSocket, Stream::StreamID sid);
    void unpauseSendStreams(const MultiplexedSocketPtr&parentMultiSocket);
    Address getRemoteEndpoint()const;
    Address getLocalEndpoint()const;

    // -- Statistics
    Duration averageSendLatency() const;
    Duration averageReceiveLatency() const;
    //converts 3 arrays into a contiguous array of base64 numbers, delimited with a '\0' at the end.
    static Chunk* toBase64ZeroDelim(const MemoryReference&a, const MemoryReference&b, const MemoryReference&c, const MemoryReference *bytesToPrependUnencoded=NULL);
    static UUID massageUUID(const UUID&);
class CheckCRLF {
    const Array<uint8,TCPStream::MaxWebSocketHeaderSize> *mArray;
    unsigned int mLastTransferred;
public:
    CheckCRLF(const Array<uint8,TCPStream::MaxWebSocketHeaderSize>*array) {
        mArray=array;
        mLastTransferred=0;

    }
    size_t operator() (const ErrorCode&error, size_t bytes_transferred);
};


};



} }