mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-03 09:01:54 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions
pipewire/pinos/server/client-node.c
Wim Taymans 0d0385b881 Rework buffer memory layout 
Move offset, size and stride to separate chunk structure that we
can keep in shared memory.
Add shared metadata type to hold the memory block holding the
metadata.
2016-12-15 14:57:34 +01:00

1309 lines
35 KiB
C
Raw Blame History

/* Pinos
* Copyright (C) 2015 Wim Taymans <wim.taymans@gmail.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.
*/
#include <string.h>
#include <stddef.h>
#include <stdio.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <dlfcn.h>
#include <sys/socket.h>
#include <sys/mman.h>
#include <sys/eventfd.h>
#include "pinos/client/pinos.h"
#include "pinos/client/connection.h"
#include "pinos/client/serialize.h"
#include "pinos/client/transport.h"
#include "pinos/server/core.h"
#include "pinos/server/client-node.h"
#include "spa/include/spa/node.h"
#define MAX_INPUTS 64
#define MAX_OUTPUTS 64
#define MAX_BUFFERS 64
#define CHECK_IN_PORT_ID(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_INPUTS)
#define CHECK_OUT_PORT_ID(this,d,p) ((d) == SPA_DIRECTION_OUTPUT && (p) < MAX_OUTPUTS)
#define CHECK_PORT_ID(this,d,p) (CHECK_IN_PORT_ID(this,d,p) || CHECK_OUT_PORT_ID(this,d,p))
#define CHECK_FREE_IN_PORT(this,d,p) (CHECK_IN_PORT_ID(this,d,p) && !(this)->in_ports[p].valid)
#define CHECK_FREE_OUT_PORT(this,d,p) (CHECK_OUT_PORT_ID(this,d,p) && !(this)->out_ports[p].valid)
#define CHECK_FREE_PORT(this,d,p) (CHECK_FREE_IN_PORT (this,d,p) || CHECK_FREE_OUT_PORT (this,d,p))
#define CHECK_IN_PORT(this,d,p) (CHECK_IN_PORT_ID(this,d,p) && (this)->in_ports[p].valid)
#define CHECK_OUT_PORT(this,d,p) (CHECK_OUT_PORT_ID(this,d,p) && (this)->out_ports[p].valid)
#define CHECK_PORT(this,d,p) (CHECK_IN_PORT (this,d,p) || CHECK_OUT_PORT (this,d,p))
#define CHECK_PORT_BUFFER(this,b,p) (b < p->n_buffers)
typedef struct _SpaProxy SpaProxy;
typedef struct _ProxyBuffer ProxyBuffer;
struct _ProxyBuffer {
SpaBuffer *outbuf;
SpaBuffer buffer;
SpaMeta metas[4];
SpaData datas[4];
off_t offset;
size_t size;
bool outstanding;
};
typedef struct {
bool valid;
SpaPortInfo *info;
SpaFormat *format;
unsigned int n_formats;
SpaFormat **formats;
void *io;
unsigned int n_buffers;
ProxyBuffer buffers[MAX_BUFFERS];
uint32_t buffer_mem_id;
PinosMemblock buffer_mem;
} SpaProxyPort;
struct _SpaProxy
{
SpaNode node;
PinosNode *pnode;
SpaIDMap *map;
SpaLog *log;
SpaLoop *main_loop;
SpaLoop *data_loop;
SpaNodeEventCallback event_cb;
void *user_data;
PinosResource *resource;
SpaSource data_source;
unsigned int max_inputs;
unsigned int n_inputs;
unsigned int max_outputs;
unsigned int n_outputs;
SpaProxyPort in_ports[MAX_INPUTS];
SpaProxyPort out_ports[MAX_OUTPUTS];
uint32_t seq;
};
typedef struct
{
PinosClientNode this;
PinosCore *core;
SpaProxy proxy;
PinosListener transport_changed;
PinosListener loop_changed;
int data_fd;
} PinosClientNodeImpl;
static void
send_async_complete (SpaProxy *this, uint32_t seq, SpaResult res)
{
SpaNodeEventAsyncComplete ac;
ac.event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE;
ac.event.size = sizeof (ac);
ac.seq = seq;
ac.res = res;
this->event_cb (&this->node, &ac.event, this->user_data);
}
static SpaResult
clear_buffers (SpaProxy *this, SpaProxyPort *port)
{
if (port->n_buffers) {
spa_log_info (this->log, "proxy %p: clear buffers", this);
pinos_memblock_free (&port->buffer_mem);
port->n_buffers = 0;
}
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_get_props (SpaNode *node,
SpaProps **props)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_proxy_node_set_props (SpaNode *node,
const SpaProps *props)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_proxy_node_send_command (SpaNode *node,
SpaNodeCommand *command)
{
SpaProxy *this;
SpaResult res = SPA_RESULT_OK;
if (node == NULL || command == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
switch (command->type) {
case SPA_NODE_COMMAND_INVALID:
return SPA_RESULT_INVALID_COMMAND;
case SPA_NODE_COMMAND_START:
case SPA_NODE_COMMAND_PAUSE:
case SPA_NODE_COMMAND_FLUSH:
case SPA_NODE_COMMAND_DRAIN:
case SPA_NODE_COMMAND_MARKER:
{
PinosMessageNodeCommand cnc;
/* send start */
cnc.seq = this->seq++;
cnc.command = command;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_NODE_COMMAND,
&cnc,
true);
res = SPA_RESULT_RETURN_ASYNC (cnc.seq);
break;
}
case SPA_NODE_COMMAND_CLOCK_UPDATE:
{
PinosMessageNodeCommand cnc;
/* send start */
cnc.command = command;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_NODE_COMMAND,
&cnc,
true);
break;
}
}
return res;
}
static SpaResult
spa_proxy_node_set_event_callback (SpaNode *node,
SpaNodeEventCallback event,
void *user_data)
{
SpaProxy *this;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
this->event_cb = event;
this->user_data = user_data;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_get_n_ports (SpaNode *node,
unsigned int *n_input_ports,
unsigned int *max_input_ports,
unsigned int *n_output_ports,
unsigned int *max_output_ports)
{
SpaProxy *this;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (n_input_ports)
*n_input_ports = this->n_inputs;
if (max_input_ports)
*max_input_ports = this->max_inputs;
if (n_output_ports)
*n_output_ports = this->n_outputs;
if (max_output_ports)
*max_output_ports = this->max_outputs;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_get_port_ids (SpaNode *node,
unsigned int n_input_ports,
uint32_t *input_ids,
unsigned int n_output_ports,
uint32_t *output_ids)
{
SpaProxy *this;
int c, i;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (input_ids) {
for (c = 0, i = 0; i < MAX_INPUTS && c < n_input_ports; i++) {
if (this->in_ports[i].valid)
input_ids[c++] = i;
}
}
if (output_ids) {
for (c = 0, i = 0; i < MAX_OUTPUTS && c < n_output_ports; i++) {
if (this->out_ports[i].valid)
output_ids[c++] = i;
}
}
return SPA_RESULT_OK;
}
static void
do_update_port (SpaProxy *this,
PinosMessagePortUpdate *pu)
{
SpaProxyPort *port;
unsigned int i;
size_t size;
if (pu->direction == SPA_DIRECTION_INPUT) {
port = &this->in_ports[pu->port_id];
} else {
port = &this->out_ports[pu->port_id];
}
if (pu->change_mask & PINOS_MESSAGE_PORT_UPDATE_POSSIBLE_FORMATS) {
for (i = 0; i < port->n_formats; i++)
free (port->formats[i]);
port->n_formats = pu->n_possible_formats;
port->formats = realloc (port->formats, port->n_formats * sizeof (SpaFormat *));
for (i = 0; i < port->n_formats; i++) {
size = pinos_serialize_format_get_size (pu->possible_formats[i]);
port->formats[i] = pinos_serialize_format_copy_into (malloc (size), pu->possible_formats[i]);
}
}
if (pu->change_mask & PINOS_MESSAGE_PORT_UPDATE_FORMAT) {
if (port->format)
free (port->format);
size = pinos_serialize_format_get_size (pu->format);
port->format = pinos_serialize_format_copy_into (malloc (size), pu->format);
}
if (pu->change_mask & PINOS_MESSAGE_PORT_UPDATE_PROPS) {
}
if (pu->change_mask & PINOS_MESSAGE_PORT_UPDATE_INFO && pu->info) {
if (port->info)
free (port->info);
size = pinos_serialize_port_info_get_size (pu->info);
port->info = pinos_serialize_port_info_copy_into (malloc (size), pu->info);
}
if (!port->valid) {
spa_log_info (this->log, "proxy %p: adding port %d", this, pu->port_id);
port->format = NULL;
port->valid = true;
if (pu->direction == SPA_DIRECTION_INPUT)
this->n_inputs++;
else
this->n_outputs++;
}
}
static void
clear_port (SpaProxy *this,
SpaProxyPort *port,
SpaDirection direction,
uint32_t port_id)
{
PinosMessagePortUpdate pu;
pu.change_mask = PINOS_MESSAGE_PORT_UPDATE_POSSIBLE_FORMATS |
PINOS_MESSAGE_PORT_UPDATE_FORMAT |
PINOS_MESSAGE_PORT_UPDATE_PROPS |
PINOS_MESSAGE_PORT_UPDATE_INFO;
pu.direction = direction;
pu.port_id = port_id;
pu.n_possible_formats = 0;
pu.possible_formats = NULL;
pu.format = NULL;
pu.props = NULL;
pu.info = NULL;
do_update_port (this, &pu);
clear_buffers (this, port);
}
static void
do_uninit_port (SpaProxy *this,
SpaDirection direction,
uint32_t port_id)
{
SpaProxyPort *port;
spa_log_info (this->log, "proxy %p: removing port %d", this, port_id);
if (direction == SPA_DIRECTION_INPUT) {
port = &this->in_ports[port_id];
this->n_inputs--;
} else {
port = &this->out_ports[port_id];
this->n_outputs--;
}
clear_port (this, port, direction, port_id);
port->valid = false;
}
static SpaResult
spa_proxy_node_add_port (SpaNode *node,
SpaDirection direction,
uint32_t port_id)
{
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_FREE_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
clear_port (this, port, direction, port_id);
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_remove_port (SpaNode *node,
SpaDirection direction,
uint32_t port_id)
{
SpaProxy *this;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
do_uninit_port (this, direction, port_id);
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_enum_formats (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaFormat **format,
const SpaFormat *filter,
void **state)
{
SpaProxy *this;
SpaProxyPort *port;
int index;
if (node == NULL || format == NULL || state == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
index = (*state == NULL ? 0 : *(int*)state);
if (index >= port->n_formats)
return SPA_RESULT_ENUM_END;
*format = port->formats[index];
*(int*)state = ++index;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_set_format (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaPortFormatFlags flags,
const SpaFormat *format)
{
SpaProxy *this;
PinosMessageSetFormat sf;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
sf.seq = this->seq++;
sf.direction = direction;
sf.port_id = port_id;
sf.flags = flags;
sf.format = (SpaFormat *) format;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_SET_FORMAT,
&sf,
true);
return SPA_RESULT_RETURN_ASYNC (sf.seq);
}
static SpaResult
spa_proxy_node_port_get_format (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
const SpaFormat **format)
{
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || format == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
if (!port->format)
return SPA_RESULT_NO_FORMAT;
*format = port->format;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_get_info (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
const SpaPortInfo **info)
{
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || info == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
*info = port->info;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_get_props (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaProps **props)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_proxy_node_port_set_props (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
const SpaProps *props)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_proxy_node_port_set_input (SpaNode *node,
uint32_t port_id,
SpaPortInput *input)
{
SpaProxy *this;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, SPA_DIRECTION_INPUT, port_id))
return SPA_RESULT_INVALID_PORT;
this->in_ports[port_id].io = input;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_set_output (SpaNode *node,
uint32_t port_id,
SpaPortOutput *output)
{
SpaProxy *this;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, SPA_DIRECTION_OUTPUT, port_id))
return SPA_RESULT_INVALID_PORT;
this->out_ports[port_id].io = output;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_use_buffers (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
SpaProxy *this;
SpaProxyPort *port;
unsigned int i, j;
PinosMessageAddMem am;
PinosMessageUseBuffers ub;
size_t n_mem;
PinosMessageBuffer *mb;
SpaMetaShared *msh;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
spa_log_info (this->log, "proxy %p: use buffers %p %u", this, buffers, n_buffers);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
if (!port->format)
return SPA_RESULT_NO_FORMAT;
clear_buffers (this, port);
if (n_buffers > 0) {
mb = alloca (n_buffers * sizeof (PinosMessageBuffer));
} else {
mb = NULL;
}
n_mem = 0;
for (i = 0; i < n_buffers; i++) {
ProxyBuffer *b = &port->buffers[i];
msh = spa_buffer_find_meta (buffers[i], SPA_META_TYPE_SHARED);
if (msh == NULL) {
spa_log_error (this->log, "missing shared metadata on buffer %d", i);
return SPA_RESULT_ERROR;
}
b->outbuf = buffers[i];
memcpy (&b->buffer, buffers[i], sizeof (SpaBuffer));
b->buffer.datas = b->datas;
b->buffer.metas = b->metas;
am.direction = direction;
am.port_id = port_id;
am.mem_id = n_mem++;
am.type = SPA_DATA_TYPE_MEMFD;
am.memfd = msh->fd;
am.flags = msh->flags;
am.offset = msh->offset;
am.size = msh->size;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_ADD_MEM,
&am,
false);
mb[i].buffer = &b->buffer;
mb[i].mem_id = am.mem_id;
mb[i].offset = 0;
mb[i].size = am.size;
for (j = 0; j < buffers[i]->n_metas; j++) {
memcpy (&b->buffer.metas[j], &buffers[i]->metas[j], sizeof (SpaMeta));
}
for (j = 0; j < buffers[i]->n_datas; j++) {
SpaData *d = &buffers[i]->datas[j];
memcpy (&b->buffer.datas[j], d, sizeof (SpaData));
switch (d->type) {
case SPA_DATA_TYPE_DMABUF:
case SPA_DATA_TYPE_MEMFD:
am.direction = direction;
am.port_id = port_id;
am.mem_id = n_mem;
am.type = d->type;
am.memfd = d->fd;
am.flags = d->flags;
am.offset = d->offset;
am.size = d->size;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_ADD_MEM,
&am,
false);
b->buffer.datas[j].type = SPA_DATA_TYPE_ID;
b->buffer.datas[j].data = SPA_UINT32_TO_PTR (n_mem);
n_mem++;
break;
case SPA_DATA_TYPE_MEMPTR:
b->buffer.datas[j].data = SPA_INT_TO_PTR (b->size);
b->size += d->size;
break;
default:
b->buffer.datas[j].type = SPA_DATA_TYPE_INVALID;
b->buffer.datas[j].data = 0;
spa_log_error (this->log, "invalid memory type %d", d->type);
break;
}
}
}
port->n_buffers = n_buffers;
ub.seq = this->seq++;
ub.direction = direction;
ub.port_id = port_id;
ub.n_buffers = n_buffers;
ub.buffers = mb;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_USE_BUFFERS,
&ub,
true);
return SPA_RESULT_RETURN_ASYNC (ub.seq);
}
static SpaResult
spa_proxy_node_port_alloc_buffers (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || buffers == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
if (!CHECK_PORT (this, direction, port_id))
return SPA_RESULT_INVALID_PORT;
port = direction == SPA_DIRECTION_INPUT ? &this->in_ports[port_id] : &this->out_ports[port_id];
if (!port->format)
return SPA_RESULT_NO_FORMAT;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static void
copy_meta_in (SpaProxy *this, SpaProxyPort *port, uint32_t buffer_id)
{
#if 0
ProxyBuffer *b = &port->buffers[buffer_id];
unsigned int i;
for (i = 0; i < b->outbuf->n_metas; i++) {
SpaMeta *sm = &b->buffer.metas[i];
SpaMeta *dm = &b->outbuf->metas[i];
memcpy (dm->data, sm->data, dm->size);
}
for (i = 0; i < b->outbuf->n_datas; i++) {
b->outbuf->datas[i].size = b->buffer.datas[i].size;
if (b->outbuf->datas[i].type == SPA_DATA_TYPE_MEMPTR) {
spa_log_info (this->log, "memcpy in %zd", b->buffer.datas[i].size);
memcpy (b->outbuf->datas[i].data, b->datas[i].data, b->buffer.datas[i].size);
}
}
#endif
}
static void
copy_meta_out (SpaProxy *this, SpaProxyPort *port, uint32_t buffer_id)
{
#if 0
ProxyBuffer *b = &port->buffers[buffer_id];
unsigned int i;
for (i = 0; i < b->outbuf->n_metas; i++) {
SpaMeta *sm = &b->outbuf->metas[i];
SpaMeta *dm = &b->buffer.metas[i];
memcpy (dm->data, sm->data, dm->size);
}
for (i = 0; i < b->outbuf->n_datas; i++) {
b->buffer.datas[i].size = b->outbuf->datas[i].size;
if (b->datas[i].type == SPA_DATA_TYPE_MEMPTR) {
spa_log_info (this->log, "memcpy out %zd", b->outbuf->datas[i].size);
memcpy (b->datas[i].data, b->outbuf->datas[i].data, b->outbuf->datas[i].size);
}
}
#endif
}
static SpaResult
spa_proxy_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id)
{
SpaProxy *this;
SpaNodeEventReuseBuffer rb;
PinosNode *pnode;
uint8_t cmd;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
pnode = this->pnode;
if (!CHECK_OUT_PORT (this, SPA_DIRECTION_OUTPUT, port_id))
return SPA_RESULT_INVALID_PORT;
rb.event.type = SPA_NODE_EVENT_TYPE_REUSE_BUFFER;
rb.event.size = sizeof (rb);
rb.port_id = port_id;
rb.buffer_id = buffer_id;
pinos_transport_add_event (pnode->transport, &rb.event);
cmd = PINOS_TRANSPORT_CMD_HAVE_EVENT;
write (this->data_source.fd, &cmd, 1);
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_send_command (SpaNode *node,
SpaDirection direction,
uint32_t port_id,
SpaNodeCommand *command)
{
SpaProxy *this;
SpaResult res = SPA_RESULT_OK;
if (node == NULL || command == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
switch (command->type) {
case SPA_NODE_COMMAND_INVALID:
return SPA_RESULT_INVALID_COMMAND;
case SPA_NODE_COMMAND_START:
case SPA_NODE_COMMAND_PAUSE:
case SPA_NODE_COMMAND_FLUSH:
case SPA_NODE_COMMAND_DRAIN:
case SPA_NODE_COMMAND_MARKER:
break;
default:
spa_log_warn (this->log, "unhandled command %d", command->type);
res = SPA_RESULT_NOT_IMPLEMENTED;
break;
}
return res;
}
static SpaResult
spa_proxy_node_process_input (SpaNode *node)
{
SpaProxy *this;
unsigned int i;
bool have_error = false;
uint8_t cmd;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
for (i = 0; i < this->n_inputs; i++) {
SpaProxyPort *port = &this->in_ports[i];
SpaPortInput *input;
if ((input = port->io) == NULL)
continue;
if (!CHECK_PORT_BUFFER (this, input->buffer_id, port)) {
input->status = SPA_RESULT_INVALID_BUFFER_ID;
have_error = true;
continue;
}
copy_meta_out (this, port, input->buffer_id);
}
if (have_error)
return SPA_RESULT_ERROR;
cmd = PINOS_TRANSPORT_CMD_HAVE_DATA;
write (this->data_source.fd, &cmd, 1);
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_process_output (SpaNode *node)
{
SpaProxy *this;
unsigned int i;
bool have_error = false;
if (node == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = SPA_CONTAINER_OF (node, SpaProxy, node);
for (i = 0; i < this->n_outputs; i++) {
SpaProxyPort *port = &this->out_ports[i];
SpaPortOutput *output;
if ((output = port->io) == NULL)
continue;
copy_meta_in (this, port, output->buffer_id);
if (output->status != SPA_RESULT_OK)
have_error = true;
}
if (have_error)
return SPA_RESULT_ERROR;
return SPA_RESULT_OK;
}
static SpaResult
handle_node_event (SpaProxy *this,
SpaNodeEvent *event)
{
switch (event->type) {
case SPA_NODE_EVENT_TYPE_INVALID:
break;
case SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE:
case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
case SPA_NODE_EVENT_TYPE_NEED_INPUT:
case SPA_NODE_EVENT_TYPE_REUSE_BUFFER:
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:
this->event_cb (&this->node, event, this->user_data);
break;
}
return SPA_RESULT_OK;
}
static SpaResult
client_node_dispatch_func (void *object,
PinosMessageType type,
void *message,
void *data)
{
PinosClientNode *node = data;
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (node, PinosClientNodeImpl, this);
SpaProxy *this = &impl->proxy;
switch (type) {
default:
spa_log_error (this->log, "proxy %p: got unexpected command %d", this, type);
break;
case PINOS_MESSAGE_NODE_UPDATE:
{
PinosMessageNodeUpdate *nu = message;
if (nu->change_mask & PINOS_MESSAGE_NODE_UPDATE_MAX_INPUTS)
this->max_inputs = nu->max_input_ports;
if (nu->change_mask & PINOS_MESSAGE_NODE_UPDATE_MAX_OUTPUTS)
this->max_outputs = nu->max_output_ports;
spa_log_info (this->log, "proxy %p: got node update %d, max_in %u, max_out %u", this, type,
this->max_inputs, this->max_outputs);
break;
}
case PINOS_MESSAGE_PORT_UPDATE:
{
PinosMessagePortUpdate *pu = message;
bool remove;
spa_log_info (this->log, "proxy %p: got port update %d", this, type);
if (!CHECK_PORT_ID (this, pu->direction, pu->port_id))
break;
remove = (pu->change_mask == 0);
if (remove) {
do_uninit_port (this, pu->direction, pu->port_id);
} else {
do_update_port (this, pu);
}
break;
}
case PINOS_MESSAGE_PORT_STATUS_CHANGE:
{
spa_log_warn (this->log, "proxy %p: command not implemented %d", this, type);
break;
}
case PINOS_MESSAGE_NODE_STATE_CHANGE:
{
PinosMessageNodeStateChange *sc = message;
SpaNodeState old = this->node.state;
spa_log_info (this->log, "proxy %p: got node state change %d -> %d", this, old, sc->state);
this->node.state = sc->state;
if (old == SPA_NODE_STATE_INIT)
send_async_complete (this, 0, SPA_RESULT_OK);
break;
}
case PINOS_MESSAGE_ADD_MEM:
break;
case PINOS_MESSAGE_USE_BUFFERS:
break;
case PINOS_MESSAGE_NODE_EVENT:
{
PinosMessageNodeEvent *cne = message;
handle_node_event (this, cne->event);
break;
}
}
return SPA_RESULT_OK;
}
static void
proxy_on_data_fd_events (SpaSource *source)
{
SpaProxy *this = source->data;
PinosNode *pnode = this->pnode;
if (source->rmask & SPA_IO_IN) {
uint8_t cmd;
read (this->data_source.fd, &cmd, 1);
if (cmd & PINOS_TRANSPORT_CMD_HAVE_EVENT) {
SpaNodeEvent event;
while (pinos_transport_next_event (pnode->transport, &event) == SPA_RESULT_OK) {
SpaNodeEvent *ev = alloca (event.size);
pinos_transport_parse_event (pnode->transport, ev);
this->event_cb (&this->node, ev, this->user_data);
}
}
if (cmd & PINOS_TRANSPORT_CMD_HAVE_DATA) {
SpaNodeEventHaveOutput ho;
ho.event.type = SPA_NODE_EVENT_TYPE_HAVE_OUTPUT;
ho.event.size = sizeof (ho);
ho.port_id = 0;
this->event_cb (&this->node, &ho.event, this->user_data);
}
if (cmd & PINOS_TRANSPORT_CMD_NEED_DATA) {
SpaNodeEventNeedInput ni;
ni.event.type = SPA_NODE_EVENT_TYPE_NEED_INPUT;
ni.event.size = sizeof (ni);
ni.port_id = 0;
this->event_cb (&this->node, &ni.event, this->user_data);
}
}
}
static const SpaNode proxy_node = {
sizeof (SpaNode),
NULL,
SPA_NODE_STATE_INIT,
spa_proxy_node_get_props,
spa_proxy_node_set_props,
spa_proxy_node_send_command,
spa_proxy_node_set_event_callback,
spa_proxy_node_get_n_ports,
spa_proxy_node_get_port_ids,
spa_proxy_node_add_port,
spa_proxy_node_remove_port,
spa_proxy_node_port_enum_formats,
spa_proxy_node_port_set_format,
spa_proxy_node_port_get_format,
spa_proxy_node_port_get_info,
spa_proxy_node_port_get_props,
spa_proxy_node_port_set_props,
spa_proxy_node_port_use_buffers,
spa_proxy_node_port_alloc_buffers,
spa_proxy_node_port_set_input,
spa_proxy_node_port_set_output,
spa_proxy_node_port_reuse_buffer,
spa_proxy_node_port_send_command,
spa_proxy_node_process_input,
spa_proxy_node_process_output,
};
static SpaResult
proxy_init (SpaProxy *this,
SpaDict *info,
const SpaSupport *support,
unsigned int n_support)
{
unsigned int i;
for (i = 0; i < n_support; i++) {
if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
this->log = support[i].data;
else if (strcmp (support[i].uri, SPA_LOOP__MainLoop) == 0)
this->main_loop = support[i].data;
else if (strcmp (support[i].uri, SPA_LOOP__DataLoop) == 0)
this->data_loop = support[i].data;
}
if (this->data_loop == NULL) {
spa_log_error (this->log, "a main-loop is needed");
}
if (this->data_loop == NULL) {
spa_log_error (this->log, "a data-loop is needed");
}
this->node = proxy_node;
this->data_source.func = proxy_on_data_fd_events;
this->data_source.data = this;
this->data_source.fd = -1;
this->data_source.mask = SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP;
this->data_source.rmask = 0;
return SPA_RESULT_RETURN_ASYNC (this->seq++);
}
static void
on_transport_changed (PinosListener *listener,
PinosNode *node)
{
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (listener, PinosClientNodeImpl, transport_changed);
PinosClientNode *this = &impl->this;
PinosTransportInfo info;
PinosMessageTransportUpdate tu;
pinos_transport_get_info (node->transport, &info);
tu.memfd = info.memfd;
tu.offset = info.offset;
tu.size = info.size;
pinos_resource_send_message (this->resource,
PINOS_MESSAGE_TRANSPORT_UPDATE,
&tu,
true);
}
static void
on_loop_changed (PinosListener *listener,
PinosNode *node)
{
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (listener, PinosClientNodeImpl, loop_changed);
impl->proxy.data_loop = node->data_loop->loop->loop;
}
static SpaResult
proxy_clear (SpaProxy *this)
{
unsigned int i;
for (i = 0; i < MAX_INPUTS; i++) {
if (this->in_ports[i].valid)
clear_port (this, &this->in_ports[i], SPA_DIRECTION_INPUT, i);
}
for (i = 0; i < MAX_OUTPUTS; i++) {
if (this->out_ports[i].valid)
clear_port (this, &this->out_ports[i], SPA_DIRECTION_OUTPUT, i);
}
if (this->data_source.fd != -1) {
spa_loop_remove_source (this->data_loop, &this->data_source);
close (this->data_source.fd);
}
return SPA_RESULT_OK;
}
static void
client_node_resource_destroy (PinosResource *resource)
{
pinos_client_node_destroy (resource->object);
}
/**
* pinos_client_node_new:
* @daemon: a #PinosDaemon
* @name: a name
* @properties: extra properties
*
* Create a new #PinosNode.
*
* Returns: a new #PinosNode
*/
PinosClientNode *
pinos_client_node_new (PinosClient *client,
uint32_t id,
const char *name,
PinosProperties *properties)
{
PinosClientNodeImpl *impl;
PinosClientNode *this;
impl = calloc (1, sizeof (PinosClientNodeImpl));
this = &impl->this;
impl->core = client->core;
impl->data_fd = -1;
pinos_log_debug ("client-node %p: new", impl);
pinos_signal_init (&this->destroy_signal);
proxy_init (&impl->proxy, NULL, client->core->support, client->core->n_support);
this->node = pinos_node_new (client->core,
name,
&impl->proxy.node,
NULL,
properties);
impl->proxy.pnode = this->node;
pinos_signal_add (&this->node->transport_changed,
&impl->transport_changed,
on_transport_changed);
pinos_signal_add (&this->node->loop_changed,
&impl->loop_changed,
on_loop_changed);
this->resource = pinos_resource_new (client,
id,
client->core->uri.client_node,
this,
(PinosDestroy) client_node_resource_destroy);
impl->proxy.resource = this->resource;
this->resource->dispatch_func = client_node_dispatch_func;
this->resource->dispatch_data = this;
return this;
}
static void
on_node_remove (void *object,
void *data,
SpaResult res,
uint32_t id)
{
PinosClientNode *this = object;
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (this, PinosClientNodeImpl, this);
pinos_log_debug ("client-node %p: finalize", this);
proxy_clear (&impl->proxy);
if (impl->data_fd != -1)
close (impl->data_fd);
free (impl);
}
void
pinos_client_node_destroy (PinosClientNode * this)
{
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (this, PinosClientNodeImpl, this);
pinos_log_debug ("client-node %p: destroy", impl);
pinos_signal_emit (&this->destroy_signal, this);
pinos_node_destroy (this->node);
pinos_main_loop_defer (this->node->core->main_loop,
this,
SPA_RESULT_WAIT_SYNC,
(PinosDeferFunc) on_node_remove,
this);
}
/**
* pinos_client_node_get_data_socket:
* @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: %SPA_RESULT_OK on success
*/
SpaResult
pinos_client_node_get_data_socket (PinosClientNode *this,
int *fd)
{
PinosClientNodeImpl *impl = SPA_CONTAINER_OF (this, PinosClientNodeImpl, this);
if (impl->data_fd == -1) {
int fd[2];
if (socketpair (AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, fd) != 0)
return SPA_RESULT_ERRNO;
impl->proxy.data_source.fd = fd[0];
spa_loop_add_source (impl->proxy.data_loop, &impl->proxy.data_source);
pinos_log_debug ("client-node %p: add data fd %d", this, fd[0]);
impl->data_fd = fd[1];
}
*fd = impl->data_fd;
return SPA_RESULT_OK;
}
Reference in a new issue View git blame Copy permalink