[spice-streaming-agent,v3,5/5] Move all stream-related functions within SpiceStream class

Submitted by Christophe de Dinechin on Nov. 10, 2017, 1:38 p.m.

Details

Message ID 20171110133855.31117-6-christophe@dinechin.org
State New
Headers show
Series "Add a SpiceStream class for stream access" ( rev: 2 ) in Spice

Not browsing as part of any series.

Commit Message

Christophe de Dinechin Nov. 10, 2017, 1:38 p.m.
From: Christophe de Dinechin <dinechin@redhat.com>

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>
---
 src/spice-streaming-agent.cpp | 83 ++++++++++++++++++++++++-------------------
 1 file changed, 47 insertions(+), 36 deletions(-)

Patch hide | download patch | download mbox

diff --git a/src/spice-streaming-agent.cpp b/src/spice-streaming-agent.cpp
index a33f7cf..3a6e470 100644
--- a/src/spice-streaming-agent.cpp
+++ b/src/spice-streaming-agent.cpp
@@ -51,31 +51,44 @@  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;
+    if (streamfd >= 0)
+      close(streamfd);
+    streamfd = -1;
   }
-  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(const XFixesCursorImage &image);  
+
+private:
+  size_t write_all(const void *buf, const size_t len);
+
+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] = {
@@ -97,7 +110,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;
 
@@ -121,7 +134,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];
@@ -163,7 +176,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;
@@ -185,12 +198,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;
         }
@@ -204,7 +216,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;
@@ -219,13 +231,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);
@@ -236,7 +248,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);
@@ -245,7 +257,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",
@@ -304,7 +316,7 @@  static void usage(const char *progname)
     exit(1);
 }
 
-static void send_cursor(const XFixesCursorImage &image)
+void SpiceStream::send_cursor(const XFixesCursorImage &image)
 {
     if (image.width >= 1024 || image.height >= 1024)
         return;
@@ -334,10 +346,10 @@  static void send_cursor(const XFixesCursorImage &image)
         pixels[i] = image.pixels[i];
 
     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;
 
@@ -355,23 +367,25 @@  static void cursor_changes(Display *display, int event_base)
             continue;
 
         last_serial = cursor->cursor_serial;
-        send_cursor(*cursor);
+        stream->send_cursor(*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(1) < 0) {
+            if (stream.read_command(1) < 0) {
                 syslog(LOG_ERR, "FAILED to read command\n");
                 return;
             }
@@ -406,7 +420,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) {
@@ -417,12 +431,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(0) < 0) {
+            if (stream.read_command(0) < 0) {
                 syslog(LOG_ERR, "FAILED to read command\n");
                 return;
             }
@@ -516,12 +530,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());