src/acedia_util.cpp

/*                                                                            *\
**                          Copyright (C) 2009-2010                           **
**       Dominik Charousset < Dominik.Charousset (at) haw-hamburg.de >        **
**                                                                            **
**            MMM       MMMMMM   MMMMMMMMM MMMMMMM   MMM       MMM            **
**           MMMMM     MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM      MMMMM           **
**          MMM MMM   MMM        MMMM      MMM   MMM MMM     MMM MMM          **
**         MMM   MMM  MMM        MMMMMMMMM MMM   MMM MMM    MMM   MMM         **
**        MMMMMMMMMMM MMMM   MMM MMMM      MMM   MMM MMM   MMMMMMMMMMM        **
**       MMM       MMM MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM  MMM       MMM       **
**      MMM         MMM MMMMMM   MMMMMMMMM MMMMMMM   MMM MMM         MMM      **
**                                                                            **
**                                                                            **
** This file is part of the acedia library.                                   **
**                                                                            **
** This library is free software: you can redistribute it and/or modify       **
** it under the terms of the GNU Lesser General Public License as published   **
** by the Free Software Foundation, either version 3 of the License, or       **
** (at your option) any later version.                                        **
**                                                                            **
** This library 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               **
** GNU Lesser General Public License for more details.                        **
**                                                                            **
** You should have received a copy of the GNU Lesser General Public License   **
** along with acedia. If not, see <http://www.gnu.org/licenses/>.             **
\*                                                                            */

#define ACEDIA_REDUCED_ANNOUNCE

#include "acedia_util.hpp"

#include "exceptions.hpp"
#include <boost/thread.hpp>

/*
#include <iostream>
using std::cout;
using std::endl;
using std::cerr;
#define TRACE_CALL(funcArgs) cout << __PRETTY_FUNCTION__ << " -> " << funcArgs << endl;
*/

namespace acedia
{
    namespace details
    {
        static const ::ssize_t ui32Size = (::ssize_t) sizeof(boost::uint32_t);
        static const int INIT_BUF_SIZE = 512;

        struct SocketIOStreamPrivate
        {

            NativeSocketType m_socket;
            char *m_buffer;
            boost::uint32_t m_bufferSize;

            // receive data until @p numBytes are stored into the buffer
            void readAll(SocketRecvPtr where, boost::uint32_t numBytes, boost::uint32_t offset = 0)
            {
                // keep compiler happy
                SocketRecvPtr offsetWhere = (SocketRecvPtr) (((char*) where) + offset);
                ::ssize_t x = ::recv(m_socket, offsetWhere,
                                     numBytes, 0);
                if (x < 0)
                {
                    ACEDIA_THROW(IOException, "Failed to read from socket");
                }
                else if (x == 0)
                {
                    ACEDIA_THROW(IOException, "Cannot read from a closed socket");
                }
                else if (x < (::ssize_t) numBytes)
                {
                    readAll(where, numBytes - x, offset + x);
                }
                // else: x == numBytes; all fine
            }

            void write(SocketSendPtr buf, ::size_t len)
            {
                ::ssize_t x = send(m_socket, buf, len, 0);
                if (x != (::ssize_t) len)
                {
                    if (x < 0)
                    {
/*
                        switch (errno)
                        {
                         case EBADF: sockerr = "EBADF" << endl; break;
                         case ECONNABORTED: cout << "ECONNABORTED" << endl; break;
                         case ECONNRESET: cout << "ECONNRESET" << endl; break;
                         case EFAULT: cout << "EFAULT" << endl; break;
                         case EHOSTUNREACH: cout << "EHOSTUNREACH" << endl; break;
                         case EINVAL: cout << "EINVAL" << endl; break;
                         case EINTR: cout << "EINTR" << endl; break;
                         case ENETDOWN: cout << "ENETDOWN" << endl; break;
                         case ENOBUFS: cout << "ENOBUFS" << endl; break;
                         case ENOTCONN: cout << "ENOTCONN" << endl; break;
                         case ENOTSOCK: cout << "ENOTSOCK" << endl; break;
                         case EMSGSIZE: cout << "EMSGSIZE" << endl; break;
                         case EOPNOTSUPP: cout << "EOPNOTSUPP" << endl; break;
                         case ESHUTDOWN: cout << "ESHUTDOWN" << endl; break;
                         case ETIMEDOUT: cout << "ETIMEDOUT" << endl; break;
                         case EWOULDBLOCK: cout << "EWOULDBLOCK" << endl; break;
                        }
*/
                        ACEDIA_THROW(IOException, "Failed to write to socket");
                    }
                    else if (x == 0)
                        ACEDIA_THROW(IOException, "send() returned 0 (closed?)");
                    else
                        ACEDIA_THROW(IOException, "send() wrote too less bytes");
                }
            }

            SocketIOStreamPrivate(NativeSocketType socketDescriptor) :
                    m_socket(socketDescriptor), m_bufferSize(INIT_BUF_SIZE)
            {
                m_buffer = new char[INIT_BUF_SIZE];
            }

            ~SocketIOStreamPrivate()
            {
                delete m_buffer;
            }

            void resizeBuffer(boost::uint32_t newMinSize)
            {
                // do never allow to allocate more than 1MB
                if (newMinSize >= (1024*1024))
                    ACEDIA_THROW(IOException, "Too big message (>= 1MB)");
                else
                {
                    // always allocate 100 bytes more than requestet
                    m_bufferSize = newMinSize + 100;
                    delete m_buffer;
                    m_buffer = new char[m_bufferSize];
                }
            }

            void readToBuffer(boost::uint32_t numBytes)
            {
                if (numBytes + 1 >= m_bufferSize)
                {
                    resizeBuffer(numBytes + 1);
                }
                readAll((SocketRecvPtr) m_buffer, numBytes);
            }

            const char* buffer()
            {
                return m_buffer;
            }

         private:

            SocketIOStreamPrivate(const SocketIOStreamPrivate& other);

        };
    }

    namespace util
    {
        String eraseWhitespaces(const char *input)
        {
            String result;
            char c;
            for (unsigned int i = 0; input[i] != 0; ++i)
            {
                c = input[i];
                if (c != ' ' && c != '\t') result += c;
            }
            return result;
        }


        /**********************************************************************
         *                             spinlocks                              *
         **********************************************************************/

        void Spinlock::lock() throw()
        {
            ACEDIA_MEMORY_BARRIER();
            for (;;)
            {
                if (atomic::compareAndSwap(&m_value, 0, 1))
                {
                    ACEDIA_MEMORY_BARRIER();
                    return;
                }
            }
        }

        SpinlockGuard::SpinlockGuard(Spinlock &sl) : slock(sl)
        {
            slock.lock();
        }

        SpinlockGuard::~SpinlockGuard()
        {
            slock.unlock();
        }

        void RWSpinlock::sharedLock()
        {
            for (;;)
            {
                boost::int32_t v = m_value;
                if (v >= 0 && atomic::compareAndSwap(&m_value, v, v + 1))
                {
                    // forbid hardware (and compiler) out-of-order execution
                    // around the lock operation
                    ACEDIA_MEMORY_BARRIER();
                    return;
                }
                // this lock is locked for write, wait a bit
//              else boost::this_thread::yield();
            }
        }

        void RWSpinlock::exclusiveLock()
        {
            for (;;)
            {
                boost::int32_t v = m_value;
                if (v >= 0 && atomic::compareAndSwap(&m_value, v,
                                                     (Int32Minimum+v)))
                {
                    ACEDIA_MEMORY_BARRIER();
                    // wait if this lock is locked by one or more readers
                    while (m_value != Int32Minimum)
                    {
                        boost::this_thread::yield();
                    }
                    ACEDIA_MEMORY_BARRIER();
                    return;
                }
                else boost::this_thread::yield();
            }
        }

        SharedLockGuard::SharedLockGuard(RWSpinlock &l) : rwlock(l)
        {
            l.sharedLock();
        }

        SharedLockGuard::~SharedLockGuard() { rwlock.sharedUnlock(); }

        ExclusiveLockGuard::ExclusiveLockGuard(RWSpinlock &l) : rwlock(l)
        {
            l.exclusiveLock();
        }

        ExclusiveLockGuard::~ExclusiveLockGuard() { rwlock.exclusiveUnlock(); }


        /**********************************************************************
         *                             Semaphore                              *
         **********************************************************************/

        void Semaphore::release(boost::int32_t n)
        {
            assert(n > 0);
            (void) atomic::addAndFetch(&m_value, n);
            ACEDIA_MEMORY_BARRIER();
            m_condition.notify_all();
        }

        void Semaphore::acquire(boost::int32_t n)
        {
            assert(n > 0);
            boost::unique_lock<boost::mutex> lock(m_mutex);
            while (available() < n) m_condition.wait(lock);
            (void) atomic::addAndFetch(&m_value, -n);
        }

        bool Semaphore::tryAcquire(boost::int32_t n)
        {
            assert(n > 0);
            boost::unique_lock<boost::mutex> lock(m_mutex);
            if (available() >= n)
            {
                (void) atomic::addAndFetch(&m_value, -n);
                return true;
            }
            return false;
        }

        bool Semaphore::tryAcquire(boost::int32_t n, boost::uint16_t msTimeout)
        {
            assert(n > 0);
            boost::unique_lock<boost::mutex> lock(m_mutex);
            boost::system_time timeout = boost::get_system_time();
            timeout += boost::posix_time::milliseconds(msTimeout);
            while (available() < n)
            {
                if (!m_condition.timed_wait(lock, timeout))
                {
                    return false;
                }
            }
            (void) atomic::addAndFetch(&m_value, -n);
            return true;
        }


        /**********************************************************************
         *                utility class to simplify network IO                *
         **********************************************************************/

        boost::int64_t littleEndian64BitFun(boost::int64_t value)
        {
            boost::uint8_t* from = (reinterpret_cast<boost::uint8_t*>(&value));
            boost::uint64_t tmp = 0;
            boost::uint8_t* to = (reinterpret_cast<boost::uint8_t*>(&tmp));
            register boost::int_fast32_t j = 7; // last valid index on 8 bytes
            for (register boost::int_fast32_t i = 0; i < 8; ++i)
            {
                to[i] = from[j--];
            }
            return tmp;
        }

        boost::int64_t bigEndian64BitFun(boost::int64_t value)
        {
            return value;
        }

        EndianConverter::EndianConverter()
        {
            // check endian type at runtime
            boost::uint16_t word = 0x0001;
            boost::uint8_t* bytes = reinterpret_cast<boost::uint8_t*>(&word);
            if (bytes[0])
            {
                // we are running on a little endian machine
                // (word is 0x0100 in memory)
                m_int64Fun = littleEndian64BitFun;
            }
            else m_int64Fun = bigEndian64BitFun;
        }

        AbstractEndianAwareStream::~AbstractEndianAwareStream()     { }

        EndianAwareOutputStream& operator<<(EndianAwareOutputStream& s,
                                      const std::string& str)
        {
            boost::uint32_t len = str.length();
            s << len;
            s.write(str.c_str(), len);
            return s;
        }

        template<class StringClass>
        void readStringFromStream(EndianAwareInputStream& s, StringClass& str)
        {
            boost::uint32_t len;
            s >> len;
            // use the stack on small strings instead of heap allocation
            if (len < 128)
            {
                char stack_buf[128];
                s.read(stack_buf, len);
                stack_buf[len] = 0;
                str = stack_buf;
            }
            else
            {
                char *heap_buf = new char[len+1];
                s.read(heap_buf, len);
                heap_buf[len] = 0;
                str = heap_buf;
                delete heap_buf;
            }
        }

        void operator>>(EndianAwareInputStream& s, acedia::String& str)
        {
            readStringFromStream(s, str);
        }

        void operator>>(EndianAwareInputStream& s, std::string& str)
        {
            readStringFromStream(s, str);
        }

        SocketIOStream::SocketIOStream(NativeSocketType socketDescriptor) :
                d(new details::SocketIOStreamPrivate(socketDescriptor))
        {
        }

        SocketIOStream::SocketIOStream(const SocketIOStream& other) :
                EndianAwareStream(),
                d(new details::SocketIOStreamPrivate((other.d)->m_socket))
        {
        }

        SocketIOStream::~SocketIOStream()
        {
            delete d;
        }

        void SocketIOStream::close()
        {
            closesocket(d->m_socket);
        }

        void SocketIOStream::readStringToBuffer()
        {
            boost::uint32_t strLen;
            readInt(strLen);
            if (strLen > 0)
            {
                d->readToBuffer(strLen);
            }
        }

        const char* SocketIOStream::buffer() { return d->buffer(); }

        void SocketIOStream::write(const char* what, ::size_t size)
        {
            d->write((SocketSendPtr) what, size);
        }

        void SocketIOStream::read(char* what, ::size_t size)
        {
            d->readAll((SocketRecvPtr) what, size);
        }

        void SocketIOStream::readToBuffer(::size_t size)
        {
            d->readAll((SocketRecvPtr) d->m_buffer, size);
        }

    }
}

---