mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-06-11 02:29:27 +00:00
Code Issues 1 Projects Releases Wiki Activity

win32ipc: Add support for zero-copy rendering

Browse source 
* Extend protocol so that client can notify of releasing shared memory
* Server will hold shared memory object until it's released by client
* Add allocator/buffer pool to reuse shared memory objects and buffers

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3765>
This commit is contained in:
Seungha Yang 2023-01-19 22:22:52 +09:00 committed by GStreamer Marge Bot
parent c32b898cbb
commit c6a6c56cdf
13 changed files with 1324 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

241
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcbufferpool.cpp Normal file
View file

@ -0,0 +1,241 @@
/* GStreamer
* Copyright (C) 2023 Seungha Yang <seungha@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstwin32ipcbufferpool.h"
#include "gstwin32ipcmemory.h"
#include <gst/video/video.h>
GST_DEBUG_CATEGORY_STATIC (gst_win32_ipc_buffer_pool_debug);
#define GST_CAT_DEFAULT gst_win32_ipc_buffer_pool_debug
struct _GstWin32IpcBufferPool
{
GstBufferPool parent;
GstWin32IpcAllocator *alloc;
GstVideoInfo info;
gboolean add_videometa;
};
#define gst_win32_ipc_buffer_pool_parent_class parent_class
G_DEFINE_TYPE (GstWin32IpcBufferPool,
gst_win32_ipc_buffer_pool, GST_TYPE_BUFFER_POOL);
static void gst_win32_ipc_buffer_pool_dispose (GObject * object);
static const gchar **gst_win32_ipc_buffer_pool_get_options (GstBufferPool *
pool);
static gboolean gst_win32_ipc_buffer_pool_set_config (GstBufferPool * pool,
GstStructure * config);
static GstFlowReturn gst_win32_ipc_buffer_pool_alloc_buffer (GstBufferPool *
pool, GstBuffer ** buffer, GstBufferPoolAcquireParams * params);
static gboolean gst_win32_ipc_buffer_pool_start (GstBufferPool * pool);
static gboolean gst_win32_ipc_buffer_pool_stop (GstBufferPool * pool);
static void
gst_win32_ipc_buffer_pool_class_init (GstWin32IpcBufferPoolClass * klass)
{
GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
GstBufferPoolClass *bufferpool_class = GST_BUFFER_POOL_CLASS (klass);
gobject_class->dispose = gst_win32_ipc_buffer_pool_dispose;
bufferpool_class->get_options = gst_win32_ipc_buffer_pool_get_options;
bufferpool_class->set_config = gst_win32_ipc_buffer_pool_set_config;
bufferpool_class->alloc_buffer = gst_win32_ipc_buffer_pool_alloc_buffer;
bufferpool_class->start = gst_win32_ipc_buffer_pool_start;
bufferpool_class->stop = gst_win32_ipc_buffer_pool_stop;
GST_DEBUG_CATEGORY_INIT (gst_win32_ipc_buffer_pool_debug,
"win32_ipcbufferpool", 0, "win32_ipcbufferpool object");
}
static void
gst_win32_ipc_buffer_pool_init (GstWin32IpcBufferPool * self)
{
}
static void
gst_win32_ipc_buffer_pool_dispose (GObject * object)
{
GstWin32IpcBufferPool *self = GST_WIN32_IPC_BUFFER_POOL (object);
if (self->alloc) {
gst_win32_ipc_allocator_set_active (self->alloc, FALSE);
gst_clear_object (&self->alloc);
}
G_OBJECT_CLASS (parent_class)->dispose (object);
}
static const gchar **
gst_win32_ipc_buffer_pool_get_options (GstBufferPool * pool)
{
static const gchar *options[] = { GST_BUFFER_POOL_OPTION_VIDEO_META,
nullptr
};
return options;
}
static gboolean
gst_win32_ipc_buffer_pool_set_config (GstBufferPool * pool,
GstStructure * config)
{
GstWin32IpcBufferPool *self = GST_WIN32_IPC_BUFFER_POOL (pool);
GstVideoInfo info;
GstCaps *caps = nullptr;
gboolean ret = TRUE;
guint size, min_buffers, max_buffers;
if (!gst_buffer_pool_config_get_params (config, &caps, &size, &min_buffers,
&max_buffers)) {
GST_WARNING_OBJECT (pool, "Invalid config");
return FALSE;
}
if (!caps) {
GST_WARNING_OBJECT (pool, "No caps");
return FALSE;
}
/* now parse the caps from the config */
if (!gst_video_info_from_caps (&info, caps)) {
GST_WARNING_OBJECT (self, "Couldn't get video info from caps");
return FALSE;
}
if (size < info.size) {
GST_WARNING_OBJECT (self, "Size is smaller for the caps");
return FALSE;
}
info.size = MAX (size, info.size);
self->info = info;
GST_LOG_OBJECT (pool, "%dx%d, caps %" GST_PTR_FORMAT, info.width, info.height,
caps);
if (self->alloc) {
gst_win32_ipc_allocator_set_active (self->alloc, FALSE);
gst_clear_object (&self->alloc);
}
self->alloc = gst_win32_ipc_allocator_new (size);
if (!self->alloc) {
GST_ERROR_OBJECT (self, "Couldn't create allocator");
return FALSE;
}
self->add_videometa = gst_buffer_pool_config_has_option (config,
GST_BUFFER_POOL_OPTION_VIDEO_META);
gst_buffer_pool_config_set_params (config,
caps, info.size, min_buffers, max_buffers);
return GST_BUFFER_POOL_CLASS (parent_class)->set_config (pool, config) && ret;
}
static GstFlowReturn
gst_win32_ipc_buffer_pool_alloc_buffer (GstBufferPool * pool,
GstBuffer ** buffer, GstBufferPoolAcquireParams * params)
{
GstWin32IpcBufferPool *self = GST_WIN32_IPC_BUFFER_POOL (pool);
GstFlowReturn ret = GST_FLOW_OK;
GstBuffer *buf;
GstMemory *mem = nullptr;
GstVideoInfo *info = &self->info;
ret = gst_win32_ipc_allocator_acquire_memory (self->alloc, &mem);
if (ret != GST_FLOW_OK) {
GST_WARNING_OBJECT (self, "Couldn't acquire memory");
return ret;
}
buf = gst_buffer_new ();
gst_buffer_append_memory (buf, mem);
if (self->add_videometa) {
gst_buffer_add_video_meta_full (buf, GST_VIDEO_FRAME_FLAG_NONE,
GST_VIDEO_INFO_FORMAT (info), GST_VIDEO_INFO_WIDTH (info),
GST_VIDEO_INFO_HEIGHT (info), GST_VIDEO_INFO_N_PLANES (info),
info->offset, info->stride);
}
*buffer = buf;
return GST_FLOW_OK;
}
static gboolean
gst_win32_ipc_buffer_pool_start (GstBufferPool * pool)
{
GstWin32IpcBufferPool *self = GST_WIN32_IPC_BUFFER_POOL (pool);
gboolean ret;
GST_DEBUG_OBJECT (self, "Start");
if (!self->alloc) {
GST_ERROR_OBJECT (self, "No allocator");
return FALSE;
}
if (!gst_win32_ipc_allocator_set_active (self->alloc, TRUE)) {
GST_ERROR_OBJECT (self, "Failed to activate allocator");
return FALSE;
}
ret = GST_BUFFER_POOL_CLASS (parent_class)->start (pool);
if (!ret) {
GST_ERROR_OBJECT (self, "Failed to start");
gst_win32_ipc_allocator_set_active (self->alloc, FALSE);
return FALSE;
}
return TRUE;
}
static gboolean
gst_win32_ipc_buffer_pool_stop (GstBufferPool * pool)
{
GstWin32IpcBufferPool *self = GST_WIN32_IPC_BUFFER_POOL (pool);
GST_DEBUG_OBJECT (self, "Stop");
if (self->alloc)
gst_win32_ipc_allocator_set_active (self->alloc, FALSE);
return GST_BUFFER_POOL_CLASS (parent_class)->stop (pool);
}
GstBufferPool *
gst_win32_ipc_buffer_pool_new (void)
{
GstWin32IpcBufferPool *self;
self = (GstWin32IpcBufferPool *)
g_object_new (GST_TYPE_WIN32_IPC_BUFFER_POOL, nullptr);
gst_object_ref_sink (self);
return GST_BUFFER_POOL_CAST (self);
}

32
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcbufferpool.h Normal file
View file

@ -0,0 +1,32 @@
/* GStreamer
* Copyright (C) 2023 Seungha Yang <seungha@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#pragma once
#include <gst/gst.h>
G_BEGIN_DECLS
#define GST_TYPE_WIN32_IPC_BUFFER_POOL (gst_win32_ipc_buffer_pool_get_type())
G_DECLARE_FINAL_TYPE (GstWin32IpcBufferPool, gst_win32_ipc_buffer_pool,
GST, WIN32_IPC_BUFFER_POOL, GstBufferPool);
GstBufferPool * gst_win32_ipc_buffer_pool_new (void);
G_END_DECLS

545
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcmemory.cpp Normal file
View file

@ -0,0 +1,545 @@
/* GStreamer
* Copyright (C) 2023 Seungha Yang <seungha@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstwin32ipcmemory.h"
#include "gstwin32ipcutils.h"
#include <string>
#include <mutex>
#include <string.h>
GST_DEBUG_CATEGORY_STATIC (gst_win32_ipc_allocator_debug);
#define GST_CAT_DEFAULT gst_win32_ipc_allocator_debug
#define GST_WIN32_IPC_MEMORY_NAME "Win32IpcMemory"
#define GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING(alloc) \
(g_atomic_int_get (&alloc->flushing))
static GstWin32IpcAllocator *gc_allocator = nullptr;
struct _GstWin32IpcAllocator
{
GstAllocator parent;
guint size;
gboolean is_gc;
GstAtomicQueue *queue;
GstPoll *poll;
gchar *prefix;
LONG64 seq_num;
CRITICAL_SECTION lock;
gboolean started;
gboolean active;
/* atomic */
gint outstanding;
guint max_mems;
guint cur_mems;
gboolean flushing;
};
static void gst_win32_ipc_allocator_finalize (GObject * object);
static GstMemory *gst_win32_ipc_allocator_dummy_alloc (GstAllocator * alloc,
gsize size, GstAllocationParams * params);
static void gst_win32_ipc_allocator_free (GstAllocator * alloc,
GstMemory * mem);
static gpointer gst_win32_ipc_allocator_map (GstMemory * mem, gsize maxsize,
GstMapFlags flags);
static void gst_win32_ipc_allocator_unmap (GstMemory * mem);
static GstMemory *gst_win32_ipc_allocator_share (GstMemory * mem,
gssize offset, gssize size);
static gboolean gst_win32_ipc_allocator_start (GstWin32IpcAllocator * self);
static gboolean gst_win32_ipc_allocator_stop (GstWin32IpcAllocator * self);
static gboolean gst_win32_ipc_memory_release (GstMiniObject * mini_object);
#define gst_win32_ipc_allocator_parent_class parent_class
G_DEFINE_TYPE (GstWin32IpcAllocator, gst_win32_ipc_allocator,
GST_TYPE_ALLOCATOR);
static void
gst_win32_ipc_allocator_class_init (GstWin32IpcAllocatorClass * klass)
{
GObjectClass *object_class = G_OBJECT_CLASS (klass);
GstAllocatorClass *alloc_class = GST_ALLOCATOR_CLASS (klass);
object_class->finalize = gst_win32_ipc_allocator_finalize;
alloc_class->alloc = gst_win32_ipc_allocator_dummy_alloc;
alloc_class->free = gst_win32_ipc_allocator_free;
GST_DEBUG_CATEGORY_INIT (gst_win32_ipc_allocator_debug,
"win32ipcallocator", 0, "win32ipcallocator");
}
static void
gst_win32_ipc_allocator_init (GstWin32IpcAllocator * self)
{
GstAllocator *alloc = GST_ALLOCATOR (self);
alloc->mem_type = GST_WIN32_IPC_MEMORY_NAME;
alloc->mem_map = gst_win32_ipc_allocator_map;
alloc->mem_unmap = gst_win32_ipc_allocator_unmap;
alloc->mem_share = gst_win32_ipc_allocator_share;
GST_OBJECT_FLAG_SET (alloc, GST_ALLOCATOR_FLAG_CUSTOM_ALLOC);
InitializeCriticalSection (&self->lock);
self->poll = gst_poll_new_timer ();
self->queue = gst_atomic_queue_new (16);
self->flushing = 1;
self->active = FALSE;
self->started = FALSE;
/* 1 control write for flushing - the flush token */
gst_poll_write_control (self->poll);
/* 1 control write for marking that we are not waiting for poll - the wait token */
gst_poll_write_control (self->poll);
}
static void
gst_win32_ipc_allocator_finalize (GObject * object)
{
GstWin32IpcAllocator *self = GST_WIN32_IPC_ALLOCATOR (object);
GST_DEBUG_OBJECT (self, "Finalize");
gst_win32_ipc_allocator_stop (self);
gst_atomic_queue_unref (self->queue);
gst_poll_free (self->poll);
DeleteCriticalSection (&self->lock);
g_free (self->prefix);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static GstMemory *
gst_win32_ipc_allocator_dummy_alloc (GstAllocator * alloc, gsize size,
GstAllocationParams * params)
{
g_return_val_if_reached (nullptr);
}
static void
gst_win32_ipc_allocator_free (GstAllocator * alloc, GstMemory * mem)
{
GstWin32IpcMemory *imem = (GstWin32IpcMemory *) mem;
win32_ipc_mmf_unref (imem->mmf);
g_free (imem);
}
static gpointer
gst_win32_ipc_allocator_map (GstMemory * mem, gsize maxsize, GstMapFlags flags)
{
GstWin32IpcMemory *imem = (GstWin32IpcMemory *) mem;
return win32_ipc_mmf_get_raw (imem->mmf);
}
static void
gst_win32_ipc_allocator_unmap (GstMemory * mem)
{
/* do nothing */
}
static GstMemory *
gst_win32_ipc_allocator_share (GstMemory * mem, gssize offset, gssize size)
{
/* not supported */
return nullptr;
}
static gboolean
gst_win32_ipc_allocator_start (GstWin32IpcAllocator * self)
{
if (self->started)
return TRUE;
self->started = TRUE;
return TRUE;
}
static void
gst_win32_ipc_allocator_do_set_flushing (GstWin32IpcAllocator * self,
gboolean flushing)
{
if (GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING (self) == flushing)
return;
if (flushing) {
g_atomic_int_set (&self->flushing, 1);
/* Write the flush token to wake up any waiters */
gst_poll_write_control (self->poll);
} else {
while (!gst_poll_read_control (self->poll)) {
if (errno == EWOULDBLOCK) {
/* This should not really happen unless flushing and unflushing
* happens on different threads. Let's wait a bit to get back flush
* token from the thread that was setting it to flushing */
g_thread_yield ();
continue;
} else {
/* Critical error but GstPoll already complained */
break;
}
}
g_atomic_int_set (&self->flushing, 0);
}
}
gboolean
gst_win32_ipc_allocator_set_active (GstWin32IpcAllocator * self,
gboolean active)
{
gboolean ret = TRUE;
g_return_val_if_fail (GST_IS_WIN32_IPC_ALLOCATOR (self), FALSE);
EnterCriticalSection (&self->lock);
if (self->active == active)
goto out;
if (active) {
gst_win32_ipc_allocator_start (self);
/* flush_stop may release memory objects, setting to active to avoid running
* do_stop while activating the pool */
self->active = TRUE;
gst_win32_ipc_allocator_do_set_flushing (self, FALSE);
} else {
gint outstanding;
/* set to flushing first */
gst_win32_ipc_allocator_do_set_flushing (self, TRUE);
/* when all memory objects are in the pool, free them. Else they will be
* freed when they are released */
outstanding = g_atomic_int_get (&self->outstanding);
GST_LOG_OBJECT (self, "outstanding memories %d, (in queue %d)",
outstanding, gst_atomic_queue_length (self->queue));
if (outstanding == 0) {
if (!gst_win32_ipc_allocator_stop (self)) {
GST_ERROR_OBJECT (self, "stop failed");
ret = FALSE;
goto out;
}
}
self->active = FALSE;
}
out:
LeaveCriticalSection (&self->lock);
return ret;
}
static void
gst_win32_ipc_allocator_free_memory (GstWin32IpcAllocator * self,
GstMemory * mem)
{
g_atomic_int_add (&self->cur_mems, -1);
GST_LOG_OBJECT (self, "freeing memory %p (%u left)", mem, self->cur_mems);
GST_MINI_OBJECT_CAST (mem)->dispose = nullptr;
gst_memory_unref (mem);
}
/* must be called with the lock */
static gboolean
gst_win32_ipc_allocator_clear_queue (GstWin32IpcAllocator * self)
{
GstMemory *memory;
GST_LOG_OBJECT (self, "Clearing queue");
/* clear the pool */
while ((memory = (GstMemory *) gst_atomic_queue_pop (self->queue))) {
while (!gst_poll_read_control (self->poll)) {
if (errno == EWOULDBLOCK) {
/* We put the memory into the queue but did not finish writing control
* yet, let's wait a bit and retry */
g_thread_yield ();
continue;
} else {
/* Critical error but GstPoll already complained */
break;
}
}
gst_win32_ipc_allocator_free_memory (self, memory);
}
GST_LOG_OBJECT (self, "Clear done");
return self->cur_mems == 0;
}
static gboolean
gst_win32_ipc_allocator_stop (GstWin32IpcAllocator * self)
{
GST_DEBUG_OBJECT (self, "Stop");
if (self->started) {
if (!gst_win32_ipc_allocator_clear_queue (self))
return FALSE;
self->started = FALSE;
}
return TRUE;
}
static void
dec_outstanding (GstWin32IpcAllocator * self)
{
if (g_atomic_int_dec_and_test (&self->outstanding)) {
/* all memory objects are returned to the pool, see if we need to free them */
if (GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING (self)) {
/* take the lock so that set_active is not run concurrently */
EnterCriticalSection (&self->lock);
/* now that we have the lock, check if we have been de-activated with
* outstanding buffers */
if (!self->active)
gst_win32_ipc_allocator_stop (self);
LeaveCriticalSection (&self->lock);
}
}
}
static void
gst_win32_ipc_allocator_release_memory (GstWin32IpcAllocator * self,
GstMemory * mem)
{
GST_MINI_OBJECT_CAST (mem)->dispose = nullptr;
mem->allocator = (GstAllocator *) gst_object_ref (gc_allocator);
/* keep it around in our queue */
gst_atomic_queue_push (self->queue, mem);
gst_poll_write_control (self->poll);
dec_outstanding (self);
gst_object_unref (self);
}
static gboolean
gst_win32_ipc_memory_release (GstMiniObject * mini_object)
{
GstMemory *mem = GST_MEMORY_CAST (mini_object);
GstWin32IpcAllocator *self;
g_assert (mem->allocator != nullptr);
self = GST_WIN32_IPC_ALLOCATOR (mem->allocator);
/* Memory belongs to garbage collector, free this */
if (self->is_gc)
return TRUE;
if (GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING (self))
return TRUE;
/* return the memory to the allocator */
gst_memory_ref (mem);
gst_win32_ipc_allocator_release_memory (self, mem);
return FALSE;
}
static GstFlowReturn
gst_win32_ipc_allocator_alloc (GstWin32IpcAllocator * self, GstMemory ** mem)
{
GstWin32IpcMemory *new_mem;
Win32IpcMmf *mmf;
std::string mmf_name;
mmf_name = std::string (self->prefix) +
std::to_string (InterlockedIncrement64 (&self->seq_num));
mmf = win32_ipc_mmf_alloc (self->size, mmf_name.c_str ());
if (!mmf) {
GST_ERROR_OBJECT (self, "Couldn't allocate memory");
return GST_FLOW_ERROR;
}
memset (win32_ipc_mmf_get_raw (mmf), 0, win32_ipc_mmf_get_size (mmf));
g_atomic_int_add (&self->cur_mems, 1);
new_mem = g_new0 (GstWin32IpcMemory, 1);
gst_memory_init (GST_MEMORY_CAST (new_mem), (GstMemoryFlags) 0,
GST_ALLOCATOR_CAST (gc_allocator), nullptr, self->size, 0, 0, self->size);
new_mem->mmf = mmf;
*mem = GST_MEMORY_CAST (new_mem);
return GST_FLOW_OK;
}
static GstFlowReturn
gst_win32_ipc_allocator_acquire_memory_internal (GstWin32IpcAllocator * self,
GstMemory ** memory)
{
GstFlowReturn result;
while (TRUE) {
if (GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING (self)) {
GST_DEBUG_OBJECT (self, "We are flushing");
return GST_FLOW_FLUSHING;
}
/* try to get a memory from the queue */
*memory = (GstMemory *) gst_atomic_queue_pop (self->queue);
if (*memory) {
while (!gst_poll_read_control (self->poll)) {
if (errno == EWOULDBLOCK) {
/* We put the memory into the queue but did not finish writing control
* yet, let's wait a bit and retry */
g_thread_yield ();
continue;
} else {
/* Critical error but GstPoll already complained */
break;
}
}
result = GST_FLOW_OK;
GST_LOG_OBJECT (self, "acquired memory %p", *memory);
break;
}
/* no memory, try to allocate some more */
GST_LOG_OBJECT (self, "no memory, trying to allocate");
result = gst_win32_ipc_allocator_alloc (self, memory);
if (result == GST_FLOW_OK)
/* we have a memory, return it */
break;
if (G_UNLIKELY (result != GST_FLOW_EOS))
/* something went wrong, return error */
break;
/* now we release the control socket, we wait for a memory release or
* flushing */
if (!gst_poll_read_control (self->poll)) {
if (errno == EWOULDBLOCK) {
/* This means that we have two threads trying to allocate memory
* already, and the other one already got the wait token. This
* means that we only have to wait for the poll now and not write the
* token afterwards: we will be woken up once the other thread is
* woken up and that one will write the wait token it removed */
GST_LOG_OBJECT (self, "waiting for free memory or flushing");
gst_poll_wait (self->poll, GST_CLOCK_TIME_NONE);
} else {
/* This is a critical error, GstPoll already gave a warning */
result = GST_FLOW_ERROR;
break;
}
} else {
/* We're the first thread waiting, we got the wait token and have to
* write it again later
* OR
* We're a second thread and just consumed the flush token and block all
* other threads, in which case we must not wait and give it back
* immediately */
if (!GST_WIN32_IPC_ALLOCATOR_IS_FLUSHING (self)) {
GST_LOG_OBJECT (self, "waiting for free memory or flushing");
gst_poll_wait (self->poll, GST_CLOCK_TIME_NONE);
}
gst_poll_write_control (self->poll);
}
}
return result;
}
gboolean
gst_is_win32_ipc_memory (GstMemory * mem)
{
return mem != nullptr && mem->allocator != nullptr &&
GST_IS_WIN32_IPC_ALLOCATOR (mem->allocator);
}
static void
gst_win32_ipc_allocator_init_once (void)
{
static std::once_flag once_flag;
std::call_once (once_flag,[]() {
gc_allocator = (GstWin32IpcAllocator *)
g_object_new (GST_TYPE_WIN32_IPC_ALLOCATOR, nullptr);
gst_object_ref_sink (gc_allocator);
GST_OBJECT_FLAG_SET (gc_allocator, GST_OBJECT_FLAG_MAY_BE_LEAKED);
gc_allocator->is_gc = TRUE;
});
}
GstWin32IpcAllocator *
gst_win32_ipc_allocator_new (guint size)
{
GstWin32IpcAllocator *self;
g_return_val_if_fail (size != 0, nullptr);
gst_win32_ipc_allocator_init_once ();
self = (GstWin32IpcAllocator *)
g_object_new (GST_TYPE_WIN32_IPC_ALLOCATOR, nullptr);
self->size = size;
self->prefix = gst_win32_ipc_get_mmf_prefix ();
gst_object_ref_sink (self);
return self;
}
GstFlowReturn
gst_win32_ipc_allocator_acquire_memory (GstWin32IpcAllocator * alloc,
GstMemory ** memory)
{
GstFlowReturn ret;
g_return_val_if_fail (GST_IS_WIN32_IPC_ALLOCATOR (alloc), GST_FLOW_ERROR);
g_return_val_if_fail (memory != nullptr, GST_FLOW_ERROR);
*memory = nullptr;
g_atomic_int_inc (&alloc->outstanding);
ret = gst_win32_ipc_allocator_acquire_memory_internal (alloc, memory);
if (ret == GST_FLOW_OK) {
GstMemory *mem = *memory;
/* Replace default allocator with ours */
gst_object_unref (mem->allocator);
mem->allocator = (GstAllocator *) gst_object_ref (alloc);
GST_MINI_OBJECT_CAST (mem)->dispose = gst_win32_ipc_memory_release;
} else {
dec_outstanding (alloc);
}
return ret;
}

50
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcmemory.h Normal file
View file

@ -0,0 +1,50 @@
/* GStreamer
* Copyright (C) 2023 Seungha Yang <seungha@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#pragma once
#include <gst/gst.h>
#include "protocol/win32ipcmmf.h"
G_BEGIN_DECLS
#define GST_TYPE_WIN32_IPC_ALLOCATOR (gst_win32_ipc_allocator_get_type())
G_DECLARE_FINAL_TYPE (GstWin32IpcAllocator, gst_win32_ipc_allocator,
GST, WIN32_IPC_ALLOCATOR, GstAllocator);
typedef struct _GstWin32IpcMemory GstWin32IpcMemory;
struct _GstWin32IpcMemory
{
GstMemory mem;
Win32IpcMmf *mmf;
};
gboolean gst_is_win32_ipc_memory (GstMemory * mem);
GstWin32IpcAllocator * gst_win32_ipc_allocator_new (guint size);
gboolean gst_win32_ipc_allocator_set_active (GstWin32IpcAllocator * alloc,
gboolean active);
GstFlowReturn gst_win32_ipc_allocator_acquire_memory (GstWin32IpcAllocator * alloc,
GstMemory ** memory);
G_END_DECLS

131
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcvideosink.cpp
View file

@ -39,6 +39,8 @@
#include "gstwin32ipcvideosink.h"
#include "gstwin32ipcutils.h"
#include "gstwin32ipcbufferpool.h"
#include "gstwin32ipcmemory.h"
#include "protocol/win32ipcpipeserver.h"
#include <string>
#include <string.h>
@ -65,13 +67,13 @@ struct _GstWin32IpcVideoSink
GstVideoInfo info;
Win32IpcPipeServer *pipe;
gchar *mmf_prefix;
guint64 seq_num;
LARGE_INTEGER frequency;
Win32IpcMmf *mmf;
Win32IpcVideoInfo minfo;
GstBufferPool *fallback_pool;
GstBuffer *prepared_buffer;
/* properties */
gchar *pipe_name;
};
@ -222,9 +224,6 @@ gst_win32_ipc_video_sink_start (GstBaseSink * sink)
return FALSE;
}
self->mmf_prefix = gst_win32_ipc_get_mmf_prefix ();
self->seq_num = 0;
return TRUE;
}
@ -236,8 +235,12 @@ gst_win32_ipc_video_sink_stop (GstBaseSink * sink)
GST_DEBUG_OBJECT (self, "Stop");
g_clear_pointer (&self->pipe, win32_ipc_pipe_server_unref);
g_clear_pointer (&self->mmf_prefix, g_free);
g_clear_pointer (&self->mmf, win32_ipc_mmf_unref);
gst_clear_buffer (&self->prepared_buffer);
if (self->fallback_pool) {
gst_buffer_pool_set_active (self->fallback_pool, FALSE);
gst_clear_object (&self->fallback_pool);
}
return TRUE;
}
@ -247,7 +250,7 @@ gst_win32_ipc_video_sink_unlock_stop (GstBaseSink * sink)
{
GstWin32IpcVideoSink *self = GST_WIN32_IPC_VIDEO_SINK (sink);
g_clear_pointer (&self->mmf, win32_ipc_mmf_unref);
gst_clear_buffer (&self->prepared_buffer);
return TRUE;
}
@ -281,6 +284,7 @@ static gboolean
gst_win32_ipc_video_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
{
GstWin32IpcVideoSink *self = GST_WIN32_IPC_VIDEO_SINK (sink);
GstStructure *config;
if (!gst_video_info_from_caps (&self->info, caps)) {
GST_WARNING_OBJECT (self, "Invalid caps");
@ -297,6 +301,19 @@ gst_win32_ipc_video_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
self->minfo.par_n = self->info.par_n;
self->minfo.par_d = self->info.par_d;
if (self->fallback_pool) {
gst_buffer_pool_set_active (self->fallback_pool, FALSE);
gst_object_unref (self->fallback_pool);
}
self->fallback_pool = gst_win32_ipc_buffer_pool_new ();
config = gst_buffer_pool_get_config (self->fallback_pool);
gst_buffer_pool_config_add_option (config, GST_BUFFER_POOL_OPTION_VIDEO_META);
gst_buffer_pool_config_set_params (config, caps, (guint) self->info.size,
0, 0);
gst_buffer_pool_set_config (self->fallback_pool, config);
gst_buffer_pool_set_active (self->fallback_pool, TRUE);
return TRUE;
}
@ -326,7 +343,7 @@ gst_win32_ipc_video_sink_propose_allocation (GstBaseSink * sink,
if (need_pool) {
GstStructure *config;
pool = gst_video_buffer_pool_new ();
pool = gst_win32_ipc_buffer_pool_new ();
config = gst_buffer_pool_get_config (pool);
gst_buffer_pool_config_add_option (config,
GST_BUFFER_POOL_OPTION_VIDEO_META);
@ -355,43 +372,83 @@ static GstFlowReturn
gst_win32_ipc_video_sink_prepare (GstBaseSink * sink, GstBuffer * buf)
{
GstWin32IpcVideoSink *self = GST_WIN32_IPC_VIDEO_SINK (sink);
std::string mmf_name;
GstVideoFrame frame;
GstMapInfo info;
GstVideoFrame frame, mmf_frame;
GstMemory *mem;
GstFlowReturn ret;
g_clear_pointer (&self->mmf, win32_ipc_mmf_unref);
gst_clear_buffer (&self->prepared_buffer);
if (!gst_video_frame_map (&frame, &self->info, buf, GST_MAP_READ)) {
GST_ERROR_OBJECT (self, "Couldn't map frame");
return GST_FLOW_ERROR;
}
mmf_name = std::string (self->mmf_prefix) + std::to_string (self->seq_num);
self->seq_num++;
mem = gst_buffer_peek_memory (buf, 0);
if (gst_is_win32_ipc_memory (mem) && gst_buffer_n_memory (buf) == 1) {
GST_LOG_OBJECT (self, "Upstream memory is mmf");
self->mmf = win32_ipc_mmf_alloc (GST_VIDEO_FRAME_SIZE (&frame),
mmf_name.c_str ());
if (!self->mmf) {
GST_ERROR_OBJECT (self, "Couldn't create memory with name %s",
mmf_name.c_str ());
self->prepared_buffer = gst_buffer_ref (buf);
self->minfo.size = GST_VIDEO_FRAME_SIZE (&frame);
for (guint i = 0; i < GST_VIDEO_FRAME_N_PLANES (&frame); i++) {
self->minfo.offset[i] = GST_VIDEO_FRAME_PLANE_OFFSET (&frame, i);
self->minfo.stride[i] = GST_VIDEO_FRAME_PLANE_STRIDE (&frame, i);
}
gst_video_frame_unmap (&frame);
return GST_FLOW_OK;
}
GST_LOG_OBJECT (self, "Copying into mmf buffer");
ret = gst_buffer_pool_acquire_buffer (self->fallback_pool,
&self->prepared_buffer, nullptr);
if (ret != GST_FLOW_OK) {
GST_ERROR_OBJECT (self, "Couldn't acquire buffer");
gst_video_frame_unmap (&frame);
return GST_FLOW_ERROR;
}
self->minfo.size = GST_VIDEO_FRAME_SIZE (&frame);
for (guint i = 0; i < GST_VIDEO_FRAME_N_PLANES (&frame); i++) {
self->minfo.offset[i] = GST_VIDEO_FRAME_PLANE_OFFSET (&frame, i);
self->minfo.stride[i] = GST_VIDEO_FRAME_PLANE_STRIDE (&frame, i);
if (!gst_video_frame_map (&mmf_frame, &self->info, self->prepared_buffer,
GST_MAP_WRITE)) {
GST_ERROR_OBJECT (self, "Couldn't map mmf frame");
gst_video_frame_unmap (&frame);
gst_clear_buffer (&self->prepared_buffer);
return GST_FLOW_ERROR;
}
if (!gst_video_frame_copy (&mmf_frame, &frame)) {
GST_ERROR_OBJECT (self, "Couldn't copy buffer");
gst_video_frame_unmap (&frame);
gst_video_frame_unmap (&mmf_frame);
gst_clear_buffer (&self->prepared_buffer);
return GST_FLOW_ERROR;
}
gst_video_frame_unmap (&frame);
gst_buffer_map (buf, &info, GST_MAP_READ);
memcpy (win32_ipc_mmf_get_raw (self->mmf), info.data, self->minfo.size);
gst_buffer_unmap (buf, &info);
self->minfo.size = GST_VIDEO_FRAME_SIZE (&mmf_frame);
for (guint i = 0; i < GST_VIDEO_FRAME_N_PLANES (&mmf_frame); i++) {
self->minfo.offset[i] = GST_VIDEO_FRAME_PLANE_OFFSET (&mmf_frame, i);
self->minfo.stride[i] = GST_VIDEO_FRAME_PLANE_STRIDE (&mmf_frame, i);
}
gst_video_frame_unmap (&mmf_frame);
return GST_FLOW_OK;
}
static void
gst_win32_ipc_video_sink_mmf_free (void *user_data)
{
GstBuffer *buffer = GST_BUFFER_CAST (user_data);
GST_LOG ("Relese %" GST_PTR_FORMAT, buffer);
gst_buffer_unref (buffer);
}
static GstFlowReturn
gst_win32_ipc_video_sink_render (GstBaseSink * sink, GstBuffer * buf)
{
@ -401,6 +458,20 @@ gst_win32_ipc_video_sink_render (GstBaseSink * sink, GstBuffer * buf)
GstClockTime now_qpc;
GstClockTime buf_pts;
GstClockTime buffer_clock = GST_CLOCK_TIME_NONE;
Win32IpcMmf *mmf;
GstWin32IpcMemory *mem;
if (!self->prepared_buffer) {
GST_ERROR_OBJECT (self, "No prepared buffer");
return GST_FLOW_ERROR;
}
mem = (GstWin32IpcMemory *) gst_buffer_peek_memory (self->prepared_buffer, 0);
g_assert (mem != nullptr);
g_assert (gst_is_win32_ipc_memory (GST_MEMORY_CAST (mem)));
mmf = mem->mmf;
QueryPerformanceCounter (&cur_time);
pts = now_qpc = gst_util_uint64_scale (cur_time.QuadPart, GST_SECOND,
@ -454,7 +525,9 @@ gst_win32_ipc_video_sink_render (GstBaseSink * sink, GstBuffer * buf)
/* win32_ipc_pipe_server_send_mmf() takes ownership of mmf */
if (!win32_ipc_pipe_server_send_mmf (self->pipe,
(Win32IpcMmf *) g_steal_pointer (&self->mmf), &self->minfo)) {
win32_ipc_mmf_ref (mmf), &self->minfo,
g_steal_pointer (&self->prepared_buffer),
gst_win32_ipc_video_sink_mmf_free)) {
GST_ERROR_OBJECT (self, "Couldn't send buffer");
return GST_FLOW_ERROR;
}

34
subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcvideosrc.cpp
View file

@ -257,7 +257,11 @@ gst_win32_ipc_video_src_stop (GstBaseSrc * src)
GST_DEBUG_OBJECT (self, "Stop");
g_clear_pointer (&self->pipe, win32_ipc_pipe_client_unref);
if (self->pipe) {
win32_ipc_pipe_client_stop (self->pipe);
g_clear_pointer (&self->pipe, win32_ipc_pipe_client_unref);
}
gst_clear_caps (&self->caps);
if (self->pool) {
gst_buffer_pool_set_active (self->pool, FALSE);
@ -277,7 +281,7 @@ gst_win32_ipc_video_src_unlock (GstBaseSrc * src)
AcquireSRWLockExclusive (&self->lock);
self->flushing = TRUE;
if (self->pipe)
win32_ipc_pipe_client_shutdown (self->pipe);
win32_ipc_pipe_client_set_flushing (self->pipe, TRUE);
ReleaseSRWLockExclusive (&self->lock);
return TRUE;
@ -291,9 +295,9 @@ gst_win32_ipc_video_src_unlock_stop (GstBaseSrc * src)
GST_DEBUG_OBJECT (self, "Unlock stop");
AcquireSRWLockExclusive (&self->lock);
g_clear_pointer (&self->pipe, win32_ipc_pipe_client_unref);
gst_clear_caps (&self->caps);
self->flushing = FALSE;
if (self->pipe)
win32_ipc_pipe_client_set_flushing (self->pipe, FALSE);
ReleaseSRWLockExclusive (&self->lock);
return TRUE;
@ -393,6 +397,20 @@ error:
return FALSE;
}
struct MmfReleaseData
{
Win32IpcPipeClient *pipe;
Win32IpcMmf *mmf;
};
static void
gst_win32_ipc_video_src_release_mmf (MmfReleaseData * data)
{
win32_ipc_pipe_client_release_mmf (data->pipe, data->mmf);
win32_ipc_pipe_client_unref (data->pipe);
delete data;
}
static GstFlowReturn
gst_win32_ipc_video_src_create (GstBaseSrc * src, guint64 offset, guint size,
GstBuffer ** buf)
@ -465,10 +483,14 @@ gst_win32_ipc_video_src_create (GstBaseSrc * src, guint64 offset, guint size,
}
if (self->have_video_meta || !need_video_meta) {
MmfReleaseData *data = new MmfReleaseData ();
data->pipe = win32_ipc_pipe_client_ref (self->pipe);
data->mmf = mmf;
buffer = gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY,
win32_ipc_mmf_get_raw (mmf), win32_ipc_mmf_get_size (mmf),
0, win32_ipc_mmf_get_size (mmf), mmf,
(GDestroyNotify) win32_ipc_mmf_unref);
0, win32_ipc_mmf_get_size (mmf), data,
(GDestroyNotify) gst_win32_ipc_video_src_release_mmf);
if (self->have_video_meta) {
gst_buffer_add_video_meta_full (buffer,

2
subprojects/gst-plugins-bad/sys/win32ipc/meson.build
View file

@ -4,6 +4,8 @@ win32ipc_sources = [
'protocol/win32ipcpipeserver.cpp',
'protocol/win32ipcprotocol.cpp',
'protocol/win32ipcutils.cpp',
'gstwin32ipcbufferpool.cpp',
'gstwin32ipcmemory.cpp',
'gstwin32ipcutils.cpp',
'gstwin32ipcvideosink.cpp',
'gstwin32ipcvideosrc.cpp',

191
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcpipeclient.cpp
View file

@ -66,8 +66,10 @@ struct ClientConnection : public OVERLAPPED
struct Win32IpcPipeClient
{
explicit Win32IpcPipeClient (const std::string & n)
: name (n), ref_count(1), last_err (ERROR_SUCCESS)
: name (n), ref_count(1), last_err (ERROR_SUCCESS), flushing (FALSE)
, stopped (FALSE), io_pending (FALSE)
{
release_event = CreateEventA (nullptr, FALSE, FALSE, nullptr);
cancellable = CreateEventA (nullptr, TRUE, FALSE, nullptr);
conn.pipe = INVALID_HANDLE_VALUE;
conn.self = this;
@ -76,7 +78,21 @@ struct Win32IpcPipeClient
~Win32IpcPipeClient ()
{
GST_DEBUG ("Free client %p", this);
win32_ipc_pipe_client_shutdown (this);
SetEvent (cancellable);
if (thread) {
thread->join ();
thread = nullptr;
}
last_err = ERROR_OPERATION_ABORTED;
while (!queue.empty ()) {
MmfInfo info = queue.front ();
queue.pop ();
win32_ipc_mmf_unref (info.mmf);
}
CloseHandle (release_event);
CloseHandle (cancellable);
}
@ -84,33 +100,63 @@ struct Win32IpcPipeClient
std::condition_variable cond;
std::unique_ptr<std::thread> thread;
std::queue<MmfInfo> queue;
std::queue<std::string> unused_mmf;
std::string name;
ULONG ref_count;
HANDLE release_event;
HANDLE cancellable;
UINT last_err;
BOOL flushing;
BOOL stopped;
BOOL io_pending;
ClientConnection conn;
};
static DWORD
win32_ipc_pipe_client_send_need_data_async (Win32IpcPipeClient * self);
static DWORD
win32_ipc_pipe_client_send_release_data_async (Win32IpcPipeClient * self,
const char * mmf_name);
static VOID WINAPI
win32_ipc_pipe_client_send_read_done_finish (DWORD error_code, DWORD n_bytes,
win32_ipc_pipe_client_send_finish (DWORD error_code, DWORD n_bytes,
LPOVERLAPPED overlapped)
{
ClientConnection *conn = (ClientConnection *) overlapped;
Win32IpcPipeClient *self = conn->self;
std::string unused_mmf;
if (error_code != ERROR_SUCCESS) {
std::string msg = win32_ipc_error_message (error_code);
self->last_err = error_code;
GST_WARNING ("READ-DONE failed with 0x%x (%s)",
self->last_err, msg.c_str ());
GST_WARNING ("Failed with 0x%x (%s)", self->last_err, msg.c_str ());
goto error;
}
GST_TRACE ("READ-DONE sent");
self->lock.lock ();
if (!self->unused_mmf.empty ()) {
unused_mmf = self->unused_mmf.front ();
self->unused_mmf.pop ();
}
self->lock.unlock ();
if (unused_mmf.size () > 0) {
self->last_err = win32_ipc_pipe_client_send_release_data_async (self,
unused_mmf.c_str ());
if (self->last_err != ERROR_SUCCESS)
goto error;
return;
}
/* Don't request data anymore if we are stopped, but keep connection
* to send release data message later */
if (self->stopped) {
GST_DEBUG ("We are stopped");
self->io_pending = FALSE;
return;
}
self->last_err = win32_ipc_pipe_client_send_need_data_async (self);
if (self->last_err != ERROR_SUCCESS)
@ -123,6 +169,32 @@ error:
SetEvent (self->cancellable);
}
static DWORD
win32_ipc_pipe_client_send_release_data_async (Win32IpcPipeClient * self,
const char * mmf_name)
{
ClientConnection *conn = &self->conn;
conn->to_write = win32_ipc_pkt_build_release_data (conn->client_msg,
CONN_BUFFER_SIZE, conn->seq_num, mmf_name);
if (conn->to_write == 0) {
GST_ERROR ("Couldn't build RELEASE-DATA pkt");
return ERROR_BAD_FORMAT;
}
GST_TRACE ("Sending RELEASE-DATA");
if (!WriteFileEx (conn->pipe, conn->client_msg, conn->to_write,
(OVERLAPPED *) conn, win32_ipc_pipe_client_send_finish)) {
UINT last_err = GetLastError ();
std::string msg = win32_ipc_error_message (last_err);
GST_WARNING ("WriteFileEx failed with 0x%x (%s)", last_err, msg.c_str ());
return last_err;
}
return ERROR_SUCCESS;
}
static DWORD
win32_ipc_pipe_client_send_read_done_async (Win32IpcPipeClient * self)
{
@ -138,7 +210,7 @@ win32_ipc_pipe_client_send_read_done_async (Win32IpcPipeClient * self)
GST_TRACE ("Sending READ-DONE");
if (!WriteFileEx (conn->pipe, conn->client_msg, conn->to_write,
(OVERLAPPED *) conn, win32_ipc_pipe_client_send_read_done_finish)) {
(OVERLAPPED *) conn, win32_ipc_pipe_client_send_finish)) {
UINT last_err = GetLastError ();
std::string msg = win32_ipc_error_message (last_err);
@ -171,7 +243,7 @@ win32_ipc_pipe_client_receive_have_data_finish (DWORD error_code, DWORD n_bytes,
if (!win32_ipc_pkt_parse_have_data (conn->server_msg, n_bytes,
&conn->seq_num, mmf_name, &info)) {
self->last_err = ERROR_BAD_FORMAT;
GST_WARNING ("Couldn't parse HAVE-DATA pkg");
GST_WARNING ("Couldn't parse HAVE-DATA pkt");
goto error;
}
@ -285,6 +357,8 @@ win32_ipc_pipe_client_loop (Win32IpcPipeClient * self)
DWORD mode = PIPE_READMODE_MESSAGE;
std::unique_lock<std::mutex> lk (self->lock);
ClientConnection *conn = &self->conn;
HANDLE waitables[2];
DWORD wait_ret;
conn->pipe = CreateFileA (self->name.c_str (),
GENERIC_READ | GENERIC_WRITE, 0, nullptr, OPEN_EXISTING,
@ -319,28 +393,60 @@ win32_ipc_pipe_client_loop (Win32IpcPipeClient * self)
if (self->last_err != ERROR_SUCCESS)
goto out;
self->io_pending = TRUE;
waitables[0] = self->cancellable;
waitables[1] = self->release_event;
do {
/* Enters alertable thread state and wait for I/O completion event
* or cancellable event */
DWORD ret = WaitForSingleObjectEx (self->cancellable, INFINITE, TRUE);
if (ret == WAIT_OBJECT_0) {
wait_ret = WaitForMultipleObjectsEx (2, waitables, FALSE, INFINITE, TRUE);
if (wait_ret == WAIT_OBJECT_0) {
GST_DEBUG ("Operation cancelled");
CancelIoEx (conn->pipe, (OVERLAPPED *) &conn);
break;
} else if (ret != WAIT_IO_COMPLETION) {
GST_WARNING ("Unexpected wait return 0x%x", (UINT) ret);
CancelIoEx (conn->pipe, (OVERLAPPED *) &conn);
break;
goto out;
}
switch (wait_ret) {
case WAIT_OBJECT_0 + 1:
case WAIT_IO_COMPLETION:
{
std::string unused_mmf;
/* If I/O chain is stopped, send release data message here */
if (!self->io_pending) {
lk.lock ();
if (!self->unused_mmf.empty ()) {
unused_mmf = self->unused_mmf.front ();
self->unused_mmf.pop ();
}
lk.unlock ();
}
if (unused_mmf.size () > 0) {
GST_DEBUG ("Sending release data for %s", unused_mmf.c_str ());
self->io_pending = TRUE;
self->last_err = win32_ipc_pipe_client_send_release_data_async (self,
unused_mmf.c_str ());
if (self->last_err != ERROR_SUCCESS)
goto out;
}
break;
}
default:
GST_WARNING ("Unexpected wait return 0x%x", (UINT) wait_ret);
goto out;
}
} while (true);
out:
if (conn->pipe != INVALID_HANDLE_VALUE)
if (conn->pipe != INVALID_HANDLE_VALUE) {
CancelIoEx (conn->pipe, (OVERLAPPED *) &conn);
CloseHandle (conn->pipe);
}
lk.lock ();
self->last_err = ERROR_OPERATION_ABORTED;
conn->pipe = INVALID_HANDLE_VALUE;
self->io_pending = FALSE;
self->cond.notify_all ();
}
@ -401,24 +507,10 @@ win32_ipc_pipe_client_unref (Win32IpcPipeClient * client)
}
void
win32_ipc_pipe_client_shutdown (Win32IpcPipeClient * client)
win32_ipc_pipe_client_set_flushing (Win32IpcPipeClient * client, BOOL flushing)
{
GST_DEBUG ("Shutting down %p", client);
SetEvent (client->cancellable);
if (client->thread) {
client->thread->join ();
client->thread = nullptr;
}
std::lock_guard<std::mutex> lk (client->lock);
client->last_err = ERROR_OPERATION_ABORTED;
while (!client->queue.empty ()) {
MmfInfo info = client->queue.front ();
client->queue.pop ();
win32_ipc_mmf_unref (info.mmf);
}
client->flushing = flushing;
client->cond.notify_all ();
}
@ -432,10 +524,12 @@ win32_ipc_pipe_client_get_mmf (Win32IpcPipeClient * client, Win32IpcMmf ** mmf,
return FALSE;
}
while (client->queue.empty () && client->last_err == ERROR_SUCCESS)
while (client->queue.empty () && client->last_err == ERROR_SUCCESS &&
!client->flushing && !client->stopped) {
client->cond.wait (lk);
}
if (client->last_err != ERROR_SUCCESS || client->queue.empty ())
if (client->queue.empty ())
return FALSE;
MmfInfo mmf_info = client->queue.front ();
@ -446,3 +540,30 @@ win32_ipc_pipe_client_get_mmf (Win32IpcPipeClient * client, Win32IpcMmf ** mmf,
return TRUE;
}
void
win32_ipc_pipe_client_release_mmf (Win32IpcPipeClient * client,
Win32IpcMmf * mmf)
{
std::string name = win32_ipc_mmf_get_name (mmf);
win32_ipc_mmf_unref (mmf);
std::lock_guard<std::mutex> lk (client->lock);
if (client->last_err != ERROR_SUCCESS)
return;
GST_LOG ("Enqueue release data %s", name.c_str ());
client->unused_mmf.push (name);
SetEvent (client->release_event);
}
void
win32_ipc_pipe_client_stop (Win32IpcPipeClient * client)
{
GST_DEBUG ("Stopping %p", client);
std::lock_guard<std::mutex> lk (client->lock);
client->stopped = TRUE;
client->cond.notify_all ();
}

8
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcpipeclient.h
View file

@ -40,10 +40,16 @@ Win32IpcPipeClient * win32_ipc_pipe_client_ref (Win32IpcPipeClient * client);
void win32_ipc_pipe_client_unref (Win32IpcPipeClient * client);
void win32_ipc_pipe_client_shutdown (Win32IpcPipeClient * client);
void win32_ipc_pipe_client_set_flushing (Win32IpcPipeClient * client,
BOOL flushing);
BOOL win32_ipc_pipe_client_get_mmf (Win32IpcPipeClient * client,
Win32IpcMmf ** mmf,
Win32IpcVideoInfo * info);
void win32_ipc_pipe_client_release_mmf (Win32IpcPipeClient * client,
Win32IpcMmf * mmf);
void win32_ipc_pipe_client_stop (Win32IpcPipeClient * client);
G_END_DECLS

152
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcpipeserver.cpp
View file

@ -41,22 +41,30 @@ GST_DEBUG_CATEGORY_EXTERN (gst_win32_ipc_debug);
struct MmfInfo
{
explicit MmfInfo (Win32IpcMmf * m, const Win32IpcVideoInfo * i, UINT64 s)
explicit MmfInfo (Win32IpcMmf * m, const Win32IpcVideoInfo * i, UINT64 s,
void * u, Win32IpcMmfDestroy n)
{
mmf = m;
info = *i;
seq_num = s;
user_data = u;
notify = n;
}
~MmfInfo()
{
if (mmf)
win32_ipc_mmf_unref (mmf);
if (notify)
notify (user_data);
}
Win32IpcMmf *mmf = nullptr;
Win32IpcVideoInfo info;
UINT64 seq_num;
void *user_data;
Win32IpcMmfDestroy notify;
};
struct ServerConnection : public OVERLAPPED
@ -73,6 +81,7 @@ struct ServerConnection : public OVERLAPPED
Win32IpcPipeServer *self;
std::shared_ptr<MmfInfo> minfo;
std::vector<std::shared_ptr<MmfInfo>> used_minfo;
HANDLE pipe = INVALID_HANDLE_VALUE;
UINT8 client_msg[CONN_BUFFER_SIZE];
UINT32 to_read = 0;
@ -113,7 +122,7 @@ struct Win32IpcPipeServer
};
static void
win32_ipc_pipe_server_receive_need_data_async (ServerConnection * conn);
win32_ipc_pipe_server_wait_client_msg_async (ServerConnection * conn);
static void
win32_ipc_pipe_server_close_connection (ServerConnection * conn,
@ -139,44 +148,6 @@ win32_ipc_pipe_server_close_connection (ServerConnection * conn,
delete conn;
}
static void WINAPI
win32_ipc_pipe_server_receive_read_done_finish (DWORD error_code, DWORD n_bytes,
LPOVERLAPPED overlapped)
{
ServerConnection *conn = (ServerConnection *) overlapped;
if (error_code != ERROR_SUCCESS) {
std::string msg = win32_ipc_error_message (error_code);
GST_WARNING ("READ-DONE failed with 0x%x (%s)",
(UINT) error_code, msg.c_str ());
win32_ipc_pipe_server_close_connection (conn, TRUE);
return;
}
GST_TRACE ("Got READ-DONE %p", conn);
conn->minfo = nullptr;
/* All done, wait for need-data again */
win32_ipc_pipe_server_receive_need_data_async (conn);
}
static void
win32_ipc_pipe_server_receive_read_done_async (ServerConnection * conn)
{
GST_TRACE ("Waiting READ-DONE %p", conn);
if (!ReadFileEx (conn->pipe, conn->client_msg, CONN_BUFFER_SIZE,
(OVERLAPPED *) conn, win32_ipc_pipe_server_receive_read_done_finish)) {
UINT last_err = GetLastError ();
std::string msg = win32_ipc_error_message (last_err);
GST_WARNING ("ReadFileEx failed with 0x%x (%s)", last_err, msg.c_str ());
win32_ipc_pipe_server_close_connection (conn, TRUE);
}
}
static void WINAPI
win32_ipc_pipe_server_send_have_data_finish (DWORD error_code, DWORD n_bytes,
LPOVERLAPPED overlapped)
@ -194,7 +165,7 @@ win32_ipc_pipe_server_send_have_data_finish (DWORD error_code, DWORD n_bytes,
GST_TRACE ("HAVE-DATA done with %s",
win32_ipc_mmf_get_name (conn->minfo->mmf));
win32_ipc_pipe_server_receive_read_done_async (conn);
win32_ipc_pipe_server_wait_client_msg_async (conn);
}
static void
@ -228,11 +199,13 @@ win32_ipc_pipe_server_send_have_data_async (ServerConnection * conn)
}
static void WINAPI
win32_ipc_pipe_server_receive_need_data_finish (DWORD error_code, DWORD n_bytes,
win32_ipc_pipe_server_wait_client_msg_finish (DWORD error_code, DWORD n_bytes,
LPOVERLAPPED overlapped)
{
ServerConnection *conn = (ServerConnection *) overlapped;
UINT64 seq_num;
Win32IpcPktType type;
char mmf_name[1024];
if (error_code != ERROR_SUCCESS) {
std::string msg = win32_ipc_error_message (error_code);
@ -242,32 +215,74 @@ win32_ipc_pipe_server_receive_need_data_finish (DWORD error_code, DWORD n_bytes,
return;
}
if (!win32_ipc_pkt_parse_need_data (conn->client_msg, CONN_BUFFER_SIZE,
&seq_num)) {
GST_ERROR ("Couldn't parse NEED-DATA message");
win32_ipc_pipe_server_close_connection (conn, TRUE);
return;
type = win32_ipc_pkt_type_from_raw (conn->client_msg[0]);
switch (type) {
case WIN32_IPC_PKT_NEED_DATA:
GST_TRACE ("Got NEED-DATA %p", conn);
if (!win32_ipc_pkt_parse_need_data (conn->client_msg, CONN_BUFFER_SIZE,
&seq_num)) {
GST_ERROR ("Couldn't parse NEED-DATA message");
win32_ipc_pipe_server_close_connection (conn, TRUE);
return;
}
/* Will response later once data is available */
if (!conn->minfo) {
GST_LOG ("No data available, waiting");
conn->pending_have_data = TRUE;
return;
}
win32_ipc_pipe_server_send_have_data_async (conn);
break;
case WIN32_IPC_PKT_READ_DONE:
GST_TRACE ("Got READ-DONE %p", conn);
conn->used_minfo.push_back (conn->minfo);
conn->minfo = nullptr;
/* All done, wait for need-data again */
win32_ipc_pipe_server_wait_client_msg_async (conn);
break;
case WIN32_IPC_PKT_RELEASE_DATA:
{
GST_TRACE ("Got RELEASE-DATA %p", conn);
if (!win32_ipc_pkt_parse_release_data (conn->client_msg, CONN_BUFFER_SIZE,
&seq_num, mmf_name)) {
GST_WARNING ("Couldn't parse RELEASE-DATA mssage");
return;
}
auto it = std::find_if (conn->used_minfo.begin (),
conn->used_minfo.end (), [&](const std::shared_ptr<MmfInfo> info) -> bool {
return strcmp (mmf_name, win32_ipc_mmf_get_name (info->mmf)) == 0;
});
if (it != conn->used_minfo.end ()) {
conn->used_minfo.erase (it);
} else {
GST_WARNING ("Unknown memory name %s", mmf_name);
}
win32_ipc_pipe_server_wait_client_msg_async (conn);
break;
}
default:
GST_WARNING ("Unexpected packet type");
win32_ipc_pipe_server_close_connection (conn, TRUE);
break;
}
GST_TRACE ("Got NEED-DATA");
/* Will response later once data is available */
if (!conn->minfo) {
GST_LOG ("No data available, waiting");
conn->pending_have_data = TRUE;
return;
}
win32_ipc_pipe_server_send_have_data_async (conn);
}
static void
win32_ipc_pipe_server_receive_need_data_async (ServerConnection * conn)
win32_ipc_pipe_server_wait_client_msg_async (ServerConnection * conn)
{
GST_TRACE ("Waiting NEED-DATA");
GST_TRACE ("Waiting client message");
if (!ReadFileEx (conn->pipe, conn->client_msg, CONN_BUFFER_SIZE,
(OVERLAPPED *) conn, win32_ipc_pipe_server_receive_need_data_finish)) {
(OVERLAPPED *) conn, win32_ipc_pipe_server_wait_client_msg_finish)) {
UINT last_err = GetLastError ();
std::string msg = win32_ipc_error_message (last_err);
@ -350,13 +365,13 @@ win32_ipc_pipe_server_loop (Win32IpcPipeServer * self)
self->cond.notify_all ();
lk.unlock ();
waitables[0] = overlap.hEvent;
waitables[1] = self->enqueue_event;
waitables[2] = self->cancellable;
do {
ServerConnection *conn;
waitables[0] = overlap.hEvent;
waitables[1] = self->enqueue_event;
waitables[2] = self->cancellable;
/* Enters alertable state and wait for
* 1) Client's connection request
* (similar to socket listen/accept in async manner)
@ -393,7 +408,7 @@ win32_ipc_pipe_server_loop (Win32IpcPipeServer * self)
pipe = INVALID_HANDLE_VALUE;
self->conn.push_back (conn);
win32_ipc_pipe_server_receive_need_data_async (conn);
win32_ipc_pipe_server_wait_client_msg_async (conn);
pipe = win32_ipc_pipe_server_create_pipe (self, &overlap, &io_pending);
if (pipe == INVALID_HANDLE_VALUE)
goto out;
@ -534,10 +549,11 @@ win32_ipc_pipe_server_shutdown (Win32IpcPipeServer * server)
BOOL
win32_ipc_pipe_server_send_mmf (Win32IpcPipeServer * server, Win32IpcMmf * mmf,
const Win32IpcVideoInfo * info)
const Win32IpcVideoInfo * info, void * user_data, Win32IpcMmfDestroy notify)
{
std::lock_guard<std::mutex> lk (server->lock);
server->minfo = std::make_shared<MmfInfo> (mmf, info, server->seq_num);
server->minfo = std::make_shared<MmfInfo> (mmf, info, server->seq_num,
user_data, notify);
GST_LOG ("Enqueue mmf %s", win32_ipc_mmf_get_name (mmf));

6
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcpipeserver.h
View file

@ -34,6 +34,8 @@ G_BEGIN_DECLS
struct Win32IpcPipeServer;
typedef void (*Win32IpcMmfDestroy) (void * user_data);
Win32IpcPipeServer * win32_ipc_pipe_server_new (const char * pipe_name);
Win32IpcPipeServer * win32_ipc_pipe_server_ref (Win32IpcPipeServer * server);
@ -44,7 +46,9 @@ void win32_ipc_pipe_server_shutdown (Win32IpcPipeServer * server
BOOL win32_ipc_pipe_server_send_mmf (Win32IpcPipeServer * server,
Win32IpcMmf * mmf,
const Win32IpcVideoInfo * info);
const Win32IpcVideoInfo * info,
void * user_data,
Win32IpcMmfDestroy notify);
G_END_DECLS

55
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcprotocol.cpp
View file

@ -35,6 +35,8 @@ win32_ipc_pkt_type_to_string (Win32IpcPktType type)
return "HAVE-DATA";
case WIN32_IPC_PKT_READ_DONE:
return "READ-DONE";
case WIN32_IPC_PKT_RELEASE_DATA:
return "RELEASE-DATA";
default:
break;
}
@ -235,3 +237,56 @@ win32_ipc_pkt_parse_read_done (UINT8 * pkt, UINT32 pkt_len, UINT64 * seq_num)
return TRUE;
}
UINT32
win32_ipc_pkt_build_release_data (UINT8 * pkt, UINT32 pkt_size, UINT64 seq_num,
const char * mmf_name)
{
UINT8 *data = pkt;
size_t len;
if (!pkt || !mmf_name)
return 0;
len = strlen (mmf_name);
if (len == 0)
return 0;
len++;
data[0] = win32_ipc_pkt_type_to_raw (WIN32_IPC_PKT_RELEASE_DATA);
data++;
WRITE_UINT64 (data, seq_num);
strcpy ((char *) data, mmf_name);
data += len;
return data - pkt;
}
BOOL
win32_ipc_pkt_parse_release_data (UINT8 * pkt, UINT32 pkt_size,
UINT64 * seq_num, char * mmf_name)
{
UINT8 *data = pkt;
size_t len;
if (win32_ipc_pkt_type_from_raw (pkt[0]) != WIN32_IPC_PKT_RELEASE_DATA)
return FALSE;
data++;
READ_UINT64 (data, seq_num);
len = strnlen ((const char *) data, pkt_size - (data - pkt));
if (len == 0)
return FALSE;
len++;
strcpy (mmf_name, (const char *) data);
data += len;
return TRUE;
}

17
subprojects/gst-plugins-bad/sys/win32ipc/protocol/win32ipcprotocol.h
View file

@ -47,6 +47,12 @@ G_BEGIN_DECLS
* shared-memory | |
* +------->+ |
* |--------- READ-DONE ---------->|
* | |
* +--------+ |
* release | |
* shared-memory | |
* +--------| |
* |------- RELEASE-Data---------->|
*/
typedef enum
@ -55,6 +61,7 @@ typedef enum
WIN32_IPC_PKT_NEED_DATA,
WIN32_IPC_PKT_HAVE_DATA,
WIN32_IPC_PKT_READ_DONE,
WIN32_IPC_PKT_RELEASE_DATA,
} Win32IpcPktType;
/* Same as GstVideoFormat */
@ -239,5 +246,15 @@ BOOL win32_ipc_pkt_parse_read_done (UINT8 * pkt,
UINT32 pkt_size,
UINT64 * seq_num);
UINT32 win32_ipc_pkt_build_release_data (UINT8 * pkt,
UINT32 pkt_size,
UINT64 seq_num,
const char * mmf_name);
BOOL win32_ipc_pkt_parse_release_data (UINT8 * pkt,
UINT32 pkt_size,
UINT64 * seq_num,
char * mmf_name);
G_END_DECLS