[pulseaudio-commits] r1855 - in /branches/lennart/src: Makefile.am modules/rtp/module-rtp-recv.c modules/rtp/module-rtp-send.c modules/rtp/rtp.c modules/rtp/sap.c modules/rtp/sdp.c pulsecore/rtpoll.h
svnmailer-noreply at 0pointer.de
svnmailer-noreply at 0pointer.de
Tue Sep 18 10:41:52 PDT 2007
Author: lennart
Date: Tue Sep 18 19:41:51 2007
New Revision: 1855
URL: http://0pointer.de/cgi-bin/viewcvs.cgi?rev=1855&root=pulseaudio&view=rev
Log:
port module-rtp-send.c to lock-free core
Modified:
branches/lennart/src/Makefile.am
branches/lennart/src/modules/rtp/module-rtp-recv.c
branches/lennart/src/modules/rtp/module-rtp-send.c
branches/lennart/src/modules/rtp/rtp.c
branches/lennart/src/modules/rtp/sap.c
branches/lennart/src/modules/rtp/sdp.c
branches/lennart/src/pulsecore/rtpoll.h
Modified: branches/lennart/src/Makefile.am
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/Makefile.am?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/Makefile.am (original)
+++ branches/lennart/src/Makefile.am Tue Sep 18 19:41:51 2007
@@ -947,11 +947,11 @@
# module-esound-sink.la
# See comment at librtp.la above
-#if !OS_IS_WIN32
-#modlibexec_LTLIBRARIES += \
-# module-rtp-send.la \
-# module-rtp-recv.la
-#endif
+if !OS_IS_WIN32
+modlibexec_LTLIBRARIES += \
+ module-rtp-send.la \
+ module-rtp-recv.la
+endif
if HAVE_AF_UNIX
modlibexec_LTLIBRARIES += \
@@ -1342,15 +1342,15 @@
module_suspend_on_idle_la_CFLAGS = $(AM_CFLAGS)
# RTP modules
-#module_rtp_send_la_SOURCES = modules/rtp/module-rtp-send.c
-#module_rtp_send_la_LDFLAGS = -module -avoid-version
-#module_rtp_send_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
-#module_rtp_send_la_CFLAGS = $(AM_CFLAGS)
-
-#module_rtp_recv_la_SOURCES = modules/rtp/module-rtp-recv.c
-#module_rtp_recv_la_LDFLAGS = -module -avoid-version
-#module_rtp_recv_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
-#module_rtp_recv_la_CFLAGS = $(AM_CFLAGS)
+module_rtp_send_la_SOURCES = modules/rtp/module-rtp-send.c
+module_rtp_send_la_LDFLAGS = -module -avoid-version
+module_rtp_send_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
+module_rtp_send_la_CFLAGS = $(AM_CFLAGS)
+
+module_rtp_recv_la_SOURCES = modules/rtp/module-rtp-recv.c
+module_rtp_recv_la_LDFLAGS = -module -avoid-version
+module_rtp_recv_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
+module_rtp_recv_la_CFLAGS = $(AM_CFLAGS)
# JACK
Modified: branches/lennart/src/modules/rtp/module-rtp-recv.c
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/modules/rtp/module-rtp-recv.c?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/modules/rtp/module-rtp-recv.c (original)
+++ branches/lennart/src/modules/rtp/module-rtp-recv.c Tue Sep 18 19:41:51 2007
@@ -24,7 +24,6 @@
#include <config.h>
#endif
-#include <assert.h>
#include <stdio.h>
#include <sys/socket.h>
#include <netinet/in.h>
@@ -47,6 +46,10 @@
#include <pulsecore/modargs.h>
#include <pulsecore/namereg.h>
#include <pulsecore/sample-util.h>
+#include <pulsecore/macro.h>
+#include <pulsecore/atomic.h>
+#include <pulsecore/rtclock.h>
+#include <pulsecore/atomic.h>
#include "module-rtp-recv-symdef.h"
@@ -66,7 +69,7 @@
#define DEFAULT_SAP_ADDRESS "224.0.0.56"
#define MEMBLOCKQ_MAXLENGTH (1024*170)
#define MAX_SESSIONS 16
-#define DEATH_TIMEOUT 20000000
+#define DEATH_TIMEOUT 20
static const char* const valid_modargs[] = {
"sink",
@@ -76,102 +79,113 @@
struct session {
struct userdata *userdata;
+ PA_LLIST_FIELDS(struct session);
pa_sink_input *sink_input;
pa_memblockq *memblockq;
- pa_time_event *death_event;
-
- int first_packet;
+ pa_bool_t first_packet;
uint32_t ssrc;
uint32_t offset;
struct pa_sdp_info sdp_info;
pa_rtp_context rtp_context;
- pa_io_event* rtp_event;
+
+ pa_rtpoll_item *rtpoll_item;
+
+ pa_atomic_t timestamp;
};
struct userdata {
pa_module *module;
- pa_core *core;
pa_sap_context sap_context;
pa_io_event* sap_event;
+ pa_time_event *check_death_event;
+
+ char *sink_name;
+
+ PA_LLIST_HEAD(struct session, sessions);
pa_hashmap *by_origin;
-
- char *sink_name;
-
int n_sessions;
};
-static void session_free(struct session *s, int from_hash);
-
-static int sink_input_peek(pa_sink_input *i, pa_memchunk *chunk) {
- struct session *s;
- assert(i);
- s = i->userdata;
+static void session_free(struct session *s);
+
+/* Called from I/O thread context */
+static int sink_input_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
+ struct session *s = PA_SINK_INPUT(o)->userdata;
+
+ switch (code) {
+ case PA_SINK_INPUT_MESSAGE_GET_LATENCY:
+ *((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
+
+ /* Fall through, the default handler will add in the extra
+ * latency added by the resampler */
+ break;
+ }
+
+ return pa_sink_input_process_msg(o, code, data, offset, chunk);
+}
+
+/* Called from I/O thread context */
+static int sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
+ struct session *s;
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
return pa_memblockq_peek(s->memblockq, chunk);
}
-static void sink_input_drop(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
- struct session *s;
- assert(i);
- s = i->userdata;
-
- pa_memblockq_drop(s->memblockq, chunk, length);
-}
-
+/* Called from I/O thread context */
+static void sink_input_drop(pa_sink_input *i, size_t length) {
+ struct session *s;
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
+
+ pa_memblockq_drop(s->memblockq, length);
+}
+
+/* Called from main context */
static void sink_input_kill(pa_sink_input* i) {
struct session *s;
- assert(i);
- s = i->userdata;
-
- session_free(s, 1);
-}
-
-static pa_usec_t sink_input_get_latency(pa_sink_input *i) {
- struct session *s;
- assert(i);
- s = i->userdata;
-
- return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
-}
-
-static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
- struct session *s = userdata;
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
+
+ session_free(s);
+}
+
+/* Called from I/O thread context */
+static int rtpoll_work_cb(pa_rtpoll_item *i) {
pa_memchunk chunk;
int64_t k, j, delta;
- struct timeval tv;
-
- assert(m);
- assert(e);
- assert(s);
- assert(fd == s->rtp_context.fd);
- assert(flags == PA_IO_EVENT_INPUT);
-
- if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->core->mempool) < 0)
- return;
+ struct timespec now;
+ struct session *s;
+
+ pa_assert_se(s = pa_rtpoll_item_get_userdata(i));
+
+ if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool) < 0)
+ return 0;
if (s->sdp_info.payload != s->rtp_context.payload) {
pa_memblock_unref(chunk.memblock);
- return;
+ return 0;
}
if (!s->first_packet) {
- s->first_packet = 1;
+ s->first_packet = TRUE;
s->ssrc = s->rtp_context.ssrc;
s->offset = s->rtp_context.timestamp;
- if (s->ssrc == s->userdata->core->cookie)
- pa_log_warn("WARNING! Detected RTP packet loop!");
+ if (s->ssrc == s->userdata->module->core->cookie)
+ pa_log_warn("Detected RTP packet loop!");
} else {
if (s->ssrc != s->rtp_context.ssrc) {
pa_memblock_unref(chunk.memblock);
- return;
+ return 0;
}
}
@@ -197,25 +211,48 @@
pa_memblock_unref(chunk.memblock);
- /* Reset death timer */
- pa_gettimeofday(&tv);
- pa_timeval_add(&tv, DEATH_TIMEOUT);
- m->time_restart(s->death_event, &tv);
-}
-
-static void death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
- struct session *s = userdata;
-
- assert(m);
- assert(t);
- assert(tv);
- assert(s);
-
- session_free(s, 1);
+ pa_rtclock_get(&now);
+ pa_atomic_store(&s->timestamp, now.tv_sec);
+
+ return 1;
+}
+
+/* Called from I/O thread context */
+static void sink_input_attach(pa_sink_input *i) {
+ struct session *s;
+ struct pollfd *p;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
+
+ pa_assert(!s->rtpoll_item);
+ s->rtpoll_item = pa_rtpoll_item_new(i->sink->rtpoll, PA_RTPOLL_LATE, 1);
+
+ p = pa_rtpoll_item_get_pollfd(s->rtpoll_item, NULL);
+ p->fd = s->rtp_context.fd;
+ p->events = POLLIN;
+ p->revents = 0;
+
+ pa_rtpoll_item_set_work_callback(s->rtpoll_item, rtpoll_work_cb);
+ pa_rtpoll_item_set_userdata(s->rtpoll_item, s);
+}
+
+/* Called from I/O thread context */
+static void sink_input_detach(pa_sink_input *i) {
+ struct session *s;
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
+
+ pa_assert(s->rtpoll_item);
+ pa_rtpoll_item_free(s->rtpoll_item);
+ s->rtpoll_item = NULL;
}
static int mcast_socket(const struct sockaddr* sa, socklen_t salen) {
int af, fd = -1, r, one;
+
+ pa_assert(sa);
+ pa_assert(salen > 0);
af = sa->sa_family;
if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) {
@@ -262,27 +299,34 @@
static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) {
struct session *s = NULL;
- struct timeval tv;
char *c;
pa_sink *sink;
int fd = -1;
pa_memblock *silence;
pa_sink_input_new_data data;
+ struct timespec now;
+
+ pa_assert(u);
+ pa_assert(sdp_info);
if (u->n_sessions >= MAX_SESSIONS) {
- pa_log("session limit reached.");
- goto fail;
- }
-
- if (!(sink = pa_namereg_get(u->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
- pa_log("sink does not exist.");
+ pa_log("Session limit reached.");
+ goto fail;
+ }
+
+ if (!(sink = pa_namereg_get(u->module->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
+ pa_log("Sink does not exist.");
goto fail;
}
s = pa_xnew0(struct session, 1);
s->userdata = u;
- s->first_packet = 0;
+ s->first_packet = FALSE;
s->sdp_info = *sdp_info;
+ s->rtpoll_item = NULL;
+
+ pa_rtclock_get(&now);
+ pa_atomic_store(&s->timestamp, now.tv_sec);
if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0)
goto fail;
@@ -299,25 +343,27 @@
data.module = u->module;
pa_sink_input_new_data_set_sample_spec(&data, &sdp_info->sample_spec);
- s->sink_input = pa_sink_input_new(u->core, &data, 0);
+ s->sink_input = pa_sink_input_new(u->module->core, &data, 0);
pa_xfree(c);
if (!s->sink_input) {
- pa_log("failed to create sink input.");
+ pa_log("Failed to create sink input.");
goto fail;
}
s->sink_input->userdata = s;
+ s->sink_input->parent.process_msg = sink_input_process_msg;
s->sink_input->peek = sink_input_peek;
s->sink_input->drop = sink_input_drop;
s->sink_input->kill = sink_input_kill;
- s->sink_input->get_latency = sink_input_get_latency;
-
- silence = pa_silence_memblock_new(s->userdata->core->mempool,
- &s->sink_input->sample_spec,
- (pa_bytes_per_second(&s->sink_input->sample_spec)/128/pa_frame_size(&s->sink_input->sample_spec))*
- pa_frame_size(&s->sink_input->sample_spec));
+ s->sink_input->attach = sink_input_attach;
+ s->sink_input->detach = sink_input_detach;
+
+ silence = pa_silence_memblock_new(
+ s->userdata->module->core->mempool,
+ &s->sink_input->sample_spec,
+ pa_frame_align(pa_bytes_per_second(&s->sink_input->sample_spec)/128, &s->sink_input->sample_spec));
s->memblockq = pa_memblockq_new(
0,
@@ -330,53 +376,43 @@
pa_memblock_unref(silence);
- s->rtp_event = u->core->mainloop->io_new(u->core->mainloop, fd, PA_IO_EVENT_INPUT, rtp_event_cb, s);
-
- pa_gettimeofday(&tv);
- pa_timeval_add(&tv, DEATH_TIMEOUT);
- s->death_event = u->core->mainloop->time_new(u->core->mainloop, &tv, death_event_cb, s);
+ pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
-
- pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
-
- pa_log_info("Found new session '%s'", s->sdp_info.session_name);
-
u->n_sessions++;
+ PA_LLIST_PREPEND(struct session, s->userdata->sessions, s);
+
+ pa_sink_input_put(s->sink_input);
+
+ pa_log_info("New session '%s'", s->sdp_info.session_name);
return s;
fail:
- if (s) {
- if (fd >= 0)
- close(fd);
-
- pa_xfree(s);
- }
-
+ pa_xfree(s);
+
+ if (fd >= 0)
+ pa_close(fd);
+
return NULL;
}
-static void session_free(struct session *s, int from_hash) {
- assert(s);
+static void session_free(struct session *s) {
+ pa_assert(s);
pa_log_info("Freeing session '%s'", s->sdp_info.session_name);
- s->userdata->core->mainloop->time_free(s->death_event);
- s->userdata->core->mainloop->io_free(s->rtp_event);
-
- if (from_hash)
- pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
-
- pa_sink_input_disconnect(s->sink_input);
+ pa_sink_input_unlink(s->sink_input);
pa_sink_input_unref(s->sink_input);
pa_memblockq_free(s->memblockq);
pa_sdp_info_destroy(&s->sdp_info);
pa_rtp_context_destroy(&s->rtp_context);
- assert(s->userdata->n_sessions >= 1);
+ PA_LLIST_REMOVE(struct session, s->userdata->sessions, s);
+ pa_assert(s->userdata->n_sessions >= 1);
s->userdata->n_sessions--;
+ pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
pa_xfree(s);
}
@@ -387,11 +423,11 @@
pa_sdp_info info;
struct session *s;
- assert(m);
- assert(e);
- assert(u);
- assert(fd == u->sap_context.fd);
- assert(flags == PA_IO_EVENT_INPUT);
+ pa_assert(m);
+ pa_assert(e);
+ pa_assert(u);
+ pa_assert(fd == u->sap_context.fd);
+ pa_assert(flags == PA_IO_EVENT_INPUT);
if (pa_sap_recv(&u->sap_context, &goodbye) < 0)
return;
@@ -402,7 +438,7 @@
if (goodbye) {
if ((s = pa_hashmap_get(u->by_origin, info.origin)))
- session_free(s, 1);
+ session_free(s);
pa_sdp_info_destroy(&info);
} else {
@@ -412,18 +448,47 @@
pa_sdp_info_destroy(&info);
} else {
- struct timeval tv;
-
- pa_gettimeofday(&tv);
- pa_timeval_add(&tv, DEATH_TIMEOUT);
- m->time_restart(s->death_event, &tv);
-
+ struct timespec now;
+ pa_rtclock_get(&now);
+ pa_atomic_store(&s->timestamp, now.tv_sec);
+
pa_sdp_info_destroy(&info);
}
}
}
-int pa__init(pa_core *c, pa_module*m) {
+static void check_death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *ptv, void *userdata) {
+ struct session *s, *n;
+ struct userdata *u = userdata;
+ struct timespec now;
+ struct timeval tv;
+
+ pa_assert(m);
+ pa_assert(t);
+ pa_assert(ptv);
+ pa_assert(s);
+
+ pa_rtclock_get(&now);
+
+ pa_log_debug("Checking for dead streams ...");
+
+ for (s = u->sessions; s; s = n) {
+ int k;
+ n = s->next;
+
+ k = pa_atomic_load(&s->timestamp);
+
+ if (k + DEATH_TIMEOUT < now.tv_sec)
+ session_free(s);
+ }
+
+ /* Restart timer */
+ pa_gettimeofday(&tv);
+ pa_timeval_add(&tv, DEATH_TIMEOUT*PA_USEC_PER_SEC);
+ m->time_restart(t, &tv);
+}
+
+int pa__init(pa_module*m) {
struct userdata *u;
pa_modargs *ma = NULL;
struct sockaddr_in sa4;
@@ -432,9 +497,9 @@
socklen_t salen;
const char *sap_address;
int fd = -1;
-
- assert(c);
- assert(m);
+ struct timeval tv;
+
+ pa_assert(m);
if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
pa_log("failed to parse module arguments");
@@ -454,7 +519,7 @@
sa = (struct sockaddr*) &sa4;
salen = sizeof(sa4);
} else {
- pa_log("invalid SAP address '%s'", sap_address);
+ pa_log("Invalid SAP address '%s'", sap_address);
goto fail;
}
@@ -464,16 +529,19 @@
u = pa_xnew(struct userdata, 1);
m->userdata = u;
u->module = m;
- u->core = c;
u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
u->n_sessions = 0;
- u->sap_event = c->mainloop->io_new(c->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u);
+ u->sap_event = m->core->mainloop->io_new(m->core->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u);
u->by_origin = pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func);
pa_sap_context_init_recv(&u->sap_context, fd);
+ pa_gettimeofday(&tv);
+ pa_timeval_add(&tv, DEATH_TIMEOUT * PA_USEC_PER_SEC);
+ u->check_death_event = m->core->mainloop->time_new(m->core->mainloop, &tv, check_death_event_cb, u);
+
pa_modargs_free(ma);
return 0;
@@ -483,27 +551,34 @@
pa_modargs_free(ma);
if (fd >= 0)
- close(fd);
+ pa_close(fd);
return -1;
}
-static void free_func(void *p, PA_GCC_UNUSED void *userdata) {
- session_free(p, 0);
-}
-
-void pa__done(pa_core *c, pa_module*m) {
+void pa__done(pa_module*m) {
struct userdata *u;
- assert(c);
- assert(m);
+ struct session *s;
+
+ pa_assert(m);
if (!(u = m->userdata))
return;
- c->mainloop->io_free(u->sap_event);
+ if (u->sap_event)
+ m->core->mainloop->io_free(u->sap_event);
+
+ if (u->check_death_event)
+ m->core->mainloop->time_free(u->check_death_event);
+
pa_sap_context_destroy(&u->sap_context);
- pa_hashmap_free(u->by_origin, free_func, NULL);
+ if (u->by_origin) {
+ while ((s = pa_hashmap_get_first(u->by_origin)))
+ session_free(s);
+
+ pa_hashmap_free(u->by_origin, NULL, NULL);
+ }
pa_xfree(u->sink_name);
pa_xfree(u);
Modified: branches/lennart/src/modules/rtp/module-rtp-send.c
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/modules/rtp/module-rtp-send.c?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/modules/rtp/module-rtp-send.c (original)
+++ branches/lennart/src/modules/rtp/module-rtp-send.c Tue Sep 18 19:41:51 2007
@@ -25,7 +25,6 @@
#include <config.h>
#endif
-#include <assert.h>
#include <stdio.h>
#include <sys/socket.h>
#include <netinet/in.h>
@@ -48,6 +47,8 @@
#include <pulsecore/core-util.h>
#include <pulsecore/modargs.h>
#include <pulsecore/namereg.h>
+#include <pulsecore/sample-util.h>
+#include <pulsecore/macro.h>
#include "module-rtp-send-symdef.h"
@@ -74,7 +75,7 @@
#define DEFAULT_DESTINATION "224.0.0.56"
#define MEMBLOCKQ_MAXLENGTH (1024*170)
#define DEFAULT_MTU 1280
-#define SAP_INTERVAL 5000000
+#define SAP_INTERVAL 5
static const char* const valid_modargs[] = {
"source",
@@ -90,7 +91,6 @@
struct userdata {
pa_module *module;
- pa_core *core;
pa_source_output *source_output;
pa_memblockq *memblockq;
@@ -102,56 +102,67 @@
pa_time_event *sap_event;
};
+/* Called from I/O thread context */
+static int source_output_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
+ struct userdata *u;
+ pa_assert_se(u = PA_SOURCE_OUTPUT(o)->userdata);
+
+ switch (code) {
+ case PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY:
+ *((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &u->source_output->sample_spec);
+
+ /* Fall through, the default handler will add in the extra
+ * latency added by the resampler */
+ break;
+ }
+
+ return pa_source_output_process_msg(o, code, data, offset, chunk);
+}
+
+/* Called from I/O thread context */
static void source_output_push(pa_source_output *o, const pa_memchunk *chunk) {
struct userdata *u;
- assert(o);
- u = o->userdata;
+ pa_source_output_assert_ref(o);
+ pa_assert_se(u = o->userdata);
if (pa_memblockq_push(u->memblockq, chunk) < 0) {
- pa_log("Failed to push chunk into memblockq.");
+ pa_log_warn("Failed to push chunk into memblockq.");
return;
}
pa_rtp_send(&u->rtp_context, u->mtu, u->memblockq);
}
+/* Called from main context */
static void source_output_kill(pa_source_output* o) {
struct userdata *u;
- assert(o);
- u = o->userdata;
+ pa_source_output_assert_ref(o);
+ pa_assert_se(u = o->userdata);
pa_module_unload_request(u->module);
- pa_source_output_disconnect(u->source_output);
+ pa_source_output_unlink(u->source_output);
pa_source_output_unref(u->source_output);
u->source_output = NULL;
-}
-
-static pa_usec_t source_output_get_latency (pa_source_output *o) {
- struct userdata *u;
- assert(o);
- u = o->userdata;
-
- return pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &o->sample_spec);
}
static void sap_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
struct userdata *u = userdata;
struct timeval next;
- assert(m);
- assert(t);
- assert(tv);
- assert(u);
+ pa_assert(m);
+ pa_assert(t);
+ pa_assert(tv);
+ pa_assert(u);
pa_sap_send(&u->sap_context, 0);
pa_gettimeofday(&next);
- pa_timeval_add(&next, SAP_INTERVAL);
+ pa_timeval_add(&next, SAP_INTERVAL * PA_USEC_PER_SEC);
m->time_restart(t, &next);
}
-int pa__init(pa_core *c, pa_module*m) {
+int pa__init(pa_module*m) {
struct userdata *u;
pa_modargs *ma = NULL;
const char *dest;
@@ -173,21 +184,20 @@
int loop = 0;
pa_source_output_new_data data;
- assert(c);
- assert(m);
+ pa_assert(m);
if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
- pa_log("failed to parse module arguments");
+ pa_log("Failed to parse module arguments");
goto fail;
}
if (!(s = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source", NULL), PA_NAMEREG_SOURCE, 1))) {
- pa_log("source does not exist.");
+ pa_log("Source does not exist.");
goto fail;
}
if (pa_modargs_get_value_boolean(ma, "loop", &loop) < 0) {
- pa_log("failed to parse \"loop\" parameter.");
+ pa_log("Failed to parse \"loop\" parameter.");
goto fail;
}
@@ -195,12 +205,12 @@
pa_rtp_sample_spec_fixup(&ss);
cm = s->channel_map;
if (pa_modargs_get_sample_spec(ma, &ss) < 0) {
- pa_log("failed to parse sample specification");
+ pa_log("Failed to parse sample specification");
goto fail;
}
if (!pa_rtp_sample_spec_valid(&ss)) {
- pa_log("specified sample type not compatible with RTP");
+ pa_log("Specified sample type not compatible with RTP");
goto fail;
}
@@ -209,10 +219,10 @@
payload = pa_rtp_payload_from_sample_spec(&ss);
- mtu = (DEFAULT_MTU/pa_frame_size(&ss))*pa_frame_size(&ss);
+ mtu = pa_frame_align(DEFAULT_MTU, &ss);
if (pa_modargs_get_value_u32(ma, "mtu", &mtu) < 0 || mtu < 1 || mtu % pa_frame_size(&ss) != 0) {
- pa_log("invalid mtu.");
+ pa_log("Invalid MTU.");
goto fail;
}
@@ -223,7 +233,7 @@
}
if (port & 1)
- pa_log_warn("WARNING: port number not even as suggested in RFC3550!");
+ pa_log_warn("Port number not even as suggested in RFC3550!");
dest = pa_modargs_get_value(ma, "destination", DEFAULT_DESTINATION);
@@ -238,7 +248,7 @@
sap_sa4 = sa4;
sap_sa4.sin_port = htons(SAP_PORT);
} else {
- pa_log("invalid destination '%s'", dest);
+ pa_log("Invalid destination '%s'", dest);
goto fail;
}
@@ -267,6 +277,9 @@
pa_log("IP_MULTICAST_LOOP failed: %s", pa_cstrerror(errno));
goto fail;
}
+
+ /* If the socket queue is full, let's drop packets */
+ pa_make_nonblock_fd(fd);
pa_source_output_new_data_init(&data);
data.name = "RTP Monitor Stream";
@@ -276,21 +289,20 @@
pa_source_output_new_data_set_sample_spec(&data, &ss);
pa_source_output_new_data_set_channel_map(&data, &cm);
- if (!(o = pa_source_output_new(c, &data, 0))) {
+ if (!(o = pa_source_output_new(m->core, &data, 0))) {
pa_log("failed to create source output.");
goto fail;
}
+ o->parent.process_msg = source_output_process_msg;
o->push = source_output_push;
o->kill = source_output_kill;
- o->get_latency = source_output_get_latency;
u = pa_xnew(struct userdata, 1);
m->userdata = u;
o->userdata = u;
u->module = m;
- u->core = c;
u->source_output = o;
u->memblockq = pa_memblockq_new(
@@ -305,8 +317,7 @@
u->mtu = mtu;
k = sizeof(sa_dst);
- r = getsockname(fd, (struct sockaddr*) &sa_dst, &k);
- assert(r >= 0);
+ pa_assert_se((r = getsockname(fd, (struct sockaddr*) &sa_dst, &k)) >= 0);
n = pa_sprintf_malloc("PulseAudio RTP Stream on %s", pa_get_fqdn(hn, sizeof(hn)));
@@ -317,18 +328,20 @@
pa_xfree(n);
- pa_rtp_context_init_send(&u->rtp_context, fd, c->cookie, payload, pa_frame_size(&ss));
+ pa_rtp_context_init_send(&u->rtp_context, fd, m->core->cookie, payload, pa_frame_size(&ss));
pa_sap_context_init_send(&u->sap_context, sap_fd, p);
pa_log_info("RTP stream initialized with mtu %u on %s:%u, SSRC=0x%08x, payload=%u, initial sequence #%u", mtu, dest, port, u->rtp_context.ssrc, payload, u->rtp_context.sequence);
- pa_log_info("SDP-Data:\n%s\n"__FILE__": EOF", p);
+ pa_log_info("SDP-Data:\n%s\nEOF", p);
pa_sap_send(&u->sap_context, 0);
pa_gettimeofday(&tv);
- pa_timeval_add(&tv, SAP_INTERVAL);
- u->sap_event = c->mainloop->time_new(c->mainloop, &tv, sap_event_cb, u);
-
+ pa_timeval_add(&tv, SAP_INTERVAL * PA_USEC_PER_SEC);
+ u->sap_event = m->core->mainloop->time_new(m->core->mainloop, &tv, sap_event_cb, u);
+
+ pa_source_output_put(u->source_output);
+
pa_modargs_free(ma);
return 0;
@@ -338,31 +351,31 @@
pa_modargs_free(ma);
if (fd >= 0)
- close(fd);
+ pa_close(fd);
if (sap_fd >= 0)
- close(sap_fd);
+ pa_close(sap_fd);
if (o) {
- pa_source_output_disconnect(o);
+ pa_source_output_unlink(o);
pa_source_output_unref(o);
}
return -1;
}
-void pa__done(pa_core *c, pa_module*m) {
- struct userdata *u;
- assert(c);
- assert(m);
+void pa__done(pa_module*m) {
+ struct userdata *u;
+ pa_assert(m);
if (!(u = m->userdata))
return;
- c->mainloop->time_free(u->sap_event);
+ if (u->sap_event)
+ m->core->mainloop->time_free(u->sap_event);
if (u->source_output) {
- pa_source_output_disconnect(u->source_output);
+ pa_source_output_unlink(u->source_output);
pa_source_output_unref(u->source_output);
}
@@ -371,7 +384,8 @@
pa_sap_send(&u->sap_context, 1);
pa_sap_context_destroy(&u->sap_context);
- pa_memblockq_free(u->memblockq);
+ if (u->memblockq)
+ pa_memblockq_free(u->memblockq);
pa_xfree(u);
}
Modified: branches/lennart/src/modules/rtp/rtp.c
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/modules/rtp/rtp.c?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/modules/rtp/rtp.c (original)
+++ branches/lennart/src/modules/rtp/rtp.c Tue Sep 18 19:41:51 2007
@@ -25,7 +25,6 @@
#include <config.h>
#endif
-#include <assert.h>
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
@@ -40,12 +39,14 @@
#include <pulsecore/core-error.h>
#include <pulsecore/log.h>
+#include <pulsecore/macro.h>
+#include <pulsecore/core-util.h>
#include "rtp.h"
pa_rtp_context* pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint32_t ssrc, uint8_t payload, size_t frame_size) {
- assert(c);
- assert(fd >= 0);
+ pa_assert(c);
+ pa_assert(fd >= 0);
c->fd = fd;
c->sequence = (uint16_t) (rand()*rand());
@@ -63,11 +64,11 @@
struct iovec iov[MAX_IOVECS];
pa_memblock* mb[MAX_IOVECS];
int iov_idx = 1;
- size_t n = 0, skip = 0;
-
- assert(c);
- assert(size > 0);
- assert(q);
+ size_t n = 0;
+
+ pa_assert(c);
+ pa_assert(size > 0);
+ pa_assert(q);
if (pa_memblockq_get_length(q) < size)
return 0;
@@ -76,24 +77,26 @@
int r;
pa_memchunk chunk;
+ pa_memchunk_reset(&chunk);
+
if ((r = pa_memblockq_peek(q, &chunk)) >= 0) {
size_t k = n + chunk.length > size ? size - n : chunk.length;
- if (chunk.memblock) {
- iov[iov_idx].iov_base = (void*)((uint8_t*) pa_memblock_acquire(chunk.memblock) + chunk.index);
- iov[iov_idx].iov_len = k;
- mb[iov_idx] = chunk.memblock;
- iov_idx ++;
-
- n += k;
- }
-
- skip += k;
+ pa_assert(chunk.memblock);
+
+ iov[iov_idx].iov_base = ((uint8_t*) pa_memblock_acquire(chunk.memblock) + chunk.index);
+ iov[iov_idx].iov_len = k;
+ mb[iov_idx] = chunk.memblock;
+ iov_idx ++;
+
+ n += k;
pa_memblockq_drop(q, k);
}
- if (r < 0 || !chunk.memblock || n >= size || iov_idx >= MAX_IOVECS) {
+ pa_assert(n % c->frame_size == 0);
+
+ if (r < 0 || n >= size || iov_idx >= MAX_IOVECS) {
uint32_t header[3];
struct msghdr m;
int k, i;
@@ -125,10 +128,10 @@
} else
k = 0;
- c->timestamp += skip/c->frame_size;
+ c->timestamp += n/c->frame_size;
if (k < 0) {
- if (errno != EAGAIN) /* If the queue is full, just ignore it */
+ if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */
pa_log("sendmsg() failed: %s", pa_cstrerror(errno));
return -1;
}
@@ -137,7 +140,6 @@
break;
n = 0;
- skip = 0;
iov_idx = 1;
}
}
@@ -146,7 +148,7 @@
}
pa_rtp_context* pa_rtp_context_init_recv(pa_rtp_context *c, int fd, size_t frame_size) {
- assert(c);
+ pa_assert(c);
c->fd = fd;
c->frame_size = frame_size;
@@ -161,13 +163,13 @@
int cc;
ssize_t r;
- assert(c);
- assert(chunk);
+ pa_assert(c);
+ pa_assert(chunk);
chunk->memblock = NULL;
if (ioctl(c->fd, FIONREAD, &size) < 0) {
- pa_log("FIONREAD failed: %s", pa_cstrerror(errno));
+ pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno));
goto fail;
}
@@ -188,12 +190,14 @@
m.msg_flags = 0;
if ((r = recvmsg(c->fd, &m, 0)) != size) {
- pa_log("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
+ if (r < 0 && errno != EAGAIN && errno != EINTR)
+ pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
+
goto fail;
}
if (size < 12) {
- pa_log("RTP packet too short.");
+ pa_log_warn("RTP packet too short.");
goto fail;
}
@@ -206,17 +210,17 @@
c->ssrc = ntohl(c->ssrc);
if ((header >> 30) != 2) {
- pa_log("Unsupported RTP version.");
+ pa_log_warn("Unsupported RTP version.");
goto fail;
}
if ((header >> 29) & 1) {
- pa_log("RTP padding not supported.");
+ pa_log_warn("RTP padding not supported.");
goto fail;
}
if ((header >> 28) & 1) {
- pa_log("RTP header extensions not supported.");
+ pa_log_warn("RTP header extensions not supported.");
goto fail;
}
@@ -225,7 +229,7 @@
c->sequence = header & 0xFFFF;
if (12 + cc*4 > size) {
- pa_log("RTP packet too short. (CSRC)");
+ pa_log_warn("RTP packet too short. (CSRC)");
goto fail;
}
@@ -233,7 +237,7 @@
chunk->length = size - chunk->index;
if (chunk->length % c->frame_size != 0) {
- pa_log("Vad RTP packet size.");
+ pa_log_warn("Bad RTP packet size.");
goto fail;
}
@@ -249,7 +253,7 @@
}
uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) {
- assert(ss);
+ pa_assert(ss);
if (ss->format == PA_SAMPLE_ULAW && ss->rate == 8000 && ss->channels == 1)
return 0;
@@ -264,7 +268,7 @@
}
pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec *ss) {
- assert(ss);
+ pa_assert(ss);
switch (payload) {
case 0:
@@ -299,17 +303,17 @@
}
pa_sample_spec *pa_rtp_sample_spec_fixup(pa_sample_spec * ss) {
- assert(ss);
+ pa_assert(ss);
if (!pa_rtp_sample_spec_valid(ss))
ss->format = PA_SAMPLE_S16BE;
- assert(pa_rtp_sample_spec_valid(ss));
+ pa_assert(pa_rtp_sample_spec_valid(ss));
return ss;
}
int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
- assert(ss);
+ pa_assert(ss);
if (!pa_sample_spec_valid(ss))
return 0;
@@ -322,9 +326,9 @@
}
void pa_rtp_context_destroy(pa_rtp_context *c) {
- assert(c);
-
- close(c->fd);
+ pa_assert(c);
+
+ pa_close(c->fd);
}
const char* pa_rtp_format_to_string(pa_sample_format_t f) {
@@ -343,7 +347,7 @@
}
pa_sample_format_t pa_rtp_string_to_format(const char *s) {
- assert(s);
+ pa_assert(s);
if (!(strcmp(s, "L16")))
return PA_SAMPLE_S16BE;
Modified: branches/lennart/src/modules/rtp/sap.c
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/modules/rtp/sap.c?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/modules/rtp/sap.c (original)
+++ branches/lennart/src/modules/rtp/sap.c Tue Sep 18 19:41:51 2007
@@ -25,7 +25,6 @@
#include <config.h>
#endif
-#include <assert.h>
#include <time.h>
#include <stdlib.h>
#include <sys/types.h>
@@ -46,6 +45,7 @@
#include <pulsecore/core-error.h>
#include <pulsecore/core-util.h>
#include <pulsecore/log.h>
+#include <pulsecore/macro.h>
#include "sap.h"
#include "sdp.h"
@@ -53,9 +53,9 @@
#define MIME_TYPE "application/sdp"
pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_data) {
- assert(c);
- assert(fd >= 0);
- assert(sdp_data);
+ pa_assert(c);
+ pa_assert(fd >= 0);
+ pa_assert(sdp_data);
c->fd = fd;
c->sdp_data = sdp_data;
@@ -65,9 +65,9 @@
}
void pa_sap_context_destroy(pa_sap_context *c) {
- assert(c);
-
- close(c->fd);
+ pa_assert(c);
+
+ pa_close(c->fd);
pa_xfree(c->sdp_data);
}
@@ -85,7 +85,7 @@
return -1;
}
- assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6);
+ pa_assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6);
header = htonl(((uint32_t) 1 << 29) |
(sa->sa_family == AF_INET6 ? (uint32_t) 1 << 28 : 0) |
@@ -113,14 +113,14 @@
m.msg_flags = 0;
if ((k = sendmsg(c->fd, &m, MSG_DONTWAIT)) < 0)
- pa_log("sendmsg() failed: %s\n", pa_cstrerror(errno));
+ pa_log_warn("sendmsg() failed: %s\n", pa_cstrerror(errno));
return k;
}
pa_sap_context* pa_sap_context_init_recv(pa_sap_context *c, int fd) {
- assert(c);
- assert(fd >= 0);
+ pa_assert(c);
+ pa_assert(fd >= 0);
c->fd = fd;
c->sdp_data = NULL;
@@ -136,11 +136,11 @@
int six, ac;
ssize_t r;
- assert(c);
- assert(goodbye);
+ pa_assert(c);
+ pa_assert(goodbye);
if (ioctl(c->fd, FIONREAD, &size) < 0) {
- pa_log("FIONREAD failed: %s", pa_cstrerror(errno));
+ pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno));
goto fail;
}
@@ -159,12 +159,12 @@
m.msg_flags = 0;
if ((r = recvmsg(c->fd, &m, 0)) != size) {
- pa_log("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
+ pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
goto fail;
}
if (size < 4) {
- pa_log("SAP packet too short.");
+ pa_log_warn("SAP packet too short.");
goto fail;
}
@@ -172,17 +172,17 @@
header = ntohl(header);
if (header >> 29 != 1) {
- pa_log("Unsupported SAP version.");
+ pa_log_warn("Unsupported SAP version.");
goto fail;
}
if ((header >> 25) & 1) {
- pa_log("Encrypted SAP not supported.");
+ pa_log_warn("Encrypted SAP not supported.");
goto fail;
}
if ((header >> 24) & 1) {
- pa_log("Compressed SAP not supported.");
+ pa_log_warn("Compressed SAP not supported.");
goto fail;
}
@@ -191,7 +191,7 @@
k = 4 + (six ? 16 : 4) + ac*4;
if (size < k) {
- pa_log("SAP packet too short (AD).");
+ pa_log_warn("SAP packet too short (AD).");
goto fail;
}
@@ -202,7 +202,7 @@
e += sizeof(MIME_TYPE);
size -= sizeof(MIME_TYPE);
} else if ((unsigned) size < sizeof(PA_SDP_HEADER)-1 || strncmp(e, PA_SDP_HEADER, sizeof(PA_SDP_HEADER)-1)) {
- pa_log("Invalid SDP header.");
+ pa_log_warn("Invalid SDP header.");
goto fail;
}
Modified: branches/lennart/src/modules/rtp/sdp.c
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/modules/rtp/sdp.c?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/modules/rtp/sdp.c (original)
+++ branches/lennart/src/modules/rtp/sdp.c Tue Sep 18 19:41:51 2007
@@ -25,7 +25,6 @@
#include <config.h>
#endif
-#include <assert.h>
#include <time.h>
#include <stdlib.h>
#include <sys/types.h>
@@ -38,33 +37,29 @@
#include <pulsecore/core-util.h>
#include <pulsecore/log.h>
+#include <pulsecore/macro.h>
#include "sdp.h"
#include "rtp.h"
-
char *pa_sdp_build(int af, const void *src, const void *dst, const char *name, uint16_t port, uint8_t payload, const pa_sample_spec *ss) {
uint32_t ntp;
- char buf_src[64], buf_dst[64];
+ char buf_src[64], buf_dst[64], un[64];
const char *u, *f, *a;
- assert(src);
- assert(dst);
- assert(af == AF_INET || af == AF_INET6);
-
- f = pa_rtp_format_to_string(ss->format);
- assert(f);
-
- if (!(u = getenv("USER")))
- if (!(u = getenv("USERNAME")))
- u = "-";
+ pa_assert(src);
+ pa_assert(dst);
+ pa_assert(af == AF_INET || af == AF_INET6);
+
+ pa_assert_se(f = pa_rtp_format_to_string(ss->format));
+
+ if (!(u = pa_get_user_name(un, sizeof(un))))
+ u = "-";
ntp = time(NULL) + 2208988800U;
- a = inet_ntop(af, src, buf_src, sizeof(buf_src));
- assert(a);
- a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst));
- assert(a);
+ pa_assert_se(a = inet_ntop(af, src, buf_src, sizeof(buf_src)));
+ pa_assert_se(a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst)));
return pa_sprintf_malloc(
PA_SDP_HEADER
@@ -86,8 +81,8 @@
static pa_sample_spec *parse_sdp_sample_spec(pa_sample_spec *ss, char *c) {
unsigned rate, channels;
- assert(ss);
- assert(c);
+ pa_assert(ss);
+ pa_assert(c);
if (pa_startswith(c, "L16/")) {
ss->format = PA_SAMPLE_S16BE;
@@ -123,8 +118,8 @@
uint16_t port = 0;
int ss_valid = 0;
- assert(t);
- assert(i);
+ pa_assert(t);
+ pa_assert(i);
i->origin = i->session_name = NULL;
i->salen = 0;
@@ -258,7 +253,7 @@
}
void pa_sdp_info_destroy(pa_sdp_info *i) {
- assert(i);
+ pa_assert(i);
pa_xfree(i->origin);
pa_xfree(i->session_name);
Modified: branches/lennart/src/pulsecore/rtpoll.h
URL: http://0pointer.de/cgi-bin/viewcvs.cgi/branches/lennart/src/pulsecore/rtpoll.h?rev=1855&root=pulseaudio&r1=1854&r2=1855&view=diff
==============================================================================
--- branches/lennart/src/pulsecore/rtpoll.h (original)
+++ branches/lennart/src/pulsecore/rtpoll.h Tue Sep 18 19:41:51 2007
@@ -103,7 +103,6 @@
* entered the sleeping poll */
void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i));
-
void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata);
void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i);
More information about the pulseaudio-commits
mailing list