From 530014a3cce53e86dce8917e98a4e86d02f176aa Mon Sep 17 00:00:00 2001 From: Mike Buland Date: Thu, 26 Apr 2007 15:06:49 +0000 Subject: Merged Ito and put it in the BU namespace. I should probably clean up the formatting on the comments, some of the lines wrap, but I'm not too worried about it right now. I also fixed up the doxygen config and build.conf files so that everything is building nice and smooth now. --- src/exceptions.cpp | 1 + src/exceptions.h | 1 + src/ito.cpp | 40 +++++++++ src/ito.h | 98 ++++++++++++++++++++ src/itoatom.h | 56 ++++++++++++ src/itocondition.cpp | 42 +++++++++ src/itocondition.h | 81 +++++++++++++++++ src/itomutex.cpp | 27 ++++++ src/itomutex.h | 61 +++++++++++++ src/itoqueue.h | 231 ++++++++++++++++++++++++++++++++++++++++++++++++ src/main.dox | 14 +++ src/serversocket.cpp | 148 +++++++++++++++++++++++++++++++ src/serversocket.h | 31 +++++++ src/socket.cpp | 16 +--- src/socket.h | 4 +- src/tests/itoqueue1.cpp | 110 +++++++++++++++++++++++ src/tests/itoqueue2.cpp | 83 +++++++++++++++++ 17 files changed, 1031 insertions(+), 13 deletions(-) create mode 100644 src/ito.cpp create mode 100644 src/ito.h create mode 100644 src/itoatom.h create mode 100644 src/itocondition.cpp create mode 100644 src/itocondition.h create mode 100644 src/itomutex.cpp create mode 100644 src/itomutex.h create mode 100644 src/itoqueue.h create mode 100644 src/main.dox create mode 100644 src/serversocket.cpp create mode 100644 src/serversocket.h create mode 100644 src/tests/itoqueue1.cpp create mode 100644 src/tests/itoqueue2.cpp (limited to 'src') diff --git a/src/exceptions.cpp b/src/exceptions.cpp index a512105..d9f4e70 100644 --- a/src/exceptions.cpp +++ b/src/exceptions.cpp @@ -5,6 +5,7 @@ namespace Bu { subExceptionDef( XmlException ) subExceptionDef( FileException ) + subExceptionDef( SocketException ) subExceptionDef( ConnectionException ) subExceptionDef( PluginException ) subExceptionDef( UnsupportedException ) diff --git a/src/exceptions.h b/src/exceptions.h index 3efa19f..f146b73 100644 --- a/src/exceptions.h +++ b/src/exceptions.h @@ -8,6 +8,7 @@ namespace Bu { subExceptionDecl( XmlException ) subExceptionDecl( FileException ) + subExceptionDecl( SocketException ) subExceptionDecl( ConnectionException ) subExceptionDecl( PluginException ) subExceptionDecl( UnsupportedException ) diff --git a/src/ito.cpp b/src/ito.cpp new file mode 100644 index 0000000..001ca06 --- /dev/null +++ b/src/ito.cpp @@ -0,0 +1,40 @@ +#include "ito.h" + +Bu::Ito::Ito() +{ +} + +Bu::Ito::~Ito() +{ +} + +bool Bu::Ito::start() +{ + nHandle = pthread_create( &ptHandle, NULL, threadRunner, this ); + + return true; +} + +bool Bu::Ito::stop() +{ + pthread_exit( &ptHandle ); + + return true; +} + +void *Bu::Ito::threadRunner( void *pThread ) +{ + return ((Ito *)pThread)->run(); +} + +bool Bu::Ito::join() +{ + pthread_join( ptHandle, NULL ); + return true; +} + +void Bu::Ito::yield() +{ + pthread_yield(); +} + diff --git a/src/ito.h b/src/ito.h new file mode 100644 index 0000000..01253f5 --- /dev/null +++ b/src/ito.h @@ -0,0 +1,98 @@ +#ifndef ITO_H +#define ITO_H + +#include + +namespace Bu +{ + /** + * Simple thread class. This wraps the basic pthread (posix threads) system in + * an object oriented sort of way. It allows you to create a class with + * standard member variables and callable functions that can be run in it's own + * thread, one per class instance. + *@author Mike Buland + */ + class Ito + { + public: + /** + * Construct an Ito thread. + */ + Ito(); + + /** + * Destroy an Ito thread. + */ + virtual ~Ito(); + + /** + * Begin thread execution. This will call the overridden run function, + * which will simply execute in it's own thread until the function exits, + * the thread is killed, or the thread is cancelled (optionally). The + * thread started in this manner has access to all of it's class variables, + * but be sure to protect possible multiple-access with ItoMutex objects. + *@returns True if starting the thread was successful. False if something + * went wrong and the thread has not started. + */ + bool start(); + + /** + * Forcibly kill a thread. This is not generally considered a good thing to + * do, but in those rare cases you need it, it's invaluable. The problem + * with stopping (or killing) a thread is that it stops it the moment you + * call stop, no matter what it's doing. The object oriented approach to + * this will help clean up any class variables that were used, but anything + * not managed as a member variable will probably create a memory leak type + * of situation. Instead of stop, consider using cancel, which can be + * handled by the running thread in a graceful manner. + *@returns True if the thread was stopped, false otherwise. When this + * function returns the thread may not have stopped, to ensure that the + * thread has really stopped, call join. + */ + bool stop(); + + /** + * The workhorse of the Ito class. This is the function that will run in + * the thread, when this function exits the thread dies and is cleaned up + * by the system. Make sure to read up on ItoMutex, ItoCondition, and + * cancel to see how to control and protect everything you do in a safe way + * within this function. + *@returns I'm not sure right now, but this is the posix standard form. + */ + virtual void *run()=0; + + /** + * Join the thread in action. This function performs what is commonly + * called a thread join. That is that it effectively makes the calling + * thread an the Ito thread contained in the called object one in the same, + * and pauses the calling thread until the called thread exits. That is, + * when called from, say, your main(), mythread.join() will not return until + * the thread mythread has exited. This is very handy at the end of + * programs to ensure all of your data was cleaned up. + *@returns True if the thread was joined, false if the thread couldn't be + * joined, usually because it isn't running to begin with. + */ + bool join(); + + private: + pthread_t ptHandle; /**< Internal handle to the posix thread. */ + int nHandle; /**< Numeric handle to the posix thread. */ + + protected: + /** + * This is the hidden-heard of the thread system. While run is what the + * user gets to override, and everything said about it is true, this is + * the function that actually makes up the thread, it simply calls the + * run member function in an OO-friendly way. This is what allows us to + * use member variables from within the thread itself. + *@param Should always be this. + *@returns This is specified by posix, I'm not sure yet. + */ + static void *threadRunner( void *pThread ); + + void yield(); + + }; +} + +#endif diff --git a/src/itoatom.h b/src/itoatom.h new file mode 100644 index 0000000..96090f2 --- /dev/null +++ b/src/itoatom.h @@ -0,0 +1,56 @@ +#ifndef ITO_QUEUE_H +#define ITO_QUEUE_H + +#include + +#include "itomutex.h" +#include "itocondition.h" + +/** + * A thread-safe wrapper class. + *@author Mike Buland + */ +template +class ItoAtom +{ +public: + /** + * Construct an empty queue. + */ + ItoAtom() + { + } + + ItoAtom( const T &src ) : + data( src ) + { + } + + ~ItoQueue() + { + } + + T get() + { + mOperate.lock(); + mOperate.unlock(); + return data; + } + + void set( const T &val ) + { + mOperate.lock(); + data = val; + cBlock.signal(); + mOperate.unlock(); + } + +private: + Item *pStart; /**< The start of the queue, the next element to dequeue. */ + Item *pEnd; /**< The end of the queue, the last element to dequeue. */ + + ItoMutex mOperate; /**< The master mutex, used on all operations. */ + ItoCondition cBlock; /**< The condition for blocking dequeues. */ +}; + +#endif diff --git a/src/itocondition.cpp b/src/itocondition.cpp new file mode 100644 index 0000000..d8f5375 --- /dev/null +++ b/src/itocondition.cpp @@ -0,0 +1,42 @@ +#include + +#include "itocondition.h" + +Bu::ItoCondition::ItoCondition() +{ + pthread_cond_init( &cond, NULL ); +} + +Bu::ItoCondition::~ItoCondition() +{ + pthread_cond_destroy( &cond ); +} + +int Bu::ItoCondition::wait() +{ + return pthread_cond_wait( &cond, &mutex ); +} + +int Bu::ItoCondition::wait( int nSec, int nUSec ) +{ + struct timeval now; + struct timespec timeout; + struct timezone tz; + + gettimeofday( &now, &tz ); + timeout.tv_sec = now.tv_sec + nSec + ((now.tv_usec + nUSec)/1000000); + timeout.tv_nsec = ((now.tv_usec + nUSec)%1000000)*1000; + + return pthread_cond_timedwait( &cond, &mutex, &timeout ); +} + +int Bu::ItoCondition::signal() +{ + return pthread_cond_signal( &cond ); +} + +int Bu::ItoCondition::broadcast() +{ + return pthread_cond_broadcast( &cond ); +} + diff --git a/src/itocondition.h b/src/itocondition.h new file mode 100644 index 0000000..4771b22 --- /dev/null +++ b/src/itocondition.h @@ -0,0 +1,81 @@ +#ifndef ITO_CONDITION_H +#define ITO_CONDITION_H + +#include + +#include "itomutex.h" + +namespace Bu +{ + /** + * Ito condition. This is a fairly simple condition mechanism. As you may + * notice this class inherits from the ItoMutex class, this is because all + * conditions must be within a locked block. The standard usage of a condition + * is to pause one thread, perhaps indefinately, until another thread signals + * that it is alright to procede. + *
+ * Standard usage for the thread that wants to wait is as follows: + * 
+	 * ItoCondition cond;
+	 * ... // Perform setup and enter your run loop
+	 * cond.lock();
+	 * while( !isFinished() ) // Could be anything you're waiting for
+	 *     cond.wait();
+	 * ...  // Take care of what you have to.
+	 * cond.unlock();
+	 *
+ * The usage for the triggering thread is much simpler, when it needs to tell + * the others that it's time to grab some data it calls either signal or + * broadcast. See both of those functions for the difference. + *@author Mike Buland + */ + class ItoCondition : public ItoMutex + { + public: + /** + * Create a condition. + */ + ItoCondition(); + + /** + * Destroy a condition. + */ + ~ItoCondition(); + + /** + * Wait forever, or until signalled. This has to be called from within a + * locked section, i.e. before calling this this object's lock function + * should be called. + */ + int wait(); + + /** + * Wait for a maximum of nSec seconds and nUSec micro-seconds or until + * signalled. This is a little more friendly function if you want to + * perform other operations in the thrad loop that calls this function. + * Like the other wait function, this must be inside a locked section. + *@param nSec The seconds to wait. + *@param nUSec the micro-seconds to wait. + */ + int wait( int nSec, int nUSec ); + + /** + * Notify the next thread waiting on this condition that they can go ahead. + * This only signals one thread, the next one in the condition queue, that + * it is safe to procede with whatever operation was being waited on. + */ + int signal(); + + /** + * Notify all threads waiting on this condition that they can go ahead now. + * This function is slower than signal, but more effective in certain + * situations where you may not know how many threads should be activated. + */ + int broadcast(); + + private: + pthread_cond_t cond; /**< Internal condition reference. */ + }; +} + +#endif diff --git a/src/itomutex.cpp b/src/itomutex.cpp new file mode 100644 index 0000000..dc51af9 --- /dev/null +++ b/src/itomutex.cpp @@ -0,0 +1,27 @@ +#include "itomutex.h" + +Bu::ItoMutex::ItoMutex() +{ + pthread_mutex_init( &mutex, NULL ); +} + +Bu::ItoMutex::~ItoMutex() +{ + pthread_mutex_destroy( &mutex ); +} + +int Bu::ItoMutex::lock() +{ + return pthread_mutex_lock( &mutex ); +} + +int Bu::ItoMutex::unlock() +{ + return pthread_mutex_unlock( &mutex ); +} + +int Bu::ItoMutex::trylock() +{ + return pthread_mutex_trylock( &mutex ); +} + diff --git a/src/itomutex.h b/src/itomutex.h new file mode 100644 index 0000000..80956b8 --- /dev/null +++ b/src/itomutex.h @@ -0,0 +1,61 @@ +#ifndef ITO_MUTEX_H +#define ITO_MUTEX_H + +#include + +namespace Bu +{ + /** + * Simple mutex wrapper. Currently this doesn't do anything extra for you + * except keep all of the functionality together in an OO sorta' way and keep + * you from having to worry about cleaning up your mutexes properly, or initing + * them. + *@author Mike Buland + */ + class ItoMutex + { + public: + /** + * Create an unlocked mutex. + */ + ItoMutex(); + + /** + * Destroy a mutex. This can only be done when a mutex is unlocked. + * Failure to unlock before destroying a mutex object could cause it to + * wait for the mutex to unlock, the odds of which are usually farily low + * at deconstruction time. + */ + ~ItoMutex(); + + /** + * Lock the mutex. This causes all future calls to lock on this instance + * of mutex to block until the first thread that called mutex unlocks it. + * At that point the next thread that called lock will get a chance to go + * to work. Because of the nature of a mutex lock it is a very bad idea to + * do any kind of serious or rather time consuming computation within a + * locked section. This can cause thread-deadlock and your program may + * hang. + */ + int lock(); + + /** + * Unlock the mutex. This allows the next thread that asked for a lock to + * lock the mutex and continue with execution. + */ + int unlock(); + + /** + * Try to lock the mutex. This is the option to go with if you cannot avoid + * putting lengthy operations within a locked section. trylock will attempt + * to lock the mutex, if the mutex is already locked this function returns + * immediately with an error code. + */ + int trylock(); + + protected: + pthread_mutex_t mutex; /**< The internal mutex reference. */ + }; +} + +#endif diff --git a/src/itoqueue.h b/src/itoqueue.h new file mode 100644 index 0000000..322698d --- /dev/null +++ b/src/itoqueue.h @@ -0,0 +1,231 @@ +#ifndef ITO_QUEUE_H +#define ITO_QUEUE_H + +#include + +#include "itomutex.h" +#include "itocondition.h" + +namespace Bu +{ + /** + * A thread-safe queue class. This class is a very simple queue with some cool + * extra functionality for use with the Ito system. The main extra that it + * provides is the option to either dequeue without blocking, with infinite + * blocking, or with timed blocking, which will return a value if something is + * enqueued within the specified time limit, or NULL if the time limit is + * exceded. + *@author Mike Buland + */ + template + class ItoQueue + { + private: + /** + * Helper struct. Keeps track of linked-list items for the queue data. + */ + typedef struct Item + { + T pData; + Item *pNext; + } Item; + + public: + /** + * Construct an empty queue. + */ + ItoQueue() : + pStart( NULL ), + pEnd( NULL ), + nSize( 0 ) + { + } + + /** + * Destroy the queue. This function will simply free all contained + * structures. If you stored pointers in the queue, this will lose the + * pointers without cleaning up the memory they pointed to. Make sure + * you're queue is empty before allowing it to be destroyed! + */ + ~ItoQueue() + { + Item *pCur = pStart; + while( pCur ) + { + Item *pTmp = pCur->pNext; + delete pCur; + pCur = pTmp; + } + } + + /** + * Enqueue a pieces of data. The new data will go at the end of the queue, + * and unless another piece of data is enqueued, will be the last piece of + * data to be dequeued. + *@param pData The data to enqueue. If this is not a primitive data type + * it's probably best to use a pointer type. + */ + void enqueue( T pData ) + { + mOperate.lock(); + + if( pStart == NULL ) + { + pStart = pEnd = new Item; + pStart->pData = pData; + pStart->pNext = NULL; + nSize++; + } + else + { + pEnd->pNext = new Item; + pEnd = pEnd->pNext; + pEnd->pData = pData; + pEnd->pNext = NULL; + nSize++; + } + + cBlock.signal(); + + mOperate.unlock(); + } + + /** + * Dequeue the first item from the queue. This function can operate in two + * different modes, blocking and non-blocking. In non-blocking mode it will + * return immediately weather there was data in the queue or not. If there + * was data it will remove it from the queue and return it to the caller. + * In blocking mode it will block forever wating for data to be enqueued. + * When data finally is enqueued this function will return immediately with + * the new data. The only way this function should ever return a null in + * blocking mode is if the calling thread was cancelled. It's probably a + * good idea to check for NULL return values even if you use blocking, just + * to be on the safe side. + *@param bBlock Set to true to enable blocking, leave as false to work in + * non-blocking mode. + *@returns The next piece of data in the queue, or NULL if no data was in + * the queue. + */ + T dequeue( bool bBlock=false ) + { + mOperate.lock(); + if( pStart == NULL ) + { + mOperate.unlock(); + + if( bBlock ) + { + cBlock.lock(); + + while( pStart == NULL ) + cBlock.wait(); + + T tmp = dequeue( false ); + + cBlock.unlock(); + return tmp; + + } + + return NULL; + } + else + { + T pTmp = pStart->pData; + Item *pDel = pStart; + pStart = pStart->pNext; + delete pDel; + nSize--; + + mOperate.unlock(); + return pTmp; + } + } + + /** + * Operates just like the other dequeue function in blocking mode with one + * twist. This function will block for at most nSec seconds and nUSec + * micro-seconds. If the timer is up and no data is available, this will + * just return NULL. If data is enqueued before the timeout expires, it + * will dequeue and exit immediately. + *@param nSec The number of seconds to wait, max. + *@param nUSec The number of micro-seconds to wait, max. + *@returns The next piece of data in the queue, or NULL if the timeout was + * exceeded. + */ + T dequeue( int nSec, int nUSec ) + { + mOperate.lock(); + if( pStart == NULL ) + { + mOperate.unlock(); + + cBlock.lock(); + + cBlock.wait( nSec, nUSec ); + + if( pStart == NULL ) + { + cBlock.unlock(); + return NULL; + } + + mOperate.lock(); + T pTmp = pStart->pData; + Item *pDel = pStart; + pStart = pStart->pNext; + delete pDel; + nSize--; + mOperate.unlock(); + + cBlock.unlock(); + return pTmp; + } + else + { + T pTmp = pStart->pData; + Item *pDel = pStart; + pStart = pStart->pNext; + delete pDel; + nSize--; + + mOperate.unlock(); + return pTmp; + } + } + + /** + * Checks to see if the queue has data in it or not. Note that there is no + * function to determine the length of the queue. This data isn't kept + * track of. If you really need to know, fix this. + *@returns True if the queue is empty, false if it has data in it. + */ + bool isEmpty() + { + mOperate.lock(); + bool bEmpty = (pStart == NULL ); + mOperate.unlock(); + + return bEmpty; + } + + long getSize() + { + mOperate.lock(); + long nRet = nSize; + mOperate.unlock(); + + return nRet; + } + + private: + Item *pStart; /**< The start of the queue, the next element to dequeue. */ + Item *pEnd; /**< The end of the queue, the last element to dequeue. */ + long nSize; /**< The number of items in the queue. */ + + ItoMutex mOperate; /**< The master mutex, used on all operations. */ + ItoCondition cBlock; /**< The condition for blocking dequeues. */ + }; +} + +#endif diff --git a/src/main.dox b/src/main.dox new file mode 100644 index 0000000..668d2e3 --- /dev/null +++ b/src/main.dox @@ -0,0 +1,14 @@ +/** + *@mainpage libbu++ utility library + * + *@section secIntro Introduction + * + * Libbu++ is a C++ library of general utility classes and functions. They + * cover a wide range of topics from streams and sockets to data structures to + * data serialization and xml handling to threading. + * + */ + +/** + *@namespace Bu The core libbu++ namespace, to ensure things don't get muddied. + */ diff --git a/src/serversocket.cpp b/src/serversocket.cpp new file mode 100644 index 0000000..c53c80d --- /dev/null +++ b/src/serversocket.cpp @@ -0,0 +1,148 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "serversocket.h" +#include "exceptions.h" + +Bu::ServerSocket::ServerSocket( int nPort, int nPoolSize ) : + nPort( nPort ) +{ + /* Create the socket and set it up to accept connections. */ + struct sockaddr_in name; + + /* Give the socket a name. */ + name.sin_family = AF_INET; + name.sin_port = htons( nPort ); + + // I think this specifies who we will accept connections from, + // a good thing to make configurable later on + name.sin_addr.s_addr = htonl( INADDR_ANY ); + + startServer( name, nPoolSize ); +} + +Bu::ServerSocket::ServerSocket(const FString &sAddr,int nPort, int nPoolSize) : + nPort( nPort ) +{ + /* Create the socket and set it up to accept connections. */ + struct sockaddr_in name; + + /* Give the socket a name. */ + name.sin_family = AF_INET; + name.sin_port = htons( nPort ); + + inet_aton( sAddr.getStr(), &name.sin_addr ); + + startServer( name, nPoolSize ); +} + +Bu::ServerSocket::~ServerSocket() +{ +} + +void Bu::ServerSocket::startServer( struct sockaddr_in &name, int nPoolSize ) +{ + /* Create the socket. */ + nServer = socket( PF_INET, SOCK_STREAM, 0 ); + if( nServer < 0 ) + { + throw Bu::SocketException("Couldn't create a listen socket."); + } + + int opt = 1; + setsockopt( + nServer, + SOL_SOCKET, + SO_REUSEADDR, + (char *)&opt, + sizeof( opt ) + ); + + if( bind( nServer, (struct sockaddr *) &name, sizeof(name) ) < 0 ) + { + throw Bu::SocketException("Couldn't bind to the listen socket."); + } + + if( listen( nServer, nPoolSize ) < 0 ) + { + throw Bu::SocketException( + "Couldn't begin listening to the server socket." + ); + } + + FD_ZERO( &fdActive ); + /* Initialize the set of active sockets. */ + FD_SET( nServer, &fdActive ); +} + +int Bu::ServerSocket::accept( int nTimeoutSec, int nTimeoutUSec ) +{ + fd_set fdRead = fdActive; + + struct timeval xT; + + xT.tv_sec = nTimeoutSec; + xT.tv_usec = nTimeoutUSec; + + if( TEMP_FAILURE_RETRY(select( nServer+1, &fdRead, NULL, NULL, &xT )) < 0 ) + { + throw SocketException( + "Error scanning for new connections: %s", strerror( errno ) + ); + } + + if( FD_ISSET( nServer, &fdRead ) ) + { + struct sockaddr_in clientname; + size_t size; + int nClient; + + size = sizeof( clientname ); +#ifdef __CYGWIN__ + nClient = ::accept( nServer, (struct sockaddr *)&clientname, + (int *)&size + ); +#else + nClient = ::accept( nServer, (struct sockaddr *)&clientname, &size ); +#endif + if( nClient < 0 ) + { + throw SocketException( + "Error accepting a new connection: %s", strerror( errno ) + ); + } + char tmpa[20]; + inet_ntop( AF_INET, (void *)&clientname.sin_addr, tmpa, 20 ); + //"New connection from host %s, port %hd.", + // tmpa, ntohs (clientname.sin_port) ); + + { + int flags; + + flags = fcntl( nClient, F_GETFL, 0 ); + flags |= O_NONBLOCK; + if( fcntl( nClient, F_SETFL, flags ) < 0) + { + throw SocketException( + "Error setting option on client socket: %s", + strerror( errno ) + ); + } + } + + return nClient; + } + + return -1; +} + diff --git a/src/serversocket.h b/src/serversocket.h new file mode 100644 index 0000000..9a26e2d --- /dev/null +++ b/src/serversocket.h @@ -0,0 +1,31 @@ +#ifndef SERVER_SOCKET_H +#define SERVER_SOCKET_H + +#include +#include "fstring.h" +#include "socket.h" + +namespace Bu +{ + /** + * + */ + class ServerSocket + { + public: + ServerSocket( int nPort, int nPoolSize=40 ); + ServerSocket( const FString &sAddr, int nPort, int nPoolSize=40 ); + virtual ~ServerSocket(); + + int accept( int nTimeoutSec, int nTimeoutUSec ); + + private: + void startServer( struct sockaddr_in &name, int nPoolSize ); + + fd_set fdActive; + int nServer; + int nPort; + }; +} + +#endif diff --git a/src/socket.cpp b/src/socket.cpp index c4f914b..455b5c8 100644 --- a/src/socket.cpp +++ b/src/socket.cpp @@ -118,6 +118,7 @@ void Bu::Socket::close() //} } +/* void Bu::Socket::read() { char buffer[RBS]; @@ -132,7 +133,6 @@ void Bu::Socket::read() if( nbytes < 0 && errno != 0 && errno != EAGAIN ) { //printf("errno: %d, %s\n", errno, strerror( errno ) ); - /* Read error. */ //perror("readInput"); throw ConnectionException( excodeReadError, @@ -146,15 +146,13 @@ void Bu::Socket::read() break; nTotalRead += nbytes; sReadBuf.append( buffer, nbytes ); - /* Data read. */ if( nbytes < RBS ) { break; } - /* New test, if data is divisible by RBS bytes on some libs the - * read could block, this keeps it from happening. - */ + // New test, if data is divisible by RBS bytes on some libs the + // read could block, this keeps it from happening. { fd_set rfds; FD_ZERO(&rfds); @@ -171,13 +169,7 @@ void Bu::Socket::read() } } } - - /* - if( pProtocol != NULL && nTotalRead > 0 ) - { - pProtocol->onNewData(); - }*/ -} +}*/ size_t Bu::Socket::read( void *pBuf, size_t nBytes ) { diff --git a/src/socket.h b/src/socket.h index 3d0125d..568cad6 100644 --- a/src/socket.h +++ b/src/socket.h @@ -19,7 +19,7 @@ namespace Bu virtual ~Socket(); virtual void close(); - virtual void read(); + //virtual void read(); virtual size_t read( void *pBuf, size_t nBytes ); virtual size_t write( const void *pBuf, size_t nBytes ); @@ -33,6 +33,8 @@ namespace Bu virtual bool canWrite(); virtual bool canSeek(); + + private: int nSocket; bool bActive; diff --git a/src/tests/itoqueue1.cpp b/src/tests/itoqueue1.cpp new file mode 100644 index 0000000..f73f4d3 --- /dev/null +++ b/src/tests/itoqueue1.cpp @@ -0,0 +1,110 @@ +#include +#include "bu/ito.h" +#include "bu/itoqueue.h" + +class Reader : public Bu::Ito +{ +public: + Reader( Bu::ItoQueue &q, int id ) : + q( q ), + id( id ) + { + } + + void *run() + { + for( int i = 0; i < 10; i++ ) + { + std::string *pStr = q.dequeue( true ); + if( pStr == NULL ) + { + printf("Null received...\n"); + } + else + { + printf("[%d] read: %s\n", id, pStr->c_str() ); + delete pStr; + } + usleep( (int)(((double)rand())/((double)RAND_MAX)*2000000.0) ); + } + + return NULL; + } + +private: + Bu::ItoQueue &q; + int id; +}; + +class Writer : public Bu::Ito +{ +public: + Writer( Bu::ItoQueue &q, int id, const char *strbase ) : + q( q ), + strbase( strbase ), + id( id ) + { + } + + void *run() + { + for( int i = 0; i < 11; i++ ) + { + usleep( (int)(((double)rand())/((double)RAND_MAX)*2000000.0) ); + q.enqueue( new std::string( strbase ) ); + printf("[%d] write: %s\n", id, strbase ); + } + + return NULL; + } + +private: + Bu::ItoQueue &q; + const char *strbase; + int id; +}; + +int main() +{ + Writer *wr[5]; + Reader *rd[5]; + const char bob[][7]={ + {"Test 1"}, + {"Test 2"}, + {"Test 3"}, + {"Test 4"}, + {"Test 5"} + }; + + Bu::ItoQueue q; + + for( int j = 0; j < 5; j++ ) + { + wr[j] = new Writer( q, j, bob[j] ); + rd[j] = new Reader( q, j ); + } + + for( int j = 0; j < 5; j++ ) + { + rd[j]->start(); + } + + for( int j = 0; j < 5; j++ ) + { + wr[j]->start(); + } + + for( int j = 0; j < 5; j++ ) + { + rd[j]->join(); + } + + for( int j = 0; j < 5; j++ ) + { + delete wr[j]; + delete rd[j]; + } + + return 0; +} + diff --git a/src/tests/itoqueue2.cpp b/src/tests/itoqueue2.cpp new file mode 100644 index 0000000..f4b5e19 --- /dev/null +++ b/src/tests/itoqueue2.cpp @@ -0,0 +1,83 @@ +#include +#include "bu/ito.h" +#include "bu/itoqueue.h" +#include + +class Reader : public Bu::Ito +{ +public: + Reader( Bu::ItoQueue &q, int id ) : + q( q ), + id( id ) + { + } + + void *run() + { + for( int i = 0; i < 10; i++ ) + { + std::string *pStr = q.dequeue( 0, 500000 ); + if( pStr == NULL ) + { + printf("Null received...\n"); + i--; + } + else + { + printf("[%d] read: %s\n", id, pStr->c_str() ); + delete pStr; + } + } + + return NULL; + } + +private: + Bu::ItoQueue &q; + int id; +}; + +class Writer : public Bu::Ito +{ +public: + Writer( Bu::ItoQueue &q, int id, const char *strbase ) : + q( q ), + strbase( strbase ), + id( id ) + { + } + + void *run() + { + for( int i = 0; i < 11; i++ ) + { + sleep( 2 ); + printf("[%d] write: %s\n", id, strbase ); + q.enqueue( new std::string( strbase ) ); + } + + return NULL; + } + +private: + Bu::ItoQueue &q; + const char *strbase; + int id; +}; + +int main() +{ + printf("ETIMEDOUT: %d\n", ETIMEDOUT ); + Bu::ItoQueue q; + Writer wr( q, 0, "writer" ); + Reader rd( q, 0 ); + + rd.start(); + wr.start(); + + rd.join(); + wr.join(); + + return 0; +} + -- cgit v1.2.3