go with non-blocking sockets for tcp connections, adding the ability

to reassemble packets that arrive in separate recv() calls.

xwords4/relay/udpqueue.cpp

@@ -37,9 +37,41 @@ UdpThreadClosure::logStats()
     }
 }
 
+bool
+PartialPacket::stillGood() const
+{
+    return 0 == m_errno
+        || EAGAIN == m_errno
+        || EWOULDBLOCK == m_errno;
+}
+
+bool
+PartialPacket::readAtMost( int len )
+{
+    bool success = false;
+    uint8_t tmp[len];
+    ssize_t nRead = recv( m_sock, tmp, len, 0 );
+    if ( 0 > nRead ) {          // error case
+        m_errno = errno;
+        logf( XW_LOGERROR, "%s(len=%d): recv failed: %d (%s)", __func__, len,
+              m_errno, strerror(m_errno) );
+    } else if ( 0 == nRead ) {  // remote socket closed
+        logf( XW_LOGINFO, "%s: remote closed", __func__ );
+        m_errno = -1;           // so stillGood will fail
+    } else {
+        m_errno = 0;
+        success = len == nRead;
+        int curSize = m_buf.size();
+        m_buf.resize( nRead + curSize );
+        memcpy( &m_buf[curSize], tmp, nRead );
+    }
+    return success;
+}
+
 UdpQueue::UdpQueue() 
 {
     m_nextID = 0;
+    pthread_mutex_init ( &m_partialsMutex, NULL );
     pthread_mutex_init ( &m_queueMutex, NULL );
     pthread_cond_init( &m_queueCondVar, NULL );
 
@@ -54,6 +86,7 @@ UdpQueue::~UdpQueue()
 {
     pthread_cond_destroy( &m_queueCondVar );
     pthread_mutex_destroy ( &m_queueMutex );
+    pthread_mutex_destroy ( &m_partialsMutex );
 }
 
 UdpQueue* 
@@ -65,52 +98,85 @@ UdpQueue::get()
     return s_instance;
 }
 
+// return false if socket should no longer be used
 bool
 UdpQueue::handle( const AddrInfo* addr, QueueCallback cb )
 {
-    bool success = false;
+    PartialPacket* packet;
+
     int sock = addr->socket();
-    unsigned short msgLen;
-    ssize_t nRead = recv( sock, &msgLen, sizeof(msgLen), MSG_WAITALL );
-    if ( 0 == nRead ) {
-        logf( XW_LOGINFO, "%s: recv(sock=%d) => 0: remote closed", __func__, sock );
-    } else if ( nRead != sizeof(msgLen) ) {
-        logf( XW_LOGERROR, "%s: first recv => %d: %s", __func__, 
-              nRead, strerror(errno) );
+
+    // Hang onto this mutex for as long as we may be writing to the packet
+    // since having it deleted while in use would be bad.
+    MutexLock ml( &m_partialsMutex );
+
+    map<int, PartialPacket*>::iterator iter = m_partialPackets.find( sock );
+    if ( m_partialPackets.end() == iter ) {
+        packet = new PartialPacket( sock );
+        m_partialPackets.insert( pair<int, PartialPacket*>( sock, packet ) );
     } else {
-        msgLen = ntohs( msgLen );
-        if ( MAX_MSG_LEN <= msgLen ) {
-            logf( XW_LOGERROR, "%s: message of len %d too large; dropping", __func__, msgLen );
-        } else {
-            unsigned char buf[msgLen];
-            nRead = recv( sock, buf, msgLen, MSG_WAITALL );
-            if ( nRead == msgLen ) {
-                logf( XW_LOGINFO, "%s: read %d bytes on socket %d", __func__, nRead, sock );
-                handle( addr, buf, msgLen, cb );
-                success = true;
-            } else {
-                logf( XW_LOGERROR, "%s: second recv failed: %s", __func__, 
-                      strerror(errno) );
-            }
+        packet = iter->second;
+    }
+
+    // First see if we've read the length bytes
+    if ( packet->readSoFar() < sizeof( packet->m_len ) ) {
+        if ( packet->readAtMost( sizeof(packet->m_len) - packet->readSoFar() ) ) {
+            packet->m_len = ntohs(*(unsigned short*)packet->data());
         }
     }
-    return success;
+
+    if ( packet->readSoFar() >= sizeof( packet->m_len ) ) {
+        assert( 0 < packet->m_len );
+        int leftToRead = 
+            packet->m_len - (packet->readSoFar() - sizeof(packet->m_len));
+        if ( packet->readAtMost( leftToRead ) ) {
+            handle( addr, packet->data() + sizeof(packet->m_len), 
+                    packet->m_len, cb );
+            packet = NULL;
+            newSocket_locked( sock );
+        }
+    }
+
+    return NULL == packet || packet->stillGood();
 }
 
 void 
-UdpQueue::handle( const AddrInfo* addr, unsigned char* buf, int len, 
+UdpQueue::handle( const AddrInfo* addr, const uint8_t* buf, int len, 
                   QueueCallback cb )
 {
     UdpThreadClosure* utc = new UdpThreadClosure( addr, buf, len, cb );
     MutexLock ml( &m_queueMutex );
     int id = ++m_nextID;
     utc->setID( id );
-    logf( XW_LOGINFO, "%s: enqueuing packet %d", __func__, id );
+    logf( XW_LOGINFO, "%s: enqueuing packet %d (len %d)", __func__, id, len );
     m_queue.push_back( utc );
 
     pthread_cond_signal( &m_queueCondVar );
 }
 
+void
+UdpQueue::newSocket_locked( int sock )
+{
+    map<int, PartialPacket*>::iterator iter = m_partialPackets.find( sock );
+    if ( m_partialPackets.end() != iter ) {
+        delete iter->second;
+        m_partialPackets.erase( iter );
+    }
+}
+void
+UdpQueue::newSocket( int sock )
+{
+    MutexLock ml( &m_partialsMutex );
+    newSocket_locked( sock );
+}
+
+void
+UdpQueue::newSocket( const AddrInfo* addr )
+{
+    assert( addr->isTCP() );
+    newSocket( addr->socket() );
+}
+
 void* 
 UdpQueue::thread_main()
 {

xwords4/relay/udpqueue.h

@@ -36,9 +36,9 @@ typedef void (*QueueCallback)( UdpThreadClosure* closure );
 
 class UdpThreadClosure {
 public:
-    UdpThreadClosure( const AddrInfo* addr, unsigned char* buf, 
+    UdpThreadClosure( const AddrInfo* addr, const uint8_t* buf, 
                       int len, QueueCallback cb )
-        : m_buf(new unsigned char[len])
+        : m_buf(new uint8_t[len])
         , m_len(len)
         , m_addr(*addr)
         , m_cb(cb)
@@ -60,7 +60,7 @@ public:
     int getID( void ) { return m_id; }
 
  private:
-    unsigned char* m_buf;
+    uint8_t* m_buf;
     int m_len;
     AddrInfo m_addr;
     QueueCallback m_cb;
@@ -69,25 +69,49 @@ public:
     int m_id;
 };
 
+class PartialPacket {
+ public:
+    PartialPacket(int sock) {
+        m_sock = sock;
+        m_len = 0;
+        m_errno = 0;
+    }
+    bool stillGood() const ;
+    bool readAtMost( int len );
+    size_t readSoFar() const { return m_buf.size(); }
+    const uint8_t* data() const { return m_buf.data(); }
+
+    unsigned short m_len;
+ private:
+
+    vector<uint8_t> m_buf;
+    int m_sock;
+    int m_errno;
+};
+
 class UdpQueue {
  public:
     static UdpQueue* get();
     UdpQueue();
     ~UdpQueue();
     bool handle( const AddrInfo* addr, QueueCallback cb );
-    void handle( const AddrInfo* addr, unsigned char* buf, int len,
+    void handle( const AddrInfo* addr, const uint8_t* buf, int len,
                  QueueCallback cb );
-    void forgetSocket( const AddrInfo* addr );
+    void newSocket( int socket );
+    void newSocket( const AddrInfo* addr );
 
  private:
+    void newSocket_locked( int sock );
     static void* thread_main_static( void* closure );
     void* thread_main();
 
+    pthread_mutex_t m_partialsMutex;
     pthread_mutex_t m_queueMutex;
     pthread_cond_t m_queueCondVar;
     deque<UdpThreadClosure*> m_queue;
     // map<int, vector<UdpThreadClosure*> > m_bySocket;
     int m_nextID;
+    map<int, PartialPacket*> m_partialPackets;
 };
 
 #endif

xwords4/relay/xwrelay.cpp

@@ -453,8 +453,8 @@ send_with_length_unsafe( const AddrInfo* addr, const unsigned char* buf,
         if ( addr->isCurrent() ) {
             int socket = addr->socket();
             unsigned short len = htons( bufLen );
-            ssize_t nSent = send( socket, &len, 2, 0 );
-            if ( nSent == 2 ) {
+            ssize_t nSent = send( socket, &len, sizeof(len), 0 );
+            if ( nSent == sizeof(len) ) {
                 nSent = send( socket, buf, bufLen, 0 );
                 if ( nSent == ssize_t(bufLen) ) {
                     logf( XW_LOGINFO, "sent %d bytes on socket %d", nSent, socket );
@@ -1408,7 +1408,7 @@ udp_thread_proc( UdpThreadClosure* utc )
 static void
 read_udp_packet( int udpsock )
 {
-    unsigned char buf[MAX_MSG_LEN];
+    uint8_t buf[MAX_MSG_LEN];
     AddrInfo::AddrUnion saddr;
     memset( &saddr, 0, sizeof(saddr) );
     socklen_t fromlen = sizeof(saddr.addr_in);
@@ -1448,6 +1448,8 @@ string_printf( string& str, const char* fmt, ... )
     }
 }
 
+// Going with non-blocking instead
+#if 0
 static void
 set_timeouts( int sock )
 {
@@ -1472,6 +1474,7 @@ set_timeouts( int sock )
         assert( 0 );
     }
 }
+#endif
 
 static void
 enable_keepalive( int sock )
@@ -1897,15 +1900,17 @@ main( int argc, char** argv )
                               errno, strerror(errno) );
                         assert( 0 ); // we're leaking files or load has grown
                     } else {
-			// I've seen a bug where we accept but never service
-			// connections.  Sockets are not closed, and so the
-			// number goes up.  Probably need a watchdog instead,
-			// but this will work around it.
+                        // I've seen a bug where we accept but never service
+                        // connections.  Sockets are not closed, and so the
+                        // number goes up.  Probably need a watchdog instead,
+                        // but this will work around it.
                         assert( g_maxsocks > newSock );
 
                         /* Set timeout so send and recv won't block forever */
-                        set_timeouts( newSock );
-                        
+                        // set_timeouts( newSock );
+
+                        int err = fcntl( newSock, F_SETFL, O_NONBLOCK );
+                        assert( 0 == err );
                         enable_keepalive( newSock );
 
                         logf( XW_LOGINFO, 
@@ -1918,6 +1923,7 @@ main( int argc, char** argv )
                                           perGame ? game_thread_proc
                                           : proxy_thread_proc,
                                           &addr );
+                        UdpQueue::get()->newSocket( &addr );
                     }
                     --retval;
                 }