Async stream WIP

Author: Stefan Eilemann

apps/DesktopStreamer/Stream.cpp

@@ -193,8 +193,13 @@ class Stream::Impl
         deflectImage.compressionQuality = std::max(1, std::min(quality, 100));
         deflectImage.subsampling = subsamp;
 
-        if (!_stream.send(deflectImage) || !_stream.finishFrame())
-            return "Streaming failure, connection closed";
+        for (auto& future : _lastSend)
+            if (!future.get())
+                return "Streaming failure, connection closed";
+
+        _lastSend.clear();
+        _lastSend.push_back(_stream.send(deflectImage));
+        _lastSend.push_back(_stream.finishFrame());
         return std::string();
     }
 
@@ -293,6 +298,9 @@ class Stream::Impl
     EventQueue _events;
 #endif
 
+    using Futures = std::vector<deflect::Stream::Future>;
+    Futures _lastSend;
+
     QTimer _mouseActiveTimer;
     QPoint _mousePos;
 };

apps/SimpleStreamer/main.cpp

@@ -264,7 +264,7 @@ bool send(const Image& image, const deflect::View view)
                                          : deflect::COMPRESSION_OFF;
     deflectImage.compressionQuality = deflectCompressionQuality;
     deflectImage.view = view;
-    return deflectStream->send(deflectImage);
+    return deflectStream->send(deflectImage).get();
 }
 
 bool timeout(const float sec)

deflect/Stream.cpp

@@ -84,20 +84,22 @@ const std::string& Stream::getHost() const
     return _impl->socket.getHost();
 }
 
-bool Stream::send(const ImageWrapper& image)
+Stream::Future Stream::send(const ImageWrapper& image)
 {
     return _impl->send(image);
 }
 
-bool Stream::finishFrame()
+Stream::Future Stream::finishFrame()
 {
     return _impl->finishFrame();
 }
 
+#if 0
 Stream::Future Stream::asyncSend(const ImageWrapper& image)
 {
     return _impl->asyncSend(image);
 }
+#endif
 
 bool Stream::registerForEvents(const bool exclusive)
 {

deflect/Stream.h

@@ -61,9 +61,9 @@ class StreamPrivate;
 /**
  * Stream visual data to a deflect::Server.
  *
- * A Stream can be subdivided into one or more images. This allows to have
- * different applications each responsible for sending one part of the global
- * image.
+ * A Stream can be subdivided into one or more images and eye passes. This
+ * allows to have different applications each responsible for sending one part
+ * of the global image.
  *
  * The methods in this class are reentrant (all instances are independant) but
  * are not thread-safe.
@@ -123,49 +123,31 @@ class Stream
     /** Future signaling success of asyncSend(). @version 1.5 */
     using Future = std::future<bool>;
 
-    /**
-     * Send an image and finish the frame asynchronously.
-     *
-     * The send (and the optional compression) and finishFrame() are executed in
-     * a different thread. The result of this operation can be obtained by the
-     * returned future object.
-     *
-     * @param image The image to send. Note that the image is not copied, so the
-     *              referenced must remain valid until the send is finished
-     * @return true if the image data could be sent, false otherwise
-     * @see send()
-     * @version 1.1
-     */
-    DEFLECT_API Future asyncSend(const ImageWrapper& image);
-    //@}
+    DEFLECT_API Future asyncSend(const ImageWrapper& image); //<! @deprecated
 
-    /** @name Synchronous send API */
-    //@{
     /**
-     * Send an image synchronously.
+     * Send an image asynchronously.
      *
-     * @note A call to send() while an asyncSend() is pending is undefined.
      * @param image The image to send
      * @return true if the image data could be sent, false otherwise
      * @version 1.0
      * @sa finishFrame()
      */
-    DEFLECT_API bool send(const ImageWrapper& image);
+    DEFLECT_API Future send(const ImageWrapper& image);
 
     /**
-     * Notify that all the images for this frame have been sent.
+     * Asynchronously notify that all the images for this frame have been sent.
      *
      * This method must be called everytime this Stream instance has finished
      * sending its image(s) for the current frame. The receiver will display
      * the images once all the senders which use the same identifier have
-     * finished a frame.
+     * finished a frame. This is only to be called once per frame, even for
+     * stereo rendering.
      *
-     * @note A call to finishFrame() while an asyncSend() is pending is
-     *       undefined.
-     * @see send()
+     * @sa send()
      * @version 1.0
      */
-    DEFLECT_API bool finishFrame();
+    DEFLECT_API Future finishFrame();
     //@}
 
     /**
@@ -240,17 +222,19 @@ class Stream
     DEFLECT_API Event getEvent();
 
     /**
-     * Send size hints to the stream host to indicate sizes that should be
-     * respected by resize operations on the host side.
+     * Send size hints to the stream server to indicate sizes that should be
+     * respected by resize operations on the server side.
      *
-     * @param hints the new size hints for the host
+     * @note do not use while asynchronous send operations are pending.
+     * @param hints the new size hints for the server
      * @version 1.2
      */
     DEFLECT_API void sendSizeHints(const SizeHints& hints);
 
     /**
      * Send data to the Server.
      *
+     * @note do not use while asynchronous send operations are pending.
      * @param data the pointer to the data buffer.
      * @param count the number of bytes to send.
      * @return true if the data could be sent, false otherwise

deflect/StreamPrivate.cpp

@@ -54,7 +54,6 @@
 
 namespace
 {
-const unsigned int SEGMENT_SIZE = 512;
 const char* STREAM_ID_ENV_VAR = "DEFLECT_ID";
 const char* STREAM_HOST_ENV_VAR = "DEFLECT_HOST";
 }
@@ -92,9 +91,8 @@ StreamPrivate::StreamPrivate(const std::string& id_, const std::string& host,
     : id(_getStreamId(id_))
     , socket(_getStreamHost(host), port)
     , registeredForEvents(false)
+    , _sendWorker(*this)
 {
-    imageSegmenter.setNominalSegmentDimensions(SEGMENT_SIZE, SEGMENT_SIZE);
-
     if (!socket.isConnected())
         return;
 
@@ -108,8 +106,7 @@ StreamPrivate::StreamPrivate(const std::string& id_, const std::string& host,
 
 StreamPrivate::~StreamPrivate()
 {
-    _sendWorker.reset();
-
+    _sendWorker.stop();
     if (!socket.isConnected())
         return;
 
@@ -129,37 +126,19 @@ void StreamPrivate::sendClose()
     socket.send(mh, QByteArray());
 }
 
-bool StreamPrivate::send(const ImageWrapper& image)
+Stream::Future StreamPrivate::send(const ImageWrapper& image)
 {
-    if (image.compressionPolicy != COMPRESSION_ON && image.pixelFormat != RGBA)
-    {
-        std::cerr << "Currently, RAW images can only be sent in RGBA format. "
-                     "Other formats support remain to be implemented."
-                  << std::endl;
-        return false;
-    }
-
-    if (!sendImageView(image.view))
-        return false;
-
-    const auto sendFunc = std::bind(&StreamPrivate::sendPixelStreamSegment,
-                                    this, std::placeholders::_1);
-    return imageSegmenter.generate(image, sendFunc);
+    return _sendWorker.enqueueImage(image, false);
 }
 
 Stream::Future StreamPrivate::asyncSend(const ImageWrapper& image)
 {
-    if (!_sendWorker)
-        _sendWorker.reset(new StreamSendWorker(*this));
-
-    return _sendWorker->enqueueImage(image);
+    return _sendWorker.enqueueImage(image, true);
 }
 
-bool StreamPrivate::finishFrame()
+Stream::Future StreamPrivate::finishFrame()
 {
-    // Open a window for the PixelStream
-    const MessageHeader mh(MESSAGE_TYPE_PIXELSTREAM_FINISH_FRAME, 0, id);
-    return socket.send(mh, QByteArray());
+    return _sendWorker.enqueueFinish();
 }
 
 bool StreamPrivate::sendImageView(const View view)
@@ -191,6 +170,11 @@ bool StreamPrivate::sendPixelStreamSegment(const Segment& segment)
     return socket.send(mh, message);
 }
 
+bool StreamPrivate::sendFinish()
+{
+    return socket.send({MESSAGE_TYPE_PIXELSTREAM_FINISH_FRAME, 0, id}, {});
+}
+
 bool StreamPrivate::sendSizeHints(const SizeHints& hints)
 {
     const MessageHeader mh(MESSAGE_TYPE_SIZE_HINTS, sizeof(hints), id);

deflect/StreamPrivate.h

@@ -45,22 +45,18 @@
 #include <deflect/api.h>
 
 #include "Event.h"
-#include "ImageSegmenter.h"
 #include "MessageHeader.h"
-#include "Socket.h" // member
-#include "Stream.h" // Stream::Future
+#include "Socket.h"           // member
+#include "Stream.h"           // Stream::Future
+#include "StreamSendWorker.h" // member
 
 #include <functional>
 #include <memory>
 #include <string>
 
 namespace deflect
 {
-class StreamSendWorker;
-
-/**
- * Private implementation for the Stream class.
- */
+/** Private implementation for the Stream class. */
 class StreamPrivate
 {
 public:
@@ -89,17 +85,11 @@ class StreamPrivate
      */
     bool close();
 
-    /** @sa Stream::send */
-    bool send(const ImageWrapper& image);
-
-    /** @sa Stream::asyncSend */
+    /** @deprecated async-send-and-finish for mono images. */
     Stream::Future asyncSend(const ImageWrapper& image);
 
-    /** @sa Stream::finishFrame */
-    bool finishFrame();
-
-    /** Send the view for the image to be sent with sendPixelStreamSegment. */
-    bool sendImageView(View view);
+    /** @sa Stream::send */
+    Stream::Future send(const ImageWrapper& image);
 
     /**
      * Send a Segment through the Stream.
@@ -108,6 +98,9 @@ class StreamPrivate
      */
     DEFLECT_API bool sendPixelStreamSegment(const Segment& segment);
 
+    /** @sa Stream::finishFrame */
+    Stream::Future finishFrame();
+
     /** @sa Stream::sendSizeHints */
     bool sendSizeHints(const SizeHints& hints);
 
@@ -120,16 +113,20 @@ class StreamPrivate
     /** The communication socket instance */
     Socket socket;
 
-    /** The image segmenter */
-    ImageSegmenter imageSegmenter;
-
     /** Has a successful event registration reply been received */
     bool registeredForEvents;
 
     std::function<void()> disconnectedCallback;
 
 private:
-    std::unique_ptr<StreamSendWorker> _sendWorker;
+    friend class StreamSendWorker;
+
+    /** Send the view for the image to be sent with sendPixelStreamSegment. */
+    bool sendImageView(View view);
+
+    bool sendFinish(); //<! Send the finish frame message
+
+    StreamSendWorker _sendWorker;
 };
 }
 #endif