mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-26 07:00:13 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

work on stream negotiation and start

Browse source 
Add more buffer types to add and remove memory shared memory between the
server and client. We would like to send buffers only once and then
simply reference them by index.
Do format negotiation and stream start with a START message.
This commit is contained in:
Wim Taymans 2016-07-21 18:38:24 +02:00
parent 31041a4e16
commit af3de36416
23 changed files with 1246 additions and 349 deletions
Download patch file Download diff file Expand all files Collapse all files

52
pinos/gst/gstpinosdepay.c
View file

@ -115,12 +115,12 @@ gst_pinos_depay_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
}
static void
release_fds (GstPinosDepay *this, GstBuffer *buffer)
reuse_fds (GstPinosDepay *this, GstBuffer *buffer)
{
GArray *fdids;
guint i;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosPacketReuseMem r;
PinosBuffer pbuf;
gsize size;
gpointer data;
@ -136,8 +136,8 @@ release_fds (GstPinosDepay *this, GstBuffer *buffer)
for (i = 0; i < fdids->len; i++) {
r.id = g_array_index (fdids, guint32, i);
GST_LOG ("release fd index %d", r.id);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
GST_LOG ("reuse mem id %d", r.id);
pinos_buffer_builder_add_reuse_mem (&b, &r);
}
pinos_buffer_builder_end (&b, &pbuf);
g_array_unref (fdids);
@ -204,24 +204,52 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
GST_BUFFER_OFFSET (outbuf) = hdr.seq;
break;
}
case PINOS_PACKET_TYPE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_ADD_MEM:
{
GstMemory *fdmem = NULL;
PinosPacketFDPayload p;
PinosPacketAddMem p;
int fd;
if (!pinos_buffer_iter_parse_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_add_mem (&it, &p))
goto error;
fd = g_unix_fd_list_get (fds, p.fd_index, &err);
if (fd == -1)
goto error;
fdmem = gst_fd_allocator_alloc (depay->fd_allocator, fd,
p.offset + p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, p.offset, p.size);
g_hash_table_insert (depay->mem_ids, GINT_TO_POINTER (p.id), fdmem);
break;
}
case PINOS_PACKET_TYPE_REMOVE_MEM:
{
PinosPacketRemoveMem p;
if (!pinos_buffer_iter_parse_remove_mem (&it, &p))
goto error;
g_hash_table_remove (depay->mem_ids, GINT_TO_POINTER (p.id));
break;
}
case PINOS_PACKET_TYPE_PROCESS_MEM:
{
GstMemory *fdmem = NULL;
PinosPacketProcessMem p;
if (!pinos_buffer_iter_parse_process_mem (&it, &p))
goto error;
fdmem = g_hash_table_lookup (depay->mem_ids, GINT_TO_POINTER (p.id));
if (fdmem == NULL)
goto error;
if (outbuf == NULL)
outbuf = gst_buffer_new ();
fdmem = gst_fd_allocator_alloc (depay->fd_allocator, fd,
p.offset + p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, p.offset, p.size);
fdmem = gst_memory_share (fdmem, p.offset, p.size);
gst_buffer_append_memory (outbuf, fdmem);
if (fdids == NULL)
@ -261,7 +289,7 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (outbuf),
fdids_quark, fdids, NULL);
gst_mini_object_weak_ref (GST_MINI_OBJECT_CAST (outbuf),
(GstMiniObjectNotify) release_fds, g_object_ref (depay));
(GstMiniObjectNotify) reuse_fds, g_object_ref (depay));
}
return gst_pad_push (depay->srcpad, outbuf);
}
@ -326,6 +354,7 @@ gst_pinos_depay_finalize (GObject * object)
gst_caps_replace (&depay->caps, NULL);
g_object_unref (depay->fd_allocator);
g_hash_table_unref (depay->mem_ids);
G_OBJECT_CLASS (gst_pinos_depay_parent_class)->finalize (object);
}
@ -376,4 +405,5 @@ gst_pinos_depay_init (GstPinosDepay * depay)
gst_element_add_pad (GST_ELEMENT (depay), depay->sinkpad);
depay->fd_allocator = gst_fd_allocator_new ();
depay->mem_ids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, (GDestroyNotify) gst_memory_unref);
}

1
pinos/gst/gstpinosdepay.h
View file

@ -43,6 +43,7 @@ struct _GstPinosDepay
GstPad *srcpad, *sinkpad;
GstAllocator *fd_allocator;
GHashTable *mem_ids;
};
struct _GstPinosDepayClass

36
pinos/gst/gstpinospay.c
View file

@ -218,17 +218,17 @@ client_buffer_received (GstPinosPay *pay, GstBuffer *buffer,
pinos_buffer_iter_init (&it, &pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_RELEASE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_REUSE_MEM:
{
PinosPacketReleaseFDPayload p;
PinosPacketReuseMem p;
gint id;
if (!pinos_buffer_iter_parse_release_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_reuse_mem (&it, &p))
continue;
id = p.id;
GST_LOG ("fd index %d for client %s is released", id, client_path);
GST_LOG ("fd index %d for client %s is reused", id, client_path);
pinos_fd_manager_remove (pay->fdmanager, client_path, id);
break;
}
@ -358,7 +358,7 @@ release_fds (GstPinosPay *pay, GstBuffer *buffer)
GArray *fdids;
guint i;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosPacketReuseMem r;
PinosBuffer pbuf;
gsize size;
gpointer data;
@ -375,7 +375,7 @@ release_fds (GstPinosPay *pay, GstBuffer *buffer)
for (i = 0; i < fdids->len; i++) {
r.id = g_array_index (fdids, guint32, i);
GST_LOG ("release fd index %d", r.id);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
pinos_buffer_builder_add_reuse_mem (&b, &r);
}
pinos_buffer_builder_end (&b, &pbuf);
g_array_unref (fdids);
@ -406,11 +406,11 @@ gst_pinos_pay_chain_pinos (GstPinosPay *pay, GstBuffer * buffer)
pinos_buffer_iter_init (&it, &pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_PROCESS_MEM:
{
PinosPacketFDPayload p;
PinosPacketProcessMem p;
if (!pinos_buffer_iter_parse_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_process_mem (&it, &p))
continue;
if (fdids == NULL)
@ -464,7 +464,9 @@ gst_pinos_pay_chain_other (GstPinosPay *pay, GstBuffer * buffer)
PinosBuffer pbuf;
PinosBufferBuilder builder;
PinosPacketHeader hdr;
PinosPacketFDPayload p;
PinosPacketAddMem am;
PinosPacketProcessMem p;
PinosPacketRemoveMem rm;
gsize size;
gpointer data;
GSocketControlMessage *msg;
@ -482,12 +484,18 @@ gst_pinos_pay_chain_other (GstPinosPay *pay, GstBuffer * buffer)
msg = g_unix_fd_message_new ();
fdmem = gst_pinos_pay_get_fd_memory (pay, buffer, &tmpfile);
p.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem));
p.id = pinos_fd_manager_get_id (pay->fdmanager);
am.id = pinos_fd_manager_get_id (pay->fdmanager);
am.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem));
am.offset = 0;
am.size = fdmem->size;
p.id = am.id;
p.offset = fdmem->offset;
p.size = fdmem->size;
pinos_buffer_builder_add_fd_payload (&builder, &p);
rm.id = am.id;
pinos_buffer_builder_add_add_mem (&builder, &am);
pinos_buffer_builder_add_process_mem (&builder, &p);
pinos_buffer_builder_add_remove_mem (&builder, &rm);
pinos_buffer_builder_end (&builder, &pbuf);
data = pinos_buffer_steal_data (&pbuf, &size);

38
pinos/gst/gstpinosportsink.c
View file

@ -40,9 +40,6 @@
#include "gstpinosportsink.h"
#include "gsttmpfileallocator.h"
static GQuark fdids_quark;
static GQuark orig_buffer_quark;
GST_DEBUG_CATEGORY_STATIC (pinos_port_sink_debug);
#define GST_CAT_DEFAULT pinos_port_sink_debug
@ -82,9 +79,11 @@ on_received_buffer (PinosPort *port, PinosBuffer *pbuf, GError **error, gpointer
PinosBufferIter it;
PinosBufferBuilder b;
gboolean have_out = FALSE;
guint8 buffer[1024];
gint fds[8];
if (this->pinos_input) {
pinos_buffer_builder_init (&b);
pinos_buffer_builder_init_into (&b, buffer, 1024, fds, 8);
}
pinos_buffer_iter_init (&it, pbuf);
@ -227,8 +226,9 @@ gst_pinos_port_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
PinosBufferBuilder builder;
PinosBuffer pbuf;
PinosPacketFormatChange fc;
guint8 buffer[1024];
pinos_buffer_builder_init (&builder);
pinos_buffer_builder_init_into (&builder, buffer, 1024, NULL, 0);
fc.id = 0;
fc.format = cstr = gst_caps_to_string (caps);
pinos_buffer_builder_add_format_change (&builder, &fc);
@ -297,31 +297,38 @@ gst_pinos_port_sink_render_other (GstPinosPortSink * this, GstBuffer * buffer)
PinosBuffer pbuf;
PinosBufferBuilder builder;
PinosPacketHeader hdr;
PinosPacketFDPayload p;
PinosPacketAddMem am;
PinosPacketProcessMem p;
PinosPacketRemoveMem rm;
gboolean tmpfile = TRUE;
gint fd;
guint8 send_buffer[1024];
gint send_fds[8];
hdr.flags = 0;
hdr.seq = GST_BUFFER_OFFSET (buffer);
hdr.pts = GST_BUFFER_PTS (buffer) + GST_ELEMENT_CAST (this)->base_time;
hdr.dts_offset = 0;
pinos_buffer_builder_init (&builder);
pinos_buffer_builder_init_into (&builder, send_buffer, 1024, send_fds, 8);
pinos_buffer_builder_add_header (&builder, &hdr);
fdmem = gst_pinos_port_sink_get_fd_memory (this, buffer, &tmpfile);
fd = gst_fd_memory_get_fd (fdmem);
p.fd_index = pinos_buffer_builder_add_fd (&builder, fd);
p.id = pinos_fd_manager_get_id (this->fdmanager);
am.id = pinos_fd_manager_get_id (this->fdmanager);
am.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem));
am.offset = 0;
am.size = fdmem->size;
p.id = am.id;
p.offset = fdmem->offset;
p.size = fdmem->size;
pinos_buffer_builder_add_fd_payload (&builder, &p);
rm.id = am.id;
pinos_buffer_builder_add_add_mem (&builder, &am);
pinos_buffer_builder_add_process_mem (&builder, &p);
pinos_buffer_builder_add_remove_mem (&builder, &rm);
pinos_buffer_builder_end (&builder, &pbuf);
GST_LOG ("send %d %"G_GUINT64_FORMAT" %"G_GUINT64_FORMAT" %"G_GUINT64_FORMAT,
p.id, hdr.pts, GST_BUFFER_PTS (buffer), GST_ELEMENT_CAST (this)->base_time);
pinos_buffer_builder_end (&builder, &pbuf);
if (!pinos_port_send_buffer (this->port, &pbuf, &error)) {
GST_WARNING ("send failed: %s", error->message);
g_clear_error (&error);
@ -383,9 +390,6 @@ gst_pinos_port_sink_class_init (GstPinosPortSinkClass * klass)
gstbasesink_class->propose_allocation = gst_pinos_port_sink_propose_allocation;
gstbasesink_class->render = gst_pinos_port_sink_render;
fdids_quark = g_quark_from_static_string ("GstPinosPortSinkFDIds");
orig_buffer_quark = g_quark_from_static_string ("GstPinosPortSinkOrigBuffer");
GST_DEBUG_CATEGORY_INIT (pinos_port_sink_debug, "pinosportsink", 0,
"Pinos Socket Sink");
}

67
pinos/gst/gstpinosportsrc.c
View file

@ -45,7 +45,7 @@
#include <gst/video/video.h>
static GQuark fdpayload_data_quark;
static GQuark process_mem_data_quark;
GST_DEBUG_CATEGORY_STATIC (pinos_port_src_debug);
#define GST_CAT_DEFAULT pinos_port_src_debug
@ -83,16 +83,16 @@ static gboolean gst_pinos_port_src_query (GstBaseSrc * src, GstQuery * query);
typedef struct {
GstPinosPortSrc *src;
PinosPacketFDPayload p;
} FDPayloadData;
PinosPacketProcessMem p;
} ProcessMemData;
static void
fdpayload_data_destroy (gpointer user_data)
process_mem_data_destroy (gpointer user_data)
{
FDPayloadData *data = user_data;
ProcessMemData *data = user_data;
GstPinosPortSrc *this = data->src;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosPacketReuseMem r;
PinosBuffer pbuf;
r.id = data->p.id;
@ -100,14 +100,14 @@ fdpayload_data_destroy (gpointer user_data)
GST_DEBUG_OBJECT (this, "destroy %d", r.id);
pinos_buffer_builder_init (&b);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
pinos_buffer_builder_add_reuse_mem (&b, &r);
pinos_buffer_builder_end (&b, &pbuf);
pinos_port_send_buffer (this->port, &pbuf, NULL);
pinos_buffer_unref (&pbuf);
gst_object_unref (this);
g_slice_free (FDPayloadData, data);
g_slice_free (ProcessMemData, data);
}
static gboolean
@ -145,33 +145,58 @@ on_received_buffer (PinosPort *port,
GST_BUFFER_OFFSET (buf) = hdr.seq;
break;
}
case PINOS_PACKET_TYPE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_ADD_MEM:
{
GstMemory *fdmem = NULL;
FDPayloadData data;
PinosPacketAddMem p;
int fd;
if (!pinos_buffer_iter_parse_fd_payload (&it, &data.p))
if (!pinos_buffer_iter_parse_add_mem (&it, &p))
break;
GST_DEBUG ("got fd payload id %d", data.p.id);
fd = pinos_buffer_get_fd (pbuf, data.p.fd_index);
fd = pinos_buffer_get_fd (pbuf, p.fd_index);
if (fd == -1)
break;
fdmem = gst_fd_allocator_alloc (this->fd_allocator, dup (fd),
p.offset + p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, p.offset, p.size);
g_hash_table_insert (this->mem_ids, GINT_TO_POINTER (p.id), fdmem);
break;
}
case PINOS_PACKET_TYPE_REMOVE_MEM:
{
PinosPacketRemoveMem p;
if (!pinos_buffer_iter_parse_remove_mem (&it, &p))
break;
g_hash_table_remove (this->mem_ids, GINT_TO_POINTER (p.id));
break;
}
case PINOS_PACKET_TYPE_PROCESS_MEM:
{
GstMemory *fdmem = NULL;
ProcessMemData data;
if (!pinos_buffer_iter_parse_process_mem (&it, &data.p))
break;
if (!(fdmem = g_hash_table_lookup (this->mem_ids, GINT_TO_POINTER (data.p.id))))
break;
if (buf == NULL)
buf = gst_buffer_new ();
fdmem = gst_fd_allocator_alloc (this->fd_allocator, dup (fd),
data.p.offset + data.p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, data.p.offset, data.p.size);
fdmem = gst_memory_share (fdmem, data.p.offset, data.p.size);
gst_buffer_append_memory (buf, fdmem);
data.src = gst_object_ref (this);
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (fdmem),
fdpayload_data_quark,
g_slice_dup (FDPayloadData, &data),
fdpayload_data_destroy);
process_mem_data_quark,
g_slice_dup (ProcessMemData, &data),
process_mem_data_destroy);
break;
}
case PINOS_PACKET_TYPE_FORMAT_CHANGE:
@ -313,6 +338,7 @@ gst_pinos_port_src_finalize (GObject * object)
g_object_unref (this->port);
if (this->clock)
gst_object_unref (this->clock);
g_hash_table_unref (this->mem_ids);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
@ -363,7 +389,7 @@ gst_pinos_port_src_class_init (GstPinosPortSrcClass * klass)
GST_DEBUG_CATEGORY_INIT (pinos_port_src_debug, "pinosportsrc", 0,
"Pinos Source");
fdpayload_data_quark = g_quark_from_static_string ("GstPinosPortSrcFDPayloadQuark");
process_mem_data_quark = g_quark_from_static_string ("GstPinosPortSrcProcessMemQuark");
}
static void
@ -378,6 +404,7 @@ gst_pinos_port_src_init (GstPinosPortSrc * src)
g_queue_init (&src->queue);
src->fd_allocator = gst_fd_allocator_new ();
src->mem_ids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, (GDestroyNotify) gst_memory_unref);
}
static void

1
pinos/gst/gstpinosportsrc.h
View file

@ -65,6 +65,7 @@ struct _GstPinosPortSrc {
GstClock *clock;
GstAllocator *fd_allocator;
GHashTable *mem_ids;
GQueue queue;
GCond cond;

55
pinos/gst/gstpinossink.c
View file

@ -117,7 +117,7 @@ gst_pinos_sink_finalize (GObject * object)
if (pinossink->properties)
gst_structure_free (pinossink->properties);
g_hash_table_unref (pinossink->fdids);
g_hash_table_unref (pinossink->mem_ids);
g_object_unref (pinossink->allocator);
g_free (pinossink->path);
g_free (pinossink->client_name);
@ -210,11 +210,12 @@ static void
gst_pinos_sink_init (GstPinosSink * sink)
{
sink->allocator = gst_tmpfile_allocator_new ();
sink->fdmanager = pinos_fd_manager_get (PINOS_FD_MANAGER_DEFAULT);
sink->client_name = pinos_client_name();
sink->mode = DEFAULT_PROP_MODE;
sink->fdids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
(GDestroyNotify) gst_buffer_unref);
sink->mem_ids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
(GDestroyNotify) gst_memory_unref);
}
static GstCaps *
@ -345,15 +346,15 @@ on_new_buffer (GObject *gobject,
pinos_buffer_iter_init (&it, pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_RELEASE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_REUSE_MEM:
{
PinosPacketReleaseFDPayload p;
PinosPacketReuseMem p;
if (!pinos_buffer_iter_parse_release_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_reuse_mem (&it, &p))
continue;
GST_LOG ("fd index %d is released", p.id);
g_hash_table_remove (pinossink->fdids, GINT_TO_POINTER (p.id));
GST_LOG ("mem index %d is reused", p.id);
g_hash_table_remove (pinossink->mem_ids, GINT_TO_POINTER (p.id));
break;
}
case PINOS_PACKET_TYPE_REFRESH_REQUEST:
@ -508,10 +509,11 @@ gst_pinos_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
GstMemory *mem = NULL;
GstClockTime pts, dts, base;
PinosPacketHeader hdr;
PinosPacketFDPayload p;
PinosPacketAddMem am;
PinosPacketProcessMem p;
PinosPacketRemoveMem rm;
gsize size;
gboolean tmpfile, res;
gint fd;
pinossink = GST_PINOS_SINK (bsink);
@ -559,27 +561,32 @@ gst_pinos_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
pinos_stream_buffer_builder_init (pinossink->stream, &builder);
pinos_buffer_builder_add_header (&builder, &hdr);
fd = gst_fd_memory_get_fd (mem);
p.fd_index = pinos_buffer_builder_add_fd (&builder, fd);
p.id = pinossink->id_counter++;
p.offset = 0;
p.size = size;
pinos_buffer_builder_add_fd_payload (&builder, &p);
am.id = pinos_fd_manager_get_id (pinossink->fdmanager);
am.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (mem));
am.offset = 0;
am.size = mem->size;
p.id = am.id;
p.offset = mem->offset;
p.size = mem->size;
rm.id = am.id;
pinos_buffer_builder_add_add_mem (&builder, &am);
pinos_buffer_builder_add_process_mem (&builder, &p);
pinos_buffer_builder_add_remove_mem (&builder, &rm);
pinos_buffer_builder_end (&builder, &pbuf);
GST_LOG ("sending fd index %d %d %d", p.id, p.fd_index, fd);
GST_LOG ("sending fd mem %d %d", am.id, am.fd_index);
res = pinos_stream_send_buffer (pinossink->stream, &pbuf);
pinos_buffer_steal_fds (&pbuf, NULL);
pinos_buffer_unref (&pbuf);
pinos_main_loop_unlock (pinossink->loop);
gst_memory_unref (mem);
if (res && !tmpfile) {
/* keep the buffer around until we get the release fd message */
g_hash_table_insert (pinossink->fdids, GINT_TO_POINTER (p.id), gst_buffer_ref (buffer));
}
/* keep the memory around until we get the reuse mem message */
g_hash_table_insert (pinossink->mem_ids, GINT_TO_POINTER (p.id), mem);
} else
gst_memory_unref (mem);
return GST_FLOW_OK;
not_negotiated:
@ -792,10 +799,10 @@ gst_pinos_sink_change_state (GstElement * element, GstStateChange transition)
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
g_hash_table_remove_all (this->fdids);
g_hash_table_remove_all (this->mem_ids);
break;
case GST_STATE_CHANGE_READY_TO_NULL:
g_hash_table_remove_all (this->fdids);
g_hash_table_remove_all (this->mem_ids);
gst_pinos_sink_close (this);
break;
default:

4
pinos/gst/gstpinossink.h
View file

@ -84,8 +84,8 @@ struct _GstPinosSink {
GstStructure *properties;
GstPinosSinkMode mode;
guint32 id_counter;
GHashTable *fdids;
PinosFdManager *fdmanager;
GHashTable *mem_ids;
};
struct _GstPinosSinkClass {

46
pinos/gst/gstpinossocketsink.c
View file

@ -174,12 +174,12 @@ gst_pinos_socket_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
}
static void
release_fds (GstPinosSocketSink *this, GstBuffer *buffer)
reuse_fds (GstPinosSocketSink *this, GstBuffer *buffer)
{
GArray *fdids;
guint i;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosPacketReuseMem r;
PinosBuffer pbuf;
gsize size;
gpointer data;
@ -195,8 +195,8 @@ release_fds (GstPinosSocketSink *this, GstBuffer *buffer)
for (i = 0; i < fdids->len; i++) {
r.id = g_array_index (fdids, guint32, i);
GST_LOG ("release fd index %d", r.id);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
GST_LOG ("reuse fd index %d", r.id);
pinos_buffer_builder_add_reuse_mem (&b, &r);
}
pinos_buffer_builder_end (&b, &pbuf);
g_array_unref (fdids);
@ -227,17 +227,17 @@ gst_pinos_socket_sink_render_pinos (GstPinosSocketSink * this, GstBuffer * buffe
pinos_buffer_iter_init (&it, &pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_PROCESS_MEM:
{
PinosPacketFDPayload p;
PinosPacketProcessMem p;
if (!pinos_buffer_iter_parse_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_process_mem (&it, &p))
continue;
if (fdids == NULL)
fdids = g_array_new (FALSE, FALSE, sizeof (guint32));
GST_LOG ("track fd index %d", p.id);
GST_LOG ("track fd id %d", p.id);
g_array_append_val (fdids, p.id);
break;
}
@ -270,7 +270,7 @@ gst_pinos_socket_sink_render_pinos (GstPinosSocketSink * this, GstBuffer * buffe
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (buffer),
fdids_quark, fdids, NULL);
gst_mini_object_weak_ref (GST_MINI_OBJECT_CAST (buffer),
(GstMiniObjectNotify) release_fds, g_object_ref (this));
(GstMiniObjectNotify) reuse_fds, g_object_ref (this));
}
gst_burst_cache_queue_buffer (this->cache, gst_buffer_ref (buffer));
@ -310,7 +310,9 @@ gst_pinos_socket_sink_render_other (GstPinosSocketSink * this, GstBuffer * buffe
PinosBuffer pbuf;
PinosBufferBuilder builder;
PinosPacketHeader hdr;
PinosPacketFDPayload p;
PinosPacketAddMem am;
PinosPacketProcessMem p;
PinosPacketRemoveMem rm;
gsize size;
gpointer data;
GSocketControlMessage *msg;
@ -326,17 +328,23 @@ gst_pinos_socket_sink_render_other (GstPinosSocketSink * this, GstBuffer * buffe
pinos_buffer_builder_add_header (&builder, &hdr);
fdmem = gst_pinos_socket_sink_get_fd_memory (this, buffer, &tmpfile);
p.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem));
p.id = pinos_fd_manager_get_id (this->fdmanager);
am.id = pinos_fd_manager_get_id (this->fdmanager);
am.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem));
am.offset = 0;
am.size = fdmem->size;
p.id = am.id;
p.offset = fdmem->offset;
p.size = fdmem->size;
pinos_buffer_builder_add_fd_payload (&builder, &p);
rm.id = am.id;
pinos_buffer_builder_add_add_mem (&builder, &am);
pinos_buffer_builder_add_process_mem (&builder, &p);
pinos_buffer_builder_add_remove_mem (&builder, &rm);
pinos_buffer_builder_end (&builder, &pbuf);
GST_LOG ("send %d %"G_GUINT64_FORMAT" %"G_GUINT64_FORMAT" %"G_GUINT64_FORMAT,
p.id, hdr.pts, GST_BUFFER_PTS (buffer), GST_ELEMENT_CAST (this)->base_time);
pinos_buffer_builder_end (&builder, &pbuf);
data = pinos_buffer_steal_data (&pbuf, &size);
fds = pinos_buffer_steal_fds (&pbuf, &n_fds);
pinos_buffer_unref (&pbuf);
@ -548,17 +556,17 @@ myreader_receive_buffer (GstPinosSocketSink *this, MyReader *myreader)
pinos_buffer_iter_init (&it, &pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_RELEASE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_REUSE_MEM:
{
PinosPacketReleaseFDPayload p;
PinosPacketReuseMem p;
gint id;
if (!pinos_buffer_iter_parse_release_fd_payload (&it, &p))
if (!pinos_buffer_iter_parse_reuse_mem (&it, &p))
continue;
id = p.id;
GST_LOG ("fd index %d for client %s is released", id, client_path);
GST_LOG ("fd id %d for client %s is reused", id, client_path);
pinos_fd_manager_remove (this->fdmanager, client_path, id);
break;
}

80
pinos/gst/gstpinossrc.c
View file

@ -45,7 +45,7 @@
#include <gst/video/video.h>
static GQuark fdpayload_data_quark;
static GQuark process_mem_data_quark;
GST_DEBUG_CATEGORY_STATIC (pinos_src_debug);
#define GST_CAT_DEFAULT pinos_src_debug
@ -183,6 +183,7 @@ gst_pinos_src_finalize (GObject * object)
gst_object_unref (pinossrc->clock);
g_free (pinossrc->path);
g_free (pinossrc->client_name);
g_hash_table_unref (pinossrc->mem_ids);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
@ -254,7 +255,7 @@ gst_pinos_src_class_init (GstPinosSrcClass * klass)
GST_DEBUG_CATEGORY_INIT (pinos_src_debug, "pinossrc", 0,
"Pinos Source");
fdpayload_data_quark = g_quark_from_static_string ("GstPinosSrcFDPayloadQuark");
process_mem_data_quark = g_quark_from_static_string ("GstPinosSrcProcessMemQuark");
}
static void
@ -269,6 +270,7 @@ gst_pinos_src_init (GstPinosSrc * src)
src->fd_allocator = gst_fd_allocator_new ();
src->client_name = pinos_client_name ();
src->mem_ids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, (GDestroyNotify) gst_memory_unref);
}
static GstCaps *
@ -320,16 +322,16 @@ gst_pinos_src_src_fixate (GstBaseSrc * bsrc, GstCaps * caps)
typedef struct {
GstPinosSrc *src;
PinosPacketFDPayload p;
} FDPayloadData;
PinosPacketProcessMem p;
} ProcessMemData;
static void
fdpayload_data_destroy (gpointer user_data)
process_mem_data_destroy (gpointer user_data)
{
FDPayloadData *data = user_data;
ProcessMemData *data = user_data;
GstPinosSrc *pinossrc = data->src;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosPacketReuseMem r;
PinosBuffer pbuf;
r.id = data->p.id;
@ -339,7 +341,7 @@ fdpayload_data_destroy (gpointer user_data)
GST_OBJECT_LOCK (pinossrc);
if (pinossrc->stream_state == PINOS_STREAM_STATE_STREAMING) {
pinos_stream_buffer_builder_init (pinossrc->stream, &b);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
pinos_buffer_builder_add_reuse_mem (&b, &r);
pinos_buffer_builder_end (&b, &pbuf);
GST_DEBUG_OBJECT (pinossrc, "send release-fd for %d", r.id);
@ -349,7 +351,7 @@ fdpayload_data_destroy (gpointer user_data)
GST_OBJECT_UNLOCK (pinossrc);
gst_object_unref (pinossrc);
g_slice_free (FDPayloadData, data);
g_slice_free (ProcessMemData, data);
}
static void
@ -390,33 +392,59 @@ on_new_buffer (GObject *gobject,
GST_BUFFER_OFFSET (buf) = hdr.seq;
break;
}
case PINOS_PACKET_TYPE_FD_PAYLOAD:
case PINOS_PACKET_TYPE_ADD_MEM:
{
GstMemory *fdmem = NULL;
FDPayloadData data;
PinosPacketAddMem p;
int fd;
if (!pinos_buffer_iter_parse_fd_payload (&it, &data.p))
if (!pinos_buffer_iter_parse_add_mem (&it, &p))
goto parse_failed;
GST_DEBUG ("got fd payload id %d", data.p.id);
fd = pinos_buffer_get_fd (pbuf, data.p.fd_index);
fd = pinos_buffer_get_fd (pbuf, p.fd_index);
if (fd == -1)
goto no_fds;
goto parse_failed;
fdmem = gst_fd_allocator_alloc (pinossrc->fd_allocator, dup (fd),
p.offset + p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, p.offset, p.size);
g_hash_table_insert (pinossrc->mem_ids, GINT_TO_POINTER (p.id), fdmem);
break;
}
case PINOS_PACKET_TYPE_REMOVE_MEM:
{
PinosPacketRemoveMem p;
if (!pinos_buffer_iter_parse_remove_mem (&it, &p))
goto parse_failed;
g_hash_table_remove (pinossrc->mem_ids, GINT_TO_POINTER (p.id));
break;
}
case PINOS_PACKET_TYPE_PROCESS_MEM:
{
GstMemory *fdmem = NULL;
ProcessMemData data;
if (!pinos_buffer_iter_parse_process_mem (&it, &data.p))
goto parse_failed;
GST_DEBUG ("got mem id %d", data.p.id);
if (!(fdmem = g_hash_table_lookup (pinossrc->mem_ids, GINT_TO_POINTER (data.p.id))))
goto parse_failed;
if (buf == NULL)
buf = gst_buffer_new ();
fdmem = gst_fd_allocator_alloc (pinossrc->fd_allocator, dup (fd),
data.p.offset + data.p.size, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (fdmem, data.p.offset, data.p.size);
fdmem = gst_memory_share (fdmem, data.p.offset, data.p.size);
gst_buffer_append_memory (buf, fdmem);
data.src = gst_object_ref (pinossrc);
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (fdmem),
fdpayload_data_quark,
g_slice_dup (FDPayloadData, &data),
fdpayload_data_destroy);
process_mem_data_quark,
g_slice_dup (ProcessMemData, &data),
process_mem_data_destroy);
break;
}
case PINOS_PACKET_TYPE_FORMAT_CHANGE:
@ -456,16 +484,6 @@ parse_failed:
pinos_main_loop_signal (pinossrc->loop, FALSE);
return;
}
no_fds:
{
pinos_buffer_iter_end (&it);
pinos_buffer_unref (pbuf);
gst_buffer_unref (buf);
GST_ELEMENT_ERROR (pinossrc, RESOURCE, FAILED, ("fd not found in buffer"), (NULL));
pinos_main_loop_signal (pinossrc->loop, FALSE);
return;
}
}
static void

1
pinos/gst/gstpinossrc.h
View file

@ -71,6 +71,7 @@ struct _GstPinosSrc {
GstAllocator *fd_allocator;
GstStructure *properties;
GHashTable *mem_ids;
GQueue queue;
GstClock *clock;
};