Move all stream-related functions within SpiceStream class

This incidentally fixes a race condition processing X events,
where we could possibly start sending cursor events to the
stream before it was actually open.

Signed-off-by: Christophe de Dinechin <dinechin@redhat.com>

1 files changed, 50 insertions, 38 deletions

diff --git a/src/spice-streaming-agent.cpp b/src/spice-streaming-agent.cpp
index 97581c4..9eabb68 100644
--- a/src/spice-streaming-agent.cpp
+++ b/src/spice-streaming-agent.cpp
@@ -52,31 +52,46 @@ struct SpiceStreamDataMessage
     StreamMsgData msg;
 };
 
-struct Stream
+class SpiceStream
 {
-    Stream(const char *name, int &fd): fd(fd)
+public:
+    SpiceStream(const char *name): streamfd(open(name, O_RDWR))
     {
-        fd = open(name, O_RDWR);
-        if (fd < 0)
+        if (streamfd < 0)
             throw std::runtime_error("failed to open streaming device");
     }
-    ~Stream()
+    ~SpiceStream()
     {
-        if (fd >= 0)
-            close(fd);
-        fd = -1;
+        close(streamfd);
     }
-    int &fd;
+    
+public:
+    bool have_something_to_read(int *pfd, int timeout);
+    int read_command_from_stdin(void);
+    int read_command_from_device(void);
+    int read_command(bool blocking);
+    int send_format(unsigned w, unsigned h, unsigned c);
+    int send_frame(const void *buf, const unsigned size);
+    void send_cursor(unsigned width, unsigned height, int hotspot_x, int hotspot_y,
+                     std::function<void(uint32_t *)> fill_cursor);
+
+private:
+    size_t write_all(const void *buf, const size_t len);
+
+    SpiceStream(const SpiceStream &) = delete;
+    SpiceStream &operator=(const SpiceStream &) = delete;
+
+private:
+    int streamfd;
 };
 
 static int streaming_requested;
 static bool quit;
-static int streamfd = -1;
 static bool stdin_ok;
 static int log_binary = 0;
 static std::mutex stream_mtx;
 
-static int have_something_to_read(int *pfd, int timeout)
+bool SpiceStream::have_something_to_read(int *pfd, int timeout)
 {
     int nfds;
     struct pollfd pollfds[2] = {
@@ -98,7 +113,7 @@ static int have_something_to_read(int *pfd, int timeout)
     return *pfd != -1;
 }
 
-static int read_command_from_stdin(void)
+int SpiceStream::read_command_from_stdin(void)
 {
     char buffer[64], *p, *save = NULL;
 
@@ -122,7 +137,7 @@ static int read_command_from_stdin(void)
     return 1;
 }
 
-static int read_command_from_device(void)
+int SpiceStream::read_command_from_device(void)
 {
     StreamDevHeader hdr;
     uint8_t msg[64];
@@ -164,7 +179,7 @@ static int read_command_from_device(void)
     return 1;
 }
 
-static int read_command(bool blocking)
+int SpiceStream::read_command(bool blocking)
 {
     int fd, n=1;
     int timeout = blocking?-1:0;
@@ -186,12 +201,11 @@ static int read_command(bool blocking)
     return n;
 }
 
-static size_t
-write_all(int fd, const void *buf, const size_t len)
+size_t SpiceStream::write_all(const void *buf, const size_t len)
 {
     size_t written = 0;
     while (written < len) {
-        int l = write(fd, (const char *) buf + written, len - written);
+        int l = write(streamfd, (const char *) buf + written, len - written);
         if (l < 0 && errno == EINTR) {
             continue;
         }
@@ -205,7 +219,7 @@ write_all(int fd, const void *buf, const size_t len)
     return written;
 }
 
-static int spice_stream_send_format(unsigned w, unsigned h, unsigned c)
+int SpiceStream::send_format(unsigned w, unsigned h, unsigned c)
 {
 
     SpiceStreamFormatMessage msg;
@@ -220,13 +234,13 @@ static int spice_stream_send_format(unsigned w, unsigned h, unsigned c)
     msg.msg.codec = c;
     syslog(LOG_DEBUG, "writing format\n");
     std::lock_guard<std::mutex> stream_guard(stream_mtx);
-    if (write_all(streamfd, &msg, msgsize) != msgsize) {
+    if (write_all(&msg, msgsize) != msgsize) {
         return EXIT_FAILURE;
     }
     return EXIT_SUCCESS;
 }
 
-static int spice_stream_send_frame(const void *buf, const unsigned size)
+int SpiceStream::send_frame(const void *buf, const unsigned size)
 {
     SpiceStreamDataMessage msg;
     const size_t msgsize = sizeof(msg);
@@ -237,7 +251,7 @@ static int spice_stream_send_frame(const void *buf, const unsigned size)
     msg.hdr.type = STREAM_TYPE_DATA;
     msg.hdr.size = size; /* includes only the body? */
     std::lock_guard<std::mutex> stream_guard(stream_mtx);
-    n = write_all(streamfd, &msg, msgsize);
+    n = write_all(&msg, msgsize);
     syslog(LOG_DEBUG,
            "wrote %ld bytes of header of data msg with frame of size %u bytes\n",
            n, msg.hdr.size);
@@ -246,7 +260,7 @@ static int spice_stream_send_frame(const void *buf, const unsigned size)
                n, msgsize);
         return EXIT_FAILURE;
     }
-    n = write_all(streamfd, buf, size);
+    n = write_all(buf, size);
     syslog(LOG_DEBUG, "wrote data msg body of size %ld\n", n);
     if (n != size) {
         syslog(LOG_WARNING, "write_all header: wrote %ld expected %u\n",
@@ -302,9 +316,8 @@ static void usage(const char *progname)
     exit(1);
 }
 
-static void
-send_cursor(unsigned width, unsigned height, int hotspot_x, int hotspot_y,
-            std::function<void(uint32_t *)> fill_cursor)
+void SpiceStream::send_cursor(unsigned width, unsigned height, int hotspot_x, int hotspot_y,
+                              std::function<void(uint32_t *)> fill_cursor)
 {
     if (width >= STREAM_MSG_CURSOR_SET_MAX_WIDTH ||
         height >= STREAM_MSG_CURSOR_SET_MAX_HEIGHT)
@@ -334,10 +347,10 @@ send_cursor(unsigned width, unsigned height, int hotspot_x, int hotspot_y,
     fill_cursor(pixels);
 
     std::lock_guard<std::mutex> stream_guard(stream_mtx);
-    write_all(streamfd, msg.get(), cursor_size);
+    write_all(msg.get(), cursor_size);
 }
 
-static void cursor_changes(Display *display, int event_base)
+static void cursor_changes(Display *display, int event_base, SpiceStream *stream)
 {
     unsigned long last_serial = 0;
 
@@ -359,23 +372,25 @@ static void cursor_changes(Display *display, int event_base)
             for (unsigned i = 0; i < cursor->width * cursor->height; ++i)
                 pixels[i] = cursor->pixels[i];
         };
-        send_cursor(cursor->width, cursor->height, cursor->xhot, cursor->yhot, fill_cursor);
+        stream->send_cursor(cursor->width, cursor->height, cursor->xhot, cursor->yhot, fill_cursor);
     }
 }
 
 static void
-do_capture(const char *streamport, FILE *f_log)
+do_capture(const char *streamport, FILE *f_log, Display *display, int event_base)
 {
     std::unique_ptr<FrameCapture> capture(agent.GetBestFrameCapture());
     if (!capture)
         throw std::runtime_error("cannot find a suitable capture system");
 
-    Stream stream(streamport, streamfd);
+    SpiceStream stream(streamport);
+    std::thread cursor_th(cursor_changes, display, event_base, &stream);
+    cursor_th.detach();
 
     unsigned int frame_count = 0;
     while (! quit) {
         while (!quit && !streaming_requested) {
-            if (read_command(true) < 0) {
+            if (stream.read_command(true) < 0) {
                 syslog(LOG_ERR, "FAILED to read command\n");
                 return;
             }
@@ -410,7 +425,7 @@ do_capture(const char *streamport, FILE *f_log)
 
                 syslog(LOG_DEBUG, "wXh %uX%u  codec=%u\n", width, height, codec);
 
-                if (spice_stream_send_format(width, height, codec) == EXIT_FAILURE)
+                if (stream.send_format(width, height, codec) == EXIT_FAILURE)
                     throw std::runtime_error("FAILED to send format message");
             }
             if (f_log) {
@@ -421,12 +436,12 @@ do_capture(const char *streamport, FILE *f_log)
                     hexdump(frame.buffer, frame.buffer_size, f_log);
                 }
             }
-            if (spice_stream_send_frame(frame.buffer, frame.buffer_size) == EXIT_FAILURE) {
+            if (stream.send_frame(frame.buffer, frame.buffer_size) == EXIT_FAILURE) {
                 syslog(LOG_ERR, "FAILED to send a frame\n");
                 break;
             }
             //usleep(1);
-            if (read_command(false) < 0) {
+            if (stream.read_command(false) < 0) {
                 syslog(LOG_ERR, "FAILED to read command\n");
                 return;
             }
@@ -520,12 +535,9 @@ int main(int argc, char* argv[])
     Window rootwindow = DefaultRootWindow(display);
     XFixesSelectCursorInput(display, rootwindow, XFixesDisplayCursorNotifyMask);
 
-    std::thread cursor_th(cursor_changes, display, event_base);
-    cursor_th.detach();
-
     int ret = EXIT_SUCCESS;
     try {
-        do_capture(streamport, f_log);
+        do_capture(streamport, f_log, display, event_base);
     }
     catch (std::runtime_error &err) {
         syslog(LOG_ERR, "%s\n", err.what());