Neither the Bu::Client class or the Bu::ProtocolTelnet class were threadsafe

and both really need to be.  Bu::Client is used in multi-threaded applications
and has been for a while, so it's a no brainer.  It may need a little bit more
safety added around variable access, but all the buffers are safe now.

ProtocolTelnet has it's own issus, and I should probably rework a few parts of
it before too long so we can better accomidate things like line editing.

4 files changed, 45 insertions, 0 deletions

diff --git a/src/stable/client.cpp b/src/stable/client.cpp
index a9deb00..ca86f37 100644
--- a/src/stable/client.cpp
+++ b/src/stable/client.cpp
@@ -12,6 +12,7 @@
 #include "bu/protocol.h"
 #include "bu/clientlink.h"
 #include "bu/clientlinkfactory.h"
+#include "bu/mutexlocker.h"
 
 /** Read buffer size. */
 #define RBS     (2000) // 1500 is the nominal MTU for ethernet, it's a good guess
@@ -39,6 +40,7 @@ Bu::Client::~Client()
 
 void Bu::Client::processInput()
 {
+    mRead.lock();
     char buf[RBS];
     Bu::size nRead, nTotal=0;
 
@@ -67,6 +69,7 @@ void Bu::Client::processInput()
             break;
         }
     }
+    mRead.unlock();
 
     if( nTotal == 0 )
     {
@@ -82,6 +85,7 @@ void Bu::Client::processInput()
 
 void Bu::Client::processOutput()
 {
+    mWrite.lock();
     char buf[RBS];
     if( qbWrite.getSize() > 0 )
     {
@@ -91,6 +95,7 @@ void Bu::Client::processOutput()
         qbWrite.seek( nReal );
         pTopStream->flush();
     }
+    mWrite.unlock();
 }
 
 void Bu::Client::setProtocol( Protocol *pProto )
@@ -128,71 +133,85 @@ bool Bu::Client::isOpen()
 
 Bu::size Bu::Client::write( const Bu::String &sData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( sData.getStr(), sData.getSize() );
 }
 
 Bu::size Bu::Client::write( const void *pData, Bu::size nBytes )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( pData, nBytes );
 }
 
 Bu::size Bu::Client::write( int8_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int16_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int32_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int64_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint8_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint16_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint32_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint64_t nData )
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::read( void *pData, Bu::size nBytes )
 {
+    Bu::MutexLocker l( mWrite );
     return qbRead.read( pData, nBytes );
 }
 
 Bu::size Bu::Client::peek( void *pData, int nBytes, int nOffset )
 {
+    Bu::MutexLocker l( mWrite );
     return qbRead.peek( pData, nBytes, nOffset );
 }
 
 Bu::size Bu::Client::getInputSize()
 {
+    Bu::MutexLocker l( mWrite );
     return qbRead.getSize();
 }
 
 Bu::size Bu::Client::getOutputSize()
 {
+    Bu::MutexLocker l( mWrite );
     return qbWrite.getSize();
 }
 
@@ -240,6 +259,7 @@ Bu::size Bu::Client::tell()
 
 void Bu::Client::seek( Bu::size offset )
 {
+    Bu::MutexLocker l( mRead );
     return qbRead.seek( offset );
 }
 
@@ -265,6 +285,7 @@ void Bu::Client::flush()
 
 bool Bu::Client::canRead()
 {
+    Bu::MutexLocker l( mRead );
     return qbRead.getSize() > 0;
 }
 
diff --git a/src/stable/client.h b/src/stable/client.h
index 89e8144..ac882eb 100644
--- a/src/stable/client.h
+++ b/src/stable/client.h
@@ -13,6 +13,7 @@
 #include "bu/config.h"
 #include "bu/string.h"
 #include "bu/queuebuf.h"
+#include "bu/mutex.h"
 
 namespace Bu
 {
@@ -127,6 +128,8 @@ namespace Bu
         Bu::QueueBuf qbWrite;
         bool bWantsDisconnect;
         class Bu::ClientLinkFactory *pfLink;
+        Bu::Mutex mRead;
+        Bu::Mutex mWrite;
     };
 }
 
diff --git a/src/stable/protocoltelnet.cpp b/src/stable/protocoltelnet.cpp
index 1461b0e..153af57 100644
--- a/src/stable/protocoltelnet.cpp
+++ b/src/stable/protocoltelnet.cpp
@@ -7,6 +7,7 @@
 
 #include "bu/protocoltelnet.h"
 #include "bu/client.h"
+#include "bu/mutexlocker.h"
 
 /* We apparently at least want defs for the lower 13, not sure we care about
  * the rest of the chars, maybe escape.
@@ -81,11 +82,16 @@ Bu::ProtocolTelnet::~ProtocolTelnet()
 
 void Bu::ProtocolTelnet::onNewConnection( Bu::Client *pClient )
 {
+    mRead.lock();
+    mWrite.lock();
     this->pClient = pClient;
+    mWrite.unlock();
+    mRead.unlock();
 }
 
 void Bu::ProtocolTelnet::onNewData( Bu::Client *pClient )
 {
+    MutexLocker l( mRead );
     char bc;
     int iLeft;
     while( (iLeft = pClient->getInputSize()) )
@@ -233,7 +239,9 @@ void Bu::ProtocolTelnet::onNewData( Bu::Client *pClient )
                         sDataBuf += bc;
                         if( oEcho.isLocalSet() && bEcho )
                         {
+                            mWrite.lock();
                             pClient->write( &bc, 1 );
+                            mWrite.unlock();
 #ifdef __TELNET_DEBUG
                             printf("%c", bc );
                             fflush( stdout );
@@ -246,7 +254,9 @@ void Bu::ProtocolTelnet::onNewData( Bu::Client *pClient )
                     sDataBuf += bc;
                     if( oEcho.isLocalSet() && bEcho )
                     {
+                        mWrite.lock();
                         pClient->write( &bc, 1 );
+                        mWrite.unlock();
                     }
                 }
             }
@@ -291,6 +301,7 @@ void Bu::ProtocolTelnet::write( const Bu::String &sData )
 
 void Bu::ProtocolTelnet::write( const char *pData, int iSize )
 {
+    Bu::MutexLocker l( mWrite );
     int iLast = 0, j;
     for( j = iLast; j < iSize; j++ )
     {
@@ -323,11 +334,13 @@ void Bu::ProtocolTelnet::write( char cData )
 
 void Bu::ProtocolTelnet::flush()
 {
+    Bu::MutexLocker l( mWrite );
     pClient->flush();
 }
 
 void Bu::ProtocolTelnet::onWill( char cCode )
 {
+    Bu::MutexLocker l( mWrite );
     try
     {
         Option *pOpt = hOpts[cCode];
@@ -353,6 +366,7 @@ void Bu::ProtocolTelnet::onWill( char cCode )
 
 void Bu::ProtocolTelnet::onWont( char cCode )
 {
+    Bu::MutexLocker l( mWrite );
     try
     {
         Option *pOpt = hOpts[cCode];
@@ -370,6 +384,7 @@ void Bu::ProtocolTelnet::onWont( char cCode )
 
 void Bu::ProtocolTelnet::onDo( char cCode )
 {
+    Bu::MutexLocker l( mWrite );
     try
     {
         Option *pOpt = hOpts[cCode];
@@ -395,6 +410,7 @@ void Bu::ProtocolTelnet::onDo( char cCode )
 
 void Bu::ProtocolTelnet::onDont( char cCode )
 {
+    Bu::MutexLocker l( mWrite );
     try
     {
         Option *pOpt = hOpts[cCode];
@@ -467,7 +483,9 @@ void Bu::ProtocolTelnet::onCtlChar( char cChr )
                 {
                     sDataBuf.resize( sDataBuf.getSize()-1 );
                     char buf[3] = { CH_BS, ' ', CH_BS };
+                    mWrite.lock();
                     pClient->write( buf, 3 );
+                    mWrite.unlock();
                 }
             }
             break;
diff --git a/src/stable/protocoltelnet.h b/src/stable/protocoltelnet.h
index 60faef4..19298dc 100644
--- a/src/stable/protocoltelnet.h
+++ b/src/stable/protocoltelnet.h
@@ -11,6 +11,7 @@
 #include "bu/protocol.h"
 #include "bu/hash.h"
 #include "bu/string.h"
+#include "bu/mutex.h"
 
 // #define __TELNET_DEBUG
 
@@ -210,6 +211,8 @@ namespace Bu
 #endif
 
     private:
+        Mutex mRead;
+        Mutex mWrite;
         Client *pClient;
 
         Bu::String sDataBuf;    /**< Buffer for regular line data. */