gstreamer/subprojects/gst-plugins-bad/sys/d3d11/gstd3d11ipcclient.cpp
Seungha Yang 4449881b08 d3d11: Add support for IPC
Adding d3d11ipcsrc and d3d11ipcsink elements so that Direct3D11 textures
can be shared between processes

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/4514>
2023-05-06 03:19:09 +09:00

1098 lines
30 KiB
C++

/* 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 "gstd3d11ipcclient.h"
#include <gst/d3d11/gstd3d11-private.h>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <atomic>
#include <memory>
#include <wrl.h>
/* *INDENT-OFF* */
using namespace Microsoft::WRL;
/* *INDENT-ON* */
GST_DEBUG_CATEGORY_STATIC (gst_d3d11_ipc_client_debug);
#define GST_CAT_DEFAULT gst_d3d11_ipc_client_debug
GType
gst_d3d11_ipc_io_mode_get_type (void)
{
static GType type = 0;
static const GEnumValue io_modes[] = {
{GST_D3D11_IPC_IO_COPY, "Copy remote memory", "copy"},
{GST_D3D11_IPC_IO_IMPORT, "Import remote memory", "import"},
{0, nullptr, nullptr}
};
GST_D3D11_CALL_ONCE_BEGIN {
type = g_enum_register_static ("GstD3D11IpcIOMode", io_modes);
} GST_D3D11_CALL_ONCE_END;
return type;
}
/* *INDENT-OFF* */
struct GstD3D11IpcClientConn : public OVERLAPPED
{
GstD3D11IpcClientConn (GstD3D11IpcClient * client, HANDLE pipe_handle)
: client (client), pipe (pipe_handle)
{
OVERLAPPED *parent = static_cast<OVERLAPPED *> (this);
parent->Internal = 0;
parent->InternalHigh = 0;
parent->Offset = 0;
parent->OffsetHigh = 0;
client_msg.resize (GST_D3D11_IPC_PKT_HEADER_SIZE);
server_msg.resize (GST_D3D11_IPC_PKT_HEADER_SIZE);
}
~GstD3D11IpcClientConn ()
{
if (pipe != INVALID_HANDLE_VALUE) {
CancelIo (pipe);
CloseHandle (pipe);
}
}
GstD3D11IpcClient *client;
HANDLE pipe = INVALID_HANDLE_VALUE;
GstD3D11IpcPktType type;
std::vector<guint8> client_msg;
std::vector<guint8> server_msg;
};
struct GstD3D11IpcImportData
{
~GstD3D11IpcImportData ()
{
auto dump = gst_d3d11_ipc_wstring_to_string (name);
GST_LOG_OBJECT (client, "Release handle \"%s\"", dump.c_str ());
gst_object_unref (client);
}
GstD3D11IpcClient *client;
ComPtr<ID3D11Texture2D> texture;
ComPtr<IDXGIKeyedMutex> mutex;
GstD3D11IpcMemLayout layout;
std::wstring name;
};
struct GstD3D11IpcReleaseData
{
GstD3D11IpcClient *self;
std::shared_ptr<GstD3D11IpcImportData> imported;
};
struct GstD3D11IpcClientPrivate
{
GstD3D11IpcClientPrivate ()
{
wakeup_event = CreateEvent (nullptr, FALSE, FALSE, nullptr);
cancellable = CreateEvent (nullptr, TRUE, FALSE, nullptr);
shutdown = false;
io_pending = true;
}
~GstD3D11IpcClientPrivate ()
{
gst_clear_caps (&caps);
if (pool) {
gst_buffer_pool_set_active (pool, FALSE);
gst_object_unref (pool);
}
gst_clear_object (&device);
CloseHandle (wakeup_event);
CloseHandle (cancellable);
}
std::string address;
GstD3D11IpcIOMode io_mode = GST_D3D11_IPC_IO_COPY;
GstClockTime timeout;
HANDLE wakeup_event;
HANDLE cancellable;
std::mutex lock;
std::condition_variable cond;
GstD3D11Device *device = nullptr;
GstCaps *caps = nullptr;
GstBufferPool *pool = nullptr;
GstVideoInfo info;
bool server_eos = false;
bool flushing = false;
bool aborted = false;
bool sent_fin = false;
std::atomic<bool> shutdown;
std::atomic<bool> io_pending;
GThread *loop_thread = nullptr;
std::queue <GstSample *> samples;
std::shared_ptr<GstD3D11IpcClientConn> conn;
std::queue<std::wstring> unused_data;
std::vector<std::weak_ptr<GstD3D11IpcImportData>> imported;
};
/* *INDENT-ON* */
struct _GstD3D11IpcClient
{
GstObject parent;
GstD3D11IpcClientPrivate *priv;
};
static void gst_d3d11_ipc_client_dispose (GObject * object);
static void gst_d3d11_ipc_client_finalize (GObject * object);
static void gst_d3d11_ipc_client_continue (GstD3D11IpcClient * self);
static void gst_d3d11_ipc_client_send_msg (GstD3D11IpcClient * self);
#define gst_d3d11_ipc_client_parent_class parent_class
G_DEFINE_TYPE (GstD3D11IpcClient, gst_d3d11_ipc_client, GST_TYPE_OBJECT);
static void
gst_d3d11_ipc_client_class_init (GstD3D11IpcClientClass * klass)
{
GObjectClass *object_class = G_OBJECT_CLASS (klass);
object_class->dispose = gst_d3d11_ipc_client_dispose;
object_class->finalize = gst_d3d11_ipc_client_finalize;
GST_DEBUG_CATEGORY_INIT (gst_d3d11_ipc_client_debug, "d3d11ipcclient",
0, "d3d11ipcclient");
}
static void
gst_d3d11_ipc_client_init (GstD3D11IpcClient * self)
{
self->priv = new GstD3D11IpcClientPrivate ();
}
static void
gst_d3d11_ipc_client_dispose (GObject * object)
{
GstD3D11IpcClient *self = GST_D3D11_IPC_CLIENT (object);
GstD3D11IpcClientPrivate *priv = self->priv;
GST_DEBUG_OBJECT (self, "dispose");
SetEvent (priv->cancellable);
g_clear_pointer (&priv->loop_thread, g_thread_join);
G_OBJECT_CLASS (parent_class)->dispose (object);
}
static void
gst_d3d11_ipc_client_finalize (GObject * object)
{
GstD3D11IpcClient *self = GST_D3D11_IPC_CLIENT (object);
GST_DEBUG_OBJECT (self, "finalize");
delete self->priv;
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
gst_d3d11_ipc_client_abort (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
std::lock_guard < std::mutex > lk (priv->lock);
priv->aborted = true;
priv->cond.notify_all ();
}
static bool
gst_d3d11_client_update_caps (GstD3D11IpcClient * self, GstCaps * caps)
{
GstD3D11IpcClientPrivate *priv = self->priv;
if (!caps)
return true;
gst_clear_caps (&priv->caps);
priv->caps = caps;
if (priv->pool) {
gst_buffer_pool_set_active (priv->pool, FALSE);
gst_clear_object (&priv->pool);
}
if (!gst_video_info_from_caps (&priv->info, caps)) {
GST_ERROR_OBJECT (self, "Invalid caps");
return false;
}
if (priv->io_mode == GST_D3D11_IPC_IO_COPY) {
GstStructure *config;
GstD3D11AllocationParams *params;
guint bind_flags = 0;
GstD3D11Format device_format;
if (!gst_d3d11_device_get_format (priv->device,
GST_VIDEO_INFO_FORMAT (&priv->info), &device_format)) {
GST_ERROR_OBJECT (self, "Couldn't get device format");
return false;
}
if ((device_format.format_support[0] &
(guint) D3D11_FORMAT_SUPPORT_SHADER_SAMPLE) != 0) {
bind_flags |= D3D11_BIND_SHADER_RESOURCE;
}
if ((device_format.format_support[0] &
(guint) D3D11_FORMAT_SUPPORT_RENDER_TARGET) != 0) {
bind_flags |= D3D11_BIND_RENDER_TARGET;
}
priv->pool = gst_d3d11_buffer_pool_new (priv->device);
config = gst_buffer_pool_get_config (priv->pool);
gst_buffer_pool_config_add_option (config,
GST_BUFFER_POOL_OPTION_VIDEO_META);
gst_buffer_pool_config_set_params (config, priv->caps,
GST_VIDEO_INFO_SIZE (&priv->info), 0, 0);
params = gst_d3d11_allocation_params_new (priv->device, &priv->info,
GST_D3D11_ALLOCATION_FLAG_DEFAULT, bind_flags, 0);
gst_buffer_pool_config_set_d3d11_allocation_params (config, params);
gst_d3d11_allocation_params_free (params);
if (!gst_buffer_pool_set_config (priv->pool, config)) {
GST_ERROR_OBJECT (self, "Couldn't set pool config");
gst_clear_object (&priv->pool);
return false;
}
if (!gst_buffer_pool_set_active (priv->pool, TRUE)) {
GST_ERROR_OBJECT (self, "Couldn't active pool");
gst_clear_object (&priv->pool);
return false;
}
}
return true;
}
static bool
gst_d3d11_ipc_client_config_data (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
gint64 prev_luid, luid;
GstCaps *caps = nullptr;
auto conn = priv->conn;
std::lock_guard < std::mutex > lk (priv->lock);
g_object_get (priv->device, "adapter-luid", &prev_luid, nullptr);
if (!gst_d3d11_ipc_pkt_parse_config (conn->server_msg, luid, &caps)) {
GST_ERROR_OBJECT (self, "Couldn't parse CONFIG-DATA");
return false;
}
if (prev_luid != luid) {
GstD3D11Device *device = gst_d3d11_device_new_for_adapter_luid (luid,
D3D11_CREATE_DEVICE_BGRA_SUPPORT);
if (!device) {
GST_ERROR_OBJECT (self, "Couldn't create device");
return false;
}
gst_object_unref (priv->device);
priv->device = device;
}
if (!gst_d3d11_client_update_caps (self, caps))
return false;
priv->cond.notify_all ();
return true;
}
static void
gst_d3d11_ipc_client_release_imported_data (GstD3D11IpcReleaseData * data)
{
GstD3D11IpcClient *self = data->self;
GstD3D11IpcClientPrivate *priv = self->priv;
auto name = data->imported->name;
auto handle_dump = gst_d3d11_ipc_wstring_to_string (name);
GST_LOG_OBJECT (self, "Releasing data \"%s\"", handle_dump.c_str ());
data->imported = nullptr;
priv->lock.lock ();
priv->unused_data.push (name);
priv->lock.unlock ();
SetEvent (priv->wakeup_event);
gst_object_unref (data->self);
delete data;
}
static bool
gst_d3d11_ipc_client_have_data (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
GstBuffer *buffer;
GstMemory *mem;
GstSample *sample;
GstClockTime pts;
GstCaps *caps = nullptr;
GstD3D11IpcMemLayout layout;
std::shared_ptr < GstD3D11IpcImportData > import_data;
std::unique_lock < std::mutex > lk (priv->lock);
std::wstring name;
auto conn = priv->conn;
ComPtr < ID3D11Texture2D > texture;
HRESULT hr;
if (!gst_d3d11_ipc_pkt_parse_have_data (conn->server_msg,
pts, layout, name, &caps)) {
GST_ERROR_OBJECT (self, "Couldn't parse HAVE-DATA packet");
return false;
}
if (!gst_d3d11_client_update_caps (self, caps))
return false;
auto handle_dump = gst_d3d11_ipc_wstring_to_string (name);
GST_LOG_OBJECT (self, "Importing handle \"%s\"", handle_dump.c_str ());
/* Check if this memory handle was imported already */
/* *INDENT-OFF* */
if (priv->io_mode == GST_D3D11_IPC_IO_IMPORT && !priv->imported.empty ()) {
GST_LOG_OBJECT (self, "Checking already imported handles, size %"
G_GSIZE_FORMAT, priv->imported.size ());
for (auto it = priv->imported.begin (); it != priv->imported.end (); ) {
auto data = it->lock ();
if (!data) {
it = priv->imported.erase (it);
} else {
if (data->name == name) {
GST_DEBUG_OBJECT (self, "Already imported handle");
import_data = data;
break;
}
it++;
}
}
}
/* *INDENT-ON* */
if (!import_data) {
ID3D11Device *device = gst_d3d11_device_get_device_handle (priv->device);
ComPtr < ID3D11Device1 > device1;
ComPtr < IDXGIKeyedMutex > mutex;
hr = device->QueryInterface (IID_PPV_ARGS (&device1));
if (!gst_d3d11_result (hr, priv->device)) {
GST_ERROR_OBJECT (self, "ID3D11Device1 interface is not available");
return false;
}
hr = device1->OpenSharedResourceByName (name.c_str (),
DXGI_SHARED_RESOURCE_READ, IID_PPV_ARGS (&texture));
if (!gst_d3d11_result (hr, priv->device)) {
GST_ERROR_OBJECT (self, "Couldn't open resource");
return false;
}
hr = texture->QueryInterface (IID_PPV_ARGS (&mutex));
if (!gst_d3d11_result (hr, priv->device)) {
GST_ERROR_OBJECT (self, "couldn't get keyed mutex interface");
return false;
}
import_data = std::make_shared < GstD3D11IpcImportData > ();
import_data->client = (GstD3D11IpcClient *) gst_object_ref (self);
import_data->texture = texture;
import_data->mutex = mutex;
import_data->layout = layout;
import_data->name = name;
if (priv->io_mode == GST_D3D11_IPC_IO_IMPORT)
priv->imported.push_back (import_data);
} else {
texture = import_data->texture;
}
if (priv->io_mode == GST_D3D11_IPC_IO_COPY) {
ID3D11DeviceContext *context =
gst_d3d11_device_get_device_context_handle (priv->device);
ID3D11Texture2D *dst_texture;
D3D11_BOX src_box = { 0, };
D3D11_TEXTURE2D_DESC dst_desc, src_desc;
GstMapInfo info;
hr = import_data->mutex->AcquireSync (0, INFINITE);
if (hr != S_OK) {
GST_ERROR_OBJECT (self, "Couldn't acquire sync");
return false;
}
gst_buffer_pool_acquire_buffer (priv->pool, &buffer, nullptr);
mem = gst_buffer_peek_memory (buffer, 0);
gst_memory_map (mem, &info, (GstMapFlags) (GST_MAP_WRITE | GST_MAP_D3D11));
dst_texture = (ID3D11Texture2D *) info.data;
dst_texture->GetDesc (&dst_desc);
texture->GetDesc (&src_desc);
src_box.left = 0;
src_box.top = 0;
src_box.front = 0;
src_box.back = 1;
src_box.right = MIN (src_desc.Width, dst_desc.Width);
src_box.bottom = MIN (src_desc.Height, dst_desc.Height);
context->CopySubresourceRegion (dst_texture, 0, 0, 0, 0,
texture.Get (), 0, &src_box);
import_data->mutex->ReleaseSync (0);
gst_memory_unmap (mem, &info);
priv->unused_data.push (name);
} else {
GstMemory *mem;
gint stride[GST_VIDEO_MAX_PLANES];
gsize offset[GST_VIDEO_MAX_PLANES];
for (guint i = 0; i < GST_VIDEO_MAX_PLANES; i++) {
stride[i] = import_data->layout.pitch;
offset[i] = import_data->layout.offset[i];
}
auto data = new GstD3D11IpcReleaseData ();
data->self = (GstD3D11IpcClient *) gst_object_ref (self);
data->imported = import_data;
mem = gst_d3d11_allocator_alloc_wrapped (nullptr, priv->device,
texture.Get (), import_data->layout.size, data,
(GDestroyNotify) gst_d3d11_ipc_client_release_imported_data);
GST_MINI_OBJECT_FLAG_SET (mem, GST_MEMORY_FLAG_READONLY);
buffer = gst_buffer_new ();
gst_buffer_append_memory (buffer, mem);
gst_buffer_add_video_meta_full (buffer, GST_VIDEO_FRAME_FLAG_NONE,
GST_VIDEO_INFO_FORMAT (&priv->info), GST_VIDEO_INFO_WIDTH (&priv->info),
GST_VIDEO_INFO_HEIGHT (&priv->info),
GST_VIDEO_INFO_N_PLANES (&priv->info), offset, stride);
}
GST_BUFFER_PTS (buffer) = pts;
GST_BUFFER_DTS (buffer) = GST_CLOCK_TIME_NONE;
GST_BUFFER_DURATION (buffer) = GST_CLOCK_TIME_NONE;
sample = gst_sample_new (buffer, priv->caps, nullptr, nullptr);
gst_buffer_unref (buffer);
/* Drops too old samples */
std::queue < GstSample * >drop_queue;
while (priv->samples.size () > 2) {
drop_queue.push (priv->samples.front ());
priv->samples.pop ();
}
priv->samples.push (sample);
priv->cond.notify_all ();
lk.unlock ();
import_data = nullptr;
while (!drop_queue.empty ()) {
auto old = drop_queue.front ();
gst_sample_unref (old);
drop_queue.pop ();
}
return true;
}
static void
gst_d3d11_ipc_client_wait_msg_finish (GstD3D11IpcClient * client)
{
GstD3D11IpcClientPrivate *priv = client->priv;
GstD3D11IpcPacketHeader header;
auto conn = priv->conn;
if (!gst_d3d11_ipc_pkt_identify (conn->server_msg, header)) {
GST_ERROR_OBJECT (client, "Broken header");
gst_d3d11_ipc_client_abort (client);
return;
}
switch (header.type) {
case GstD3D11IpcPktType::CONFIG:
GST_LOG_OBJECT (client, "Got CONFIG");
if (!gst_d3d11_ipc_client_config_data (client)) {
gst_d3d11_ipc_client_abort (client);
return;
}
gst_d3d11_ipc_client_continue (client);
break;
case GstD3D11IpcPktType::HAVE_DATA:
GST_LOG_OBJECT (client, "Got HAVE-DATA");
gst_d3d11_device_lock (priv->device);
if (!gst_d3d11_ipc_client_have_data (client)) {
gst_d3d11_device_unlock (priv->device);
gst_d3d11_ipc_client_abort (client);
return;
}
gst_d3d11_device_unlock (priv->device);
GST_LOG_OBJECT (client, "Sending READ-DONE");
gst_d3d11_ipc_pkt_build_read_done (conn->client_msg);
conn->type = GstD3D11IpcPktType::READ_DONE;
gst_d3d11_ipc_client_send_msg (client);
break;
case GstD3D11IpcPktType::EOS:
GST_DEBUG_OBJECT (client, "Got EOS");
priv->server_eos = true;
priv->lock.lock ();
priv->cond.notify_all ();
priv->lock.unlock ();
gst_d3d11_ipc_client_continue (client);
break;
default:
GST_WARNING_OBJECT (client, "Unexpected packet type");
gst_d3d11_ipc_client_abort (client);
break;
}
}
static void WINAPI
gst_d3d11_ipc_client_payload_finish (DWORD error_code, DWORD size,
OVERLAPPED * overlap)
{
GstD3D11IpcClientConn *conn =
static_cast < GstD3D11IpcClientConn * >(overlap);
GstD3D11IpcClient *self = conn->client;
if (error_code != ERROR_SUCCESS) {
auto err = gst_d3d11_ipc_win32_error_to_string (error_code);
GST_WARNING_OBJECT (self, "ReadFileEx callback failed with 0x%x (%s)",
(guint) error_code, err.c_str ());
gst_d3d11_ipc_client_abort (self);
}
gst_d3d11_ipc_client_wait_msg_finish (self);
}
static void WINAPI
gst_d3d11_ipc_client_win32_wait_header_finish (DWORD error_code, DWORD size,
OVERLAPPED * overlap)
{
GstD3D11IpcClientConn *conn =
static_cast < GstD3D11IpcClientConn * >(overlap);
GstD3D11IpcClient *self = conn->client;
GstD3D11IpcPacketHeader header;
if (error_code != ERROR_SUCCESS) {
auto err = gst_d3d11_ipc_win32_error_to_string (error_code);
GST_WARNING_OBJECT (self, "ReadFileEx callback failed with 0x%x (%s)",
(guint) error_code, err.c_str ());
gst_d3d11_ipc_client_abort (self);
return;
}
if (!gst_d3d11_ipc_pkt_identify (conn->server_msg, header)) {
GST_ERROR_OBJECT (self, "Broken header");
gst_d3d11_ipc_client_abort (self);
return;
}
if (header.payload_size == 0) {
gst_d3d11_ipc_client_wait_msg_finish (self);
return;
}
GST_LOG_OBJECT (self, "Reading payload");
if (!ReadFileEx (conn->pipe, &conn->server_msg[0] +
GST_D3D11_IPC_PKT_HEADER_SIZE, header.payload_size, conn,
gst_d3d11_ipc_client_payload_finish)) {
guint last_err = GetLastError ();
auto err = gst_d3d11_ipc_win32_error_to_string (last_err);
GST_WARNING_OBJECT (self, "ReadFileEx failed with 0x%x (%s)",
last_err, err.c_str ());
gst_d3d11_ipc_client_abort (self);
}
}
static void
gst_d3d11_ipc_client_wait_msg (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
auto conn = priv->conn;
priv->io_pending = true;
if (!ReadFileEx (conn->pipe, &conn->server_msg[0],
GST_D3D11_IPC_PKT_HEADER_SIZE, conn.get (),
gst_d3d11_ipc_client_win32_wait_header_finish)) {
guint last_err = GetLastError ();
auto err = gst_d3d11_ipc_win32_error_to_string (last_err);
GST_WARNING_OBJECT (self, "ReadFileEx failed with 0x%x (%s)",
last_err, err.c_str ());
gst_d3d11_ipc_client_abort (self);
}
}
static void WINAPI
gst_cuda_ipc_client_win32_send_msg_finish (DWORD error_code, DWORD size,
OVERLAPPED * overlap)
{
GstD3D11IpcClientConn *conn =
static_cast < GstD3D11IpcClientConn * >(overlap);
GstD3D11IpcClient *self = conn->client;
if (error_code != ERROR_SUCCESS) {
auto err = gst_d3d11_ipc_win32_error_to_string (error_code);
GST_WARNING_OBJECT (self, "WriteFileEx callback failed with 0x%x (%s)",
(guint) error_code, err.c_str ());
gst_d3d11_ipc_client_abort (self);
return;
}
switch (conn->type) {
case GstD3D11IpcPktType::NEED_DATA:
GST_LOG_OBJECT (self, "Sent NEED-DATA");
gst_d3d11_ipc_client_wait_msg (self);
break;
case GstD3D11IpcPktType::READ_DONE:
GST_LOG_OBJECT (self, "Sent READ-DONE");
gst_d3d11_ipc_client_continue (self);
break;
case GstD3D11IpcPktType::RELEASE_DATA:
GST_LOG_OBJECT (self, "Sent RELEASE-DATA");
gst_d3d11_ipc_client_continue (self);
break;
case GstD3D11IpcPktType::FIN:
GST_DEBUG_OBJECT (self, "Sent FIN");
gst_d3d11_ipc_client_abort (self);
break;
default:
GST_ERROR_OBJECT (self, "Unexpected msg type");
gst_d3d11_ipc_client_abort (self);
break;
}
}
static void
gst_d3d11_ipc_client_send_msg (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
auto conn = priv->conn;
priv->io_pending = true;
if (!WriteFileEx (conn->pipe, &conn->client_msg[0],
conn->client_msg.size (), conn.get (),
gst_cuda_ipc_client_win32_send_msg_finish)) {
guint last_err = GetLastError ();
auto err = gst_d3d11_ipc_win32_error_to_string (last_err);
GST_WARNING_OBJECT (self, "WriteFileEx failed with 0x%x (%s)",
last_err, err.c_str ());
gst_d3d11_ipc_client_abort (self);
}
}
static void
gst_d3d11_ipc_client_run_gc (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
for (auto it = priv->imported.begin (); it != priv->imported.end ();) {
auto data = it->lock ();
if (!data) {
it = priv->imported.erase (it);
} else {
it++;
}
}
}
static void
gst_d3d11_ipc_client_continue (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
std::unique_lock < std::mutex > lk (priv->lock);
auto conn = priv->conn;
if (!conn) {
GST_WARNING_OBJECT (self, "No connection was made");
priv->aborted = true;
priv->cond.notify_all ();
return;
}
if (priv->aborted) {
priv->cond.notify_all ();
GST_DEBUG_OBJECT (self, "Operation was aborted");
return;
}
if (!priv->unused_data.empty ()) {
auto name = priv->unused_data.front ();
priv->unused_data.pop ();
auto handle_dump = gst_d3d11_ipc_wstring_to_string (name);
GST_LOG_OBJECT (self, "Sending RELEASE-DATA \"%s\"", handle_dump.c_str ());
gst_d3d11_ipc_pkt_build_release_data (conn->client_msg, name);
conn->type = GstD3D11IpcPktType::RELEASE_DATA;
lk.unlock ();
gst_d3d11_ipc_client_send_msg (self);
return;
}
if (priv->shutdown) {
auto drop_queue = priv->samples;
while (!priv->samples.empty ())
priv->samples.pop ();
lk.unlock ();
while (!drop_queue.empty ()) {
auto sample = drop_queue.front ();
gst_sample_unref (sample);
drop_queue.pop ();
}
lk.lock ();
}
if (priv->server_eos || priv->shutdown) {
gst_d3d11_ipc_client_run_gc (self);
GST_DEBUG_OBJECT (self, "Remaining imported memory %" G_GSIZE_FORMAT,
priv->imported.size ());
if (priv->imported.empty ()) {
GST_DEBUG_OBJECT (self, "Drained");
if (priv->sent_fin) {
priv->aborted = true;
priv->cond.notify_all ();
} else {
lk.unlock ();
priv->sent_fin = true;
gst_d3d11_ipc_pkt_build_fin (conn->client_msg);
conn->type = GstD3D11IpcPktType::FIN;
GST_DEBUG_OBJECT (self, "Sending FIN");
gst_d3d11_ipc_client_send_msg (self);
return;
}
} else {
priv->io_pending = false;
}
return;
}
lk.unlock ();
gst_d3d11_ipc_pkt_build_need_data (conn->client_msg);
conn->type = GstD3D11IpcPktType::NEED_DATA;
GST_LOG_OBJECT (self, "Sending NEED-DATA");
gst_d3d11_ipc_client_send_msg (self);
}
static gpointer
gst_d3d11_ipc_client_loop_thread_func (GstD3D11IpcClient * self)
{
GstD3D11IpcClientPrivate *priv = self->priv;
DWORD mode = PIPE_READMODE_MESSAGE;
guint wait_ret;
HANDLE pipe = INVALID_HANDLE_VALUE;
GstClockTime start_time = gst_d3d11_ipc_clock_now ();
HANDLE waitables[] = { priv->cancellable, priv->wakeup_event };
std::wstring address = gst_d3d11_ipc_string_to_wstring (priv->address);
#if (_WIN32_WINNT >= _WIN32_WINNT_WIN8)
CREATEFILE2_EXTENDED_PARAMETERS params;
memset (&params, 0, sizeof (CREATEFILE2_EXTENDED_PARAMETERS));
params.dwSize = sizeof (CREATEFILE2_EXTENDED_PARAMETERS);
params.dwFileAttributes = 0;
params.dwFileFlags = FILE_FLAG_OVERLAPPED;
params.dwSecurityQosFlags = SECURITY_IMPERSONATION;
#endif
GST_DEBUG_OBJECT (self, "Starting loop thread");
std::unique_lock < std::mutex > lk (priv->lock);
do {
GstClockTime diff;
if (priv->flushing) {
GST_DEBUG_OBJECT (self, "We are flushing");
priv->aborted = true;
priv->cond.notify_all ();
goto out;
}
#if (_WIN32_WINNT >= _WIN32_WINNT_WIN8)
pipe = CreateFile2 (address.c_str (), GENERIC_READ | GENERIC_WRITE, 0,
OPEN_EXISTING, &params);
#else
pipe = CreateFileW (address.c_str (),
GENERIC_READ | GENERIC_WRITE, 0, nullptr, OPEN_EXISTING,
FILE_FLAG_OVERLAPPED, nullptr);
#endif
if (pipe != INVALID_HANDLE_VALUE)
break;
if (priv->timeout > 0) {
diff = gst_d3d11_ipc_clock_now () - start_time;
if (diff > priv->timeout) {
GST_WARNING_OBJECT (self, "Timeout");
priv->aborted = true;
priv->cond.notify_all ();
goto out;
}
}
/* Retry per 100ms */
GST_DEBUG_OBJECT (self, "Sleep for next retry");
priv->cond.wait_for (lk, std::chrono::milliseconds (100));
} while (true);
if (!SetNamedPipeHandleState (pipe, &mode, nullptr, nullptr)) {
guint last_err = GetLastError ();
auto err = gst_d3d11_ipc_win32_error_to_string (last_err);
GST_WARNING_OBJECT (self, "SetNamedPipeHandleState failed with 0x%x (%s)",
last_err, err.c_str ());
CloseHandle (pipe);
priv->aborted = true;
priv->cond.notify_all ();
goto out;
}
priv->conn = std::make_shared < GstD3D11IpcClientConn > (self, pipe);
priv->cond.notify_all ();
lk.unlock ();
gst_d3d11_ipc_client_wait_msg (self);
do {
/* Enters alertable thread state and wait for I/O completion event
* or cancellable event */
wait_ret = WaitForMultipleObjectsEx (G_N_ELEMENTS (waitables), waitables,
FALSE, INFINITE, TRUE);
if (wait_ret == WAIT_OBJECT_0) {
GST_DEBUG ("Operation cancelled");
goto out;
}
switch (wait_ret) {
case WAIT_IO_COMPLETION:
break;
case WAIT_OBJECT_0 + 1:
if (!priv->io_pending)
gst_d3d11_ipc_client_continue (self);
break;
default:
GST_WARNING ("Unexpected wait return 0x%x", wait_ret);
gst_d3d11_ipc_client_abort (self);
goto out;
}
} while (true);
out:
while (!priv->samples.empty ()) {
auto sample = priv->samples.front ();
gst_sample_unref (sample);
priv->samples.pop ();
}
priv->conn = nullptr;
GST_DEBUG_OBJECT (self, "Exit loop thread");
return nullptr;
}
GstFlowReturn
gst_d3d11_ipc_client_run (GstD3D11IpcClient * client)
{
GstD3D11IpcClientPrivate *priv;
g_return_val_if_fail (GST_IS_D3D11_IPC_CLIENT (client), GST_FLOW_ERROR);
priv = client->priv;
std::unique_lock < std::mutex > lk (priv->lock);
if (!priv->loop_thread) {
priv->loop_thread = g_thread_new ("d3d11-ipc-client",
(GThreadFunc) gst_d3d11_ipc_client_loop_thread_func, client);
while (!priv->caps && !priv->aborted && !priv->flushing)
priv->cond.wait (lk);
}
if (priv->flushing) {
GST_DEBUG_OBJECT (client, "We are flushing");
return GST_FLOW_FLUSHING;
} else if (priv->aborted || !priv->caps) {
GST_DEBUG_OBJECT (client, "Aborted");
return GST_FLOW_ERROR;
}
return GST_FLOW_OK;
}
GstCaps *
gst_d3d11_ipc_client_get_caps (GstD3D11IpcClient * client)
{
GstCaps *caps = nullptr;
GstD3D11IpcClientPrivate *priv;
g_return_val_if_fail (GST_IS_D3D11_IPC_CLIENT (client), nullptr);
priv = client->priv;
if (gst_d3d11_ipc_client_run (client) != GST_FLOW_OK)
return nullptr;
std::lock_guard < std::mutex > lk (priv->lock);
if (priv->caps)
caps = gst_caps_ref (priv->caps);
return caps;
}
void
gst_d3d11_ipc_client_stop (GstD3D11IpcClient * client)
{
GstD3D11IpcClientPrivate *priv;
g_return_if_fail (GST_IS_D3D11_IPC_CLIENT (client));
priv = client->priv;
GST_DEBUG_OBJECT (client, "Stopping");
priv->shutdown = true;
SetEvent (priv->wakeup_event);
std::unique_lock < std::mutex > lk (priv->lock);
while (!priv->aborted)
priv->cond.wait (lk);
lk.unlock ();
GST_DEBUG_OBJECT (client, "Terminating");
SetEvent (priv->cancellable);
g_clear_pointer (&priv->loop_thread, g_thread_join);
GST_DEBUG_OBJECT (client, "Stopped");
}
void
gst_d3d11_ipc_client_set_flushing (GstD3D11IpcClient * client, bool flushing)
{
GstD3D11IpcClientPrivate *priv;
g_return_if_fail (GST_IS_D3D11_IPC_CLIENT (client));
priv = client->priv;
std::lock_guard < std::mutex > lk (priv->lock);
priv->flushing = flushing;
priv->cond.notify_all ();
}
GstFlowReturn
gst_d3d11_ipc_client_get_sample (GstD3D11IpcClient * client,
GstSample ** sample)
{
GstD3D11IpcClientPrivate *priv;
g_return_val_if_fail (GST_IS_D3D11_IPC_CLIENT (client), GST_FLOW_ERROR);
g_return_val_if_fail (sample, GST_FLOW_ERROR);
priv = client->priv;
GST_LOG_OBJECT (client, "Waiting for sample");
std::unique_lock < std::mutex > lk (priv->lock);
while (!priv->flushing && !priv->aborted && !priv->server_eos &&
priv->samples.empty ()) {
priv->cond.wait (lk);
}
if (!priv->samples.empty ()) {
*sample = priv->samples.front ();
priv->samples.pop ();
GST_LOG_OBJECT (client, "Have sample");
return GST_FLOW_OK;
}
if (priv->flushing) {
GST_DEBUG_OBJECT (client, "Flushing");
return GST_FLOW_FLUSHING;
}
GST_DEBUG_OBJECT (client, "EOS");
return GST_FLOW_EOS;
}
GstD3D11IpcClient *
gst_d3d11_ipc_client_new (const std::string & address, GstD3D11Device * device,
GstD3D11IpcIOMode io_mode, guint timeout)
{
GstD3D11IpcClient *self;
GstD3D11IpcClientPrivate *priv;
g_return_val_if_fail (GST_IS_D3D11_DEVICE (device), nullptr);
self = (GstD3D11IpcClient *)
g_object_new (GST_TYPE_D3D11_IPC_CLIENT, nullptr);
gst_object_ref_sink (self);
priv = self->priv;
priv->address = address;
priv->timeout = timeout * GST_SECOND;
priv->io_mode = io_mode;
priv->device = (GstD3D11Device *) gst_object_ref (device);
return self;
}