daemon: implement double-buffered SHM

Refs #70056

Change-Id: I914339024e3570c3ebf2d1772265d31f0a4e2d7f

src/media/video/shm_header.h

@@ -38,16 +38,21 @@
 
 #include <semaphore.h>
 
-struct SHMHeader {
-    sem_t notification;
-    sem_t mutex;
-
-    unsigned buffer_gen;
-    int buffer_size;
-    /* The header will be aligned on 16-byte boundaries */
-    char padding[8];
+// Implementation note: double-buffering
+// Shared memory is divided in two regions, each representing one frame.
+// First byte of each frame is warranted to by aligned on 16 bytes.
+// One region is marked readable: this region can be safely read.
+// The other region is writeable: only the producer can use it.
 
-    char data[];
+struct SHMHeader {
+    sem_t mutex;                // Lock it before any operations on following fields.
+    sem_t frameGenMutex;        // unlocked by producer when frameGen modified
+    unsigned frameGen;          // monotonically incremented when a producer changes readOffset
+    unsigned frameSize;         // size in bytes of 1 frame
+    unsigned mapSize;           // size to map if you need to see all data
+    unsigned readOffset;        // offset of readable frame in data
+    unsigned writeOffset;       // offset of writable frame in data
+    char data[];                // the whole shared memory
 };
 
 #endif

src/media/video/sinkclient.cpp

@@ -48,6 +48,8 @@
 #include "media_buffer.h"
 #include "logger.h"
 #include "noncopyable.h"
+#include "client/ring_signal.h"
+#include "dring/videomanager_interface.h"
 
 #include <sys/mman.h>
 #include <fcntl.h>
@@ -85,159 +87,179 @@ class SemGuardLock {
 class ShmHolder
 {
     public:
-        ShmHolder(const std::string& shm_name={});
+        ShmHolder(const std::string& name={});
         ~ShmHolder();
 
-        std::string openedName() const noexcept {
-            return opened_name_;
+        std::string name() const noexcept {
+            return openedName_;
         }
 
-        void render_frame(VideoFrame& src);
+        void renderFrame(VideoFrame& src) noexcept;
 
     private:
-        bool resize_area(std::size_t desired_length);
-        void alloc_area(std::size_t desired_length) noexcept;
+        bool resizeArea(std::size_t desired_length) noexcept;
+        char* getShmAreaDataPtr() noexcept;
 
-        std::string shm_name_;
-        std::string opened_name_;
-        std::size_t shm_area_len_ {0};
-        SHMHeader* shm_area_ {static_cast<SHMHeader*>(MAP_FAILED)};
+        void unMapShmArea() noexcept {
+            if (area_ != MAP_FAILED and ::munmap(area_, areaSize_) < 0) {
+                RING_ERR("ShmHolder[%s]: munmap(%u) failed with errno %d",
+                         openedName_.c_str(), areaSize_, errno);
+            }
+        }
+
+        SHMHeader* area_ {static_cast<SHMHeader*>(MAP_FAILED)};
+        std::size_t areaSize_ {0};
+        std::string openedName_;
         int fd_ {-1};
 };
 
-void
-ShmHolder::alloc_area(std::size_t desired_length) noexcept
-{
-    shm_area_ = static_cast<SHMHeader*>(::mmap(nullptr, desired_length,
-                                               PROT_READ | PROT_WRITE,
-                                               MAP_SHARED, fd_, 0));
-}
-
-ShmHolder::ShmHolder(const std::string& shm_name) : shm_name_ {shm_name}
+ShmHolder::ShmHolder(const std::string& name)
 {
     static constexpr int flags = O_RDWR | O_CREAT | O_TRUNC | O_EXCL;
     static constexpr int perms = S_IRUSR | S_IWUSR;
 
-    if (not shm_name_.empty()) {
-        fd_ = ::shm_open(shm_name_.c_str(), flags, perms);
-        if (fd_ < 0) {
+    static auto shmFailedWithErrno = [this](const std::string& what) {
         std::ostringstream msg;
-            msg << "could not open shm area \""
-                << shm_name_.c_str()
-                << "\"";
+        msg << "ShmHolder[" << openedName_ << "]: "
+        << what << " failed, errno=" << errno;
         throw std::runtime_error {msg.str()};
-        }
+    };
+
+    if (not name.empty()) {
+        openedName_ = name;
+        fd_ = ::shm_open(openedName_.c_str(), flags, perms);
+        if (fd_ < 0)
+            shmFailedWithErrno("shm_open");
     } else {
         for (int i = 0; fd_ < 0; ++i) {
-            std::ostringstream name;
-            name << PACKAGE_NAME << "_shm_" << getpid() << "_" << i;
-            shm_name_ = name.str();
-            fd_ = ::shm_open(shm_name_.c_str(), flags, perms);
+            std::ostringstream tmpName;
+            tmpName << PACKAGE_NAME << "_shm_" << getpid() << "_" << i;
+            openedName_ = tmpName.str();
+            fd_ = ::shm_open(openedName_.c_str(), flags, perms);
             if (fd_ < 0 and errno != EEXIST)
-                throw std::runtime_error {"shm_open() failed"};
-        }
+                shmFailedWithErrno("shm_open");
         }
-
-    RING_DBG("Using name %s", shm_name_.c_str());
-    opened_name_ = shm_name_;
-
-    shm_area_len_ = sizeof(SHMHeader);
-
-    if (::ftruncate(fd_, shm_area_len_)) {
-        RING_ERR("Could not make shm area large enough for header");
-        strErr();
-        throw std::runtime_error {"could not make shm area large enough for header"};
     }
 
-    alloc_area(shm_area_len_);
+    // Set size enough for header only (no frame data)
+    if (!resizeArea(0))
+        shmFailedWithErrno("resizeArea");
 
-    if (shm_area_ == MAP_FAILED)
-        throw std::runtime_error {"could not map shm area, mmap failed"};
+    // Header fields initialization
+    std::memset(area_, 0, areaSize_);
 
-    std::memset(shm_area_, 0, shm_area_len_);
+    if (::sem_init(&area_->mutex, 1, 1) < 0)
+        shmFailedWithErrno("sem_init(mutex)");
 
-    if (::sem_init(&shm_area_->notification, 1, 0) != 0)
-        throw std::runtime_error {"sem_init: notification initialization failed"};
+    if (::sem_init(&area_->frameGenMutex, 1, 0) < 0)
+        shmFailedWithErrno("sem_init(frameGenMutex)");
 
-    if (::sem_init(&shm_area_->mutex, 1, 1) != 0)
-        throw std::runtime_error {"sem_init: mutex initialization failed"};
+    RING_DBG("ShmHolder: new holder '%s'", openedName_.c_str());
 }
 
 ShmHolder::~ShmHolder()
 {
-    if (fd_ >= 0 and ::close(fd_) == -1)
-        strErr();
+    if (fd_ < 0)
+        return;
 
-    if (not opened_name_.empty())
-        ::shm_unlink(opened_name_.c_str());
+    ::close(fd_);
+    ::shm_unlink(openedName_.c_str());
 
-    if (shm_area_ != MAP_FAILED) {
-        ::sem_post(&shm_area_->notification);
-        if (::munmap(shm_area_, shm_area_len_) < 0)
-            strErr();
-    }
+    if (area_ == MAP_FAILED)
+        return;
+
+    area_->frameSize = 0;
+    ::sem_post(&area_->frameGenMutex);
+    unMapShmArea();
 }
 
 bool
-ShmHolder::resize_area(size_t desired_length)
+ShmHolder::resizeArea(std::size_t frameSize) noexcept
 {
-    if (desired_length <= shm_area_len_)
+    // aligned on 16-byte boundary frameSize
+    frameSize = (frameSize + 15) & ~15;
+
+    if (area_ != MAP_FAILED and frameSize == area_->frameSize)
         return true;
 
-    if (::munmap(shm_area_, shm_area_len_)) {
-        RING_ERR("Could not unmap shared area");
-        strErr();
+    // full area size: +15 to take care of maximum padding size
+    const auto areaSize = sizeof(SHMHeader) + 2 * frameSize + 15;
+    RING_DBG("ShmHolder[%s]: new sizes: f=%u, a=%u", openedName_.c_str(),
+             frameSize, areaSize);
+
+    unMapShmArea();
+
+    if (::ftruncate(fd_, areaSize) < 0) {
+        RING_ERR("ShmHolder[%s]: ftruncate(%u) failed with errno %d",
+                 openedName_.c_str(), areaSize, errno);
         return false;
     }
 
-    if (::ftruncate(fd_, desired_length)) {
-        RING_ERR("Could not resize shared area");
-        strErr();
+    area_ = static_cast<SHMHeader*>(::mmap(nullptr, areaSize,
+                                           PROT_READ | PROT_WRITE,
+                                           MAP_SHARED, fd_, 0));
+
+    if (area_ == MAP_FAILED) {
+        areaSize_ = 0;
+        RING_ERR("ShmHolder[%s]: mmap(%u) failed with errno %d",
+                 openedName_.c_str(), areaSize, errno);
         return false;
     }
 
-    alloc_area(desired_length);
+    areaSize_ = areaSize;
 
-    if (shm_area_ == MAP_FAILED) {
-        shm_area_len_ = 0;
-        RING_ERR("Could not remap shared area");
-        return false;
+    if (frameSize) {
+        SemGuardLock lk {area_->mutex};
+
+        area_->frameSize = frameSize;
+        area_->mapSize = areaSize;
+
+        // Compute aligned IO pointers
+        // Note: we not using std::align as not implemented in 4.9
+        // https://gcc.gnu.org/bugzilla/show_bug.cgi?id=57350
+        auto p = reinterpret_cast<std::uintptr_t>(area_->data);
+        area_->writeOffset = ((p + 15) & ~15) - p;
+        area_->readOffset = area_->writeOffset + frameSize;
     }
 
-    shm_area_len_ = desired_length;
     return true;
 }
 
 void
-ShmHolder::render_frame(VideoFrame& src)
+ShmHolder::renderFrame(VideoFrame& src) noexcept
 {
-    VideoFrame dst;
-    VideoScaler scaler;
-
-    const int width = src.width();
-    const int height = src.height();
-    const int format = VIDEO_PIXFMT_BGRA;
-    const auto bytes = videoFrameSize(format, width, height);
-
-    if (!resize_area(sizeof(SHMHeader) + bytes)) {
-        RING_ERR("Could not resize area");
+    const auto width = src.width();
+    const auto height = src.height();
+    const auto format = VIDEO_PIXFMT_BGRA;
+    const auto frameSize = videoFrameSize(format, width, height);
+
+    if (!resizeArea(frameSize)) {
+        RING_ERR("ShmHolder[%s]: could not resize area",
+                 openedName_.c_str());
         return;
     }
 
-    SemGuardLock lk{shm_area_->mutex};
+    {
+        VideoFrame dst;
+        VideoScaler scaler;
 
-    dst.setFromMemory(shm_area_->data, format, width, height);
+        dst.setFromMemory(area_->data + area_->writeOffset, format, width, height);
         scaler.scale(src, dst);
+    }
 
-    shm_area_->buffer_size = bytes;
-    ++shm_area_->buffer_gen;
-    sem_post(&shm_area_->notification);
+    {
+        SemGuardLock lk {area_->mutex};
+
+        ++area_->frameGen;
+        std::swap(area_->readOffset, area_->writeOffset);
+        ::sem_post(&area_->frameGenMutex);
+    }
 }
 
 std::string
 SinkClient::openedName() const noexcept
 {
-    return shm_->openedName();
+    return shm_->name();
 }
 
 bool
@@ -247,7 +269,6 @@ SinkClient::start() noexcept
         try {
             shm_ = std::make_shared<ShmHolder>();
         } catch (const std::runtime_error& e) {
-            strErr();
             RING_ERR("SHMHolder ctor failure: %s", e.what());
         }
     }
@@ -308,7 +329,7 @@ SinkClient::update(Observable<std::shared_ptr<VideoFrame>>* /*obs*/,
 #endif
 
 #if HAVE_SHM
-    shm_->render_frame(*f.get());
+    shm_->renderFrame(*f.get());
 #endif
 
     if (target_) {

src/media/video/video_mixer.cpp

@@ -195,7 +195,7 @@ void VideoMixer::start_sink()
 {
     if (sink_->start()) {
         if (this->attach(sink_.get())) {
-            emitSignal<DRing::VideoSignal::DecodingStarted>(id_,
+            emitSignal<DRing::VideoSignal::DecodingStarted>(sink_->getId(),
                                                             sink_->openedName(),
                                                             width_, height_,
                                                             true);
@@ -208,13 +208,12 @@ void VideoMixer::start_sink()
 
 void VideoMixer::stop_sink()
 {
-    if (this->detach(sink_.get())) {
-        emitSignal<DRing::VideoSignal::DecodingStopped>(id_,
+    this->detach(sink_.get());
+    emitSignal<DRing::VideoSignal::DecodingStopped>(sink_->getId(),
                                                     sink_->openedName(),
                                                     true);
     sink_->stop();
 }
-}
 
 int VideoMixer::getWidth() const
 { return width_; }

src/media/video/video_receive_thread.cpp

@@ -143,9 +143,8 @@ void VideoReceiveThread::process()
 
 void VideoReceiveThread::cleanup()
 {
-    if (detach(sink_.get()))
-        emitSignal<DRing::VideoSignal::DecodingStopped>(id_,
-                                                        sink_->openedName(),
+    detach(sink_.get());
+    emitSignal<DRing::VideoSignal::DecodingStopped>(id_, sink_->openedName(),
                                                     false);
     sink_->stop();
 
@@ -206,9 +205,9 @@ void VideoReceiveThread::enterConference()
         return;
 
     if (detach(sink_.get())) {
-        emitSignal<DRing::VideoSignal::DecodingStopped>(id_,
+        emitSignal<DRing::VideoSignal::DecodingStopped>(sink_->getId(),
                                                         sink_->openedName(),
-                                                        false);
+                                                        true);
         RING_DBG("RX: shm sink <%s> detached", sink_->openedName().c_str());
     }
 }
@@ -220,7 +219,7 @@ void VideoReceiveThread::exitConference()
 
     if (dstWidth_ > 0 && dstHeight_ > 0) {
         if (attach(sink_.get())) {
-            emitSignal<DRing::VideoSignal::DecodingStarted>(id_,
+            emitSignal<DRing::VideoSignal::DecodingStarted>(sink_->getId(),
                                                             sink_->openedName(),
                                                             dstWidth_,
                                                             dstHeight_, false);