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

System: More work on making system functions pluggable

Browse source 
Move the epoll functions to the system functions and make the loop
use those. Use simple mask for events instead of enum.
Add the used system api in pw_loop.
Add System API to spa_support and use it where possible.
Pass the system API used in the realtime loops in spa_support as
well and use this in the realtime paths.
Improve bootstrapping, load only the log and cpu interfaces because
those can/need to be shared between instances. Let the core load
the other interfaces.
Add keys to configure the System and Loop implementations used in
pw_loop.
This commit is contained in:
Wim Taymans 2019-06-06 15:21:40 +02:00
parent 86dc0496a5
commit db88e9f954
22 changed files with 455 additions and 342 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pipewire-jack

@ -1 +1 @@
Subproject commit 8646c4976070a7c8cfbf710a294be45e15206b6c
Subproject commit 21e904d694bf203b7fc759075db94d55acb87138

2
pipewire-pulseaudio

@ -1 +1 @@
Subproject commit 531185845afd0844ee1e513e9c941e4f1501ef2d
Subproject commit 9ebbcd46f9bf87f27c97078c3736fef1407ee259

18
spa/include/spa/support/loop.h
View file

@ -32,6 +32,7 @@ extern "C" {
#include <spa/utils/defs.h>
#include <spa/utils/hook.h>
#include <spa/utils/result.h>
#include <spa/support/system.h>
#define SPA_VERSION_LOOP 0
struct spa_loop { struct spa_interface iface; };
@ -41,13 +42,6 @@ struct spa_loop_control { struct spa_interface iface; };
struct spa_loop_utils { struct spa_interface iface; };
struct spa_source;
enum spa_io {
SPA_IO_IN = (1 << 0),
SPA_IO_OUT = (1 << 1),
SPA_IO_HUP = (1 << 2),
SPA_IO_ERR = (1 << 3),
};
typedef void (*spa_source_func_t) (struct spa_source *source);
struct spa_source {
@ -55,8 +49,8 @@ struct spa_source {
spa_source_func_t func;
void *data;
int fd;
enum spa_io mask;
enum spa_io rmask;
uint32_t mask;
uint32_t rmask;
};
typedef int (*spa_invoke_func_t) (struct spa_loop *loop,
@ -202,7 +196,7 @@ struct spa_loop_control_methods {
#define spa_loop_control_leave(l) spa_loop_control_method_v(l,leave,0)
#define spa_loop_control_iterate(l,...) spa_loop_control_method_r(l,iterate,0,__VA_ARGS__)
typedef void (*spa_source_io_func_t) (void *data, int fd, enum spa_io mask);
typedef void (*spa_source_io_func_t) (void *data, int fd, uint32_t mask);
typedef void (*spa_source_idle_func_t) (void *data);
typedef void (*spa_source_event_func_t) (void *data, uint64_t count);
typedef void (*spa_source_timer_func_t) (void *data, uint64_t expirations);
@ -219,11 +213,11 @@ struct spa_loop_utils_methods {
struct spa_source *(*add_io) (void *object,
int fd,
enum spa_io mask,
uint32_t mask,
bool close,
spa_source_io_func_t func, void *data);
int (*update_io) (void *object, struct spa_source *source, enum spa_io mask);
int (*update_io) (void *object, struct spa_source *source, uint32_t mask);
struct spa_source *(*add_idle) (void *object,
bool enabled,

29
spa/include/spa/support/system.h
View file

@ -29,23 +29,36 @@
extern "C" {
#endif
#include <sys/timerfd.h>
#include <spa/utils/defs.h>
#include <spa/utils/hook.h>
#include <spa/utils/result.h>
/**
* a collection of system functions
* a collection of core system functions
*/
#define SPA_VERSION_SYSTEM 0
struct spa_system { struct spa_interface iface; };
/* IO events */
#define SPA_IO_IN (1 << 0)
#define SPA_IO_OUT (1 << 1)
#define SPA_IO_HUP (1 << 2)
#define SPA_IO_ERR (1 << 3)
/* flags */
#define SPA_FD_CLOEXEC (1<<0)
#define SPA_FD_NONBLOCK (1<<1)
#define SPA_FD_EVENT_SEMAPHORE (1<<2)
#define SPA_FD_TIMER_ABSTIME (1<<3)
#define SPA_FD_TIMER_CANCEL_ON_SET (1<<4)
struct spa_poll_event {
uint32_t events;
void *data;
};
struct spa_system_methods {
#define SPA_VERSION_SYSTEM_METHODS 0
uint32_t version;
@ -62,6 +75,14 @@ struct spa_system_methods {
int (*clock_getres) (void *object,
int clockid, struct timespec *res);
/* poll */
int (*pollfd_create) (void *object, int flags);
int (*pollfd_add) (void *object, int pfd, int fd, uint32_t events, void *data);
int (*pollfd_mod) (void *object, int pfd, int fd, uint32_t events, void *data);
int (*pollfd_del) (void *object, int pfd, int fd);
int (*pollfd_wait) (void *object, int pfd,
struct spa_poll_event *ev, int n_ev, int timeout);
/* timers */
int (*timerfd_create) (void *object, int clockid, int flags);
int (*timerfd_settime) (void *object,
@ -101,6 +122,12 @@ struct spa_system_methods {
#define spa_system_clock_gettime(s,...) spa_system_method_r(s,clock_gettime,0,__VA_ARGS__)
#define spa_system_clock_getres(s,...) spa_system_method_r(s,clock_getres,0,__VA_ARGS__)
#define spa_system_pollfd_create(s,...) spa_system_method_r(s,pollfd_create,0,__VA_ARGS__)
#define spa_system_pollfd_add(s,...) spa_system_method_r(s,pollfd_add,0,__VA_ARGS__)
#define spa_system_pollfd_mod(s,...) spa_system_method_r(s,pollfd_mod,0,__VA_ARGS__)
#define spa_system_pollfd_del(s,...) spa_system_method_r(s,pollfd_del,0,__VA_ARGS__)
#define spa_system_pollfd_wait(s,...) spa_system_method_r(s,pollfd_wait,0,__VA_ARGS__)
#define spa_system_timerfd_create(s,...) spa_system_method_r(s,timerfd_create,0,__VA_ARGS__)
#define spa_system_timerfd_settime(s,...) spa_system_method_r(s,timerfd_settime,0,__VA_ARGS__)
#define spa_system_timerfd_gettime(s,...) spa_system_method_r(s,timerfd_gettime,0,__VA_ARGS__)

18
spa/plugins/alsa/alsa-utils.c
View file

@ -7,9 +7,9 @@
#include <sys/time.h>
#include <math.h>
#include <limits.h>
#include <sys/timerfd.h>
#include <spa/pod/filter.h>
#include <spa/support/system.h>
#include <spa/control/control.h>
#include "alsa-utils.h"
@ -34,7 +34,8 @@ static int spa_alsa_open(struct state *state)
SND_PCM_NO_AUTO_RESAMPLE |
SND_PCM_NO_AUTO_CHANNELS | SND_PCM_NO_AUTO_FORMAT), "open failed");
state->timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
state->timerfd = spa_system_timerfd_create(state->data_system,
CLOCK_MONOTONIC, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
state->opened = true;
state->sample_count = 0;
state->sample_time = 0;
@ -52,7 +53,7 @@ int spa_alsa_close(struct state *state)
spa_log_info(state->log, "%p: Device '%s' closing", state, state->props.device);
CHECK(snd_pcm_close(state->hndl), "close failed");
close(state->timerfd);
spa_system_close(state->data_system, state->timerfd);
state->opened = false;
return err;
@ -495,7 +496,8 @@ static int set_timeout(struct state *state, uint64_t time)
ts.it_value.tv_nsec = time % SPA_NSEC_PER_SEC;
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
timerfd_settime(state->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);
spa_system_timerfd_settime(state->data_system,
state->timerfd, SPA_FD_TIMER_ABSTIME, &ts, NULL);
}
return 0;
}
@ -998,13 +1000,13 @@ static void alsa_on_timeout_event(struct spa_source *source)
uint64_t expire;
int res;
if (state->started && read(state->timerfd, &expire, sizeof(uint64_t)) != sizeof(uint64_t))
if (state->started && spa_system_timerfd_read(state->data_system, state->timerfd, &expire) < 0)
spa_log_warn(state->log, "error reading timerfd: %s", strerror(errno));
if (state->position)
state->threshold = state->position->size;
clock_gettime(CLOCK_MONOTONIC, &state->now);
spa_system_clock_gettime(state->data_system, CLOCK_MONOTONIC, &state->now);
if ((res = get_status(state, &delay)) < 0)
return;
@ -1049,7 +1051,7 @@ static int set_timers(struct state *state)
ts.it_value.tv_nsec = 1;
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
timerfd_settime(state->timerfd, 0, &ts, NULL);
spa_system_timerfd_settime(state->data_system, state->timerfd, 0, &ts, NULL);
return 0;
}
@ -1160,7 +1162,7 @@ static int do_remove_source(struct spa_loop *loop,
ts.it_value.tv_nsec = 0;
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
timerfd_settime(state->timerfd, 0, &ts, NULL);
spa_system_timerfd_settime(state->data_system, state->timerfd, 0, &ts, NULL);
return 0;
}

24
spa/plugins/audiotestsrc/audiotestsrc.c
View file

@ -27,9 +27,9 @@
#include <unistd.h>
#include <string.h>
#include <stdio.h>
#include <sys/timerfd.h>
#include <spa/support/log.h>
#include <spa/support/system.h>
#include <spa/support/loop.h>
#include <spa/utils/list.h>
#include <spa/utils/keys.h>
@ -113,6 +113,7 @@ struct impl {
struct spa_log *log;
struct spa_loop *data_loop;
struct spa_system *data_system;
uint64_t info_all;
struct spa_node_info info;
@ -295,7 +296,8 @@ static void set_timer(struct impl *this, bool enabled)
this->timerspec.it_value.tv_sec = 0;
this->timerspec.it_value.tv_nsec = 0;
}
timerfd_settime(this->timer_source.fd, TFD_TIMER_ABSTIME, &this->timerspec, NULL);
spa_system_timerfd_settime(this->data_system,
this->timer_source.fd, SPA_FD_TIMER_ABSTIME, &this->timerspec, NULL);
}
}
@ -304,7 +306,7 @@ static void read_timer(struct impl *this)
uint64_t expirations;
if (this->async || this->props.live) {
if (read(this->timer_source.fd, &expirations, sizeof(uint64_t)) != sizeof(uint64_t))
if (spa_system_timerfd_read(this->data_system, this->timer_source.fd, &expirations) < 0)
perror("read timerfd");
}
}
@ -972,7 +974,7 @@ static int impl_clear(struct spa_handle *handle)
if (this->data_loop)
spa_loop_remove_source(this->data_loop, &this->timer_source);
close(this->timer_source.fd);
spa_system_close(this->data_system, this->timer_source.fd);
return 0;
}
@ -1004,10 +1006,17 @@ impl_init(const struct spa_handle_factory *factory,
this = (struct impl *) handle;
for (i = 0; i < n_support; i++) {
if (support[i].type == SPA_TYPE_INTERFACE_Log)
switch (support[i].type) {
case SPA_TYPE_INTERFACE_Log:
this->log = support[i].data;
else if (support[i].type == SPA_TYPE_INTERFACE_DataLoop)
break;
case SPA_TYPE_INTERFACE_DataLoop:
this->data_loop = support[i].data;
break;
case SPA_TYPE_INTERFACE_DataSystem:
this->data_system = support[i].data;
break;
}
}
spa_hook_list_init(&this->hooks);
@ -1031,7 +1040,8 @@ impl_init(const struct spa_handle_factory *factory,
this->timer_source.func = on_output;
this->timer_source.data = this;
this->timer_source.fd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC);
this->timer_source.fd = spa_system_timerfd_create(this->data_system,
CLOCK_MONOTONIC, SPA_FD_CLOEXEC);
this->timer_source.mask = SPA_IO_IN;
this->timer_source.rmask = 0;
this->timerspec.it_value.tv_sec = 0;

40
spa/plugins/bluez5/a2dp-sink.c
View file

@ -25,12 +25,12 @@
#include <unistd.h>
#include <stddef.h>
#include <stdio.h>
#include <sys/timerfd.h>
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <spa/support/loop.h>
#include <spa/support/log.h>
#include <spa/support/system.h>
#include <spa/utils/list.h>
#include <spa/utils/keys.h>
#include <spa/monitor/device.h>
@ -90,8 +90,8 @@ struct impl {
struct spa_node node;
struct spa_log *log;
struct spa_loop *main_loop;
struct spa_loop *data_loop;
struct spa_system *data_system;
struct spa_hook_list hooks;
struct spa_callbacks callbacks;
@ -249,7 +249,7 @@ static int set_timers(struct impl *this)
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
res = timerfd_settime(this->timerfd, 0, &ts, NULL);
res = spa_system_timerfd_settime(this->data_system, this->timerfd, 0, &ts, NULL);
this->source.mask = SPA_IO_IN;
spa_loop_update_source(this->data_loop, &this->source);
return res;
@ -632,7 +632,7 @@ static int flush_data(struct impl *this, uint64_t now_time)
&this->now, &ts.it_value);
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
timerfd_settime(this->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);
spa_system_timerfd_settime(this->data_system, this->timerfd, SPA_FD_TIMER_ABSTIME, &ts, NULL);
this->source.mask = SPA_IO_IN;
spa_loop_update_source(this->data_loop, &this->source);
} else {
@ -658,7 +658,7 @@ static void a2dp_on_flush(struct spa_source *source)
return;
}
clock_gettime(CLOCK_MONOTONIC, &this->now);
spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
now_time = this->now.tv_sec * SPA_NSEC_PER_SEC + this->now.tv_nsec;
flush_data(this, now_time);
@ -672,10 +672,10 @@ static void a2dp_on_timeout(struct spa_source *source)
uint64_t exp, now_time;
struct spa_io_buffers *io = port->io;
if (this->started && read(this->timerfd, &exp, sizeof(uint64_t)) != sizeof(uint64_t))
if (this->started && spa_system_timerfd_read(this->data_system, this->timerfd, &exp) < 0)
spa_log_warn(this->log, "error reading timerfd: %s", strerror(errno));
clock_gettime(CLOCK_MONOTONIC, &this->now);
spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
now_time = SPA_TIMESPEC_TO_NSEC(&this->now);
spa_log_trace(this->log, NAME" %p: timeout %ld %ld", this,
@ -851,7 +851,7 @@ static int do_remove_source(struct spa_loop *loop,
ts.it_value.tv_nsec = 0;
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
timerfd_settime(this->timerfd, 0, &ts, NULL);
spa_system_timerfd_settime(this->data_system, this->timerfd, 0, &ts, NULL);
if (this->flush_source.loop)
spa_loop_remove_source(this->data_loop, &this->flush_source);
@ -1311,7 +1311,7 @@ static int impl_node_process(void *object)
io = port->io;
spa_return_val_if_fail(io != NULL, -EIO);
clock_gettime(CLOCK_MONOTONIC, &this->now);
spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
now_time = SPA_TIMESPEC_TO_NSEC(&this->now);
if (!spa_list_is_empty(&port->ready))
@ -1393,6 +1393,8 @@ static int impl_get_interface(struct spa_handle *handle, uint32_t type, void **i
static int impl_clear(struct spa_handle *handle)
{
struct impl *this = (struct impl *) handle;
spa_system_close(this->data_system, this->timerfd);
return 0;
}
@ -1423,19 +1425,24 @@ impl_init(const struct spa_handle_factory *factory,
this = (struct impl *) handle;
for (i = 0; i < n_support; i++) {
if (support[i].type == SPA_TYPE_INTERFACE_Log)
switch (support[i].type) {
case SPA_TYPE_INTERFACE_Log:
this->log = support[i].data;
else if (support[i].type == SPA_TYPE_INTERFACE_DataLoop)
break;
case SPA_TYPE_INTERFACE_DataLoop:
this->data_loop = support[i].data;
else if (support[i].type == SPA_TYPE_INTERFACE_MainLoop)
this->main_loop = support[i].data;
break;
case SPA_TYPE_INTERFACE_DataSystem:
this->data_system = support[i].data;
break;
}
}
if (this->data_loop == NULL) {
spa_log_error(this->log, "a data loop is needed");
return -EINVAL;
}
if (this->main_loop == NULL) {
spa_log_error(this->log, "a main loop is needed");
if (this->data_system == NULL) {
spa_log_error(this->log, "a data system is needed");
return -EINVAL;
}
@ -1482,7 +1489,8 @@ impl_init(const struct spa_handle_factory *factory,
spa_bt_transport_add_listener(this->transport,
&this->transport_listener, &transport_events, this);
this->timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
this->timerfd = spa_system_timerfd_create(this->data_system,
CLOCK_MONOTONIC, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
return 0;
}

22
spa/plugins/bluez5/a2dp-source.c
View file

@ -33,6 +33,7 @@
#include <spa/support/loop.h>
#include <spa/support/log.h>
#include <spa/support/system.h>
#include <spa/utils/list.h>
#include <spa/utils/keys.h>
#include <spa/monitor/device.h>
@ -91,8 +92,8 @@ struct impl {
struct spa_node node;
struct spa_log *log;
struct spa_loop *main_loop;
struct spa_loop *data_loop;
struct spa_system *data_system;
struct spa_hook_list hooks;
struct spa_callbacks callbacks;
@ -392,7 +393,7 @@ static void a2dp_on_ready_read(struct spa_source *source)
}
/* update the current pts */
clock_gettime(CLOCK_MONOTONIC, &this->now);
spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
again:
/* read data from socket */
@ -1113,19 +1114,24 @@ impl_init(const struct spa_handle_factory *factory,
this = (struct impl *) handle;
for (i = 0; i < n_support; i++) {
if (support[i].type == SPA_TYPE_INTERFACE_Log)
switch (support[i].type) {
case SPA_TYPE_INTERFACE_Log:
this->log = support[i].data;
else if (support[i].type == SPA_TYPE_INTERFACE_DataLoop)
break;
case SPA_TYPE_INTERFACE_DataLoop:
this->data_loop = support[i].data;
else if (support[i].type == SPA_TYPE_INTERFACE_MainLoop)
this->main_loop = support[i].data;
break;
case SPA_TYPE_INTERFACE_DataSystem:
this->data_system = support[i].data;
break;
}
}
if (this->data_loop == NULL) {
spa_log_error(this->log, "a data loop is needed");
return -EINVAL;
}
if (this->main_loop == NULL) {
spa_log_error(this->log, "a main loop is needed");
if (this->data_system == NULL) {
spa_log_error(this->log, "a data system is needed");
return -EINVAL;
}

8
spa/plugins/support/dbus.c
View file

@ -76,9 +76,9 @@ static void dispatch_status(DBusConnection *conn, DBusDispatchStatus status, voi
status == DBUS_DISPATCH_COMPLETE ? false : true);
}
static inline enum spa_io dbus_to_io(DBusWatch *watch)
static inline uint32_t dbus_to_io(DBusWatch *watch)
{
enum spa_io mask;
uint32_t mask;
unsigned int flags;
/* no watch flags for disabled watches */
@ -96,7 +96,7 @@ static inline enum spa_io dbus_to_io(DBusWatch *watch)
return mask;
}
static inline unsigned int io_to_dbus(enum spa_io mask)
static inline unsigned int io_to_dbus(uint32_t mask)
{
unsigned int flags = 0;
@ -112,7 +112,7 @@ static inline unsigned int io_to_dbus(enum spa_io mask)
}
static void
handle_io_event(void *userdata, int fd, enum spa_io mask)
handle_io_event(void *userdata, int fd, uint32_t mask)
{
DBusWatch *watch = userdata;

11
spa/plugins/support/logger.c
View file

@ -27,10 +27,10 @@
#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <sys/eventfd.h>
#include <spa/support/log.h>
#include <spa/support/loop.h>
#include <spa/support/system.h>
#include <spa/support/plugin.h>
#include <spa/utils/ringbuffer.h>
#include <spa/utils/type.h>
@ -93,14 +93,13 @@ impl_log_logv(void *object,
if (SPA_UNLIKELY(do_trace)) {
uint32_t index;
uint64_t count = 1;
spa_ringbuffer_get_write_index(&impl->trace_rb, &index);
spa_ringbuffer_write_data(&impl->trace_rb, impl->trace_data, TRACE_BUFFER,
index & (TRACE_BUFFER - 1), location, size);
spa_ringbuffer_write_update(&impl->trace_rb, index + size);
if (write(impl->source.fd, &count, sizeof(uint64_t)) != sizeof(uint64_t))
if (spa_system_eventfd_write(impl->system, impl->source.fd, 1) < 0)
fprintf(impl->file, "error signaling eventfd: %s\n", strerror(errno));
} else
fputs(location, impl->file);
@ -129,7 +128,7 @@ static void on_trace_event(struct spa_source *source)
uint32_t index;
uint64_t count;
if (read(source->fd, &count, sizeof(uint64_t)) != sizeof(uint64_t))
if (spa_system_eventfd_read(impl->system, source->fd, &count) < 0)
fprintf(impl->file, "failed to read event fd: %s", strerror(errno));
while ((avail = spa_ringbuffer_get_read_index(&impl->trace_rb, &index)) > 0) {
@ -247,10 +246,10 @@ impl_init(const struct spa_handle_factory *factory,
if (this->file == NULL)
this->file = stderr;
if (loop) {
if (loop != NULL && this->system != NULL) {
this->source.func = on_trace_event;
this->source.data = this;
this->source.fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
this->source.fd = spa_system_eventfd_create(this->system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
this->source.mask = SPA_IO_IN;
this->source.rmask = 0;
spa_loop_add_source(loop, &this->source);

97
spa/plugins/support/loop.c
View file

@ -28,7 +28,6 @@
#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <pthread.h>
#include <spa/support/loop.h>
@ -72,7 +71,7 @@ struct impl {
struct spa_list destroy_list;
struct spa_hook_list hooks_list;
int epoll_fd;
int poll_fd;
pthread_t thread;
struct spa_source *wakeup;
@ -100,83 +99,38 @@ struct source_impl {
};
/** \endcond */
static inline uint32_t spa_io_to_epoll(enum spa_io mask)
{
uint32_t events = 0;
if (mask & SPA_IO_IN)
events |= EPOLLIN;
if (mask & SPA_IO_OUT)
events |= EPOLLOUT;
if (mask & SPA_IO_ERR)
events |= EPOLLERR;
if (mask & SPA_IO_HUP)
events |= EPOLLHUP;
return events;
}
static inline enum spa_io spa_epoll_to_io(uint32_t events)
{
enum spa_io mask = 0;
if (events & EPOLLIN)
mask |= SPA_IO_IN;
if (events & EPOLLOUT)
mask |= SPA_IO_OUT;
if (events & EPOLLHUP)
mask |= SPA_IO_HUP;
if (events & EPOLLERR)
mask |= SPA_IO_ERR;
return mask;
}
static int loop_add_source(void *object, struct spa_source *source)
{
struct impl *impl = object;
source->loop = &impl->loop;
if (source->fd != -1) {
struct epoll_event ep;
if (SPA_UNLIKELY(source->fd == -1))
return 0;
spa_zero(ep);
ep.events = spa_io_to_epoll(source->mask);
ep.data.ptr = source;
if (epoll_ctl(impl->epoll_fd, EPOLL_CTL_ADD, source->fd, &ep) < 0)
return errno;
}
return 0;
return spa_system_pollfd_add(impl->system, impl->poll_fd, source->fd, source->mask, source);
}
static int loop_update_source(void *object, struct spa_source *source)
{
struct impl *impl = object;
if (source->fd != -1) {
struct epoll_event ep;
if (SPA_UNLIKELY(source->fd == -1))
return 0;
spa_zero(ep);
ep.events = spa_io_to_epoll(source->mask);
ep.data.ptr = source;
if (epoll_ctl(impl->epoll_fd, EPOLL_CTL_MOD, source->fd, &ep) < 0)
return errno;
}
return 0;
return spa_system_pollfd_mod(impl->system, impl->poll_fd, source->fd, source->mask, source);
}
static int loop_remove_source(void *object, struct spa_source *source)
{
struct impl *impl = object;
if (source->fd != -1)
epoll_ctl(impl->epoll_fd, EPOLL_CTL_DEL, source->fd, NULL);
source->loop = NULL;
return 0;
if (SPA_UNLIKELY(source->fd == -1))
return 0;
return spa_system_pollfd_del(impl->system, impl->poll_fd, source->fd);
}
static int
@ -288,7 +242,7 @@ static void wakeup_func(void *data, uint64_t count)
static int loop_get_fd(void *object)
{
struct impl *impl = object;
return impl->epoll_fd;
return impl->poll_fd;
}
static void
@ -325,12 +279,13 @@ static int loop_iterate(void *object, int timeout)
{
struct impl *impl = object;
struct spa_loop *loop = &impl->loop;
struct epoll_event ep[32];
struct spa_poll_event ep[32];
int i, nfds, save_errno = 0;
spa_loop_control_hook_before(&impl->hooks_list);
if (SPA_UNLIKELY((nfds = epoll_wait(impl->epoll_fd, ep, SPA_N_ELEMENTS(ep), timeout)) < 0))
nfds = spa_system_pollfd_wait(impl->system, impl->poll_fd, ep, SPA_N_ELEMENTS(ep), timeout);
if (SPA_UNLIKELY(nfds < 0))
save_errno = errno;
spa_loop_control_hook_after(&impl->hooks_list);
@ -342,11 +297,11 @@ static int loop_iterate(void *object, int timeout)
* some callback might also want to look at other sources it manages and
* can then reset the rmask to suppress the callback */
for (i = 0; i < nfds; i++) {
struct spa_source *s = ep[i].data.ptr;
s->rmask = spa_epoll_to_io(ep[i].events);
struct spa_source *s = ep[i].data;
s->rmask = ep[i].events;
}
for (i = 0; i < nfds; i++) {
struct spa_source *s = ep[i].data.ptr;
struct spa_source *s = ep[i].data;
if (s->rmask && s->fd != -1 && s->loop == loop)
s->func(s);
}
@ -363,7 +318,7 @@ static void source_io_func(struct spa_source *source)
static struct spa_source *loop_add_io(void *object,
int fd,
enum spa_io mask,
uint32_t mask,
bool close, spa_source_io_func_t func, void *data)
{
struct impl *impl = object;
@ -389,7 +344,7 @@ static struct spa_source *loop_add_io(void *object,
return &source->source;
}
static int loop_update_io(void *object, struct spa_source *source, enum spa_io mask)
static int loop_update_io(void *object, struct spa_source *source, uint32_t mask)
{
source->mask = mask;
return loop_update_source(object, source);
@ -694,7 +649,7 @@ static int impl_clear(struct spa_handle *handle)
process_destroy(impl);
spa_system_close(impl->system, impl->ack_fd);
close(impl->epoll_fd);
spa_system_close(impl->system, impl->poll_fd);
return 0;
}
@ -746,9 +701,13 @@ impl_init(const struct spa_handle_factory *factory,
break;
}
}
if (impl->system == NULL) {
spa_log_error(impl->log, NAME " %p: a System is needed", impl);
return -EINVAL;
}
impl->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
if (impl->epoll_fd == -1)
impl->poll_fd = spa_system_pollfd_create(impl->system, SPA_FD_CLOEXEC);
if (impl->poll_fd < 0)
return errno;
spa_list_init(&impl->source_list);

99
spa/plugins/support/system.c
View file

@ -28,6 +28,7 @@
#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <sys/ioctl.h>
#include <sys/timerfd.h>
#include <sys/eventfd.h>
@ -43,7 +44,6 @@
struct impl {
struct spa_handle handle;
struct spa_system system;
struct spa_log *log;
};
@ -89,6 +89,90 @@ static int impl_clock_getres(void *object,
return clock_getres(clockid, res);
}
/* poll */
static inline uint32_t spa_io_to_epoll(uint32_t mask)
{
uint32_t events = 0;
if (mask & SPA_IO_IN)
events |= EPOLLIN;
if (mask & SPA_IO_OUT)
events |= EPOLLOUT;
if (mask & SPA_IO_ERR)
events |= EPOLLERR;
if (mask & SPA_IO_HUP)
events |= EPOLLHUP;
return events;
}
static inline uint32_t spa_epoll_to_io(uint32_t events)
{
uint32_t mask = 0;
if (events & EPOLLIN)
mask |= SPA_IO_IN;
if (events & EPOLLOUT)
mask |= SPA_IO_OUT;
if (events & EPOLLHUP)
mask |= SPA_IO_HUP;
if (events & EPOLLERR)
mask |= SPA_IO_ERR;
return mask;
}
static int impl_pollfd_create(void *object, int flags)
{
int fl = 0;
if (flags & SPA_FD_CLOEXEC)
fl |= EPOLL_CLOEXEC;
return epoll_create1(fl);
}
static int impl_pollfd_add(void *object, int pfd, int fd, uint32_t events, void *data)
{
struct epoll_event ep;
spa_zero(ep);
ep.events = spa_io_to_epoll(events);
ep.data.ptr = data;
return epoll_ctl(pfd, EPOLL_CTL_ADD, fd, &ep);
}
static int impl_pollfd_mod(void *object, int pfd, int fd, uint32_t events, void *data)
{
struct epoll_event ep;
spa_zero(ep);
ep.events = spa_io_to_epoll(events);
ep.data.ptr = data;
return epoll_ctl(pfd, EPOLL_CTL_MOD, fd, &ep);
}
static int impl_pollfd_del(void *object, int pfd, int fd)
{
return epoll_ctl(pfd, EPOLL_CTL_DEL, fd, NULL);
}
static int impl_pollfd_wait(void *object, int pfd,
struct spa_poll_event *ev, int n_ev, int timeout)
{
struct epoll_event ep[n_ev];
int i, nfds;
if (SPA_UNLIKELY((nfds = epoll_wait(pfd, ep, SPA_N_ELEMENTS(ep), timeout)) < 0))
return nfds;
for (i = 0; i < nfds; i++) {
ev[i].events = spa_epoll_to_io(ep[i].events);
ev[i].data = ep[i].data.ptr;
}
return nfds;
}
/* timers */
static int impl_timerfd_create(void *object, int clockid, int flags)
{
@ -121,8 +205,7 @@ static int impl_timerfd_gettime(void *object,
}
static int impl_timerfd_read(void *object, int fd, uint64_t *expirations)
{
int res;
if ((res = read(fd, expirations, sizeof(uint64_t))) != sizeof(uint64_t))
if (read(fd, expirations, sizeof(uint64_t)) != sizeof(uint64_t))
return -errno;
return 0;
}
@ -195,6 +278,11 @@ static const struct spa_system_methods impl_system = {
.close = impl_close,
.clock_gettime = impl_clock_gettime,
.clock_getres = impl_clock_getres,
.pollfd_create = impl_pollfd_create,
.pollfd_add = impl_pollfd_add,
.pollfd_mod = impl_pollfd_mod,
.pollfd_del = impl_pollfd_del,
.pollfd_wait = impl_pollfd_wait,
.timerfd_create = impl_timerfd_create,
.timerfd_settime = impl_timerfd_settime,
.timerfd_gettime = impl_timerfd_gettime,
@ -261,8 +349,11 @@ impl_init(const struct spa_handle_factory *factory,
&impl_system, impl);
for (i = 0; i < n_support; i++) {
if (support[i].type == SPA_TYPE_INTERFACE_Log)
switch (support[i].type) {
case SPA_TYPE_INTERFACE_Log:
impl->log = support[i].data;
break;
}
}
spa_log_debug(impl->log, NAME " %p: initialized", impl);

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

@ -28,8 +28,8 @@
#include <errno.h>
#include <unistd.h>
#include <time.h>
#include <sys/eventfd.h>
#include <spa/support/system.h>
#include <spa/node/node.h>
#include <spa/node/utils.h>
#include <spa/pod/filter.h>
@ -127,6 +127,7 @@ struct node {
struct spa_log *log;
struct spa_loop *data_loop;
struct spa_system *data_system;
struct spa_hook_list hooks;
struct spa_callbacks callbacks;
@ -978,15 +979,14 @@ static int impl_node_process(void *object)
struct impl *impl = this->impl;
struct pw_node *n = impl->this.node;
struct timespec ts;
uint64_t cmd = 1;
spa_log_trace_fp(this->log, "%p: send process %p", this, impl->this.node->driver_node);
clock_gettime(CLOCK_MONOTONIC, &ts);
spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &ts);
n->rt.activation->status = TRIGGERED;
n->rt.activation->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
if (write(this->writefd, &cmd, sizeof(cmd)) != sizeof(cmd))
if (spa_system_eventfd_write(this->data_system, this->writefd, 1) < 0)
spa_log_warn(this->log, "node %p: error %m", this);
return SPA_STATUS_OK;
@ -1118,7 +1118,8 @@ static void node_on_data_fd_events(struct spa_source *source)
if (source->rmask & SPA_IO_IN) {
uint64_t cmd;
if (read(this->data_source.fd, &cmd, sizeof(cmd)) != sizeof(cmd) || cmd != 1)
if (spa_system_eventfd_read(this->data_system,
this->data_source.fd, &cmd) < 0 || cmd != 1)
spa_log_warn(this->log, "node %p: read %"PRIu64" failed %m", this, cmd);
spa_log_trace_fp(this->log, "node %p: got ready", this);
@ -1162,7 +1163,8 @@ node_init(struct node *this,
case SPA_TYPE_INTERFACE_DataLoop:
this->data_loop = support[i].data;
break;
default:
case SPA_TYPE_INTERFACE_DataSystem:
this->data_system = support[i].data;
break;
}
}
@ -1170,6 +1172,10 @@ node_init(struct node *this,
spa_log_error(this->log, "a data-loop is needed");
return -EINVAL;
}
if (this->data_system == NULL) {
spa_log_error(this->log, "a data-system is needed");
return -EINVAL;
}
this->node.iface = SPA_INTERFACE_INIT(
SPA_TYPE_INTERFACE_Node,
@ -1292,13 +1298,14 @@ static void node_initialized(void *data)
struct pw_client_node *this = &impl->this;
struct pw_node *node = this->node;
struct pw_global *global;
struct spa_system *data_system = impl->node.data_system;
size_t size;
if (this->resource == NULL)
return;
impl->fds[0] = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
impl->fds[1] = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
impl->fds[0] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
impl->fds[1] = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
impl->node.data_source.fd = impl->fds[0];
impl->node.writefd = impl->fds[1];
impl->other_fds[0] = impl->fds[1];
@ -1325,6 +1332,7 @@ static void node_initialized(void *data)
static void node_free(void *data)
{
struct impl *impl = data;
struct spa_system *data_system = impl->node.data_system;
pw_log_debug("client-node %p: free", &impl->this);
node_clear(&impl->node);
@ -1338,9 +1346,9 @@ static void node_free(void *data)
pw_map_clear(&impl->io_map);
if (impl->fds[0] != -1)
close(impl->fds[0]);
spa_system_close(data_system, impl->fds[0]);
if (impl->fds[1] != -1)
close(impl->fds[1]);
spa_system_close(data_system, impl->fds[1]);
free(impl);
}

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

@ -194,7 +194,7 @@ client_busy_changed(void *data, bool busy)
{
struct client_data *c = data;
struct pw_client *client = c->client;
enum spa_io mask = SPA_IO_ERR | SPA_IO_HUP;
uint32_t mask = SPA_IO_ERR | SPA_IO_HUP;
c->busy = busy;
@ -210,7 +210,7 @@ client_busy_changed(void *data, bool busy)
}
static void
connection_data(void *data, int fd, enum spa_io mask)
connection_data(void *data, int fd, uint32_t mask)
{
struct client_data *this = data;
struct pw_client *client = this->client;
@ -370,7 +370,7 @@ static bool lock_socket(struct server *s)
}
static void
socket_data(void *data, int fd, enum spa_io mask)
socket_data(void *data, int fd, uint32_t mask)
{
struct server *s = data;
struct pw_client *client;
@ -467,7 +467,7 @@ static int impl_steal_fd(struct pw_protocol_client *client)
}
static void
on_remote_data(void *data, int fd, enum spa_io mask)
on_remote_data(void *data, int fd, uint32_t mask)
{
struct client *impl = data;
struct pw_remote *this = impl->this.remote;

31
src/pipewire/core.c
View file

@ -445,7 +445,8 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop,
struct pw_core *this;
const char *name;
void *dbus_iface = NULL;
int res;
uint32_t n_support;
int res = 0;
impl = calloc(1, sizeof(struct impl) + user_data_size);
if (impl == NULL)
@ -465,31 +466,33 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop,
this->properties = properties;
this->data_loop_impl = pw_data_loop_new(properties);
this->data_loop_impl = pw_data_loop_new(pw_properties_copy(properties));
if (this->data_loop_impl == NULL)
goto no_data_loop;
this->data_loop = pw_data_loop_get_loop(this->data_loop_impl);
this->data_system = this->data_loop->system;
this->main_loop = main_loop;
pw_array_init(&this->factory_lib, 32);
pw_map_init(&this->globals, 128, 32);
this->support[0] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_DataLoop, this->data_loop->loop);
this->support[1] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_MainLoop, this->main_loop->loop);
this->support[2] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_LoopUtils, this->main_loop->utils);
this->support[3] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_Log, pw_log_get());
n_support = pw_get_support(this->support, SPA_N_ELEMENTS(this->support));
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_System, this->main_loop->system);
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_Loop, this->main_loop->loop);
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_LoopUtils, this->main_loop->utils);
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_DataSystem, this->data_system);
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_DataLoop, this->data_loop->loop);
impl->dbus_handle = pw_load_spa_handle("support/libspa-dbus", "dbus", NULL, 4, this->support);
if (impl->dbus_handle) {
if ((res = spa_handle_get_interface(impl->dbus_handle,
SPA_TYPE_INTERFACE_DBus, &dbus_iface)) < 0)
impl->dbus_handle = pw_load_spa_handle("support/libspa-dbus", "dbus", NULL, n_support, this->support);
if (impl->dbus_handle == NULL ||
(res = spa_handle_get_interface(impl->dbus_handle,
SPA_TYPE_INTERFACE_DBus, &dbus_iface)) < 0) {
pw_log_warn("can't load dbus interface: %s", spa_strerror(res));
} else {
this->support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_DBus, dbus_iface);
}
this->support[4] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_DBus, dbus_iface);
this->support[5] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_CPU,
pw_get_support_interface(SPA_TYPE_INTERFACE_CPU));
this->n_support = 6;
this->n_support = n_support;
pw_data_loop_start(this->data_loop_impl);

5
src/pipewire/keys.h
View file

@ -61,6 +61,11 @@ extern "C" {
#define PW_KEY_SEC_GID "pipewire.sec.gid" /**< client gid, set by protocol*/
#define PW_KEY_SEC_LABEL "pipewire.sec.label" /**< client security label, set by protocol*/
#define PW_KEY_LOOP_LIBRARY_SYSTEM "loop.library.system" /**< name of the system library to use for
* a loop. */
#define PW_KEY_LOOP_LIBRARY_LOOP "loop.library.loop" /**< name of the loop library to use for a
* a loop. */
#define PW_KEY_CORE_MONITORS "core.monitors" /**< the apis monitored by core. */
/* remote keys */

95
src/pipewire/loop.c
View file

@ -38,7 +38,9 @@
struct impl {
struct pw_loop this;
struct spa_handle *handle;
struct spa_handle *system_handle;
struct spa_handle *loop_handle;
struct pw_properties *properties;
};
/** \endcond */
@ -52,63 +54,97 @@ struct pw_loop *pw_loop_new(struct pw_properties *properties)
int res;
struct impl *impl;
struct pw_loop *this;
const struct spa_handle_factory *factory;
void *iface;
const struct spa_support *support;
struct spa_support support[32];
uint32_t n_support;
const char *lib;
support = pw_get_support(&n_support);
if (support == NULL)
return NULL;
n_support = pw_get_support(support, 32);
factory = pw_get_support_factory("loop");
if (factory == NULL)
return NULL;
impl = calloc(1, sizeof(struct impl) + spa_handle_factory_get_size(factory, NULL));
impl = calloc(1, sizeof(struct impl));
if (impl == NULL)
return NULL;
impl->handle = SPA_MEMBER(impl, sizeof(struct impl), struct spa_handle);
this = &impl->this;
impl->properties = properties;
if ((res = spa_handle_factory_init(factory,
impl->handle,
NULL,
support,
n_support)) < 0) {
fprintf(stderr, "can't make factory instance: %d\n", res);
goto failed;
if (properties)
lib = pw_properties_get(properties, PW_KEY_LOOP_LIBRARY_SYSTEM);
else
lib = NULL;
impl->system_handle = pw_load_spa_handle(lib,
"system",
properties ? &properties->dict : NULL,
n_support, support);
if (impl->system_handle == NULL) {
pw_log_error("can't make system handle");
goto out_free;
}
if ((res = spa_handle_get_interface(impl->handle,
if ((res = spa_handle_get_interface(impl->system_handle,
SPA_TYPE_INTERFACE_System,
&iface)) < 0) {
fprintf(stderr, "can't get System interface %d\n", res);
goto out_free_system;
}
this->system = iface;
support[n_support++] = SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_System, iface);
if (properties)
lib = pw_properties_get(properties, PW_KEY_LOOP_LIBRARY_LOOP);
else
lib = NULL;
impl->loop_handle = pw_load_spa_handle(lib,
"loop",
properties ? &properties->dict : NULL,
n_support, support);
if (impl->loop_handle == NULL) {
pw_log_error("can't make loop handle");
goto out_free_system;
}
if ((res = spa_handle_get_interface(impl->system_handle,
SPA_TYPE_INTERFACE_System,
&iface)) < 0) {
fprintf(stderr, "can't get System interface %d\n", res);
goto out_free_loop;
}
this->system = iface;
if ((res = spa_handle_get_interface(impl->loop_handle,
SPA_TYPE_INTERFACE_Loop,
&iface)) < 0) {
fprintf(stderr, "can't get Loop interface %d\n", res);
goto failed;
goto out_free_loop;
}
this->loop = iface;
if ((res = spa_handle_get_interface(impl->handle,
if ((res = spa_handle_get_interface(impl->loop_handle,
SPA_TYPE_INTERFACE_LoopControl,
&iface)) < 0) {
fprintf(stderr, "can't get LoopControl interface %d\n", res);
goto failed;
goto out_free_loop;
}
this->control = iface;
if ((res = spa_handle_get_interface(impl->handle,
if ((res = spa_handle_get_interface(impl->loop_handle,
SPA_TYPE_INTERFACE_LoopUtils,
&iface)) < 0) {
fprintf(stderr, "can't get LoopUtils interface %d\n", res);
goto failed;
goto out_free_loop;
}
this->utils = iface;
return this;
failed:
out_free_loop:
pw_unload_spa_handle(impl->loop_handle);
out_free_system:
pw_unload_spa_handle(impl->system_handle);
out_free:
free(impl);
return NULL;
}
@ -122,6 +158,9 @@ void pw_loop_destroy(struct pw_loop *loop)
{
struct impl *impl = SPA_CONTAINER_OF(loop, struct impl, this);
spa_handle_clear(impl->handle);
if (impl->properties)
pw_properties_free(impl->properties);
pw_unload_spa_handle(impl->loop_handle);
pw_unload_spa_handle(impl->system_handle);
free(impl);
}

19
src/pipewire/node.c
View file

@ -30,6 +30,7 @@
#include <time.h>
#include <sys/eventfd.h>
#include <spa/support/system.h>
#include <spa/pod/parser.h>
#include <spa/node/utils.h>
#include <spa/debug/types.h>
@ -671,6 +672,7 @@ static inline int resume_node(struct pw_node *this, int status)
struct pw_node_target *t;
struct timespec ts;
struct pw_node_activation *activation = this->rt.activation;
struct spa_system *data_system = this->core->data_system;
uint64_t nsec;
if (status & SPA_STATUS_HAVE_BUFFER) {
@ -678,7 +680,7 @@ static inline int resume_node(struct pw_node *this, int status)
spa_node_process(p->mix);
}
clock_gettime(CLOCK_MONOTONIC, &ts);
spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
nsec = SPA_TIMESPEC_TO_NSEC(&ts);
activation->status = FINISHED;
activation->finish_time = nsec;
@ -708,11 +710,12 @@ static inline int process_node(void *data)
struct timespec ts;
struct pw_port *p;
struct pw_node_activation *a = this->rt.activation;
struct spa_system *data_system = this->core->data_system;
int status;
pw_log_trace_fp("node %p: process", this);
clock_gettime(CLOCK_MONOTONIC, &ts);
spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
a->status = AWAKE;
a->awake_time = SPA_TIMESPEC_TO_NSEC(&ts);
@ -723,7 +726,7 @@ static inline int process_node(void *data)
a->state[0].status = status;
if (this == this->driver_node && !this->exported) {
clock_gettime(CLOCK_MONOTONIC, &ts);
spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
a->status = FINISHED;
a->signal_time = a->finish_time;
a->finish_time = SPA_TIMESPEC_TO_NSEC(&ts);
@ -742,6 +745,7 @@ static inline int process_node(void *data)
static void node_on_fd_events(struct spa_source *source)
{
struct pw_node *this = source->data;
struct spa_system *data_system = this->core->data_system;
if (source->rmask & (SPA_IO_ERR | SPA_IO_HUP)) {
pw_log_warn("node %p: got socket error %08x", this, source->rmask);
@ -751,7 +755,7 @@ static void node_on_fd_events(struct spa_source *source)
if (source->rmask & SPA_IO_IN) {
uint64_t cmd;
if (read(this->source.fd, &cmd, sizeof(cmd)) != sizeof(cmd) || cmd != 1)
if (spa_system_eventfd_read(data_system, this->source.fd, &cmd) < 0 || cmd != 1)
pw_log_warn("node %p: read %"PRIu64" failed %m", this, cmd);
pw_log_trace_fp("node %p: got process", this);
@ -768,6 +772,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
struct impl *impl;
struct pw_node *this;
size_t size;
struct spa_system *data_system = core->data_system;
char *n;
impl = calloc(1, sizeof(struct impl) + user_data_size);
@ -795,7 +800,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
size = sizeof(struct pw_node_activation);
this->source.fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
this->source.fd = spa_system_eventfd_create(data_system, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
if (this->source.fd == -1)
goto clean_impl;
@ -856,7 +861,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
clean_impl:
if (this->source.func != NULL)
close(this->source.fd);
spa_system_close(this->core->data_system, this->source.fd);
if (properties)
pw_properties_free(properties);
free(impl);
@ -1197,7 +1202,7 @@ void pw_node_destroy(struct pw_node *node)
clear_info(node);
close(node->source.fd);
spa_system_close(node->core->data_system, node->source.fd);
free(impl);
}

229
src/pipewire/pipewire.c
View file

@ -41,6 +41,8 @@
#define MAX_SUPPORT 32
#define SUPPORTLIB "support/libspa-support"
struct plugin {
struct spa_list link;
char *filename;
@ -53,7 +55,7 @@ struct plugin {
struct handle {
struct spa_list link;
struct plugin *plugin;
const char *factory_name;
char *factory_name;
int ref;
struct spa_handle handle;
};
@ -65,9 +67,9 @@ struct registry {
struct support {
char **categories;
const char *plugin_dir;
struct plugin *support_plugin;
struct spa_support support[MAX_SUPPORT];
const char *support_lib;
struct registry *registry;
struct spa_support support[MAX_SUPPORT];
uint32_t n_support;
};
@ -164,52 +166,13 @@ static const struct spa_handle_factory *find_factory(struct plugin *plugin, cons
return NULL;
}
static struct handle *
load_handle(struct plugin *plugin,
const char *factory_name,
const struct spa_dict *info,
uint32_t n_support,
struct spa_support support[n_support])
{
int res;
struct handle *handle;
const struct spa_handle_factory *factory;
factory = find_factory(plugin, factory_name);
if (factory == NULL)
goto not_found;
handle = calloc(1, sizeof(struct handle) + spa_handle_factory_get_size(factory, info));
if (handle == NULL)
goto alloc_failed;
if ((res = spa_handle_factory_init(factory,
&handle->handle, info,
support, n_support)) < 0) {
fprintf(stderr, "can't make factory instance %s: %d\n", factory_name, res);
goto init_failed;
}
handle->ref = 1;
handle->plugin = plugin;
handle->factory_name = factory_name;
spa_list_append(&plugin->handles, &handle->link);
return handle;
init_failed:
free(handle);
alloc_failed:
not_found:
return NULL;
}
static void unref_handle(struct handle *handle)
{
if (--handle->ref == 0) {
spa_list_remove(&handle->link);
spa_handle_clear(&handle->handle);
unref_plugin(handle->plugin);
free(handle->factory_name);
free(handle);
}
}
@ -230,30 +193,13 @@ static void configure_debug(struct support *support, const char *str)
pw_free_strv(level);
}
/** Get a support interface
* \param type the interface type
* \return the interface or NULL when not configured
*/
SPA_EXPORT
void *pw_get_support_interface(uint32_t type)
uint32_t pw_get_support(struct spa_support *support, uint32_t max_support)
{
return spa_support_find(global_support.support, global_support.n_support, type);
}
SPA_EXPORT
const struct spa_handle_factory *pw_get_support_factory(const char *factory_name)
{
struct plugin *plugin = global_support.support_plugin;
if (plugin == NULL)
return NULL;
return find_factory(plugin, factory_name);
}
SPA_EXPORT
const struct spa_support *pw_get_support(uint32_t *n_support)
{
*n_support = global_support.n_support;
return global_support.support;
uint32_t i, n = SPA_MIN(global_support.n_support, max_support);
for (i = 0; i < n; i++)
support[i] = global_support.support[i];
return n;
}
SPA_EXPORT
@ -264,31 +210,53 @@ struct spa_handle *pw_load_spa_handle(const char *lib,
const struct spa_support support[])
{
struct support *sup = &global_support;
struct spa_support extra_support[MAX_SUPPORT];
uint32_t extra_n_support;
struct plugin *plugin;
struct handle *handle;
uint32_t i;
const struct spa_handle_factory *factory;
int res;
extra_n_support = sup->n_support;
memcpy(extra_support, sup->support,
sizeof(struct spa_support) * sup->n_support);
for (i = 0; i < n_support; i++) {
extra_support[extra_n_support++] =
SPA_SUPPORT_INIT(support[i].type, support[i].data);
}
pw_log_debug("load \"%s\", \"%s\"", lib, factory_name);
spa_return_val_if_fail(factory_name != NULL, NULL);
if (lib == NULL)
lib = sup->support_lib;
if (lib == NULL)
return NULL;
if ((plugin = open_plugin(sup->registry, sup->plugin_dir, lib)) == NULL) {
pw_log_warn("can't load '%s'", lib);
return NULL;
goto out;
}
handle = load_handle(plugin, factory_name, info, extra_n_support, extra_support);
factory = find_factory(plugin, factory_name);
if (factory == NULL)
goto out_unref_plugin;
handle = calloc(1, sizeof(struct handle) + spa_handle_factory_get_size(factory, info));
if (handle == NULL)
return NULL;
goto out;
if ((res = spa_handle_factory_init(factory,
&handle->handle, info,
support, n_support)) < 0) {
pw_log_warn("can't make factory instance %s: %d\n", factory_name, res);
goto out_free_handle;
}
handle->ref = 1;
handle->plugin = plugin;
handle->factory_name = strdup(factory_name);
spa_list_append(&plugin->handles, &handle->link);
return &handle->handle;
out_free_handle:
free(handle);
out_unref_plugin:
unref_plugin(plugin);
out:
return NULL;
}
static struct handle *find_handle(struct spa_handle *handle)
@ -319,6 +287,29 @@ int pw_unload_spa_handle(struct spa_handle *handle)
return 0;
}
static void *add_interface(struct support *support,
const char *factory_name,
uint32_t type,
const struct spa_dict *info)
{
struct spa_handle *handle;
void *iface = NULL;
int res = -ENOENT;
handle = pw_load_spa_handle(support->support_lib,
factory_name, info,
support->n_support, support->support);
if (handle == NULL ||
(res = spa_handle_get_interface(handle, type, &iface)) < 0) {
fprintf(stderr, "can't get %d interface %d\n", type, res);
} else {
support->support[support->n_support++] =
SPA_SUPPORT_INIT(type, iface);
}
return iface;
}
/** Initialize PipeWire
*
* \param argc pointer to argc
@ -335,74 +326,46 @@ SPA_EXPORT
void pw_init(int *argc, char **argv[])
{
const char *str;
struct handle *handle;
void *iface;
struct support *support = &global_support;
struct plugin *plugin;
struct spa_dict_item items[2];
uint32_t n_items;
struct spa_dict info;
struct spa_dict_item items[1];
int res = 0;
struct support *support = &global_support;
struct spa_log *log;
char level[32];
if (support->registry != NULL)
return;
if ((str = getenv("PIPEWIRE_DEBUG")))
configure_debug(support, str);
if ((str = getenv("SPA_PLUGIN_DIR")) == NULL)
str = PLUGINDIR;
support->plugin_dir = str;
if ((str = getenv("SPA_SUPPORT_LIB")) == NULL)
str = SUPPORTLIB;
support->support_lib = str;
spa_list_init(&global_registry.plugins);
support->registry = &global_registry;
if (support->n_support > 0)
return;
n_items = 0;
items[n_items++] = SPA_DICT_ITEM_INIT(SPA_KEY_LOG_COLORS, "1");
snprintf(level, sizeof(level), "%d", pw_log_level);
items[n_items++] = SPA_DICT_ITEM_INIT(SPA_KEY_LOG_LEVEL, level);
info = SPA_DICT_INIT(items, n_items);
plugin = open_plugin(support->registry, support->plugin_dir, "support/libspa-support");
if (plugin == NULL) {
fprintf(stderr, "can't open support library");
return;
}
log = add_interface(support, "logger", SPA_TYPE_INTERFACE_Log, &info);
if (log)
pw_log_set(log);
support->support_plugin = plugin;
n_items = 0;
if ((str = getenv("PIPEWIRE_CPU")))
items[n_items++] = SPA_DICT_ITEM_INIT(SPA_KEY_CPU_FORCE, str);
info = SPA_DICT_INIT(items, n_items);
items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_LOG_COLORS, "1");
info = SPA_DICT_INIT(items, 1);
handle = load_handle(plugin, "logger", &info, support->n_support, support->support);
if (handle == NULL ||
(res = spa_handle_get_interface(&handle->handle,
SPA_TYPE_INTERFACE_Log, &iface)) < 0) {
fprintf(stderr, "can't get Log interface %d\n", res);
}
else {
support->support[support->n_support++] =
SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_Log, iface);
pw_log_set(iface);
}
handle = load_handle(plugin, "system", NULL, support->n_support, support->support);
if (handle == NULL ||
(res = spa_handle_get_interface(&handle->handle,
SPA_TYPE_INTERFACE_System, &iface)) < 0) {
fprintf(stderr, "can't get System interface %d\n", res);
}
else {
support->support[support->n_support++] =
SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_System, iface);
}
handle = load_handle(plugin, "cpu", NULL, support->n_support, support->support);
if (handle == NULL ||
(res = spa_handle_get_interface(&handle->handle,
SPA_TYPE_INTERFACE_CPU, &iface)) < 0) {
fprintf(stderr, "can't get CPU interface %d\n", res);
}
else {
struct spa_cpu *cpu = iface;
if ((str = getenv("PIPEWIRE_CPU")))
spa_cpu_force_flags(cpu, strtoul(str, NULL, 0));
support->support[support->n_support++] =
SPA_SUPPORT_INIT(SPA_TYPE_INTERFACE_CPU, iface);
}
add_interface(support, "cpu", SPA_TYPE_INTERFACE_CPU, &info);
pw_log_info("version %s", pw_get_library_version());
}

9
src/pipewire/pipewire.h
View file

@ -137,8 +137,7 @@ pw_fill_stream_properties(struct pw_core *core, struct pw_properties *properties
enum pw_direction
pw_direction_reverse(enum pw_direction direction);
void *
pw_get_support_interface(uint32_t type);
uint32_t pw_get_support(struct spa_support *support, uint32_t max_support);
struct spa_handle *pw_load_spa_handle(const char *lib,
const char *factory_name,
@ -148,12 +147,6 @@ struct spa_handle *pw_load_spa_handle(const char *lib,
int pw_unload_spa_handle(struct spa_handle *handle);
const struct spa_handle_factory *
pw_get_support_factory(const char *factory_name);
const struct spa_support *
pw_get_support(uint32_t *n_support);
#ifdef __cplusplus
}
#endif

1
src/pipewire/private.h
View file

@ -232,6 +232,7 @@ struct pw_core {
struct pw_loop *main_loop; /**< main loop for control */
struct pw_loop *data_loop; /**< data loop for data passing */
struct pw_data_loop *data_loop_impl;
struct spa_system *data_system; /**< data system for data passing */
struct spa_support support[16]; /**< support for spa plugins */
uint32_t n_support; /**< number of support items */

2
src/tools/pipewire-cli.c
View file

@ -1312,7 +1312,7 @@ static bool parse(struct data *data, char *buf, size_t size, char **error)
return false;
}
static void do_input(void *data, int fd, enum spa_io mask)
static void do_input(void *data, int fd, uint32_t mask)
{
struct data *d = data;
char buf[4096], *error;