gstreamer/subprojects/gst-plugins-bad/sys/win32ipc/gstwin32ipcmemory.cpp

/* 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;
}