[Spice-devel] [PATCH spice-streaming-agent 8/9] Encapsulate the stream port fd and locking
Frediano Ziglio
fziglio at redhat.com
Thu May 3 11:35:21 UTC 2018
> Wrap the streaming virtio port along with the mutex to lock it in a
> class. Pass the class temporarily around to functions that need it until
> the functions too are consolidated into the class.
>
> Signed-off-by: Lukáš Hrázký <lhrazky at redhat.com>
The mutex is supposed to avoid situations like:
1- data header
2- cursor header
3- data payload
4- cursor payload
protecting single write instead of full messages allow these
situations.
To fix this I would either:
1- document that synchronization should be done outside the class
(either removing the mutex from class or using it)
2- add a lock/unlock;
3- add a StreamPortLocked class and move write/read to it. The locker
class will hold the lock and make sure read/write are done with
the mutex held. This also allows to make the mutex field private
(using friend class) and prevent usage or read/write without
the mutex held.
(I prefer 3 but all are good).
Frediano
> ---
> src/spice-streaming-agent.cpp | 104
> +++++++++++++++++-------------------------
> src/stream-port.cpp | 25 ++++++++++
> src/stream-port.hpp | 14 ++++++
> 3 files changed, 81 insertions(+), 62 deletions(-)
>
> diff --git a/src/spice-streaming-agent.cpp b/src/spice-streaming-agent.cpp
> index 692f067..2fdd02f 100644
> --- a/src/spice-streaming-agent.cpp
> +++ b/src/spice-streaming-agent.cpp
> @@ -32,7 +32,6 @@
> #include <exception>
> #include <stdexcept>
> #include <memory>
> -#include <mutex>
> #include <thread>
> #include <vector>
> #include <string>
> @@ -60,12 +59,10 @@ static bool streaming_requested = false;
> static bool quit_requested = false;
> static bool log_binary = false;
> static std::set<SpiceVideoCodecType> client_codecs;
> -static int streamfd = -1;
> -static std::mutex stream_mtx;
>
> -static int have_something_to_read(int timeout)
> +static int have_something_to_read(StreamPort &stream_port, int timeout)
> {
> - struct pollfd pollfd = {streamfd, POLLIN, 0};
> + struct pollfd pollfd = {stream_port.fd, POLLIN, 0};
>
> if (poll(&pollfd, 1, timeout) < 0) {
> syslog(LOG_ERR, "poll FAILED\n");
> @@ -79,7 +76,7 @@ static int have_something_to_read(int timeout)
> return 0;
> }
>
> -static void handle_stream_start_stop(uint32_t len)
> +static void handle_stream_start_stop(StreamPort &stream_port, uint32_t len)
> {
> uint8_t msg[256];
>
> @@ -88,7 +85,7 @@ static void handle_stream_start_stop(uint32_t len)
> "(longer than " +
> std::to_string(sizeof(msg)) + ")");
> }
>
> - read_all(streamfd, msg, len);
> + stream_port.read(msg, len);
> streaming_requested = (msg[0] != 0); /* num_codecs */
> syslog(LOG_INFO, "GOT START_STOP message -- request to %s streaming\n",
> streaming_requested ? "START" : "STOP");
> @@ -98,7 +95,7 @@ static void handle_stream_start_stop(uint32_t len)
> }
> }
>
> -static void handle_stream_capabilities(uint32_t len)
> +static void handle_stream_capabilities(StreamPort &stream_port, uint32_t
> len)
> {
> uint8_t caps[STREAM_MSG_CAPABILITIES_MAX_BYTES];
>
> @@ -106,7 +103,7 @@ static void handle_stream_capabilities(uint32_t len)
> throw std::runtime_error("capability message too long");
> }
>
> - read_all(streamfd, caps, len);
> + stream_port.read(caps, len);
> // we currently do not support extensions so just reply so
> StreamDevHeader hdr = {
> STREAM_DEVICE_PROTOCOL,
> @@ -114,10 +111,10 @@ static void handle_stream_capabilities(uint32_t len)
> STREAM_TYPE_CAPABILITIES,
> 0
> };
> - write_all(streamfd, &hdr, sizeof(hdr));
> + stream_port.write(&hdr, sizeof(hdr));
> }
>
> -static void handle_stream_error(size_t len)
> +static void handle_stream_error(StreamPort &stream_port, size_t len)
> {
> if (len < sizeof(StreamMsgNotifyError)) {
> throw std::runtime_error("Received NotifyError message size " +
> std::to_string(len) +
> @@ -131,7 +128,7 @@ static void handle_stream_error(size_t len)
>
> size_t len_to_read = std::min(len, sizeof(msg) - 1);
>
> - read_all(streamfd, &msg, len_to_read);
> + stream_port.read(&msg, len_to_read);
> msg.msg[len_to_read - sizeof(StreamMsgNotifyError)] = '\0';
>
> syslog(LOG_ERR, "Received NotifyError message from the server: %d -
> %s\n",
> @@ -143,13 +140,11 @@ static void handle_stream_error(size_t len)
> }
> }
>
> -static void read_command_from_device(void)
> +static void read_command_from_device(StreamPort &stream_port)
> {
> StreamDevHeader hdr;
>
> - std::lock_guard<std::mutex> stream_guard(stream_mtx);
> -
> - read_all(streamfd, &hdr, sizeof(hdr));
> + stream_port.read(&hdr, sizeof(hdr));
>
> if (hdr.protocol_version != STREAM_DEVICE_PROTOCOL) {
> throw std::runtime_error("BAD VERSION " +
> std::to_string(hdr.protocol_version) +
> @@ -158,34 +153,34 @@ static void read_command_from_device(void)
>
> switch (hdr.type) {
> case STREAM_TYPE_CAPABILITIES:
> - return handle_stream_capabilities(hdr.size);
> + return handle_stream_capabilities(stream_port, hdr.size);
> case STREAM_TYPE_NOTIFY_ERROR:
> - return handle_stream_error(hdr.size);
> + return handle_stream_error(stream_port, hdr.size);
> case STREAM_TYPE_START_STOP:
> - return handle_stream_start_stop(hdr.size);
> + return handle_stream_start_stop(stream_port, hdr.size);
> }
> throw std::runtime_error("UNKNOWN msg of type " +
> std::to_string(hdr.type));
> }
>
> -static int read_command(bool blocking)
> +static int read_command(StreamPort &stream_port, bool blocking)
> {
> int timeout = blocking?-1:0;
> while (!quit_requested) {
> - if (!have_something_to_read(timeout)) {
> + if (!have_something_to_read(stream_port, timeout)) {
> if (!blocking) {
> return 0;
> }
> sleep(1);
> continue;
> }
> - read_command_from_device();
> + read_command_from_device(stream_port);
> break;
> }
>
> return 1;
> }
>
> -static void spice_stream_send_format(unsigned w, unsigned h, unsigned c)
> +static void spice_stream_send_format(StreamPort &stream_port, unsigned w,
> unsigned h, unsigned c)
> {
>
> SpiceStreamFormatMessage msg;
> @@ -199,11 +194,10 @@ static void spice_stream_send_format(unsigned w,
> unsigned h, unsigned c)
> msg.msg.height = h;
> msg.msg.codec = c;
> syslog(LOG_DEBUG, "writing format\n");
> - std::lock_guard<std::mutex> stream_guard(stream_mtx);
> - write_all(streamfd, &msg, msgsize);
> + stream_port.write(&msg, msgsize);
> }
>
> -static void spice_stream_send_frame(const void *buf, const unsigned size)
> +static void spice_stream_send_frame(StreamPort &stream_port, const void
> *buf, const unsigned size)
> {
> SpiceStreamDataMessage msg;
> const size_t msgsize = sizeof(msg);
> @@ -212,9 +206,8 @@ static void spice_stream_send_frame(const void *buf,
> const unsigned size)
> msg.hdr.protocol_version = STREAM_DEVICE_PROTOCOL;
> msg.hdr.type = STREAM_TYPE_DATA;
> msg.hdr.size = size; /* includes only the body? */
> - std::lock_guard<std::mutex> stream_guard(stream_mtx);
> - write_all(streamfd, &msg, msgsize);
> - write_all(streamfd, buf, size);
> + stream_port.write(&msg, msgsize);
> + stream_port.write(buf, size);
>
> syslog(LOG_DEBUG, "Sent a frame of size %u\n", size);
> }
> @@ -264,7 +257,7 @@ static void usage(const char *progname)
> }
>
> static void
> -send_cursor(unsigned width, unsigned height, int hotspot_x, int hotspot_y,
> +send_cursor(StreamPort &stream_port, 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) {
> @@ -294,11 +287,10 @@ send_cursor(unsigned width, unsigned height, int
> hotspot_x, int hotspot_y,
> uint32_t *pixels = reinterpret_cast<uint32_t *>(cursor_msg.data);
> fill_cursor(pixels);
>
> - std::lock_guard<std::mutex> stream_guard(stream_mtx);
> - write_all(streamfd, msg.get(), cursor_size);
> + stream_port.write(msg.get(), cursor_size);
> }
>
> -static void cursor_changes(Display *display, int event_base)
> +static void cursor_changes(StreamPort *stream_port, Display *display, int
> event_base)
> {
> unsigned long last_serial = 0;
>
> @@ -323,26 +315,18 @@ 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);
> + send_cursor(*stream_port, cursor->width, cursor->height,
> cursor->xhot, cursor->yhot, fill_cursor);
> }
> }
>
> static void
> -do_capture(const char *streamport, FILE *f_log)
> +do_capture(StreamPort &stream_port, FILE *f_log)
> {
> - streamfd = open(streamport, O_RDWR | O_NONBLOCK);
> - if (streamfd < 0) {
> - throw std::runtime_error("failed to open the streaming device (" +
> - std::string(streamport) + "): "
> - + strerror(errno));
> - }
> -
> unsigned int frame_count = 0;
> while (!quit_requested) {
> while (!quit_requested && !streaming_requested) {
> - if (read_command(true) < 0) {
> - syslog(LOG_ERR, "FAILED to read command\n");
> - goto done;
> + if (read_command(stream_port, true) < 0) {
> + throw std::runtime_error("FAILED to read command");
> }
> }
>
> @@ -385,7 +369,7 @@ do_capture(const char *streamport, FILE *f_log)
>
> syslog(LOG_DEBUG, "wXh %uX%u codec=%u\n", width, height,
> codec);
>
> - spice_stream_send_format(width, height, codec);
> + spice_stream_send_format(stream_port, width, height, codec);
> }
> if (f_log) {
> if (log_binary) {
> @@ -398,32 +382,25 @@ do_capture(const char *streamport, FILE *f_log)
> }
>
> try {
> - spice_stream_send_frame(frame.buffer, frame.buffer_size);
> + spice_stream_send_frame(stream_port, frame.buffer,
> frame.buffer_size);
> } catch (const WriteError& e) {
> syslog(e);
> break;
> }
>
> //usleep(1);
> - if (read_command(false) < 0) {
> - syslog(LOG_ERR, "FAILED to read command\n");
> - goto done;
> + if (read_command(stream_port, false) < 0) {
> + throw std::runtime_error("FAILED to read command");
> }
> }
> }
> -
> -done:
> - if (streamfd >= 0) {
> - close(streamfd);
> - streamfd = -1;
> - }
> }
>
> #define arg_error(...) syslog(LOG_ERR, ## __VA_ARGS__);
>
> int main(int argc, char* argv[])
> {
> - const char *streamport = "/dev/virtio-ports/org.spice-space.stream.0";
> + const char *stream_port_name =
> "/dev/virtio-ports/org.spice-space.stream.0";
> int opt;
> const char *log_filename = NULL;
> int logmask = LOG_UPTO(LOG_WARNING);
> @@ -454,7 +431,7 @@ int main(int argc, char* argv[])
> pluginsdir = optarg;
> break;
> case 'p':
> - streamport = optarg;
> + stream_port_name = optarg;
> break;
> case 'c': {
> char *p = strchr(optarg, '=');
> @@ -512,12 +489,15 @@ 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);
> + StreamPort stream_port(stream_port_name);
> +
> + std::thread cursor_th(cursor_changes, &stream_port, display,
> event_base);
> + cursor_th.detach();
> +
> + do_capture(stream_port, f_log);
> }
> catch (std::exception &err) {
> syslog(LOG_ERR, "%s\n", err.what());
> diff --git a/src/stream-port.cpp b/src/stream-port.cpp
> index ee85179..3cd4753 100644
> --- a/src/stream-port.cpp
> +++ b/src/stream-port.cpp
> @@ -8,6 +8,7 @@
> #include "error.hpp"
>
> #include <errno.h>
> +#include <fcntl.h>
> #include <poll.h>
> #include <string.h>
> #include <syslog.h>
> @@ -18,6 +19,30 @@
> namespace spice {
> namespace streaming_agent {
>
> +StreamPort::StreamPort(const std::string &port_name) :
> fd(open(port_name.c_str(), O_RDWR | O_NONBLOCK))
> +{
> + if (fd < 0) {
> + throw IOError("Failed to open the streaming device \"" + port_name +
> "\"", errno);
> + }
> +}
> +
> +StreamPort::~StreamPort()
> +{
> + close(fd);
> +}
> +
> +void StreamPort::read(void *buf, size_t len)
> +{
> + std::lock_guard<std::mutex> guard(mutex);
> + read_all(fd, buf, len);
> +}
> +
> +void StreamPort::write(const void *buf, size_t len)
> +{
> + std::lock_guard<std::mutex> guard(mutex);
> + write_all(fd, buf, len);
> +}
> +
> void read_all(int fd, void *buf, size_t len)
> {
> while (len > 0) {
> diff --git a/src/stream-port.hpp b/src/stream-port.hpp
> index b2d8352..9187cf5 100644
> --- a/src/stream-port.hpp
> +++ b/src/stream-port.hpp
> @@ -8,11 +8,25 @@
> #define SPICE_STREAMING_AGENT_STREAM_PORT_HPP
>
> #include <cstddef>
> +#include <string>
> +#include <mutex>
>
>
> namespace spice {
> namespace streaming_agent {
>
> +class StreamPort {
> +public:
> + StreamPort(const std::string &port_name);
> + ~StreamPort();
> +
> + void read(void *buf, size_t len);
> + void write(const void *buf, size_t len);
> +
> + int fd;
> + std::mutex mutex;
> +};
> +
> void read_all(int fd, void *buf, size_t len);
> void write_all(int fd, const void *buf, size_t len);
>
More information about the Spice-devel
mailing list