mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-10-29 05:40:27 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

mem: improve memory handling

Browse source 
Add a memory pool to manage blocks of memory. Make it possible
to allocate and import blocks.

Add add_mem and remove_mem to the core events to signal a client
of a block of memory. Remove the client-node add_mem.

Make a global pool for memory and a per client pool where we
import and share the memory we need with the client.

Use the new memory pool to track and map memory in clients.
This commit is contained in:
Wim Taymans 2019-07-23 17:41:25 +02:00
parent a8bc8e1b1e
commit 2caf81c97c
16 changed files with 903 additions and 689 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pipewire-jack

@ -1 +1 @@
Subproject commit 78109838a96ff94988df9eee1976f99ddcc938b2
Subproject commit 135b72ce70b5da10424f664edcb531bf530618b1

47
src/extensions/client-node.h
View file

@ -46,46 +46,23 @@ struct pw_client_node_buffer {
struct spa_buffer *buffer; /**< buffer describing metadata and buffer memory */
};
#define PW_CLIENT_NODE_PROXY_EVENT_ADD_MEM 0
#define PW_CLIENT_NODE_PROXY_EVENT_TRANSPORT 1
#define PW_CLIENT_NODE_PROXY_EVENT_SET_PARAM 2
#define PW_CLIENT_NODE_PROXY_EVENT_SET_IO 3
#define PW_CLIENT_NODE_PROXY_EVENT_EVENT 4
#define PW_CLIENT_NODE_PROXY_EVENT_COMMAND 5
#define PW_CLIENT_NODE_PROXY_EVENT_ADD_PORT 6
#define PW_CLIENT_NODE_PROXY_EVENT_REMOVE_PORT 7
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_PARAM 8
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_USE_BUFFERS 9
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_IO 10
#define PW_CLIENT_NODE_PROXY_EVENT_SET_ACTIVATION 11
#define PW_CLIENT_NODE_PROXY_EVENT_NUM 12
#define PW_CLIENT_NODE_PROXY_EVENT_TRANSPORT 0
#define PW_CLIENT_NODE_PROXY_EVENT_SET_PARAM 1
#define PW_CLIENT_NODE_PROXY_EVENT_SET_IO 2
#define PW_CLIENT_NODE_PROXY_EVENT_EVENT 3
#define PW_CLIENT_NODE_PROXY_EVENT_COMMAND 4
#define PW_CLIENT_NODE_PROXY_EVENT_ADD_PORT 5
#define PW_CLIENT_NODE_PROXY_EVENT_REMOVE_PORT 6
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_PARAM 7
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_USE_BUFFERS 8
#define PW_CLIENT_NODE_PROXY_EVENT_PORT_SET_IO 9
#define PW_CLIENT_NODE_PROXY_EVENT_SET_ACTIVATION 10
#define PW_CLIENT_NODE_PROXY_EVENT_NUM 11
/** \ref pw_client_node events */
struct pw_client_node_proxy_events {
#define PW_VERSION_CLIENT_NODE_PROXY_EVENTS 0
uint32_t version;
/**
* Memory was added to a node
*
* Memory is given to a node as an fd in \a memfd of a certain
* memory \a type.
*
* Further references to this fd will be made with the per memory
* unique identifier \a mem_id.
*
* Buffers or controls will reference the memory by \a mem_id and
* mapping the specified area will give access to the memory.
*
* \param mem_id the id of the memory
* \param type the memory type
* \param memfd the fd of the memory
* \param flags flags for the \a memfd
*/
int (*add_mem) (void *object,
uint32_t mem_id,
uint32_t type,
int memfd,
uint32_t flags);
/**
* Notify of a new transport area
*

233
src/modules/module-client-node/client-node.c
View file

@ -51,7 +51,6 @@
#define MAX_BUFFERS 64
#define MAX_AREAS 1024
#define MAX_IO 32
#define MAX_MIX 128
#define CHECK_IN_PORT_ID(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_INPUTS)
@ -70,25 +69,11 @@
#define CHECK_PORT_BUFFER(this,b,p) (b < p->n_buffers)
struct mem {
uint32_t id;
int ref;
int fd;
uint32_t type;
uint32_t flags;
};
struct buffer {
struct spa_buffer *outbuf;
struct spa_buffer buffer;
struct spa_meta metas[4];
struct spa_data datas[4];
uint32_t memid;
};
struct io {
uint32_t id;
uint32_t memid;
};
struct mix {
@ -98,7 +83,6 @@ struct mix {
struct port *port;
uint32_t n_buffers;
struct buffer buffers[MAX_BUFFERS];
struct io ios[MAX_IO];
};
struct port {
@ -133,7 +117,6 @@ struct node {
struct spa_hook_list hooks;
struct spa_callbacks callbacks;
struct io ios[MAX_IO];
struct pw_resource *resource;
@ -214,55 +197,6 @@ do_port_use_buffers(struct impl *impl,
/** \endcond */
static struct mem *ensure_mem(struct impl *impl, int fd, uint32_t type, uint32_t flags)
{
struct mem *m, *f = NULL;
pw_array_for_each(m, &impl->mems) {
if (m->ref <= 0)
f = m;
else if (m->fd == fd)
goto found;
}
if (f == NULL) {
m = pw_array_add(&impl->mems, sizeof(struct mem));
if (m == NULL)
return NULL;
m->id = pw_array_get_len(&impl->mems, struct mem) - 1;
}
else {
m = f;
}
m->fd = fd;
m->type = type;
m->flags = flags;
m->ref = 0;
pw_log_debug(NAME " %p: add mem %d", impl, m->id);
pw_client_node_resource_add_mem(impl->node.resource,
m->id,
type,
m->fd,
m->flags);
found:
m->ref++;
pw_log_debug(NAME " %p: mem %d, ref %d", impl, m->id, m->ref);
return m;
}
static inline struct mem *find_mem_fd(struct impl *impl, int fd)
{
struct mem *m;
pw_array_for_each(m, &impl->mems) {
if (m->fd == fd)
return m;
}
return NULL;
}
static struct mix *find_mix(struct port *p, uint32_t mix_id)
{
struct mix *mix;
@ -274,13 +208,6 @@ static struct mix *find_mix(struct port *p, uint32_t mix_id)
return mix;
}
static void init_ios(struct io *ios)
{
int i;
for (i = 0; i < MAX_IO; i++)
ios[i].id = SPA_ID_INVALID;
}
static void mix_init(struct mix *mix, struct port *p, uint32_t id)
{
mix->valid = true;
@ -288,7 +215,6 @@ static void mix_init(struct mix *mix, struct port *p, uint32_t id)
mix->port = p;
mix->active = false;
mix->n_buffers = 0;
init_ios(mix->ios);
}
@ -304,66 +230,17 @@ static struct mix *ensure_mix(struct impl *impl, struct port *p, uint32_t mix_id
return mix;
}
static void clear_io(struct node *node, struct io *io)
{
struct mem *m;
m = pw_array_get_unchecked(&node->impl->mems, io->memid, struct mem);
m->ref--;
io->id = SPA_ID_INVALID;
spa_log_debug(node->log, "node %p: clear io %p %d mem %u %d", node, io, io->id, io->memid, m->ref);
}
static struct io *update_io(struct node *this,
struct io *ios, uint32_t id, uint32_t memid)
{
int i;
struct io *io, *f = NULL;
for (i = 0; i < MAX_IO; i++) {
io = &ios[i];
if (io->id == SPA_ID_INVALID)
f = io;
else if (io->id == id) {
if (io->memid != memid) {
clear_io(this, io);
if (memid == SPA_ID_INVALID)
io->id = SPA_ID_INVALID;
}
goto found;
}
}
if (f == NULL || memid == SPA_ID_INVALID)
return NULL;
io = f;
io->id = id;
io->memid = memid;
spa_log_debug(this->log, "node %p: add io %p %s %d", this, io,
spa_debug_type_find_name(spa_type_io, id), memid);
found:
return io;
}
static void clear_ios(struct node *this, struct io *ios)
{
int i;
for (i = 0; i < MAX_IO; i++) {
struct io *io = &ios[i];
if (io->id != SPA_ID_INVALID)
clear_io(this, io);
}
}
static int clear_buffers(struct node *this, struct mix *mix)
{
uint32_t i, j;
struct impl *impl = this->impl;
if (this->resource == NULL)
return 0;
for (i = 0; i < mix->n_buffers; i++) {
struct buffer *b = &mix->buffers[i];
struct mem *m;
struct pw_memblock *m;
spa_log_debug(this->log, "node %p: clear buffer %d", this, i);
@ -375,14 +252,11 @@ static int clear_buffers(struct node *this, struct mix *mix)
uint32_t id;
id = SPA_PTR_TO_UINT32(b->buffer.datas[j].data);
m = pw_array_get_unchecked(&impl->mems, id, struct mem);
m->ref--;
pw_log_debug(NAME " %p: mem %d, ref %d", impl, m->id, m->ref);
m = pw_mempool_find_id(this->resource->client->pool, id);
if (m)
pw_log_debug(NAME " %p: mem %d", impl, m->id);
}
}
m = pw_array_get_unchecked(&impl->mems, b->memid, struct mem);
m->ref--;
pw_log_debug(NAME " %p: mem %d, ref %d", impl, m->id, m->ref);
}
mix->n_buffers = 0;
return 0;
@ -396,7 +270,6 @@ static void mix_clear(struct node *this, struct mix *mix)
return;
do_port_use_buffers(this->impl, port->direction, port->id,
mix->id, NULL, 0);
clear_ios(this, mix->ios);
mix->valid = false;
}
@ -458,8 +331,7 @@ static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
{
struct node *this = object;
struct impl *impl;
struct pw_memblock *mem;
struct mem *m;
struct pw_memblock *mem, *m;
uint32_t memid, mem_offset, mem_size;
spa_return_val_if_fail(this != NULL, -EINVAL);
@ -473,19 +345,18 @@ static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
return -EIO;
if (data) {
if ((mem = pw_memblock_find(data)) == NULL)
if ((mem = pw_mempool_find_ptr(impl->core->pool, data)) == NULL)
return -EINVAL;
mem_offset = SPA_PTRDIFF(data, mem->ptr);
mem_size = mem->size;
if (mem_size - mem_offset < size)
mem_offset = SPA_PTRDIFF(data, mem->map->ptr);
if (mem_offset + size > mem->map->size)
return -EINVAL;
mem_offset += mem->offset;
mem_size = size;
m = ensure_mem(impl, mem->fd, SPA_DATA_MemFd, mem->flags);
m = pw_mempool_import_block(this->resource->client->pool, mem);
if (m == NULL)
return -errno;
mem_size = size;
memid = m->id;
}
else {
@ -493,8 +364,6 @@ static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
mem_offset = mem_size = 0;
}
update_io(this, this->ios, id, memid);
return pw_client_node_resource_set_io(this->resource,
id,
memid,
@ -762,8 +631,7 @@ static int do_port_set_io(struct impl *impl,
uint32_t id, void *data, size_t size)
{
struct node *this = &impl->node;
struct pw_memblock *mem;
struct mem *m;
struct pw_memblock *mem, *m;
uint32_t memid, mem_offset, mem_size;
struct port *port;
struct mix *mix;
@ -783,17 +651,17 @@ static int do_port_set_io(struct impl *impl,
return -EINVAL;
if (data) {
if ((mem = pw_memblock_find(data)) == NULL)
if ((mem = pw_mempool_find_ptr(impl->core->pool, data)) == NULL)
return -EINVAL;
mem_offset = SPA_PTRDIFF(data, mem->ptr);
if (mem_offset + size > mem->size)
mem_offset = SPA_PTRDIFF(data, mem->map->ptr);
if (mem_offset + size > mem->map->size)
return -EINVAL;
mem_offset += mem->offset;
m = ensure_mem(impl, mem->fd, SPA_DATA_MemFd, mem->flags);
m = pw_mempool_import_block(this->resource->client->pool, mem);
if (m == NULL)
return -errno;
memid = m->id;
mem_size = size;
}
@ -802,8 +670,6 @@ static int do_port_set_io(struct impl *impl,
mem_offset = mem_size = 0;
}
update_io(this, mix->ios, id, memid);
return pw_client_node_resource_port_set_io(this->resource,
direction, port_id,
mix_id,
@ -867,8 +733,7 @@ do_port_use_buffers(struct impl *impl,
for (i = 0; i < n_buffers; i++) {
struct buffer *b = &mix->buffers[i];
struct pw_memblock *mem;
struct mem *m;
struct pw_memblock *mem, *m;
size_t data_size;
void *baseptr;
@ -884,7 +749,7 @@ do_port_use_buffers(struct impl *impl,
else
return -EINVAL;
if ((mem = pw_memblock_find(baseptr)) == NULL)
if ((mem = pw_mempool_find_ptr(impl->core->pool, baseptr)) == NULL)
return -EINVAL;
data_size = buffers[i]->n_datas * sizeof(struct spa_chunk);
@ -897,14 +762,13 @@ do_port_use_buffers(struct impl *impl,
data_size += d->maxsize;
}
m = ensure_mem(impl, mem->fd, SPA_DATA_MemFd, mem->flags);
m = pw_mempool_import_block(this->resource->client->pool, mem);
if (m == NULL)
return -errno;
b->memid = m->id;
mb[i].buffer = &b->buffer;
mb[i].mem_id = b->memid;
mb[i].offset = SPA_PTRDIFF(baseptr, SPA_MEMBER(mem->ptr, mem->offset, void));
mb[i].mem_id = m->id;
mb[i].offset = SPA_PTRDIFF(baseptr, SPA_MEMBER(mem->map->ptr, 0, void));
mb[i].size = data_size;
spa_log_debug(this->log, "buffer %d %d %d %d", i, mb[i].mem_id,
mb[i].offset, mb[i].size);
@ -920,7 +784,8 @@ do_port_use_buffers(struct impl *impl,
if (d->type == SPA_DATA_DmaBuf ||
d->type == SPA_DATA_MemFd) {
m = ensure_mem(impl, d->fd, d->type, d->flags);
m = pw_mempool_import(this->resource->client->pool,
d->type, d->fd, d->flags);
if (m == NULL)
return -errno;
b->buffer.datas[j].data = SPA_UINT32_TO_PTR(m->id);
@ -1207,8 +1072,6 @@ node_init(struct node *this,
spa_hook_list_init(&this->hooks);
spa_list_init(&this->pending_list);
init_ios(this->ios);
this->data_source.func = node_on_data_fd_events;
this->data_source.data = this;
this->data_source.fd = -1;
@ -1222,8 +1085,6 @@ static int node_clear(struct node *this)
{
uint32_t i;
clear_ios(this, this->ios);
for (i = 0; i < this->n_params; i++)
free(this->params[i]);
free(this->params);
@ -1293,7 +1154,7 @@ void pw_client_node_registered(struct pw_client_node *this, struct pw_global *gl
struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
struct pw_node *node = this->node;
uint32_t node_id = global->id;
struct mem *m;
struct pw_memblock *m;
pw_log_debug(NAME " %p: %d", this, node_id);
pw_client_node_resource_transport(this->resource,
@ -1301,9 +1162,9 @@ void pw_client_node_registered(struct pw_client_node *this, struct pw_global *gl
impl->other_fds[0],
impl->other_fds[1]);
m = ensure_mem(impl, node->activation->fd, SPA_DATA_MemFd, node->activation->flags);
m = pw_mempool_import_block(this->resource->client->pool, node->activation);
if (m == NULL) {
pw_log_debug(NAME " %p: can't ensure mem: %m", this);
pw_log_debug(NAME " %p: can't import block: %m", this);
return;
}
@ -1344,14 +1205,13 @@ static void node_initialized(void *data)
size = sizeof(struct spa_io_buffers) * MAX_AREAS;
if (pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
PW_MEMBLOCK_FLAG_MAP_READWRITE |
PW_MEMBLOCK_FLAG_SEAL,
size,
&impl->io_areas) < 0)
if (pw_mempool_alloc(impl->core->pool,
PW_MEMBLOCK_FLAG_MAP |
PW_MEMBLOCK_FLAG_SEAL,
size, &impl->io_areas) < 0)
return;
pw_log_debug(NAME " %p: io areas %p", node, impl->io_areas->ptr);
pw_log_debug(NAME " %p: io areas %p", node, impl->io_areas->map->ptr);
if ((global = pw_node_get_global(node)) != NULL)
pw_client_node_registered(this, global);
@ -1373,9 +1233,8 @@ static void node_free(void *data)
if (this->resource)
pw_resource_destroy(this->resource);
pw_array_clear(&impl->mems);
if (impl->io_areas)
pw_memblock_free(impl->io_areas);
pw_memblock_unref(impl->io_areas);
pw_map_clear(&impl->io_map);
@ -1399,13 +1258,13 @@ static int port_init_mix(void *data, struct pw_port_mix *mix)
if (mix->id == SPA_ID_INVALID)
return -errno;
mix->io = SPA_MEMBER(impl->io_areas->ptr,
mix->io = SPA_MEMBER(impl->io_areas->map->ptr,
mix->id * sizeof(struct spa_io_buffers), void);
mix->io->buffer_id = SPA_ID_INVALID;
mix->io->status = SPA_STATUS_NEED_BUFFER;
pw_log_debug(NAME " %p: init mix io %d %p %p", impl, mix->id, mix->io,
impl->io_areas->ptr);
impl->io_areas->map->ptr);
return 0;
}
@ -1418,7 +1277,7 @@ static int port_release_mix(void *data, struct pw_port_mix *mix)
struct mix *m;
pw_log_debug(NAME " %p: remove mix io %d %p %p", impl, mix->id, mix->io,
impl->io_areas->ptr);
impl->io_areas->map->ptr);
if ((m = find_mix(port, mix->port.port_id)) == NULL || !m->valid)
return -EINVAL;
@ -1607,17 +1466,16 @@ static void node_peer_added(void *data, struct pw_node *peer)
{
struct impl *impl = data;
struct node *this = &impl->node;
struct mem *m;
struct pw_memblock *m;
if (this->resource == NULL)
return;
m = ensure_mem(impl, peer->activation->fd, SPA_DATA_MemFd, peer->activation->flags);
m = pw_mempool_import_block(this->resource->client->pool, peer->activation);
if (m == NULL) {
pw_log_debug(NAME " %p: can't ensure mem: %m", this);
return;
}
pw_log_debug(NAME " %p: peer %p %u added %u", &impl->this, peer,
peer->info.id, m->id);
@ -1633,19 +1491,18 @@ static void node_peer_removed(void *data, struct pw_node *peer)
{
struct impl *impl = data;
struct node *this = &impl->node;
struct mem *m;
struct pw_memblock *m;
if (this->resource == NULL)
return;
m = find_mem_fd(impl, peer->activation->fd);
m = pw_mempool_find_fd(this->resource->client->pool,
peer->activation->fd);
if (m == NULL) {
pw_log_warn(NAME " %p: unknown peer %p fd:%d", &impl->this, peer,
peer->source.fd);
return;
}
m->ref--;
pw_log_debug(NAME " %p: peer %p %u removed", &impl->this, peer,
peer->info.id);
@ -1655,6 +1512,7 @@ static void node_peer_removed(void *data, struct pw_node *peer)
SPA_ID_INVALID,
0,
0);
pw_memblock_unref(m);
}
static void node_driver_changed(void *data, struct pw_node *old, struct pw_node *driver)
@ -1737,7 +1595,6 @@ struct pw_client_node *pw_client_node_new(struct pw_resource *resource,
this->flags = do_register ? 0 : 1;
pw_map_init(&impl->io_map, 64, 64);
pw_array_init(&impl->mems, 64);
if ((name = pw_properties_get(properties, PW_KEY_NODE_NAME)) == NULL)
name = NAME;

47
src/modules/module-client-node/protocol-native.c
View file

@ -262,30 +262,6 @@ static int client_node_marshal_event_method(void *object, const struct spa_event
return pw_protocol_native_end_proxy(proxy, b);
}
static int client_node_demarshal_add_mem(void *object, const struct pw_protocol_native_message *msg)
{
struct pw_proxy *proxy = object;
struct spa_pod_parser prs;
uint32_t mem_id, type, memfd_idx, flags;
int memfd;
spa_pod_parser_init(&prs, msg->data, msg->size);
if (spa_pod_parser_get_struct(&prs,
SPA_POD_Int(&mem_id),
SPA_POD_Id(&type),
SPA_POD_Int(&memfd_idx),
SPA_POD_Int(&flags)) < 0)
return -EINVAL;
memfd = pw_protocol_native_get_proxy_fd(proxy, memfd_idx);
pw_proxy_notify(proxy, struct pw_client_node_proxy_events, add_mem, 0,
mem_id,
type,
memfd, flags);
return 0;
}
static int client_node_demarshal_transport(void *object, const struct pw_protocol_native_message *msg)
{
struct pw_proxy *proxy = object;
@ -564,27 +540,6 @@ static int client_node_demarshal_set_io(void *object, const struct pw_protocol_n
return 0;
}
static int
client_node_marshal_add_mem(void *object,
uint32_t mem_id,
uint32_t type,
int memfd, uint32_t flags)
{
struct pw_protocol_native_message *msg;
struct pw_resource *resource = object;
struct spa_pod_builder *b;
b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_PROXY_EVENT_ADD_MEM, &msg);
spa_pod_builder_add_struct(b,
SPA_POD_Int(mem_id),
SPA_POD_Id(type),
SPA_POD_Int(pw_protocol_native_add_resource_fd(resource, memfd)),
SPA_POD_Int(flags));
return pw_protocol_native_end_resource(resource, b);
}
static int client_node_marshal_transport(void *object, uint32_t node_id, int readfd, int writefd)
{
struct pw_protocol_native_message *msg;
@ -1053,7 +1008,6 @@ pw_protocol_native_client_node_method_demarshal[PW_CLIENT_NODE_PROXY_METHOD_NUM]
static const struct pw_client_node_proxy_events pw_protocol_native_client_node_event_marshal = {
PW_VERSION_CLIENT_NODE_PROXY_EVENTS,
.add_mem = &client_node_marshal_add_mem,
.transport = &client_node_marshal_transport,
.set_param = &client_node_marshal_set_param,
.set_io = &client_node_marshal_set_io,
@ -1070,7 +1024,6 @@ static const struct pw_client_node_proxy_events pw_protocol_native_client_node_e
static const struct pw_protocol_native_demarshal
pw_protocol_native_client_node_event_demarshal[PW_CLIENT_NODE_PROXY_EVENT_NUM] =
{
[PW_CLIENT_NODE_PROXY_EVENT_ADD_MEM] = { &client_node_demarshal_add_mem, 0 },
[PW_CLIENT_NODE_PROXY_EVENT_TRANSPORT] = { &client_node_demarshal_transport, 0 },
[PW_CLIENT_NODE_PROXY_EVENT_SET_PARAM] = { &client_node_demarshal_set_param, 0 },
[PW_CLIENT_NODE_PROXY_EVENT_SET_IO] = { &client_node_demarshal_set_io, 0 },

356
src/modules/module-client-node/remote-node.c
View file

@ -41,35 +41,22 @@
#include "extensions/client-node.h"
#define MAX_MIX 4096
#define MAX_IO 32
/** \cond */
struct mapping {
void *ptr;
struct pw_map_range map;
int prot;
};
struct mem {
uint32_t id;
int fd;
uint32_t flags;
uint32_t ref;
struct mapping map;
};
struct buffer_mem {
uint32_t mem_id;
struct mapping map;
};
struct buffer {
uint32_t id;
struct spa_buffer *buf;
struct buffer_mem *mem;
struct pw_memmap **mem;
uint32_t n_mem;
};
struct io {
uint32_t id;
struct pw_memmap *mem;
};
struct mix {
struct spa_list link;
struct pw_port *port;
@ -77,11 +64,12 @@ struct mix {
struct pw_port_mix mix;
struct pw_array buffers;
bool active;
struct io ios[MAX_IO];
};
struct link {
uint32_t node_id;
uint32_t mem_id;
struct pw_memmap *map;
struct pw_node_target target;
int signalfd;
};
@ -97,8 +85,6 @@ struct node_data {
struct spa_list mix[2];
struct spa_list free_mix;
struct pw_array mems;
struct pw_node *node;
struct spa_hook node_listener;
int do_free:1;
@ -109,6 +95,7 @@ struct node_data {
struct spa_hook proxy_listener;
struct pw_proxy *proxy;
struct io ios[MAX_IO];
struct spa_io_position *position;
struct pw_array links;
@ -116,6 +103,50 @@ struct node_data {
/** \endcond */
static void init_ios(struct io *ios)
{
int i;
for (i = 0; i < MAX_IO; i++)
ios[i].id = SPA_ID_INVALID;
}
static void clear_io(struct io *io)
{
pw_log_debug("%p clear id:%u mem:%p", io, io->id, io->mem);
pw_memmap_free(io->mem);
io->mem = NULL;
io->id = SPA_ID_INVALID;
}
static struct io *update_io(struct node_data *data, struct io *ios,
uint32_t id, struct pw_memmap *mem)
{
int i;
struct io *io, *f = NULL;
pw_log_debug("node %p: update id:%u mem:%p", data, id, mem);
for (i = 0; i < MAX_IO; i++) {
io = &ios[i];
if (io->id == SPA_ID_INVALID && f == NULL)
f = io;
else if (io->id == id) {
if (io->mem && io->mem != mem)
clear_io(io);
f = io;
break;
}
}
if (f == NULL)
return NULL;
io = f;
io->id = id;
io->mem = mem;
return io;
}
static struct link *find_activation(struct pw_array *links, uint32_t node_id)
{
struct link *l;
@ -127,112 +158,26 @@ static struct link *find_activation(struct pw_array *links, uint32_t node_id)
return NULL;
}
static struct mem *find_mem(struct node_data *data, uint32_t id)
{
struct mem *m;
pw_array_for_each(m, &data->mems) {
if (m->id == id)
return m;
}
return NULL;
}
static struct mem *find_mem_ptr(struct node_data *data, void *ptr)
{
struct mem *m;
pw_array_for_each(m, &data->mems) {
if (ptr >= m->map.ptr && ptr < SPA_MEMBER(m->map.ptr, m->map.map.size, void))
return m;
}
return NULL;
}
static void *mem_map(struct node_data *data, struct mapping *map,
int fd, int prot, uint32_t offset, uint32_t size)
{
struct mapping m;
void *ptr;
pw_map_range_init(&m.map, offset, size, data->core->sc_pagesize);
if (map->ptr == NULL || map->map.offset != m.map.offset || map->map.size != m.map.size) {
m.ptr = mmap(map->ptr, m.map.size, prot, MAP_SHARED, fd, m.map.offset);
if (m.ptr == MAP_FAILED) {
pw_log_error("remote %p: Failed to mmap memory %d size:%d: %m", data, fd, size);
return NULL;
}
map->map = m.map;
map->ptr = m.ptr;
pw_log_debug("remote %p: fd %d map %d %d %p", data, fd, m.map.offset, m.map.size, m.ptr);
}
ptr = SPA_MEMBER(map->ptr, m.map.start, void);
pw_log_debug("remote %p: fd %d ptr %p (%d %d)", data, fd, ptr, offset, size);
return ptr;
}
static void *mem_unmap(struct node_data *data, struct mapping *map)
{
if (map->ptr != NULL) {
if (munmap(map->ptr, map->map.size) < 0)
pw_log_warn("failed to unmap: %m");
}
return NULL;
}
static void clear_mem(struct node_data *data, struct mem *m)
{
if (m->fd != -1) {
bool has_ref = false;
int fd;
struct mem *m2;
pw_log_debug("remote %p: clear mem %p %d %d", data, m, m->id, m->fd);
fd = m->fd;
m->fd = -1;
m->id = SPA_ID_INVALID;
pw_array_for_each(m2, &data->mems) {
if (m2->fd == fd) {
has_ref = true;
break;
}
}
if (!has_ref) {
m->map.ptr = mem_unmap(data, &m->map);
close(fd);
}
}
}
static void clear_link(struct node_data *data, struct link *link)
{
struct mem *m;
link->node_id = SPA_ID_INVALID;
link->target.activation = NULL;
m = find_mem(data, link->mem_id);
if (m && --m->ref == 0)
clear_mem(data, m);
pw_memmap_free(link->map);
close(link->signalfd);
spa_list_remove(&link->target.link);
}
static void clean_transport(struct node_data *data)
{
struct mem *m;
struct link *l;
if (!data->have_transport)
return;
pw_array_for_each(m, &data->mems)
clear_mem(data, m);
pw_array_clear(&data->mems);
pw_array_for_each(l, &data->links)
clear_link(data, l);
pw_array_for_each(l, &data->links) {
if (l->node_id != SPA_ID_INVALID)
clear_link(data, l);
}
pw_array_clear(&data->links);
close(data->rtwritefd);
@ -248,6 +193,7 @@ static void mix_init(struct mix *mix, struct pw_port *port, uint32_t mix_id)
mix->active = false;
pw_array_init(&mix->buffers, 32);
pw_array_ensure_size(&mix->buffers, sizeof(struct buffer) * 64);
init_ios(mix->ios);
}
static int
@ -331,35 +277,6 @@ static struct mix *ensure_mix(struct node_data *data,
return mix;
}
static int client_node_add_mem(void *object,
uint32_t mem_id,
uint32_t type, int memfd, uint32_t flags)
{
struct pw_proxy *proxy = object;
struct node_data *data = proxy->user_data;
struct mem *m;
m = find_mem(data, mem_id);
if (m) {
pw_log_warn("duplicate mem %u, fd %d, flags %d",
mem_id, memfd, flags);
return -EINVAL;
}
m = pw_array_add(&data->mems, sizeof(struct mem));
if (m == NULL)
return -errno;
pw_log_debug("add mem %u, fd %d, flags %d", mem_id, memfd, flags);
m->id = mem_id;
m->fd = memfd;
m->flags = flags;
m->ref = 0;
m->map.map = PW_MAP_RANGE_INIT;
m->map.ptr = NULL;
return 0;
}
static int client_node_transport(void *object, uint32_t node_id,
int readfd, int writefd)
@ -523,39 +440,36 @@ client_node_set_io(void *object,
{
struct pw_proxy *proxy = object;
struct node_data *data = proxy->user_data;
struct mem *m;
struct pw_memmap *mm;
void *ptr;
if (memid == SPA_ID_INVALID) {
ptr = NULL;
size = 0;
mm = ptr = NULL;
}
else {
m = find_mem(data, memid);
if (m == NULL) {
pw_log_warn("unknown memory id %u", memid);
return -EINVAL;
}
ptr = mem_map(data, &m->map, m->fd,
PROT_READ|PROT_WRITE, offset, size);
if (ptr == NULL) {
pw_proxy_error(proxy, -errno, "set_io: mmap failed: %m");
mm = pw_mempool_map_id(proxy->remote->pool, memid,
PROT_READ|PROT_WRITE, offset, size);
if (mm == NULL) {
pw_log_warn("can't map memory id %u: %m", memid);
return -errno;
}
m->ref++;
ptr = mm->ptr;
}
pw_log_debug("node %p: set io %s %p", proxy,
spa_debug_type_find_name(spa_type_io, id), ptr);
if (id == SPA_IO_Position) {
if (ptr == NULL && data->position) {
m = find_mem_ptr(data, data->position);
if (m && --m->ref == 0)
clear_mem(data, m);
}
update_io(data, data->ios, id, mm);
switch (id) {
case SPA_IO_Position:
data->position = ptr;
break;
default:
break;
}
return spa_node_set_io(data->node->node, id, ptr, size);
}
@ -631,15 +545,9 @@ static int clear_buffers(struct node_data *data, struct mix *mix)
pw_array_for_each(b, &mix->buffers) {
for (i = 0; i < b->n_mem; i++) {
struct buffer_mem *bm = &b->mem[i];
struct mem *m;
pw_log_debug("port %p: clear buffer %d mem %d",
port, b->id, bm->mem_id);
m = find_mem(data, bm->mem_id);
if (m && --m->ref == 0)
clear_mem(data, m);
pw_log_debug("port %p: clear buffer %d map %p",
port, b->id, b->mem[i]);
pw_memmap_free(b->mem[i]);
}
b->n_mem = 0;
free(b->buf);
@ -714,14 +622,14 @@ client_node_port_use_buffers(void *object,
bufs = alloca(n_buffers * sizeof(struct spa_buffer *));
for (i = 0; i < n_buffers; i++) {
struct buffer_mem bmem = { 0, };
size_t size;
off_t offset;
struct mem *m;
struct pw_memmap *mm;
m = find_mem(data, buffers[i].mem_id);
if (m == NULL) {
res = -ENODEV;
mm = pw_mempool_map_id(proxy->remote->pool, buffers[i].mem_id,
prot, buffers[i].offset, buffers[i].size);
if (mm == NULL) {
res = -errno;
goto error_exit_cleanup;
}
@ -732,24 +640,17 @@ client_node_port_use_buffers(void *object,
}
bid->id = i;
bmem.mem_id = m->id;
bmem.map.ptr = mem_map(data, &bmem.map, m->fd, prot,
buffers[i].offset, buffers[i].size);
if (bmem.map.ptr == NULL) {
res = -errno;
goto error_exit_cleanup;
}
if (mlock(bmem.map.ptr, bmem.map.map.size) < 0)
pw_log_warn("Failed to mlock memory %u %u: %m",
bmem.map.map.offset, bmem.map.map.size);
if (mlock(mm->ptr, mm->size) < 0)
pw_log_warn("Failed to mlock memory %p %u: %m",
mm->ptr, mm->size);
size = sizeof(struct spa_buffer);
size += sizeof(struct buffer_mem);
size += sizeof(struct pw_memmap *);
for (j = 0; j < buffers[i].buffer->n_metas; j++)
size += sizeof(struct spa_meta);
for (j = 0; j < buffers[i].buffer->n_datas; j++) {
size += sizeof(struct spa_data);
size += sizeof(struct buffer_mem);
size += sizeof(struct pw_memmap *);
}
b = bid->buf = malloc(size);
@ -763,20 +664,18 @@ client_node_port_use_buffers(void *object,
b->datas = SPA_MEMBER(b->metas, sizeof(struct spa_meta) * b->n_metas,
struct spa_data);
bid->mem = SPA_MEMBER(b->datas, sizeof(struct spa_data) * b->n_datas,
struct buffer_mem);
struct pw_memmap *);
bid->n_mem = 0;
bid->mem[bid->n_mem++] = mm;
bid->mem[bid->n_mem++] = bmem;
m->ref++;
pw_log_debug("add buffer %d %d %u %u", m->id,
bid->id, bmem.map.map.offset, bmem.map.map.size);
pw_log_debug("add buffer %d %d %u %u", mm->block->id,
bid->id, buffers[i].offset, buffers[i].size);
offset = 0;
for (j = 0; j < b->n_metas; j++) {
struct spa_meta *m = &b->metas[j];
memcpy(m, &buffers[i].buffer->metas[j], sizeof(struct spa_meta));
m->data = SPA_MEMBER(bmem.map.ptr, offset, void);
m->data = SPA_MEMBER(mm->ptr, offset, void);
offset += SPA_ROUND_UP_N(m->size, 8);
}
@ -785,14 +684,14 @@ client_node_port_use_buffers(void *object,
memcpy(d, &buffers[i].buffer->datas[j], sizeof(struct spa_data));
d->chunk =
SPA_MEMBER(bmem.map.ptr, offset + sizeof(struct spa_chunk) * j,
SPA_MEMBER(mm->ptr, offset + sizeof(struct spa_chunk) * j,
struct spa_chunk);
if (d->type == SPA_DATA_MemFd || d->type == SPA_DATA_DmaBuf) {
if (d->type == SPA_DATA_MemId) {
uint32_t mem_id = SPA_PTR_TO_UINT32(d->data);
struct mem *bm = find_mem(data, mem_id);
struct buffer_mem bm2;
struct pw_memblock *bm;
bm = pw_mempool_find_id(proxy->remote->pool, mem_id);
if (bm == NULL) {
pw_log_error("unknown buffer mem %u", mem_id);
res = -ENODEV;
@ -800,18 +699,14 @@ client_node_port_use_buffers(void *object,
}
d->fd = bm->fd;
bm->ref++;
bm2.mem_id = bm->id;
bm2.map.ptr = NULL;
d->data = bm2.map.ptr;
bid->mem[bid->n_mem++] = bm2;
d->type = bm->type;
d->data = NULL;
pw_log_debug(" data %d %u -> fd %d maxsize %d",
j, bm->id, bm->fd, d->maxsize);
} else if (d->type == SPA_DATA_MemPtr) {
int offs = SPA_PTR_TO_INT(d->data);
d->data = SPA_MEMBER(bmem.map.ptr, offs, void);
d->data = SPA_MEMBER(mm->ptr, offs, void);
d->fd = -1;
pw_log_debug(" data %d %u -> mem %p maxsize %d",
j, bid->id, d->data, d->maxsize);
@ -848,7 +743,7 @@ client_node_port_set_io(void *object,
struct pw_proxy *proxy = object;
struct node_data *data = proxy->user_data;
struct mix *mix;
struct mem *m;
struct pw_memmap *mm;
void *ptr;
int res = 0;
@ -859,39 +754,32 @@ client_node_port_set_io(void *object,
}
if (memid == SPA_ID_INVALID) {
ptr = NULL;
mm = ptr = NULL;
size = 0;
}
else {
m = find_mem(data, memid);
if (m == NULL) {
res = -ENODEV;
goto error_exit;
}
ptr = mem_map(data, &m->map, m->fd,
PROT_READ|PROT_WRITE, offset, size);
if (ptr == NULL) {
mm = pw_mempool_map_id(proxy->remote->pool, memid,
PROT_READ|PROT_WRITE, offset, size);
if (mm == NULL) {
res = -errno;
goto error_exit;
}
m->ref++;
ptr = mm->ptr;
}
pw_log_debug("port %p: set io %s %p %p", mix->port,
pw_log_debug("port %p: set io:%s new:%p old:%p", mix->port,
spa_debug_type_find_name(spa_type_io, id), ptr, mix->mix.io);
update_io(data, mix->ios, id, mm);
if (id == SPA_IO_Buffers) {
if (ptr == NULL && mix->mix.io) {
if (ptr == NULL && mix->mix.io)
deactivate_mix(data, mix);
m = find_mem_ptr(data, mix->mix.io);
if (m && --m->ref == 0)
clear_mem(data, m);
}
mix->mix.io = ptr;
if (ptr)
activate_mix(data, mix);
}
if ((res = spa_node_port_set_io(mix->port->mix,
direction, mix_id,
id, ptr, size)) < 0) {
@ -936,28 +824,24 @@ client_node_set_activation(void *object,
struct pw_proxy *proxy = object;
struct node_data *data = proxy->user_data;
struct pw_node *node = data->node;
struct mem *m;
struct pw_memmap *mm;
struct pw_node_activation *ptr;
struct link *link;
int res = 0;
if (memid == SPA_ID_INVALID) {
ptr = NULL;
mm = NULL;
size = 0;
}
else {
m = find_mem(data, memid);
if (m == NULL) {
res = -ENODEV;
goto error_exit;
}
ptr = mem_map(data, &m->map, m->fd,
PROT_READ|PROT_WRITE, offset, size);
if (ptr == NULL) {
mm = pw_mempool_map_id(proxy->remote->pool, memid,
PROT_READ|PROT_WRITE, offset, size);
if (mm == NULL) {
res = -errno;
goto error_exit;
}
m->ref++;
ptr = mm->ptr;
}
pw_log_debug("node %p: set activation %d %p %u %u", node, node_id, ptr, offset, size);
@ -975,7 +859,7 @@ client_node_set_activation(void *object,
goto error_exit;
}
link->node_id = node_id;
link->mem_id = memid;
link->map = mm;
link->target.activation = ptr;
link->signalfd = signalfd;
link->target.signal = link_signal_func;
@ -1007,7 +891,6 @@ error_exit:
static const struct pw_client_node_proxy_events client_node_events = {
PW_VERSION_CLIENT_NODE_PROXY_EVENTS,
.add_mem = client_node_add_mem,
.transport = client_node_transport,
.set_param = client_node_set_param,
.set_io = client_node_set_io,
@ -1226,6 +1109,7 @@ static struct pw_proxy *node_export(struct pw_remote *remote, void *object, bool
data->core = pw_node_get_core(node);
data->node_proxy = (struct pw_client_node_proxy *)proxy;
data->remote_id = SPA_ID_INVALID;
init_ios(data->ios);
node->exported = true;
@ -1235,8 +1119,6 @@ static struct pw_proxy *node_export(struct pw_remote *remote, void *object, bool
for (i = 0; i < MAX_MIX; i++)
spa_list_append(&data->free_mix, &data->mix_pool[i].link);
pw_array_init(&data->mems, 64);
pw_array_ensure_size(&data->mems, sizeof(struct mem) * 64);
pw_array_init(&data->links, 64);
pw_array_ensure_size(&data->links, sizeof(struct link) * 64);

67
src/modules/module-protocol-native/protocol-native.c
View file

@ -335,6 +335,40 @@ static int core_event_demarshal_remove_id(void *object, const struct pw_protocol
return pw_proxy_notify(proxy, struct pw_core_proxy_events, remove_id, 0, id);
}
static int core_event_demarshal_add_mem(void *object, const struct pw_protocol_native_message *msg)
{
struct pw_proxy *proxy = object;
struct spa_pod_parser prs;
uint32_t id, type, idx, flags;
int fd;
spa_pod_parser_init(&prs, msg->data, msg->size);
if (spa_pod_parser_get_struct(&prs,
SPA_POD_Int(&id),
SPA_POD_Id(&type),
SPA_POD_Int(&idx),
SPA_POD_Int(&flags)) < 0)
return -EINVAL;
fd = pw_protocol_native_get_proxy_fd(proxy, idx);
return pw_proxy_notify(proxy, struct pw_core_proxy_events, add_mem, 0, id, type, fd, flags);
}
static int core_event_demarshal_remove_mem(void *object, const struct pw_protocol_native_message *msg)
{
struct pw_proxy *proxy = object;
struct spa_pod_parser prs;
uint32_t id;
spa_pod_parser_init(&prs, msg->data, msg->size);
if (spa_pod_parser_get_struct(&prs,
SPA_POD_Int(&id)) < 0)
return -EINVAL;
return pw_proxy_notify(proxy, struct pw_core_proxy_events, remove_mem, 0, id);
}
static void core_event_marshal_info(void *object, const struct pw_core_info *info)
{
struct pw_resource *resource = object;
@ -418,6 +452,35 @@ static void core_event_marshal_remove_id(void *object, uint32_t id)
pw_protocol_native_end_resource(resource, b);
}
static void core_event_marshal_add_mem(void *object, uint32_t id, uint32_t type, int fd, uint32_t flags)
{
struct pw_resource *resource = object;
struct spa_pod_builder *b;
b = pw_protocol_native_begin_resource(resource, PW_CORE_PROXY_EVENT_ADD_MEM, NULL);
spa_pod_builder_add_struct(b,
SPA_POD_Int(id),
SPA_POD_Id(type),
SPA_POD_Int(pw_protocol_native_add_resource_fd(resource, fd)),
SPA_POD_Int(flags));
pw_protocol_native_end_resource(resource, b);
}
static void core_event_marshal_remove_mem(void *object, uint32_t id)
{
struct pw_resource *resource = object;
struct spa_pod_builder *b;
b = pw_protocol_native_begin_resource(resource, PW_CORE_PROXY_EVENT_REMOVE_MEM, NULL);
spa_pod_builder_add_struct(b,
SPA_POD_Int(id));
pw_protocol_native_end_resource(resource, b);
}
static int core_method_demarshal_hello(void *object, const struct pw_protocol_native_message *msg)
{
struct pw_resource *resource = object;
@ -1835,6 +1898,8 @@ static const struct pw_core_proxy_events pw_protocol_native_core_event_marshal =
.ping = &core_event_marshal_ping,
.error = &core_event_marshal_error,
.remove_id = &core_event_marshal_remove_id,
.add_mem = &core_event_marshal_add_mem,
.remove_mem = &core_event_marshal_remove_mem,
};
static const struct pw_protocol_native_demarshal
@ -1845,6 +1910,8 @@ pw_protocol_native_core_event_demarshal[PW_CORE_PROXY_EVENT_NUM] =
[PW_CORE_PROXY_EVENT_PING] = { &core_event_demarshal_ping, 0, },
[PW_CORE_PROXY_EVENT_ERROR] = { &core_event_demarshal_error, 0, },
[PW_CORE_PROXY_EVENT_REMOVE_ID] = { &core_event_demarshal_remove_id, 0, },
[PW_CORE_PROXY_EVENT_ADD_MEM] = { &core_event_demarshal_add_mem, 0, },
[PW_CORE_PROXY_EVENT_REMOVE_MEM] = { &core_event_demarshal_remove_mem, 0, },
};
static const struct pw_protocol_marshal pw_protocol_native_core_marshal = {

33
src/pipewire/client.c
View file

@ -36,6 +36,7 @@ struct impl {
struct pw_client this;
struct spa_hook core_listener;
struct pw_array permissions;
struct spa_hook pool_listener;
};
#define pw_client_resource(r,m,v,...) pw_resource_call(r,struct pw_client_proxy_events,m,v,__VA_ARGS__)
@ -223,6 +224,29 @@ error_resource:
return -errno;
}
static void pool_added(void *data, struct pw_memblock *block)
{
struct impl *impl = data;
struct pw_client *client = &impl->this;
if (client->core_resource)
pw_core_resource_add_mem(client->core_resource,
block->id, block->type, block->fd, block->flags);
}
static void pool_removed(void *data, struct pw_memblock *block)
{
struct impl *impl = data;
struct pw_client *client = &impl->this;
if (client->core_resource)
pw_core_resource_remove_mem(client->core_resource, block->id);
}
static const struct pw_mempool_events pool_events = {
PW_VERSION_MEMPOOL_EVENTS,
.added = pool_added,
.removed = pool_removed,
};
static void
core_global_removed(void *data, struct pw_global *global)
{
@ -284,10 +308,16 @@ struct pw_client *pw_client_new(struct pw_core *core,
res = -errno;
goto error_clear_array;
}
p->id = SPA_ID_INVALID;
p->permissions = 0;
this->pool = pw_mempool_new(NULL);
if (this->pool == NULL) {
res = -errno;
goto error_clear_array;
}
pw_mempool_add_listener(this->pool, &impl->pool_listener, &pool_events, impl);
this->properties = properties;
this->permission_func = client_permission_func;
this->permission_data = impl;
@ -443,6 +473,7 @@ void pw_client_destroy(struct pw_client *client)
pw_map_clear(&client->objects);
pw_array_clear(&impl->permissions);
pw_mempool_destroy(client->pool);
pw_properties_free(client->properties);

15
src/pipewire/control.c
View file

@ -119,7 +119,7 @@ void pw_control_destroy(struct pw_control *control)
if (control->direction == SPA_DIRECTION_OUTPUT) {
if (impl->mem)
pw_memblock_free(impl->mem);
pw_memblock_unref(impl->mem);
}
free(control);
}
@ -183,11 +183,10 @@ int pw_control_add_link(struct pw_control *control, uint32_t cmix,
size = SPA_MAX(control->size, other->size);
if (impl->mem == NULL) {
if ((res = pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
PW_MEMBLOCK_FLAG_SEAL |
PW_MEMBLOCK_FLAG_MAP_READWRITE,
size,
&impl->mem)) < 0)
if ((res = pw_mempool_alloc(control->core->pool,
PW_MEMBLOCK_FLAG_SEAL |
PW_MEMBLOCK_FLAG_MAP,
size, &impl->mem)) < 0)
goto exit;
}
@ -195,7 +194,7 @@ int pw_control_add_link(struct pw_control *control, uint32_t cmix,
if (control->port) {
if ((res = port_set_io(control->port, cmix,
control->id,
impl->mem->ptr, size)) < 0) {
impl->mem->map->ptr, size)) < 0) {
pw_log_warn("control %p: set io failed %d %s", control,
res, spa_strerror(res));
goto exit;
@ -205,7 +204,7 @@ int pw_control_add_link(struct pw_control *control, uint32_t cmix,
if (other->port) {
if ((res = port_set_io(other->port, omix,
other->id, impl->mem->ptr, size)) < 0) {
other->id, impl->mem->map->ptr, size)) < 0) {
pw_log_warn("control %p: set io failed %d %s", control,
res, spa_strerror(res));
goto exit;

4
src/pipewire/core.c
View file

@ -504,6 +504,8 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop,
goto error_free;
}
this->pool = pw_mempool_new(NULL);
this->data_loop = pw_data_loop_get_loop(this->data_loop_impl);
this->data_system = this->data_loop->system;
this->main_loop = main_loop;
@ -650,6 +652,8 @@ void pw_core_destroy(struct pw_core *core)
pw_log_debug("core %p: free", core);
pw_core_emit_free(core);
pw_mempool_destroy(core->pool);
pw_data_loop_destroy(core->data_loop_impl);
pw_properties_free(core->properties);

27
src/pipewire/interfaces.h
View file

@ -83,7 +83,9 @@ struct pw_link_proxy { struct spa_interface iface; };
#define PW_CORE_PROXY_EVENT_PING 2
#define PW_CORE_PROXY_EVENT_ERROR 3
#define PW_CORE_PROXY_EVENT_REMOVE_ID 4
#define PW_CORE_PROXY_EVENT_NUM 5
#define PW_CORE_PROXY_EVENT_ADD_MEM 5
#define PW_CORE_PROXY_EVENT_REMOVE_MEM 6
#define PW_CORE_PROXY_EVENT_NUM 7
/** \struct pw_core_proxy_events
* \brief Core events
@ -149,6 +151,29 @@ struct pw_core_proxy_events {
* \param id deleted object ID
*/
void (*remove_id) (void *object, uint32_t id);
/**
* Add memory for a client
*
* Memory is given to a client as \a fd of a certain
* memory \a type.
*
* Further references to this fd will be made with the per memory
* unique identifier \a id.
*
* \param id the unique id of the memory
* \param type the memory type, one of enum spa_data_type
* \param fd the file descriptor
* \param flags extra flags
*/
void (*add_mem) (void *object, uint32_t id, uint32_t type, int fd, uint32_t flags);
/**
* Remove memory for a client
*
* \param id the memory id to remove
*/
void (*remove_mem) (void *object, uint32_t id);
};

37
src/pipewire/link.c
View file

@ -445,14 +445,14 @@ static int alloc_buffers(struct pw_link *this,
/* pointer to buffer structures */
bp = SPA_MEMBER(buffers, n_buffers * sizeof(struct spa_buffer *), struct spa_buffer);
if ((res = pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
PW_MEMBLOCK_FLAG_MAP_READWRITE |
PW_MEMBLOCK_FLAG_SEAL, n_buffers * info.mem_size,
&m)) < 0)
if ((res = pw_mempool_alloc(this->core->pool,
PW_MEMBLOCK_FLAG_SEAL |
PW_MEMBLOCK_FLAG_MAP,
n_buffers * info.mem_size, &m)) < 0)
return res;
pw_log_debug("layout buffers %p data %p", bp, m->ptr);
spa_buffer_alloc_layout_array(&info, n_buffers, buffers, bp, m->ptr);
pw_log_debug("layout buffers %p data %p", bp, m->map->ptr);
spa_buffer_alloc_layout_array(&info, n_buffers, buffers, bp, m->map->ptr);
allocation->mem = m;
allocation->n_buffers = n_buffers;
@ -748,18 +748,9 @@ do_activate_link(struct spa_loop *loop,
{
struct pw_link *this = user_data;
struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
int res;
pw_log_trace("link %p: activate", this);
if ((res = port_set_io(this, this->input, SPA_IO_Buffers, this->io,
sizeof(struct spa_io_buffers), &this->rt.in_mix)) < 0)
return res;
if ((res = port_set_io(this, this->output, SPA_IO_Buffers, this->io,
sizeof(struct spa_io_buffers), &this->rt.out_mix)) < 0)
return res;
spa_list_append(&this->output->rt.mix_list, &this->rt.out_mix.rt_link);
spa_list_append(&this->input->rt.mix_list, &this->rt.in_mix.rt_link);
@ -774,6 +765,7 @@ do_activate_link(struct spa_loop *loop,
int pw_link_activate(struct pw_link *this)
{
struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
int res;
pw_log_debug("link %p: activate %d %d", this, impl->activated, this->info.state);
@ -782,6 +774,14 @@ int pw_link_activate(struct pw_link *this)
pw_link_prepare(this);
if ((res = port_set_io(this, this->input, SPA_IO_Buffers, this->io,
sizeof(struct spa_io_buffers), &this->rt.in_mix)) < 0)
return res;
if ((res = port_set_io(this, this->output, SPA_IO_Buffers, this->io,
sizeof(struct spa_io_buffers), &this->rt.out_mix)) < 0)
return res;
if (this->info.state == PW_LINK_STATE_PAUSED) {
pw_loop_invoke(this->output->node->data_loop,
do_activate_link, SPA_ID_INVALID, NULL, 0, false, this);
@ -967,9 +967,6 @@ do_deactivate_link(struct spa_loop *loop,
pw_log_trace("link %p: disable %p and %p", this, &this->rt.in_mix, &this->rt.out_mix);
port_set_io(this, this->input, SPA_IO_Buffers, NULL, 0, &this->rt.in_mix);
port_set_io(this, this->output, SPA_IO_Buffers, NULL, 0, &this->rt.out_mix);
spa_list_remove(&this->rt.out_mix.rt_link);
spa_list_remove(&this->rt.in_mix.rt_link);
@ -995,6 +992,10 @@ int pw_link_deactivate(struct pw_link *this)
if (impl->activated) {
pw_loop_invoke(this->output->node->data_loop,
do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, this);
port_set_io(this, this->input, SPA_IO_Buffers, NULL, 0, &this->rt.in_mix);
port_set_io(this, this->output, SPA_IO_Buffers, NULL, 0, &this->rt.out_mix);
impl->activated = false;
}

538
src/pipewire/mem.c
View file

@ -35,10 +35,14 @@
#include <sys/syscall.h>
#include <spa/utils/list.h>
#include <spa/buffer/buffer.h>
#include <pipewire/log.h>
#include <pipewire/map.h>
#include <pipewire/mem.h>
#define USE_MEMFD
#ifndef HAVE_MEMFD_CREATE
/*
* No glibc wrappers exist for memfd_create(2), so provide our own.
@ -80,22 +84,109 @@ static inline int memfd_create(const char *name, unsigned int flags)
#define F_SEAL_WRITE 0x0008 /* prevent writes */
#endif
static struct spa_list _mempools = SPA_LIST_INIT(&_mempools);
#define pw_mempool_emit(p,m,v,...) spa_hook_list_call(&p->listener_list, struct pw_mempool_events, m, v, ##__VA_ARGS__)
#define pw_mempool_emit_destroy(p) pw_mempool_emit(p, destroy, 0)
#define pw_mempool_emit_added(p,b) pw_mempool_emit(p, added, 0, b)
#define pw_mempool_emit_removed(p,b) pw_mempool_emit(p, removed, 0, b)
struct mempool {
struct pw_mempool this;
struct spa_list link;
struct spa_hook_list listener_list;
struct pw_map map;
struct spa_list blocks;
uint32_t pagesize;
};
struct memblock {
struct pw_memblock mem;
struct pw_memblock this;
struct spa_list link;
struct spa_list mappings;
struct spa_list maps;
};
struct mapping {
struct memblock *block;
int ref;
uint32_t offset;
uint32_t size;
struct spa_list link;
void *ptr;
};
struct memmap {
struct pw_memmap this;
struct mapping *mapping;
struct spa_list link;
};
static struct spa_list _memblocks = SPA_LIST_INIT(&_memblocks);
struct pw_mempool *pw_mempool_new(struct pw_properties *props)
{
struct mempool *impl;
struct pw_mempool *this;
#define USE_MEMFD
impl = calloc(1, sizeof(struct mempool));
if (impl == NULL)
return NULL;
this = &impl->this;
this->props = props;
impl->pagesize = sysconf(_SC_PAGESIZE);
pw_log_debug("mempool %p: new", this);
spa_hook_list_init(&impl->listener_list);
pw_map_init(&impl->map, 64, 64);
spa_list_init(&impl->blocks);
spa_list_append(&_mempools, &impl->link);
return this;
}
void pw_mempool_destroy(struct pw_mempool *pool)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
pw_log_debug("mempool %p: destroy", pool);
pw_mempool_emit_destroy(impl);
spa_list_remove(&impl->link);
spa_list_consume(b, &impl->blocks, link)
pw_memblock_free(&b->this);
if (pool->props)
pw_properties_free(pool->props);
free(impl);
}
void pw_mempool_add_listener(struct pw_mempool *pool,
struct spa_hook *listener,
const struct pw_mempool_events *events,
void *data)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
spa_hook_list_append(&impl->listener_list, listener, events, data);
}
#if 0
/** Map a memblock
* \param mem a memblock
* \return 0 on success, < 0 on error
* \memberof pw_memblock
*/
SPA_EXPORT
int pw_memblock_map(struct pw_memblock *mem)
int pw_memblock_map_old(struct pw_memblock *mem)
{
if (mem->ptr != NULL)
return 0;
@ -147,6 +238,152 @@ int pw_memblock_map(struct pw_memblock *mem)
return 0;
}
#endif
static struct mapping * memblock_find_mapping(struct memblock *b,
uint32_t flags, uint32_t offset, uint32_t size)
{
struct mapping *m;
spa_list_for_each(m, &b->mappings, link) {
if (m->offset <= offset && (m->offset + m->size) >= (offset + size))
return m;
}
return NULL;
}
static struct mapping * memblock_map(struct memblock *b,
enum pw_memmap_flags flags, uint32_t offset, uint32_t size)
{
struct mempool *p = SPA_CONTAINER_OF(b->this.pool, struct mempool, this);
struct mapping *m;
void *ptr;
int prot = 0;
if (flags & PW_MEMMAP_FLAG_READ)
prot |= PROT_READ;
if (flags & PW_MEMMAP_FLAG_WRITE)
prot |= PROT_WRITE;
if (flags & PW_MEMMAP_FLAG_TWICE) {
errno = -ENOTSUP;
return NULL;
}
ptr = mmap(NULL, size, prot, MAP_SHARED, b->this.fd, offset);
if (ptr == MAP_FAILED) {
pw_log_error("pool %p: Failed to mmap memory %d size:%d: %m",
p, b->this.fd, size);
return NULL;
}
m = calloc(1, sizeof(struct mapping));
if (m == NULL) {
munmap(ptr, size);
return NULL;
}
m->ptr = ptr;
m->block = b;
m->offset = offset;
m->size = size;
b->this.ref++;
spa_list_append(&b->mappings, &m->link);
pw_log_debug("pool %p: fd:%d map:%p ptr:%p (%d %d)", p,
b->this.fd, m, m->ptr, offset, size);
return m;
}
static void mapping_unmap(struct mapping *m)
{
struct memblock *b = m->block;
struct mempool *p = SPA_CONTAINER_OF(b->this.pool, struct mempool, this);
pw_log_debug("pool %p: map:%p fd:%d ptr:%p size:%d", p, m, b->this.fd, m->ptr, m->size);
munmap(m->ptr, m->size);
spa_list_remove(&m->link);
free(m);
pw_memblock_unref(&b->this);
}
SPA_EXPORT
struct pw_memmap * pw_memblock_map(struct pw_memblock *block,
enum pw_memmap_flags flags, uint32_t offset, uint32_t size)
{
struct memblock *b = SPA_CONTAINER_OF(block, struct memblock, this);
struct mempool *p = SPA_CONTAINER_OF(block->pool, struct mempool, this);
struct mapping *m;
struct memmap *mm;
struct pw_map_range range;
pw_map_range_init(&range, offset, size, p->pagesize);
m = memblock_find_mapping(b, flags, range.offset, range.size);
if (m == NULL)
m = memblock_map(b, flags, range.offset, range.size);
if (m == NULL)
return NULL;
mm = calloc(1, sizeof(struct memmap));
if (mm == NULL) {
if (m->ref == 0)
mapping_unmap(m);
return NULL;
}
m->ref++;
mm->mapping = m;
mm->this.block = block;
mm->this.flags = flags;
mm->this.offset = offset;
mm->this.size = size;
mm->this.ptr = SPA_MEMBER(m->ptr, range.start, void);
spa_list_append(&b->maps, &mm->link);
pw_log_debug("pool %p: map:%p fd:%d ptr:%p (%d %d) mapping:%p ref:%d", p,
&mm->this, b->this.fd, mm->this.ptr, offset, size, m, m->ref);
return &mm->this;
}
SPA_EXPORT
struct pw_memmap * pw_mempool_map_id(struct pw_mempool *pool,
uint32_t id, enum pw_memmap_flags flags, uint32_t offset, uint32_t size)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
b = pw_map_lookup(&impl->map, id);
if (b == NULL) {
errno = -ENOENT;
return NULL;
}
return pw_memblock_map(&b->this, flags, offset, size);
}
SPA_EXPORT
int pw_memmap_free(struct pw_memmap *map)
{
struct memmap *mm = SPA_CONTAINER_OF(map, struct memmap, this);
struct mapping *m = mm->mapping;
struct memblock *b = m->block;
struct mempool *p = SPA_CONTAINER_OF(b->this.pool, struct mempool, this);
pw_log_debug("pool %p: map:%p fd:%d ptr:%p map:%p ref:%d", p,
&mm->this, b->this.fd, mm->this.ptr, m, m->ref);
if (--m->ref == 0)
mapping_unmap(m);
spa_list_remove(&mm->link);
free(mm);
return 0;
}
/** Create a new memblock
* \param flags memblock flags
@ -156,107 +393,152 @@ int pw_memblock_map(struct pw_memblock *mem)
* \memberof pw_memblock
*/
SPA_EXPORT
int pw_memblock_alloc(enum pw_memblock_flags flags, size_t size, struct pw_memblock **mem)
int pw_mempool_alloc(struct pw_mempool *pool, enum pw_memblock_flags flags,
size_t size, struct pw_memblock **mem)
{
struct memblock tmp, *p;
struct pw_memblock *m;
bool use_fd;
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
int res;
if (mem == NULL)
return -EINVAL;
spa_return_val_if_fail(pool != NULL, -EINVAL);
spa_return_val_if_fail(mem != NULL, -EINVAL);
m = &tmp.mem;
m->offset = 0;
m->flags = flags;
m->size = size;
m->ptr = NULL;
use_fd = ! !(flags & (PW_MEMBLOCK_FLAG_MAP_TWICE | PW_MEMBLOCK_FLAG_WITH_FD));
if (use_fd) {
#ifdef USE_MEMFD
m->fd = memfd_create("pipewire-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
if (m->fd == -1) {
res = -errno;
pw_log_error("Failed to create memfd: %s\n", strerror(errno));
return res;
}
#else
char filename[] = "/dev/shm/pipewire-tmpfile.XXXXXX";
m->fd = mkostemp(filename, O_CLOEXEC);
if (m->fd == -1) {
res = -errno;
pw_log_error("Failed to create temporary file: %s\n", strerror(errno));
return res;
}
unlink(filename);
#endif
if (ftruncate(m->fd, size) < 0) {
res = -errno;
pw_log_warn("Failed to truncate temporary file: %s", strerror(errno));
close(m->fd);
return res;
}
#ifdef USE_MEMFD
if (flags & PW_MEMBLOCK_FLAG_SEAL) {
unsigned int seals = F_SEAL_GROW | F_SEAL_SHRINK | F_SEAL_SEAL;
if (fcntl(m->fd, F_ADD_SEALS, seals) == -1) {
pw_log_warn("Failed to add seals: %s", strerror(errno));
}
}
#endif
if ((res = pw_memblock_map(m)) < 0)
goto mmap_failed;
} else {
if (size > 0) {
m->ptr = malloc(size);
if (m->ptr == NULL)
return -errno;
}
m->fd = -1;
}
if (!(flags & PW_MEMBLOCK_FLAG_WITH_FD) && m->fd != -1) {
close(m->fd);
m->fd = -1;
}
p = calloc(1, sizeof(struct memblock));
if (p == NULL)
b = calloc(1, sizeof(struct memblock));
if (b == NULL)
return -errno;
*p = tmp;
spa_list_append(&_memblocks, &p->link);
*mem = &p->mem;
pw_log_debug("mem %p: alloc", *mem);
b->this.ref = 1;
b->this.pool = pool;
b->this.flags = flags;
spa_list_init(&b->mappings);
spa_list_init(&b->maps);
#ifdef USE_MEMFD
b->this.fd = memfd_create("pipewire-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
if (b->this.fd == -1) {
res = -errno;
pw_log_error("Failed to create memfd: %s\n", strerror(errno));
goto error_free;
}
#else
char filename[] = "/dev/shm/pipewire-tmpfile.XXXXXX";
b->this.fd = mkostemp(filename, O_CLOEXEC);
if (b->this.fd == -1) {
res = -errno;
pw_log_error("Failed to create temporary file: %s\n", strerror(errno));
goto error_free;
}
unlink(filename);
#endif
if (ftruncate(b->this.fd, size) < 0) {
res = -errno;
pw_log_warn("Failed to truncate temporary file: %s", strerror(errno));
goto error_close;
}
#ifdef USE_MEMFD
if (flags & PW_MEMBLOCK_FLAG_SEAL) {
unsigned int seals = F_SEAL_GROW | F_SEAL_SHRINK | F_SEAL_SEAL;
if (fcntl(b->this.fd, F_ADD_SEALS, seals) == -1) {
pw_log_warn("Failed to add seals: %s", strerror(errno));
}
}
#endif
b->this.type = SPA_DATA_MemFd;
if (flags & PW_MEMBLOCK_FLAG_MAP && size > 0) {
b->this.map = pw_memblock_map(&b->this, PROT_READ|PROT_WRITE, 0, size);
if (b->this.map == NULL) {
res = -errno;
goto error_close;
}
}
b->this.id = pw_map_insert_new(&impl->map, b);
spa_list_append(&impl->blocks, &b->link);
pw_log_debug("mem %p: alloc id:%d", &b->this, b->this.id);
pw_mempool_emit_added(impl, &b->this);
*mem = &b->this;
return 0;
mmap_failed:
close(m->fd);
error_close:
close(b->this.fd);
error_free:
free(b);
return res;
}
SPA_EXPORT
int
pw_memblock_import(enum pw_memblock_flags flags,
int fd, off_t offset, size_t size,
struct pw_memblock **mem)
static struct memblock * mempool_find_fd(struct pw_mempool *pool, int fd)
{
int res;
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
if ((res = pw_memblock_alloc(0, 0, mem)) < 0)
return res;
spa_list_for_each(b, &impl->blocks, link) {
if (fd == b->this.fd) {
pw_log_debug("pool %p: found %p id:%d for fd %d", pool, &b->this, b->this.id, fd);
return b;
}
}
return NULL;
}
(*mem)->flags = flags;
(*mem)->fd = fd;
(*mem)->offset = offset;
(*mem)->size = size;
SPA_EXPORT
struct pw_memblock * pw_mempool_import(struct pw_mempool *pool,
uint32_t type, int fd, uint32_t flags)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
pw_log_debug("mem %p: import", *mem);
b = mempool_find_fd(pool, fd);
if (b != NULL) {
b->this.ref++;
return &b->this;
}
return pw_memblock_map(*mem);
b = calloc(1, sizeof(struct memblock));
if (b == NULL)
return NULL;
spa_list_init(&b->maps);
spa_list_init(&b->mappings);
b->this.ref = 1;
b->this.pool = pool;
b->this.type = type;
b->this.fd = fd;
b->this.flags = flags;
b->this.id = pw_map_insert_new(&impl->map, b);
spa_list_append(&impl->blocks, &b->link);
pw_log_debug("pool %p: import %p id:%u fd:%d", pool, b, b->this.id, fd);
pw_mempool_emit_added(impl, &b->this);
return &b->this;
}
SPA_EXPORT
struct pw_memblock * pw_mempool_import_block(struct pw_mempool *pool,
struct pw_memblock *mem)
{
return pw_mempool_import(pool, mem->type, mem->fd,
mem->flags | PW_MEMBLOCK_FLAG_DONT_CLOSE);
}
int pw_mempool_remove_id(struct pw_mempool *pool, uint32_t id)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
b = pw_map_lookup(&impl->map, id);
if (b == NULL)
return -ENOENT;
pw_memblock_unref(&b->this);
return 0;
}
/** Free a memblock
@ -264,36 +546,76 @@ pw_memblock_import(enum pw_memblock_flags flags,
* \memberof pw_memblock
*/
SPA_EXPORT
void pw_memblock_free(struct pw_memblock *mem)
void pw_memblock_free(struct pw_memblock *block)
{
struct memblock *m = (struct memblock *)mem;
struct memblock *b = SPA_CONTAINER_OF(block, struct memblock, this);
struct pw_mempool *pool = block->pool;
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memmap *mm;
if (mem == NULL)
return;
spa_return_if_fail(block != NULL);
pw_log_debug("mem %p: free %p %d", mem, mem->ptr, mem->fd);
if (mem->flags & PW_MEMBLOCK_FLAG_WITH_FD) {
if (mem->ptr)
munmap(mem->ptr, mem->size);
if (mem->fd != -1)
close(mem->fd);
} else {
free(mem->ptr);
pw_log_debug("pool %p: free mem %p id:%d fd:%d ref:%d",
pool, block, block->id, block->fd, block->ref);
block->ref++;
pw_map_remove(&impl->map, block->id);
spa_list_remove(&b->link);
pw_mempool_emit_removed(impl, block);
spa_list_consume(mm, &b->maps, link)
pw_memmap_free(&mm->this);
if (block->fd != -1 && !(block->flags & PW_MEMBLOCK_FLAG_DONT_CLOSE)) {
pw_log_debug("pool %p: close fd:%d", pool, block->fd);
close(block->fd);
}
spa_list_remove(&m->link);
free(mem);
free(b);
}
SPA_EXPORT
struct pw_memblock * pw_memblock_find(const void *ptr)
struct pw_memblock * pw_mempool_find_ptr(struct pw_mempool *pool, const void *ptr)
{
struct memblock *m;
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
struct mapping *m;
spa_list_for_each(m, &_memblocks, link) {
if (ptr >= m->mem.ptr && ptr < SPA_MEMBER(m->mem.ptr, m->mem.size, void)) {
pw_log_debug("mem %p: found for %p", &m->mem, ptr);
return &m->mem;
spa_list_for_each(b, &impl->blocks, link) {
spa_list_for_each(m, &b->mappings, link) {
if (ptr >= m->ptr && ptr < SPA_MEMBER(m->ptr, m->size, void)) {
pw_log_debug("pool %p: found %p id:%d for %p", pool,
m->block, b->this.id, ptr);
return &b->this;
}
}
}
return NULL;
}
SPA_EXPORT
struct pw_memblock * pw_mempool_find_id(struct pw_mempool *pool, uint32_t id)
{
struct mempool *impl = SPA_CONTAINER_OF(pool, struct mempool, this);
struct memblock *b;
b = pw_map_lookup(&impl->map, id);
pw_log_debug("pool %p: found %p for %d", pool, b, id);
if (b == NULL)
return NULL;
return &b->this;
}
SPA_EXPORT
struct pw_memblock * pw_mempool_find_fd(struct pw_mempool *pool, int fd)
{
struct memblock *b;
b = mempool_find_fd(pool, fd);
if (b == NULL)
return NULL;
return &b->this;
}

122
src/pipewire/mem.h
View file

@ -25,51 +25,47 @@
#ifndef PIPEWIRE_MEM_H
#define PIPEWIRE_MEM_H
#include <spa/utils/defs.h>
#include <pipewire/properties.h>
#ifdef __cplusplus
extern "C" {
#endif
/** Flags passed to \ref pw_memblock_alloc() \memberof pw_memblock */
/** Flags passed to \ref pw_mempool_alloc() \memberof pw_memblock */
enum pw_memblock_flags {
PW_MEMBLOCK_FLAG_NONE = 0,
PW_MEMBLOCK_FLAG_WITH_FD = (1 << 0),
PW_MEMBLOCK_FLAG_SEAL = (1 << 1),
PW_MEMBLOCK_FLAG_MAP_READ = (1 << 2),
PW_MEMBLOCK_FLAG_MAP_WRITE = (1 << 3),
PW_MEMBLOCK_FLAG_MAP_TWICE = (1 << 4),
PW_MEMBLOCK_FLAG_SEAL = (1 << 0),
PW_MEMBLOCK_FLAG_MAP = (1 << 1),
PW_MEMBLOCK_FLAG_DONT_CLOSE = (1 << 2),
};
#define PW_MEMBLOCK_FLAG_MAP_READWRITE (PW_MEMBLOCK_FLAG_MAP_READ | PW_MEMBLOCK_FLAG_MAP_WRITE)
enum pw_memmap_flags {
PW_MEMMAP_FLAG_NONE = 0,
PW_MEMMAP_FLAG_READ = (1 << 0), /**< map in read mode */
PW_MEMMAP_FLAG_WRITE = (1 << 1), /**< map in write mode */
PW_MEMMAP_FLAG_TWICE = (1 << 2), /**< map the same area twice afer eachother,
* creating a circular ringbuffer */
};
struct pw_memchunk;
struct pw_mempool {
struct pw_properties *props;
};
/** \class pw_memblock
* Memory block structure */
struct pw_memblock {
enum pw_memblock_flags flags; /**< flags used when allocating */
int fd; /**< memfd if any */
off_t offset; /**< offset of mappable memory */
void *ptr; /**< ptr to mapped memory */
size_t size; /**< size of mapped memory */
struct pw_mempool *pool; /**< owner pool */
enum pw_memblock_flags flags;
int ref; /**< refcount */
uint32_t id; /**< unique id */
uint32_t type; /**< type of the fd */
uint32_t size; /**< size of memory */
int fd; /**< fd */
struct pw_memmap *map; /**< optional map when PW_MEMBLOCK_FLAG_MAP was given */
};
int
pw_memblock_alloc(enum pw_memblock_flags flags, size_t size, struct pw_memblock **mem);
int
pw_memblock_import(enum pw_memblock_flags flags,
int fd, off_t offset, size_t size,
struct pw_memblock **mem);
int
pw_memblock_map(struct pw_memblock *mem);
void
pw_memblock_free(struct pw_memblock *mem);
/** Find memblock for given \a ptr */
struct pw_memblock * pw_memblock_find(const void *ptr);
/** parameters to map a memory range */
struct pw_map_range {
uint32_t start; /** offset in first page with start of data */
@ -77,6 +73,72 @@ struct pw_map_range {
uint32_t size; /** size to map */
};
/** a mapped region of a pw_memblock */
struct pw_memmap {
struct pw_memblock *block; /**< owner memblock */
enum pw_memmap_flags flags; /**< flags used when mapping */
uint32_t offset; /**< offset in memblock */
uint32_t size; /**< size in memblock */
void *ptr; /**< mapped pointer */
};
struct pw_mempool_events {
#define PW_VERSION_MEMPOOL_EVENTS 0
uint32_t version;
void (*destroy) (void *data);
void (*added) (void *data, struct pw_memblock *block);
void (*removed) (void *data, struct pw_memblock *block);
};
struct pw_mempool *pw_mempool_new(struct pw_properties *props);
void pw_mempool_add_listener(struct pw_mempool *pool,
struct spa_hook *listener,
const struct pw_mempool_events *events,
void *data);
void pw_mempool_destroy(struct pw_mempool *pool);
int pw_mempool_alloc(struct pw_mempool *pool, enum pw_memblock_flags flags,
size_t size, struct pw_memblock **mem);
struct pw_memblock * pw_mempool_import_block(struct pw_mempool *pool,
struct pw_memblock *mem);
struct pw_memblock * pw_mempool_import(struct pw_mempool *pool,
uint32_t type, int fd, enum pw_memblock_flags flags);
/** Find memblock for given \a id */
int pw_mempool_remove_id(struct pw_mempool *pool, uint32_t id);
/** Find memblock for given \a ptr */
struct pw_memblock * pw_mempool_find_ptr(struct pw_mempool *pool, const void *ptr);
/** Find memblock for given \a id */
struct pw_memblock * pw_mempool_find_id(struct pw_mempool *pool, uint32_t id);
/** Find memblock for given \a fd */
struct pw_memblock * pw_mempool_find_fd(struct pw_mempool *pool, int fd);
struct pw_memmap * pw_memblock_map(struct pw_memblock *block,
enum pw_memmap_flags flags, uint32_t offset, uint32_t size);
struct pw_memmap * pw_mempool_map_id(struct pw_mempool *pool, uint32_t id,
enum pw_memmap_flags flags, uint32_t offset, uint32_t size);
int pw_memmap_free(struct pw_memmap *map);
void pw_memblock_free(struct pw_memblock *mem);
static inline void pw_memblock_unref(struct pw_memblock *mem)
{
if (--mem->ref == 0)
pw_memblock_free(mem);
}
#define PW_MAP_RANGE_INIT (struct pw_map_range){ 0, }
/** Calculate parameters to mmap() memory into \a range so that

20
src/pipewire/node.c
View file

@ -700,9 +700,11 @@ static void dump_states(struct pw_node *driver)
spa_list_for_each(t, &driver->rt.target_list, link) {
struct pw_node_activation *a = t->activation;
if (t->node == NULL)
continue;
pw_log_warn("node %p (%s): required:%d s:%"PRIu64" a:%"PRIu64" f:%"PRIu64
" waiting:%"PRIu64" process:%"PRIu64" status:%d",
t->node, t->node ? t->node->info.name : "",
t->node, t->node->info.name,
a->state[0].required,
a->signal_time,
a->awake_time,
@ -860,13 +862,13 @@ struct pw_node *pw_node_new(struct pw_core *core,
size = sizeof(struct pw_node_activation);
if ((res = pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
PW_MEMBLOCK_FLAG_MAP_READWRITE |
PW_MEMBLOCK_FLAG_SEAL,
size,
&this->activation)) < 0)
if ((res = pw_mempool_alloc(this->core->pool,
PW_MEMBLOCK_FLAG_SEAL |
PW_MEMBLOCK_FLAG_MAP,
size, &this->activation)) < 0)
goto error_clean;
impl->work = pw_work_queue_new(this->core->main_loop);
if (impl->work == NULL) {
res = -errno;
@ -894,7 +896,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
spa_list_init(&this->rt.output_mix);
spa_list_init(&this->rt.target_list);
this->rt.activation = this->activation->ptr;
this->rt.activation = this->activation->map->ptr;
this->rt.target.activation = this->rt.activation;
this->rt.target.node = this;
this->rt.target.signal = process_node;
@ -914,7 +916,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
error_clean:
if (this->activation)
pw_memblock_free(this->activation);
pw_memblock_unref(this->activation);
if (this->source.fd != -1)
spa_system_close(this->core->data_system, this->source.fd);
if (n)
@ -1262,7 +1264,7 @@ void pw_node_destroy(struct pw_node *node)
pw_log_debug("node %p: free", node);
pw_node_emit_free(node);
pw_memblock_free(node->activation);
pw_memblock_unref(node->activation);
pw_work_queue_destroy(impl->work);

18
src/pipewire/private.h
View file

@ -99,6 +99,7 @@ struct pw_client {
struct pw_client_info info; /**< client info */
struct pw_mempool *pool; /**< client mempool */
struct pw_resource *core_resource; /**< core resource object */
struct pw_resource *client_resource; /**< client resource object */
@ -159,11 +160,13 @@ struct pw_global {
#define pw_core_emit_global_removed(c,g) pw_core_emit(c, global_removed, 0, g)
#define pw_core_resource(r,m,v,...) pw_resource_call(r, struct pw_core_proxy_events, m, v, ##__VA_ARGS__)
#define pw_core_resource_info(r,...) pw_core_resource(r,info,0,__VA_ARGS__)
#define pw_core_resource_done(r,...) pw_core_resource(r,done,0,__VA_ARGS__)
#define pw_core_resource_ping(r,...) pw_core_resource(r,ping,0,__VA_ARGS__)
#define pw_core_resource_error(r,...) pw_core_resource(r,error,0,__VA_ARGS__)
#define pw_core_resource_remove_id(r,...) pw_core_resource(r,remove_id,0,__VA_ARGS__)
#define pw_core_resource_info(r,...) pw_core_resource(r,info,0,__VA_ARGS__)
#define pw_core_resource_done(r,...) pw_core_resource(r,done,0,__VA_ARGS__)
#define pw_core_resource_ping(r,...) pw_core_resource(r,ping,0,__VA_ARGS__)
#define pw_core_resource_error(r,...) pw_core_resource(r,error,0,__VA_ARGS__)
#define pw_core_resource_remove_id(r,...) pw_core_resource(r,remove_id,0,__VA_ARGS__)
#define pw_core_resource_add_mem(r,...) pw_core_resource(r,add_mem,0,__VA_ARGS__)
#define pw_core_resource_remove_mem(r,...) pw_core_resource(r,remove_mem,0,__VA_ARGS__)
static inline void
pw_core_resource_errorv(struct pw_resource *resource, uint32_t id, int seq,
@ -198,6 +201,8 @@ struct pw_core {
struct pw_properties *properties; /**< properties of the core */
struct pw_mempool *pool; /**< global memory pool */
struct pw_map globals; /**< map of globals */
struct spa_list protocol_list; /**< list of protocols */
@ -273,7 +278,7 @@ static inline void move_allocation(struct allocation *alloc, struct allocation *
static inline void free_allocation(struct allocation *alloc)
{
if (alloc->mem) {
pw_memblock_free(alloc->mem);
pw_memblock_unref(alloc->mem);
free(alloc->buffers);
}
alloc->mem = NULL;
@ -669,6 +674,7 @@ struct pw_remote {
struct spa_list link; /**< link in core remote_list */
struct pw_properties *properties; /**< extra properties */
struct pw_mempool *pool; /**< memory pool */
struct pw_core_proxy *core_proxy; /**< proxy for the core object */
struct pw_map objects; /**< map of client side proxy objects
* indexed with the client id */

26
src/pipewire/remote.c
View file

@ -143,12 +143,36 @@ static void core_event_remove_id(void *data, uint32_t id)
}
}
static void core_event_add_mem(void *data, uint32_t id, uint32_t type, int fd, uint32_t flags)
{
struct pw_remote *this = data;
struct pw_memblock *m;
pw_log_debug("remote %p: add mem %u type:%u fd:%d flags:%u", this, id, type, fd, flags);
m = pw_mempool_import(this->pool, type, fd, flags);
if (m->id != id) {
pw_log_error("remote %p: invalid mem id %u, expected %u",
this, id, m->id);
pw_memblock_unref(m);
}
}
static void core_event_remove_mem(void *data, uint32_t id)
{
struct pw_remote *this = data;
pw_log_debug("remote %p: remove mem %u", this, id);
pw_mempool_remove_id(this->pool, id);
}
static const struct pw_core_proxy_events core_proxy_events = {
PW_VERSION_CORE_PROXY_EVENTS,
.error = core_event_error,
.ping = core_event_ping,
.done = core_event_done,
.remove_id = core_event_remove_id,
.add_mem = core_event_add_mem,
.remove_mem = core_event_remove_mem,
};
SPA_EXPORT
@ -183,6 +207,7 @@ struct pw_remote *pw_remote_new(struct pw_core *core,
pw_fill_remote_properties(core, properties);
this->properties = properties;
this->pool = pw_mempool_new(NULL);
this->state = PW_REMOTE_STATE_UNCONNECTED;
@ -264,6 +289,7 @@ void pw_remote_destroy(struct pw_remote *remote)
pw_map_clear(&remote->objects);
pw_log_debug("remote %p: free", remote);
pw_mempool_destroy(remote->pool);
pw_properties_free(remote->properties);
free(remote->error);
free(impl);