mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-12-16 08:56:45 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

make separate sockets for data and control

Browse source
This commit is contained in:
Wim Taymans 2016-10-14 19:23:05 +02:00
parent ee202e13e9
commit 184e0a300a
7 changed files with 419 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

192
pinos/client/stream.c
View file

@ -90,6 +90,9 @@ struct _PinosStreamPrivate
GSocket *socket;
GSource *socket_source;
int fd;
GSocket *rtsocket;
GSource *rtsocket_source;
int rtfd;
GSource *timeout_source;
@ -700,7 +703,7 @@ send_need_input (PinosStream *stream, uint32_t port_id)
add_need_input (stream, &builder, port_id);
spa_control_builder_end (&builder, &control);
if (spa_control_write (&control, priv->fd) < 0)
if (spa_control_write (&control, priv->rtfd) < 0)
g_warning ("stream %p: error writing control", stream);
spa_control_clear (&control);
@ -763,7 +766,7 @@ send_reuse_buffer (PinosStream *stream, uint32_t port_id, uint32_t buffer_id)
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_NODE_EVENT, &cne);
spa_control_builder_end (&builder, &control);
if (spa_control_write (&control, priv->fd) < 0)
if (spa_control_write (&control, priv->rtfd) < 0)
g_warning ("stream %p: error writing control", stream);
spa_control_clear (&control);
@ -795,7 +798,7 @@ send_process_buffer (PinosStream *stream, uint32_t port_id, uint32_t buffer_id)
spa_control_builder_end (&builder, &control);
if (spa_control_write (&control, priv->fd) < 0)
if (spa_control_write (&control, priv->rtfd) < 0)
g_warning ("stream %p: error writing control", stream);
spa_control_clear (&control);
@ -861,13 +864,44 @@ static gboolean
handle_node_event (PinosStream *stream,
SpaNodeEvent *event)
{
PinosStreamPrivate *priv = stream->priv;
switch (event->type) {
case SPA_NODE_EVENT_TYPE_INVALID:
case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
case SPA_NODE_EVENT_TYPE_NEED_INPUT:
case SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE:
case SPA_NODE_EVENT_TYPE_REUSE_BUFFER:
case SPA_NODE_EVENT_TYPE_DRAINED:
case SPA_NODE_EVENT_TYPE_MARKER:
case SPA_NODE_EVENT_TYPE_ERROR:
case SPA_NODE_EVENT_TYPE_BUFFERING:
case SPA_NODE_EVENT_TYPE_REQUEST_REFRESH:
case SPA_NODE_EVENT_TYPE_REQUEST_CLOCK_UPDATE:
g_warning ("unhandled node event %d", event->type);
break;
}
return TRUE;
}
static gboolean
handle_rtnode_event (PinosStream *stream,
SpaNodeEvent *event)
{
PinosStreamPrivate *priv = stream->priv;
switch (event->type) {
case SPA_NODE_EVENT_TYPE_INVALID:
case SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE:
case SPA_NODE_EVENT_TYPE_DRAINED:
case SPA_NODE_EVENT_TYPE_MARKER:
case SPA_NODE_EVENT_TYPE_ERROR:
case SPA_NODE_EVENT_TYPE_BUFFERING:
case SPA_NODE_EVENT_TYPE_REQUEST_REFRESH:
case SPA_NODE_EVENT_TYPE_REQUEST_CLOCK_UPDATE:
g_warning ("unexpected node event %d", event->type);
break;
case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
case SPA_NODE_EVENT_TYPE_NEED_INPUT:
g_warning ("unhandled node event %d", event->type);
break;
@ -887,14 +921,6 @@ handle_node_event (PinosStream *stream,
}
break;
}
case SPA_NODE_EVENT_TYPE_DRAINED:
case SPA_NODE_EVENT_TYPE_MARKER:
case SPA_NODE_EVENT_TYPE_ERROR:
case SPA_NODE_EVENT_TYPE_BUFFERING:
case SPA_NODE_EVENT_TYPE_REQUEST_REFRESH:
case SPA_NODE_EVENT_TYPE_REQUEST_CLOCK_UPDATE:
g_warning ("unhandled node event %d", event->type);
break;
}
return TRUE;
}
@ -1003,6 +1029,7 @@ parse_control (PinosStream *stream,
case SPA_CONTROL_CMD_PORT_UPDATE:
case SPA_CONTROL_CMD_PORT_STATUS_CHANGE:
case SPA_CONTROL_CMD_NODE_STATE_CHANGE:
case SPA_CONTROL_CMD_PROCESS_BUFFER:
g_warning ("got unexpected control %d", cmd);
break;
@ -1183,6 +1210,65 @@ parse_control (PinosStream *stream,
spa_control_clear (&control);
break;
}
case SPA_CONTROL_CMD_NODE_EVENT:
{
SpaControlCmdNodeEvent p;
if (spa_control_iter_parse_cmd (&it, &p) < 0)
break;
handle_node_event (stream, p.event);
break;
}
case SPA_CONTROL_CMD_NODE_COMMAND:
{
SpaControlCmdNodeCommand p;
if (spa_control_iter_parse_cmd (&it, &p) < 0)
break;
handle_node_command (stream, p.seq, p.command);
break;
}
case SPA_CONTROL_CMD_INVALID:
g_warning ("unhandled command %d", cmd);
break;
}
}
spa_control_iter_end (&it);
return TRUE;
}
static gboolean
parse_rtcontrol (PinosStream *stream,
SpaControl *ctrl)
{
SpaControlIter it;
PinosStreamPrivate *priv = stream->priv;
spa_control_iter_init (&it, ctrl);
while (spa_control_iter_next (&it) == SPA_RESULT_OK) {
SpaControlCmd cmd = spa_control_iter_get_cmd (&it);
switch (cmd) {
case SPA_CONTROL_CMD_INVALID:
case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
case SPA_CONTROL_CMD_PORT_STATUS_CHANGE:
case SPA_CONTROL_CMD_NODE_STATE_CHANGE:
case SPA_CONTROL_CMD_ADD_PORT:
case SPA_CONTROL_CMD_REMOVE_PORT:
case SPA_CONTROL_CMD_SET_FORMAT:
case SPA_CONTROL_CMD_SET_PROPERTY:
case SPA_CONTROL_CMD_ADD_MEM:
case SPA_CONTROL_CMD_REMOVE_MEM:
case SPA_CONTROL_CMD_USE_BUFFERS:
case SPA_CONTROL_CMD_NODE_COMMAND:
g_warning ("got unexpected control %d", cmd);
break;
case SPA_CONTROL_CMD_PROCESS_BUFFER:
{
SpaControlCmdProcessBuffer p;
@ -1212,23 +1298,9 @@ parse_control (PinosStream *stream,
if (spa_control_iter_parse_cmd (&it, &p) < 0)
break;
handle_node_event (stream, p.event);
handle_rtnode_event (stream, p.event);
break;
}
case SPA_CONTROL_CMD_NODE_COMMAND:
{
SpaControlCmdNodeCommand p;
if (spa_control_iter_parse_cmd (&it, &p) < 0)
break;
handle_node_command (stream, p.seq, p.command);
break;
}
case SPA_CONTROL_CMD_INVALID:
g_warning ("unhandled command %d", cmd);
break;
}
}
spa_control_iter_end (&it);
@ -1275,6 +1347,46 @@ on_socket_condition (GSocket *socket,
return TRUE;
}
static gboolean
on_rtsocket_condition (GSocket *socket,
GIOCondition condition,
gpointer user_data)
{
PinosStream *stream = user_data;
PinosStreamPrivate *priv = stream->priv;
switch (condition) {
case G_IO_IN:
{
SpaControl *control = &priv->recv_control;
guint8 buffer[4096];
if (spa_control_read (control,
priv->rtfd,
buffer,
sizeof (buffer),
NULL,
0) < 0) {
g_warning ("stream %p: failed to read buffer", stream);
return TRUE;
}
parse_rtcontrol (stream, control);
spa_control_clear (control);
break;
}
case G_IO_OUT:
g_warning ("can do IO\n");
break;
default:
break;
}
return TRUE;
}
static gboolean
on_timeout (gpointer user_data)
{
@ -1296,7 +1408,7 @@ on_timeout (gpointer user_data)
}
static void
handle_socket (PinosStream *stream, gint fd)
handle_socket (PinosStream *stream, gint fd, gint rtfd)
{
PinosStreamPrivate *priv = stream->priv;
GError *error = NULL;
@ -1304,12 +1416,20 @@ handle_socket (PinosStream *stream, gint fd)
priv->socket = g_socket_new_from_fd (fd, &error);
if (priv->socket == NULL)
goto socket_failed;
priv->rtsocket = g_socket_new_from_fd (rtfd, &error);
if (priv->rtsocket == NULL)
goto socket_failed;
priv->fd = g_socket_get_fd (priv->socket);
priv->socket_source = g_socket_create_source (priv->socket, G_IO_IN, NULL);
g_source_set_callback (priv->socket_source, (GSourceFunc) on_socket_condition, stream, NULL);
g_source_attach (priv->socket_source, priv->context->priv->context);
priv->rtfd = g_socket_get_fd (priv->rtsocket);
priv->rtsocket_source = g_socket_create_source (priv->rtsocket, G_IO_IN, NULL);
g_source_set_callback (priv->rtsocket_source, (GSourceFunc) on_rtsocket_condition, stream, NULL);
g_source_attach (priv->rtsocket_source, priv->context->priv->context);
priv->timeout_source = g_timeout_source_new (100);
g_source_set_callback (priv->timeout_source, (GSourceFunc) on_timeout, stream, NULL);
g_source_attach (priv->timeout_source, priv->context->priv->context);
@ -1334,6 +1454,10 @@ unhandle_socket (PinosStream *stream)
g_source_destroy (priv->socket_source);
g_clear_pointer (&priv->socket_source, g_source_unref);
}
if (priv->rtsocket_source) {
g_source_destroy (priv->rtsocket_source);
g_clear_pointer (&priv->rtsocket_source, g_source_unref);
}
}
static void
@ -1382,6 +1506,7 @@ on_node_created (GObject *source_object,
const gchar *node_path;
GUnixFDList *fd_list;
gint fd_idx, fd;
gint rtfd_idx, rtfd;
g_assert (context->priv->daemon == G_DBUS_PROXY (source_object));
@ -1391,16 +1516,19 @@ on_node_created (GObject *source_object,
if (ret == NULL)
goto create_failed;
g_variant_get (ret, "(&oh)", &node_path, &fd_idx);
g_variant_get (ret, "(&ohh)", &node_path, &fd_idx, &rtfd_idx);
g_variant_unref (ret);
if ((fd = g_unix_fd_list_get (fd_list, fd_idx, &error)) < 0)
goto fd_failed;
if ((rtfd = g_unix_fd_list_get (fd_list, rtfd_idx, &error)) < 0)
goto fd_failed;
priv->fd = fd;
priv->rtfd = rtfd;
g_object_unref (fd_list);
handle_socket (stream, priv->fd);
handle_socket (stream, priv->fd, priv->rtfd);
pinos_subscribe_get_proxy (context->priv->subscribe,
PINOS_DBUS_SERVICE,
@ -1462,7 +1590,7 @@ on_node_registered (GObject *source_object,
priv->fd = fd;
g_object_unref (fd_list);
handle_socket (stream, priv->fd);
handle_socket (stream, -1, priv->fd);
return;

1
pinos/dbus/org.pinos.xml
View file

@ -50,6 +50,7 @@
<arg type='a{sv}' name='properties' direction='in'/>
<arg type='o' name='node' direction='out'/>
<arg type='h' name='fd' direction='out'/>
<arg type='h' name='rtfd' direction='out'/>
</method>
<method name='RegisterClientNode'>

71
pinos/server/client-node.c
View file

@ -40,7 +40,9 @@
struct _PinosClientNodePrivate
{
int fd;
int rtfd;
GSocket *sockets[2];
GSocket *rtsockets[2];
SpaHandle *handle;
};
@ -104,7 +106,7 @@ pinos_client_node_set_property (GObject *_object,
*/
GSocket *
pinos_client_node_get_socket_pair (PinosClientNode *this,
GError **error)
GError **error)
{
PinosNode *node;
PinosClientNodePrivate *priv;
@ -156,6 +158,70 @@ create_failed:
}
}
/**
* pinos_client_node_get_rtsocket_pair:
* @node: a #PinosClientNode
* @error: a #GError
*
* Create or return a previously create socket pair for @node. The
* Socket for the other end is returned.
*
* Returns: a #GSocket that can be used to send/receive buffers to node.
*/
GSocket *
pinos_client_node_get_rtsocket_pair (PinosClientNode *this,
GError **error)
{
PinosNode *node;
PinosClientNodePrivate *priv;
g_return_val_if_fail (PINOS_IS_CLIENT_NODE (this), FALSE);
node = PINOS_NODE (this);
priv = this->priv;
if (priv->rtsockets[1] == NULL) {
SpaProps *props;
SpaPropValue value;
int fd[2];
if (socketpair (AF_UNIX, SOCK_STREAM, 0, fd) != 0)
goto no_sockets;
priv->rtsockets[0] = g_socket_new_from_fd (fd[0], error);
if (priv->rtsockets[0] == NULL)
goto create_failed;
priv->rtsockets[1] = g_socket_new_from_fd (fd[1], error);
if (priv->rtsockets[1] == NULL)
goto create_failed;
priv->rtfd = g_socket_get_fd (priv->rtsockets[0]);
spa_node_get_props (node->node, &props);
value.value = &priv->rtfd;
value.size = sizeof (int);
spa_props_set_value (props, spa_props_index_for_name (props, "rt-socket"), &value);
spa_node_set_props (node->node, props);
}
return g_object_ref (priv->rtsockets[1]);
/* ERRORS */
no_sockets:
{
g_set_error (error,
G_IO_ERROR,
g_io_error_from_errno (errno),
"could not create socketpair: %s", strerror (errno));
return NULL;
}
create_failed:
{
g_clear_object (&priv->rtsockets[0]);
g_clear_object (&priv->rtsockets[1]);
return NULL;
}
}
static void
pinos_client_node_dispose (GObject * object)
{
@ -171,6 +237,7 @@ pinos_client_node_dispose (GObject * object)
value.value = &fd;
value.size = sizeof (int);
spa_props_set_value (props, spa_props_index_for_name (props, "socket"), &value);
spa_props_set_value (props, spa_props_index_for_name (props, "rt-socket"), &value);
spa_node_set_props (node->node, props);
G_OBJECT_CLASS (pinos_client_node_parent_class)->dispose (object);
@ -186,6 +253,8 @@ pinos_client_node_finalize (GObject * object)
g_clear_object (&priv->sockets[0]);
g_clear_object (&priv->sockets[1]);
g_clear_object (&priv->rtsockets[0]);
g_clear_object (&priv->rtsockets[1]);
spa_handle_clear (priv->handle);
g_free (priv->handle);

2
pinos/server/client-node.h
View file

@ -69,6 +69,8 @@ PinosNode * pinos_client_node_new (PinosDaemon *daem
GSocket * pinos_client_node_get_socket_pair (PinosClientNode *node,
GError **error);
GSocket * pinos_client_node_get_rtsocket_pair (PinosClientNode *node,
GError **error);
G_END_DECLS

17
pinos/server/daemon.c
View file

@ -378,8 +378,8 @@ handle_create_client_node (PinosDaemon1 *interface,
PinosProperties *props;
GError *error = NULL;
GUnixFDList *fdlist;
GSocket *socket;
gint fdidx;
GSocket *socket, *rtsocket;
gint fdidx, rtfdidx;
sender = g_dbus_method_invocation_get_sender (invocation);
client = sender_get_client (daemon, sender, TRUE);
@ -397,6 +397,10 @@ handle_create_client_node (PinosDaemon1 *interface,
if (socket == NULL)
goto no_socket;
rtsocket = pinos_client_node_get_rtsocket_pair (PINOS_CLIENT_NODE (node), &error);
if (rtsocket == NULL)
goto no_socket;
pinos_client_add_object (client, G_OBJECT (node));
object_path = pinos_node_get_object_path (PINOS_NODE (node));
@ -405,10 +409,12 @@ handle_create_client_node (PinosDaemon1 *interface,
fdlist = g_unix_fd_list_new ();
fdidx = g_unix_fd_list_append (fdlist, g_socket_get_fd (socket), &error);
rtfdidx = g_unix_fd_list_append (fdlist, g_socket_get_fd (rtsocket), &error);
g_object_unref (socket);
g_object_unref (rtsocket);
g_dbus_method_invocation_return_value_with_unix_fd_list (invocation,
g_variant_new ("(oh)", object_path, fdidx), fdlist);
g_variant_new ("(ohh)", object_path, fdidx, rtfdidx), fdlist);
g_object_unref (fdlist);
return TRUE;
@ -943,8 +949,11 @@ poll_event (GIOChannel *source,
PollData *data = user_data;
SpaPollNotifyData d;
g_debug ("poll event %d", data->item.fds[0].fd);
d.user_data = data->item.user_data;
d.fds = data->item.fds;
d.fds[0].revents = condition;
d.n_fds = data->item.n_fds;
data->item.after_cb (&d);
@ -960,6 +969,7 @@ do_add_item (SpaPoll *poll,
GSource *source;
PollData data;
g_debug ("add main poll");
channel = g_io_channel_unix_new (item->fds[0].fd);
source = g_io_create_watch (channel, G_IO_IN);
g_io_channel_unref (channel);
@ -987,6 +997,7 @@ do_remove_item (SpaPoll *poll,
{
GSource *source;
g_debug ("remove main poll %d", item->id);
source = g_main_context_find_source_by_id (g_main_context_get_thread_default (), item->id);
g_source_destroy (source);

6
spa/lib/control.c
View file

@ -28,7 +28,7 @@
#include <spa/control.h>
#include <spa/debug.h>
#if 0
#if 1
#define SPA_DEBUG_CONTROL(format,args...) fprintf(stderr,format,##args)
#else
#define SPA_DEBUG_CONTROL(format,args...)
@ -1122,7 +1122,7 @@ spa_control_read (SpaControl *control,
}
sc->magic = SSC_MAGIC;
SPA_DEBUG_CONTROL ("control %p: read %zd bytes and %d fds\n", sc, len, sc->n_fds);
SPA_DEBUG_CONTROL ("control %p: %d read %zd bytes and %d fds\n", sc, fd, len, sc->n_fds);
return SPA_RESULT_OK;
@ -1185,7 +1185,7 @@ spa_control_write (SpaControl *control,
if (len != (ssize_t) sc->size)
return SPA_RESULT_ERROR;
SPA_DEBUG_CONTROL ("control %p: written %zd bytes and %d fds\n", sc, len, sc->n_fds);
SPA_DEBUG_CONTROL ("control %p: %d written %zd bytes and %d fds\n", sc, fd, len, sc->n_fds);
return SPA_RESULT_OK;

222
spa/plugins/remote/proxy.c
View file

@ -67,6 +67,7 @@ struct _ProxyBuffer {
typedef struct {
SpaProps props;
int socketfd;
int rtsocketfd;
} SpaProxyProps;
typedef struct {
@ -100,6 +101,7 @@ struct _SpaProxy {
URI uri;
SpaIDMap *map;
SpaLog *log;
SpaPoll *main_loop;
SpaPoll *data_loop;
SpaProxyProps props[2];
@ -109,6 +111,8 @@ struct _SpaProxy {
SpaPollFd fds[1];
SpaPollItem poll;
SpaPollFd rtfds[1];
SpaPollItem rtpoll;
unsigned int max_inputs;
unsigned int n_inputs;
@ -122,6 +126,7 @@ struct _SpaProxy {
enum {
PROP_ID_SOCKET,
PROP_ID_RT_SOCKET,
PROP_ID_LAST,
};
@ -133,12 +138,19 @@ static const SpaPropInfo prop_info[PROP_ID_LAST] =
SPA_PROP_TYPE_INT, sizeof (int),
SPA_PROP_RANGE_TYPE_NONE, 0, NULL,
NULL },
{ PROP_ID_RT_SOCKET, offsetof (SpaProxyProps, rtsocketfd),
"rt-socket",
SPA_PROP_FLAG_READWRITE,
SPA_PROP_TYPE_INT, sizeof (int),
SPA_PROP_RANGE_TYPE_NONE, 0, NULL,
NULL },
};
static void
reset_proxy_props (SpaProxyProps *props)
{
props->socketfd = -1;
props->rtsocketfd = -1;
}
static SpaResult
@ -149,14 +161,32 @@ update_poll (SpaProxy *this, int socketfd)
p = &this->props[1];
if (p->socketfd != -1) {
spa_poll_remove_item (this->data_loop, &this->poll);
}
if (p->socketfd != -1)
spa_poll_remove_item (this->main_loop, &this->poll);
p->socketfd = socketfd;
if (p->socketfd != -1) {
this->fds[0].fd = p->socketfd;
spa_poll_add_item (this->data_loop, &this->poll);
spa_poll_add_item (this->main_loop, &this->poll);
}
return res;
}
static SpaResult
update_rtpoll (SpaProxy *this, int rtsocketfd)
{
SpaProxyProps *p;
SpaResult res = SPA_RESULT_OK;
p = &this->props[1];
if (p->rtsocketfd != -1)
spa_poll_remove_item (this->data_loop, &this->rtpoll);
p->rtsocketfd = rtsocketfd;
if (p->rtsocketfd != -1) {
this->rtfds[0].fd = p->rtsocketfd;
spa_poll_add_item (this->data_loop, &this->rtpoll);
}
return res;
}
@ -232,8 +262,11 @@ spa_proxy_node_set_props (SpaNode *node,
res = spa_props_copy_values (props, &np->props);
/* compare changes */
if (op->socketfd != np->socketfd)
res = update_poll (this, np->socketfd);
if (op->rtsocketfd != np->rtsocketfd)
res = update_rtpoll (this, np->rtsocketfd);
/* commit changes */
memcpy (op, np, sizeof (*np));
@ -780,64 +813,68 @@ spa_proxy_node_port_use_buffers (SpaNode *node,
size += b->size;
}
/* make mem for the buffers */
port->buffer_mem_id = n_mem++;
port->buffer_mem_size = size;
port->buffer_mem_fd = memfd_create ("spa-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
if (n_buffers > 0) {
/* make mem for the buffers */
port->buffer_mem_id = n_mem++;
port->buffer_mem_size = size;
port->buffer_mem_fd = memfd_create ("spa-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
if (ftruncate (port->buffer_mem_fd, size) < 0) {
spa_log_error (this->log, "Failed to truncate temporary file: %s\n", strerror (errno));
close (port->buffer_mem_fd);
return SPA_RESULT_ERROR;
}
if (ftruncate (port->buffer_mem_fd, size) < 0) {
spa_log_error (this->log, "Failed to truncate temporary file: %s\n", strerror (errno));
close (port->buffer_mem_fd);
return SPA_RESULT_ERROR;
}
#if 0
{
unsigned int seals = F_SEAL_GROW | F_SEAL_SHRINK | F_SEAL_SEAL;
if (fcntl (port->buffer_mem_fd, F_ADD_SEALS, seals) == -1) {
spa_log_error (this->log, "Failed to add seals: %s\n", strerror (errno));
{
unsigned int seals = F_SEAL_GROW | F_SEAL_SHRINK | F_SEAL_SEAL;
if (fcntl (port->buffer_mem_fd, F_ADD_SEALS, seals) == -1) {
spa_log_error (this->log, "Failed to add seals: %s\n", strerror (errno));
}
}
}
#endif
p = port->buffer_mem_ptr = mmap (NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, port->buffer_mem_fd, 0);
p = port->buffer_mem_ptr = mmap (NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, port->buffer_mem_fd, 0);
for (i = 0; i < n_buffers; i++) {
ProxyBuffer *b = &port->buffers[i];
SpaBuffer *sb;
SpaMeta *sbm;
SpaData *sbd;
for (i = 0; i < n_buffers; i++) {
ProxyBuffer *b = &port->buffers[i];
SpaBuffer *sb;
SpaMeta *sbm;
SpaData *sbd;
spa_buffer_serialize (p, &b->buffer);
spa_buffer_serialize (p, &b->buffer);
sb = p;
b->buffer.datas = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->datas), SpaData);
sbm = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->metas), SpaMeta);
sbd = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->datas), SpaData);
sb = p;
b->buffer.datas = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->datas), SpaData);
sbm = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->metas), SpaMeta);
sbd = SPA_MEMBER (sb, SPA_PTR_TO_INT (sb->datas), SpaData);
for (j = 0; j < b->buffer.n_metas; j++)
b->metas[j].data = SPA_MEMBER (sb, SPA_PTR_TO_INT (sbm[j].data), void);
for (j = 0; j < b->buffer.n_metas; j++)
b->metas[j].data = SPA_MEMBER (sb, SPA_PTR_TO_INT (sbm[j].data), void);
for (j = 0; j < b->buffer.n_datas; j++) {
if (b->datas[j].type == SPA_DATA_TYPE_MEMPTR)
b->datas[j].data = SPA_MEMBER (sb, SPA_PTR_TO_INT (sbd[j].data), void);
for (j = 0; j < b->buffer.n_datas; j++) {
if (b->datas[j].type == SPA_DATA_TYPE_MEMPTR)
b->datas[j].data = SPA_MEMBER (sb, SPA_PTR_TO_INT (sbd[j].data), void);
}
p += b->size;
}
p += b->size;
}
am.direction = direction;
am.port_id = port_id;
am.mem_id = port->buffer_mem_id;
am.type = SPA_DATA_TYPE_MEMFD;
am.fd_index = spa_control_builder_add_fd (&builder, port->buffer_mem_fd, false);
am.flags = 0;
am.offset = 0;
am.size = size;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_MEM, &am);
am.direction = direction;
am.port_id = port_id;
am.mem_id = port->buffer_mem_id;
am.type = SPA_DATA_TYPE_MEMFD;
am.fd_index = spa_control_builder_add_fd (&builder, port->buffer_mem_fd, false);
am.flags = 0;
am.offset = 0;
am.size = size;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_MEM, &am);
memref = alloca (n_buffers * sizeof (SpaControlMemRef));
for (i = 0; i < n_buffers; i++) {
memref[i].mem_id = port->buffer_mem_id;
memref[i].offset = port->buffers[i].offset;
memref[i].size = port->buffers[i].size;
memref = alloca (n_buffers * sizeof (SpaControlMemRef));
for (i = 0; i < n_buffers; i++) {
memref[i].mem_id = port->buffer_mem_id;
memref[i].offset = port->buffers[i].offset;
memref[i].size = port->buffers[i].size;
}
} else {
memref = NULL;
}
port->n_buffers = n_buffers;
@ -988,7 +1025,7 @@ spa_proxy_node_port_push_input (SpaNode *node,
if (have_enough)
return SPA_RESULT_HAVE_ENOUGH_INPUT;
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
if ((res = spa_control_write (&control, this->rtfds[0].fd)) < 0)
spa_log_error (this->log, "proxy %p: error writing control\n", this);
spa_control_clear (&control);
@ -1074,7 +1111,7 @@ spa_proxy_node_port_reuse_buffer (SpaNode *node,
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_NODE_EVENT, &cne);
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
if ((res = spa_control_write (&control, this->rtfds[0].fd)) < 0)
spa_log_error (this->log, "proxy %p: error writing control %d\n", this, res);
spa_control_clear (&control);
@ -1145,6 +1182,7 @@ parse_control (SpaProxy *this,
case SPA_CONTROL_CMD_SET_FORMAT:
case SPA_CONTROL_CMD_SET_PROPERTY:
case SPA_CONTROL_CMD_NODE_COMMAND:
case SPA_CONTROL_CMD_PROCESS_BUFFER:
spa_log_error (this->log, "proxy %p: got unexpected control %d\n", this, cmd);
break;
@ -1217,6 +1255,50 @@ parse_control (SpaProxy *this,
case SPA_CONTROL_CMD_USE_BUFFERS:
break;
case SPA_CONTROL_CMD_NODE_EVENT:
{
SpaControlCmdNodeEvent cne;
if (spa_control_iter_parse_cmd (&it, &cne) < 0)
break;
handle_node_event (this, cne.event);
break;
}
}
}
spa_control_iter_end (&it);
return SPA_RESULT_OK;
}
static SpaResult
parse_rtcontrol (SpaProxy *this,
SpaControl *ctrl)
{
SpaControlIter it;
spa_control_iter_init (&it, ctrl);
while (spa_control_iter_next (&it) == SPA_RESULT_OK) {
SpaControlCmd cmd = spa_control_iter_get_cmd (&it);
switch (cmd) {
case SPA_CONTROL_CMD_INVALID:
case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
case SPA_CONTROL_CMD_NODE_STATE_CHANGE:
case SPA_CONTROL_CMD_PORT_STATUS_CHANGE:
case SPA_CONTROL_CMD_ADD_PORT:
case SPA_CONTROL_CMD_REMOVE_PORT:
case SPA_CONTROL_CMD_SET_FORMAT:
case SPA_CONTROL_CMD_SET_PROPERTY:
case SPA_CONTROL_CMD_NODE_COMMAND:
case SPA_CONTROL_CMD_ADD_MEM:
case SPA_CONTROL_CMD_REMOVE_MEM:
case SPA_CONTROL_CMD_USE_BUFFERS:
spa_log_error (this->log, "proxy %p: got unexpected control %d\n", this, cmd);
break;
case SPA_CONTROL_CMD_PROCESS_BUFFER:
{
SpaControlCmdProcessBuffer cmd;
@ -1266,7 +1348,7 @@ proxy_on_fd_events (SpaPollNotifyData *data)
uint8_t buf[1024];
int fds[16];
if ((res = spa_control_read (&control, data->fds[0].fd, buf, 1024, fds, 16)) < 0) {
if ((res = spa_control_read (&control, data->fds[0].fd, buf, sizeof (buf), fds, 16)) < 0) {
spa_log_error (this->log, "proxy %p: failed to read control: %d\n", this, res);
return 0;
}
@ -1276,6 +1358,26 @@ proxy_on_fd_events (SpaPollNotifyData *data)
return 0;
}
static int
proxy_on_rtfd_events (SpaPollNotifyData *data)
{
SpaProxy *this = data->user_data;
SpaResult res;
if (data->fds[0].revents & POLLIN) {
SpaControl control;
uint8_t buf[1024];
if ((res = spa_control_read (&control, data->fds[0].fd, buf, sizeof (buf), NULL, 0)) < 0) {
spa_log_error (this->log, "proxy %p: failed to read control: %d\n", this, res);
return 0;
}
parse_rtcontrol (this, &control);
spa_control_clear (&control);
}
return 0;
}
static const SpaNode proxy_node = {
sizeof (SpaNode),
NULL,
@ -1367,6 +1469,8 @@ proxy_init (const SpaHandleFactory *factory,
this->map = support[i].data;
else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
this->log = support[i].data;
else if (strcmp (support[i].uri, SPA_POLL__MainLoop) == 0)
this->main_loop = support[i].data;
else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
this->data_loop = support[i].data;
}
@ -1398,6 +1502,18 @@ proxy_init (const SpaHandleFactory *factory,
this->poll.after_cb = proxy_on_fd_events;
this->poll.user_data = this;
this->rtfds[0].fd = -1;
this->rtfds[0].events = POLLIN | POLLPRI | POLLERR;
this->rtfds[0].revents = 0;
this->rtpoll.id = 0;
this->rtpoll.enabled = true;
this->rtpoll.fds = this->rtfds;
this->rtpoll.n_fds = 1;
this->rtpoll.idle_cb = NULL;
this->rtpoll.before_cb = NULL;
this->rtpoll.after_cb = proxy_on_rtfd_events;
this->rtpoll.user_data = this;
return SPA_RESULT_RETURN_ASYNC (this->seq++);
}