Merge branch 'socket' of https://github.com/CendioOssman/tigervnc

12 files changed, 130 insertions, 89 deletions

diff --git a/common/network/Socket.h b/common/network/Socket.h
index 378a9006..13b12d1d 100644
--- a/common/network/Socket.h
+++ b/common/network/Socket.h
@@ -125,12 +125,17 @@ namespace network {
     //   resources to be freed.
     virtual void removeSocket(network::Socket* sock) = 0;
 
-    // processSocketEvent() tells the server there is a Socket read event.
+    // processSocketReadEvent() tells the server there is a Socket read event.
     //   The implementation can indicate that the Socket is no longer active
     //   by calling shutdown() on it.  The caller will then call removeSocket()
     //   soon after processSocketEvent returns, to allow any pre-Socket
     //   resources to be tidied up.
-    virtual void processSocketEvent(network::Socket* sock) = 0;
+    virtual void processSocketReadEvent(network::Socket* sock) = 0;
+
+    // processSocketReadEvent() tells the server there is a Socket write event.
+    //   This is only necessary if the Socket has been put in non-blocking
+    //   mode and needs this callback to flush the buffer.
+    virtual void processSocketWriteEvent(network::Socket* sock) = 0;
 
     // checkTimeouts() allows the server to check socket timeouts, etc.  The
     //   return value is the number of milliseconds to wait before
diff --git a/common/rdr/FdOutStream.cxx b/common/rdr/FdOutStream.cxx
index f4299030..e6d081a3 100644
--- a/common/rdr/FdOutStream.cxx
+++ b/common/rdr/FdOutStream.cxx
@@ -95,32 +95,14 @@ unsigned FdOutStream::getIdleTime()
 
 void FdOutStream::flush()
 {
-  int timeoutms_;
-
-  if (blocking)
-    timeoutms_ = timeoutms;
-  else
-    timeoutms_ = 0;
-
   while (sentUpTo < ptr) {
     int n = writeWithTimeout((const void*) sentUpTo,
-                             ptr - sentUpTo, timeoutms_);
+                             ptr - sentUpTo,
+                             blocking? timeoutms : 0);
 
     // Timeout?
-    if (n == 0) {
-      // If non-blocking then we're done here
-      if (!blocking)
-        break;
-
-      // Otherwise try blocking (with possible timeout)
-      if ((timeoutms_ == 0) && (timeoutms != 0)) {
-        timeoutms_ = timeoutms;
-        break;
-      }
-
-      // Proper timeout
+    if ((n == 0) && blocking)
       throw TimedOut();
-    }
 
     sentUpTo += n;
     offset += n;
@@ -183,38 +165,34 @@ int FdOutStream::writeWithTimeout(const void* data, int length, int timeoutms)
   int n;
 
   do {
+    fd_set fds;
+    struct timeval tv;
+    struct timeval* tvp = &tv;
+
+    if (timeoutms != -1) {
+      tv.tv_sec = timeoutms / 1000;
+      tv.tv_usec = (timeoutms % 1000) * 1000;
+    } else {
+      tvp = NULL;
+    }
+
+    FD_ZERO(&fds);
+    FD_SET(fd, &fds);
+    n = select(fd+1, 0, &fds, 0, tvp);
+  } while (n < 0 && errno == EINTR);
+
+  if (n < 0)
+    throw SystemException("select", errno);
+
+  if (n == 0)
+    return 0;
+
+  do {
+    n = ::write(fd, data, length);
+  } while (n < 0 && (errno == EINTR));
 
-    do {
-      fd_set fds;
-      struct timeval tv;
-      struct timeval* tvp = &tv;
-
-      if (timeoutms != -1) {
-        tv.tv_sec = timeoutms / 1000;
-        tv.tv_usec = (timeoutms % 1000) * 1000;
-      } else {
-        tvp = 0;
-      }
-
-      FD_ZERO(&fds);
-      FD_SET(fd, &fds);
-      n = select(fd+1, 0, &fds, 0, tvp);
-    } while (n < 0 && errno == EINTR);
-
-    if (n < 0) throw SystemException("select",errno);
-
-    if (n == 0) return 0;
-
-    do {
-      n = ::write(fd, data, length);
-    } while (n < 0 && (errno == EINTR));
-      
-    // NB: This outer loop simply fixes a broken Winsock2 EWOULDBLOCK
-    // condition, found only under Win98 (first edition), with slow
-    // network connections.  Should in fact never ever happen...
-  } while (n < 0 && (errno == EWOULDBLOCK));
-
-  if (n < 0) throw SystemException("write",errno);
+  if (n < 0)
+    throw SystemException("write", errno);
 
   gettimeofday(&lastWrite, NULL);
 
diff --git a/common/rfb/HTTPServer.cxx b/common/rfb/HTTPServer.cxx
index f50722ab..54becbb2 100644
--- a/common/rfb/HTTPServer.cxx
+++ b/common/rfb/HTTPServer.cxx
@@ -337,7 +337,7 @@ HTTPServer::removeSocket(network::Socket* sock) {
 }
 
 void
-HTTPServer::processSocketEvent(network::Socket* sock) {
+HTTPServer::processSocketReadEvent(network::Socket* sock) {
   std::list<Session*>::iterator i;
   for (i=sessions.begin(); i!=sessions.end(); i++) {
     if ((*i)->getSock() == sock) {
@@ -356,6 +356,23 @@ HTTPServer::processSocketEvent(network::Socket* sock) {
   throw rdr::Exception("invalid Socket in HTTPServer");
 }
 
+void
+HTTPServer::processSocketWriteEvent(network::Socket* sock) {
+  std::list<Session*>::iterator i;
+  for (i=sessions.begin(); i!=sessions.end(); i++) {
+    if ((*i)->getSock() == sock) {
+      try {
+        sock->outStream().flush();
+      } catch (rdr::Exception& e) {
+        vlog.error("untrapped: %s", e.str());
+        sock->shutdown();
+      }
+      return;
+    }
+  }
+  throw rdr::Exception("invalid Socket in HTTPServer");
+}
+
 void HTTPServer::getSockets(std::list<network::Socket*>* sockets)
 {
   sockets->clear();
diff --git a/common/rfb/HTTPServer.h b/common/rfb/HTTPServer.h
index 6412946a..d7ca69ad 100644
--- a/common/rfb/HTTPServer.h
+++ b/common/rfb/HTTPServer.h
@@ -58,11 +58,16 @@ namespace rfb {
     //   Could clean up socket-specific resources here.
     virtual void removeSocket(network::Socket* sock);
 
-    // processSocketEvent()
+    // processSocketReadEvent()
     //   The platform-specific side of the server implementation calls
     //   this method whenever data arrives on one of the active
     //   network sockets.
-    virtual void processSocketEvent(network::Socket* sock);
+    virtual void processSocketReadEvent(network::Socket* sock);
+
+    // processSocketWriteEvent()
+    //   Similar to processSocketReadEvent(), but called when it is
+    //   possible to write more data to a socket.
+    virtual void processSocketWriteEvent(network::Socket* sock);
 
     // Check for socket timeouts
     virtual int checkTimeouts();
diff --git a/common/rfb/SMsgWriter.cxx b/common/rfb/SMsgWriter.cxx
index e4215086..5040b658 100644
--- a/common/rfb/SMsgWriter.cxx
+++ b/common/rfb/SMsgWriter.cxx
@@ -285,6 +285,7 @@ void SMsgWriter::startRect(const Rect& r, int encoding)
 
 void SMsgWriter::endRect()
 {
+  os->flush();
 }
 
 void SMsgWriter::startMsg(int type)
diff --git a/common/rfb/VNCSConnectionST.cxx b/common/rfb/VNCSConnectionST.cxx
index 932f5796..97f7d286 100644
--- a/common/rfb/VNCSConnectionST.cxx
+++ b/common/rfb/VNCSConnectionST.cxx
@@ -75,8 +75,7 @@ VNCSConnectionST::VNCSConnectionST(VNCServerST* server_, network::Socket *s,
     pingCounter(0), congestionTimer(this),
     server(server_), updates(false),
     updateRenderedCursor(false), removeRenderedCursor(false),
-    continuousUpdates(false), encodeManager(this),
-    updateTimer(this), pointerEventTime(0),
+    continuousUpdates(false), encodeManager(this), pointerEventTime(0),
     accessRights(AccessDefault), startTime(time(0))
 {
   setStreams(&sock->inStream(), &sock->outStream());
@@ -190,6 +189,21 @@ void VNCSConnectionST::processMessages()
   }
 }
 
+void VNCSConnectionST::flushSocket()
+{
+  if (state() == RFBSTATE_CLOSING) return;
+  try {
+    setSocketTimeouts();
+    sock->outStream().flush();
+    // Flushing the socket might release an update that was previously
+    // delayed because of congestion.
+    if (sock->outStream().bufferUsage() == 0)
+      writeFramebufferUpdate();
+  } catch (rdr::Exception &e) {
+    close(e.str());
+  }
+}
+
 void VNCSConnectionST::pixelBufferChange()
 {
   try {
@@ -730,9 +744,7 @@ void VNCSConnectionST::supportsContinuousUpdates()
 bool VNCSConnectionST::handleTimeout(Timer* t)
 {
   try {
-    if (t == &updateTimer)
-      writeFramebufferUpdate();
-    else if (t == &congestionTimer)
+    if (t == &congestionTimer)
       updateCongestion();
     else if (t == &queryConnectTimer) {
       if (state() == RFBSTATE_QUERYING)
@@ -820,6 +832,7 @@ bool VNCSConnectionST::isCongested()
   int offset;
 
   // Stuff still waiting in the send buffer?
+  sock->outStream().flush();
   if (sock->outStream().bufferUsage() > 0)
     return true;
 
@@ -928,8 +941,6 @@ void VNCSConnectionST::writeFramebufferUpdate()
   bool needNewUpdateInfo;
   bool drawRenderedCursor;
 
-  updateTimer.stop();
-
   // We're in the middle of processing a command that's supposed to be
   // synchronised. Allowing an update to slip out right now might violate
   // that synchronisation.
@@ -949,10 +960,8 @@ void VNCSConnectionST::writeFramebufferUpdate()
 
   // Check that we actually have some space on the link and retry in a
   // bit if things are congested.
-  if (isCongested()) {
-    updateTimer.start(50);
+  if (isCongested())
     return;
-  }
 
   // In continuous mode, we will be outputting at least three distinct
   // messages. We need to aggregate these in order to not clog up TCP's
diff --git a/common/rfb/VNCSConnectionST.h b/common/rfb/VNCSConnectionST.h
index 72ffc1df..3f0163a3 100644
--- a/common/rfb/VNCSConnectionST.h
+++ b/common/rfb/VNCSConnectionST.h
@@ -65,6 +65,9 @@ namespace rfb {
     // Socket if an error occurs, via the close() call.
     void processMessages();
 
+    // flushSocket() pushes any unwritten data on to the network.
+    void flushSocket();
+
     // Called when the underlying pixelbuffer is resized or replaced.
     void pixelBufferChange();
 
@@ -202,8 +205,6 @@ namespace rfb {
     Region cuRegion;
     EncodeManager encodeManager;
 
-    Timer updateTimer;
-
     std::set<rdr::U32> pressedKeys;
 
     time_t lastEventTime;
diff --git a/common/rfb/VNCServerST.cxx b/common/rfb/VNCServerST.cxx
index 199524ec..d5010854 100644
--- a/common/rfb/VNCServerST.cxx
+++ b/common/rfb/VNCServerST.cxx
@@ -163,7 +163,7 @@ void VNCServerST::removeSocket(network::Socket* sock) {
   closingSockets.remove(sock);
 }
 
-void VNCServerST::processSocketEvent(network::Socket* sock)
+void VNCServerST::processSocketReadEvent(network::Socket* sock)
 {
   // - Find the appropriate VNCSConnectionST and process the event
   std::list<VNCSConnectionST*>::iterator ci;
@@ -176,6 +176,19 @@ void VNCServerST::processSocketEvent(network::Socket* sock)
   throw rdr::Exception("invalid Socket in VNCServerST");
 }
 
+void VNCServerST::processSocketWriteEvent(network::Socket* sock)
+{
+  // - Find the appropriate VNCSConnectionST and process the event
+  std::list<VNCSConnectionST*>::iterator ci;
+  for (ci = clients.begin(); ci != clients.end(); ci++) {
+    if ((*ci)->getSock() == sock) {
+      (*ci)->flushSocket();
+      return;
+    }
+  }
+  throw rdr::Exception("invalid Socket in VNCServerST");
+}
+
 int VNCServerST::checkTimeouts()
 {
   int timeout = 0;
diff --git a/common/rfb/VNCServerST.h b/common/rfb/VNCServerST.h
index 1e055dd9..bd84c452 100644
--- a/common/rfb/VNCServerST.h
+++ b/common/rfb/VNCServerST.h
@@ -67,11 +67,15 @@ namespace rfb {
     //   Clean up any resources associated with the Socket
     virtual void removeSocket(network::Socket* sock);
 
-    // processSocketEvent
+    // processSocketReadEvent
     //   Read more RFB data from the Socket.  If an error occurs during
     //   processing then shutdown() is called on the Socket, causing
     //   removeSocket() to be called by the caller at a later time.
-    virtual void processSocketEvent(network::Socket* sock);
+    virtual void processSocketReadEvent(network::Socket* sock);
+
+    // processSocketWriteEvent
+    //   Flush pending data from the Socket on to the network.
+    virtual void processSocketWriteEvent(network::Socket* sock);
 
     // checkTimeouts
     //   Returns the number of milliseconds left until the next idle timeout
diff --git a/unix/x0vncserver/x0vncserver.cxx b/unix/x0vncserver/x0vncserver.cxx
index 6b5d479d..8f73ac2c 100644
--- a/unix/x0vncserver/x0vncserver.cxx
+++ b/unix/x0vncserver/x0vncserver.cxx
@@ -508,8 +508,9 @@ int main(int argc, char** argv)
     PollingScheduler sched((int)pollingCycle, (int)maxProcessorUsage);
 
     while (!caughtSignal) {
+      int wait_ms;
       struct timeval tv;
-      fd_set rfds;
+      fd_set rfds, wfds;
       std::list<Socket*> sockets;
       std::list<Socket*>::iterator i;
 
@@ -517,6 +518,8 @@ int main(int argc, char** argv)
       TXWindow::handleXEvents(dpy);
 
       FD_ZERO(&rfds);
+      FD_ZERO(&wfds);
+
       FD_SET(ConnectionNumber(dpy), &rfds);
       for (std::list<TcpListener*>::iterator i = listeners.begin();
            i != listeners.end();
@@ -531,6 +534,8 @@ int main(int argc, char** argv)
           delete (*i);
         } else {
           FD_SET((*i)->getFd(), &rfds);
+          if ((*i)->outStream().bufferUsage() > 0)
+            FD_SET((*i)->getFd(), &wfds);
           clients_connected++;
         }
       }
@@ -538,23 +543,24 @@ int main(int argc, char** argv)
       if (!clients_connected)
         sched.reset();
 
+      wait_ms = 0;
+
       if (sched.isRunning()) {
-        int wait_ms = sched.millisRemaining();
+        wait_ms = sched.millisRemaining();
         if (wait_ms > 500) {
           wait_ms = 500;
         }
-        tv.tv_usec = wait_ms * 1000;
-#ifdef DEBUG
-        // fprintf(stderr, "[%d]\t", wait_ms);
-#endif
-      } else {
-        tv.tv_usec = 100000;
       }
-      tv.tv_sec = 0;
+
+      soonestTimeout(&wait_ms, server.checkTimeouts());
+
+      tv.tv_sec = wait_ms / 1000;
+      tv.tv_usec = (wait_ms % 1000) * 1000;
 
       // Do the wait...
       sched.sleepStarted();
-      int n = select(FD_SETSIZE, &rfds, 0, 0, &tv);
+      int n = select(FD_SETSIZE, &rfds, &wfds, 0,
+                     wait_ms ? &tv : NULL);
       sched.sleepFinished();
 
       if (n < 0) {
@@ -573,6 +579,7 @@ int main(int argc, char** argv)
         if (FD_ISSET((*i)->getFd(), &rfds)) {
           Socket* sock = (*i)->accept();
           if (sock) {
+            sock->outStream().setBlocking(false);
             server.addSocket(sock);
           } else {
             vlog.status("Client connection rejected");
@@ -580,7 +587,6 @@ int main(int argc, char** argv)
         }
       }
 
-      Timer::checkTimeouts();
       server.checkTimeouts();
 
       // Client list could have been changed.
@@ -593,7 +599,9 @@ int main(int argc, char** argv)
       // Process events on existing VNC connections
       for (i = sockets.begin(); i != sockets.end(); i++) {
         if (FD_ISSET((*i)->getFd(), &rfds))
-          server.processSocketEvent(*i);
+          server.processSocketReadEvent(*i);
+        if (FD_ISSET((*i)->getFd(), &wfds))
+          server.processSocketWriteEvent(*i);
       }
 
       if (desktop.isRunning() && sched.goodTimeToPoll()) {
diff --git a/unix/xserver/hw/vnc/XserverDesktop.cc b/unix/xserver/hw/vnc/XserverDesktop.cc
index 9b91d9a4..f1c9b747 100644
--- a/unix/xserver/hw/vnc/XserverDesktop.cc
+++ b/unix/xserver/hw/vnc/XserverDesktop.cc
@@ -495,7 +495,7 @@ void XserverDesktop::readWakeupHandler(fd_set* fds, int nfds)
         int fd = (*i)->getFd();
         if (FD_ISSET(fd, fds)) {
           FD_CLR(fd, fds);
-          server->processSocketEvent(*i);
+          server->processSocketReadEvent(*i);
         }
       }
 
@@ -505,7 +505,7 @@ void XserverDesktop::readWakeupHandler(fd_set* fds, int nfds)
           int fd = (*i)->getFd();
           if (FD_ISSET(fd, fds)) {
             FD_CLR(fd, fds);
-            httpServer->processSocketEvent(*i);
+            httpServer->processSocketReadEvent(*i);
           }
         }
       }
@@ -581,7 +581,7 @@ void XserverDesktop::writeWakeupHandler(fd_set* fds, int nfds)
       int fd = (*i)->getFd();
       if (FD_ISSET(fd, fds)) {
         FD_CLR(fd, fds);
-        (*i)->outStream().flush();
+        server->processSocketWriteEvent(*i);
       }
     }
 
@@ -591,7 +591,7 @@ void XserverDesktop::writeWakeupHandler(fd_set* fds, int nfds)
         int fd = (*i)->getFd();
         if (FD_ISSET(fd, fds)) {
           FD_CLR(fd, fds);
-          (*i)->outStream().flush();
+          httpServer->processSocketWriteEvent(*i);
         }
       }
     }
diff --git a/win/rfb_win32/SocketManager.cxx b/win/rfb_win32/SocketManager.cxx
index b073b8fb..5b211a0d 100644
--- a/win/rfb_win32/SocketManager.cxx
+++ b/win/rfb_win32/SocketManager.cxx
@@ -193,7 +193,7 @@ void SocketManager::processEvent(HANDLE event) {
       WSAResetEvent(event);
 
       // Call the socket server to process the event
-      ci.server->processSocketEvent(ci.sock);
+      ci.server->processSocketReadEvent(ci.sock);
       if (ci.sock->isShutdown()) {
         remSocket(ci.sock);
         return;