latency work

major main loop bugfix


git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@154 fefdeb5f-60dc-0310-8127-8f9354f1896f

21 files changed, 126 insertions, 61 deletions

diff --git a/polyp/mainloop.c b/polyp/mainloop.c
index 84505bb6f..20d14e517 100644
--- a/polyp/mainloop.c
+++ b/polyp/mainloop.c
@@ -186,6 +186,7 @@ static struct pa_time_event* mainloop_time_new(struct pa_mainloop_api*a, const s
     e->destroy_callback = NULL;
 
     pa_idxset_put(m->time_events, e, NULL);
+    
     return e;
 }
 
@@ -201,6 +202,7 @@ static void mainloop_time_restart(struct pa_time_event *e, const struct timeval
 
 static void mainloop_time_free(struct pa_time_event *e) {
     assert(e);
+
     e->dead = e->mainloop->time_events_scan_dead = 1;
 }
 
@@ -271,7 +273,7 @@ static int io_foreach(void *p, uint32_t index, int *del, void*userdata) {
     int *all = userdata;
     assert(e && del && all);
 
-    if (!*all || !e->dead)
+    if (!*all && !e->dead)
         return 0;
     
     if (e->destroy_callback)
@@ -286,7 +288,7 @@ static int time_foreach(void *p, uint32_t index, int *del, void*userdata) {
     int *all = userdata;
     assert(e && del && all);
 
-    if (!*all || !e->dead)
+    if (!*all && !e->dead)
         return 0;
     
     if (e->destroy_callback)
@@ -301,7 +303,7 @@ static int defer_foreach(void *p, uint32_t index, int *del, void*userdata) {
     int *all = userdata;
     assert(e && del && all);
 
-    if (!*all || !e->dead)
+    if (!*all && !e->dead)
         return 0;
     
     if (e->destroy_callback)
@@ -336,6 +338,8 @@ static void scan_dead(struct pa_mainloop *m) {
         pa_idxset_foreach(m->time_events, time_foreach, &all);
     if (m->defer_events_scan_dead)
         pa_idxset_foreach(m->defer_events, defer_foreach, &all);
+
+    m->io_events_scan_dead = m->time_events_scan_dead = m->defer_events_scan_dead = 0;
 }
 
 static void rebuild_pollfds(struct pa_mainloop *m) {
diff --git a/polyp/module.c b/polyp/module.c
index 1deb7cde8..eb8a8acd6 100644
--- a/polyp/module.c
+++ b/polyp/module.c
@@ -62,10 +62,10 @@ struct pa_module* pa_module_load(struct pa_core *c, const char *name, const char
     if (!(m->dl = lt_dlopenext(name)))
         goto fail;
 
-    if (!(m->init = lt_dlsym(m->dl, "pa_module_init")))
+    if (!(m->init = (int (*)(struct pa_core *c, struct pa_module*m)) lt_dlsym(m->dl, "pa_module_init")))
         goto fail;
 
-    if (!(m->done = lt_dlsym(m->dl, "pa_module_done")))
+    if (!(m->done = (void (*)(struct pa_core *c, struct pa_module*m)) lt_dlsym(m->dl, "pa_module_done")))
         goto fail;
     
     m->userdata = NULL;
diff --git a/polyp/native-common.h b/polyp/native-common.h
index d826837ae..5250532e0 100644
--- a/polyp/native-common.h
+++ b/polyp/native-common.h
@@ -46,15 +46,12 @@ enum {
     PA_COMMAND_RECORD_STREAM_KILLED,
     PA_COMMAND_STAT,
     PA_COMMAND_GET_PLAYBACK_LATENCY,
-    
     PA_COMMAND_CREATE_UPLOAD_STREAM,
     PA_COMMAND_DELETE_UPLOAD_STREAM,
     PA_COMMAND_FINISH_UPLOAD_STREAM,
     PA_COMMAND_PLAY_SAMPLE,
     PA_COMMAND_REMOVE_SAMPLE,
-
     PA_COMMAND_GET_SERVER_INFO,
-    
     PA_COMMAND_GET_SINK_INFO,
     PA_COMMAND_GET_SINK_INFO_LIST,
     PA_COMMAND_GET_SOURCE_INFO,
@@ -78,6 +75,7 @@ enum {
 
     PA_COMMAND_CORK_PLAYBACK_STREAM,
     PA_COMMAND_FLUSH_PLAYBACK_STREAM,
+    PA_COMMAND_TRIGGER_PLAYBACK_STREAM,
     
     PA_COMMAND_MAX
 };
diff --git a/polyp/pacat.c b/polyp/pacat.c
index 198776d35..9efa552a8 100644
--- a/polyp/pacat.c
+++ b/polyp/pacat.c
@@ -277,16 +277,16 @@ static void exit_signal_callback(struct pa_mainloop_api*m, struct pa_signal_even
 }
 
 /* Show the current playback latency */
-static void stream_get_latency_callback(struct pa_stream *s, uint32_t latency, void *userdata) {
+static void stream_get_latency_callback(struct pa_stream *s, const struct pa_latency_info *i, void *userdata) {
     assert(s);
 
-    if (latency == (uint32_t) -1) {
+    if (!i) {
         fprintf(stderr, "Failed to get latency: %s\n", strerror(errno));
         quit(1);
         return;
     }
 
-    fprintf(stderr, "Current latency is %u usecs.\n", latency);
+    fprintf(stderr, "Current latency is %u usecs.\n", i->buffer_usec+i->sink_usec);
 }
 
 /* Someone requested that the latency is shown */
diff --git a/polyp/pdispatch.c b/polyp/pdispatch.c
index c46d4f770..22f5da09f 100644
--- a/polyp/pdispatch.c
+++ b/polyp/pdispatch.c
@@ -60,6 +60,17 @@ static const char *command_names[PA_COMMAND_MAX] = {
     [PA_COMMAND_FINISH_UPLOAD_STREAM] = "FINISH_UPLOAD_STREAM",
     [PA_COMMAND_PLAY_SAMPLE] = "PLAY_SAMPLE",
     [PA_COMMAND_REMOVE_SAMPLE] = "REMOVE_SAMPLE",
+    [PA_COMMAND_GET_SERVER_INFO] = "GET_SERVER_INFO",
+    [PA_COMMAND_GET_SINK_INFO] = "GET_SET_INFO",
+    [PA_COMMAND_GET_SINK_INPUT_INFO] = "GET_SINK_INPUT_INFO",
+    [PA_COMMAND_SUBSCRIBE] = "SUBSCRIBE",
+    [PA_COMMAND_SUBSCRIBE_EVENT] = "SUBSCRIBE_EVENT",
+    [PA_COMMAND_SET_SINK_VOLUME] = "SET_SINK_VOLUME",
+    [PA_COMMAND_SET_SINK_INPUT_VOLUME] = "SET_SINK_INPUT_VOLUME",
+    [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = "TRIGGER_PLAYBACK_STREAM",
+    [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = "FLUSH_PLAYBACK_STREAM",
+    [PA_COMMAND_CORK_PLAYBACK_STREAM] = "CORK_PLAYBACK_STREAM",
+    
 };
 
 #endif
diff --git a/polyp/play-memchunk.c b/polyp/play-memchunk.c
index b94a05242..486f0cf65 100644
--- a/polyp/play-memchunk.c
+++ b/polyp/play-memchunk.c
@@ -25,6 +25,8 @@
 
 #include <stdlib.h>
 #include <assert.h>
+#include <stdio.h>
+#include <string.h>
 
 #include "play-memchunk.h"
 #include "sink-input.h"
@@ -64,7 +66,7 @@ static void sink_input_drop(struct pa_sink_input *i, const struct pa_memchunk*ch
     assert(i && length && i->userdata);
     c = i->userdata;
 
-    assert(chunk == c);
+    assert(!memcmp(chunk, c, sizeof(chunk)));
     assert(length <= c->length);
 
     c->length -= length;
diff --git a/polyp/polypaudio.pa b/polyp/polypaudio.pa
index c63e14795..7c31634dd 100755
--- a/polyp/polypaudio.pa
+++ b/polyp/polypaudio.pa
@@ -30,10 +30,10 @@ load module-pipe-sink
 
 #autoload_sink_add output module-oss device="/dev/dsp" sink_name=output source_name=input
 #autoload_source_add input module-oss device="/dev/dsp" sink_name=output source_name=input
-autoload_sink_add output module-oss-mmap device="/dev/dsp" sink_name=output source_name=input
-autoload_source_add input module-oss-mmap device="/dev/dsp" sink_name=output source_name=input
-#autoload_sink_add output module-alsa-sink sink_name=output
-#autoload_source_add input module-alsa-source source_name=input
+#autoload_sink_add output module-oss-mmap device="/dev/dsp" sink_name=output source_name=input
+#autoload_source_add input module-oss-mmap device="/dev/dsp" sink_name=output source_name=input
+autoload_sink_add output module-alsa-sink sink_name=output
+autoload_source_add input module-alsa-source source_name=input
 
 # Load several protocols
 load module-esound-protocol-tcp
diff --git a/polyp/polyplib-context.c b/polyp/polyplib-context.c
index 9acb2d703..7542dd9bb 100644
--- a/polyp/polyplib-context.c
+++ b/polyp/polyplib-context.c
@@ -463,7 +463,10 @@ static void set_dispatch_callbacks(struct pa_operation *o) {
 
 struct pa_operation* pa_context_drain(struct pa_context *c, void (*cb) (struct pa_context*c, void *userdata), void *userdata) {
     struct pa_operation *o;
-    assert(c && c->ref >= 1 && c->state == PA_CONTEXT_READY);
+    assert(c && c->ref >= 1);
+
+    if (c->state != PA_CONTEXT_READY)
+        return NULL;
 
     if (!pa_context_is_pending(c))
         return NULL;
diff --git a/polyp/polyplib-def.h b/polyp/polyplib-def.h
index 6420e87e4..02f5e526f 100644
--- a/polyp/polyplib-def.h
+++ b/polyp/polyplib-def.h
@@ -24,6 +24,7 @@
 
 #include <inttypes.h>
 #include "cdecl.h"
+#include "sample.h"
 
 /** \file
  * Global definitions */
@@ -128,6 +129,14 @@ enum pa_subscription_event_type {
 /** Return one if an event type t matches an event mask bitfield */
 #define pa_subscription_match_flags(m, t) (!!((m) & (1 << ((t) & PA_SUBSCRIPTION_EVENT_FACILITY_MASK))))
 
+/** A structure for latency info. See pa_stream_get_latency().  */
+struct pa_latency_info {
+    pa_usec_t buffer_usec;    /**< Time in usecs the current buffer takes to play */
+    pa_usec_t sink_usec;      /**< Time in usecs a sample takes to be played on the sink. The total latency is buffer_usec+sink_usec. */
+    int playing;              /**< Non-zero when the stream is currently playing */
+    int queue_length;         /**< Queue size in bytes. */  
+};
+
 PA_C_DECL_END
 
 #endif
diff --git a/polyp/polyplib-internal.h b/polyp/polyplib-internal.h
index fd6cd38fb..8c5d3166a 100644
--- a/polyp/polyplib-internal.h
+++ b/polyp/polyplib-internal.h
@@ -34,9 +34,9 @@
 #include "llist.h"
 #include "native-common.h"
 
-#define DEFAULT_MAXLENGTH 204800
-#define DEFAULT_TLENGTH 10240
-#define DEFAULT_PREBUF 4096
+#define DEFAULT_TLENGTH (10240*2)
+#define DEFAULT_MAXLENGTH (DEFAULT_TLENGTH*2)
+#define DEFAULT_PREBUF DEFAULT_TLENGTH
 #define DEFAULT_MINREQ 1024
 #define DEFAULT_FRAGSIZE 1024
 
diff --git a/polyp/polyplib-introspect.h b/polyp/polyplib-introspect.h
index 0cc525496..0c3051843 100644
--- a/polyp/polyplib-introspect.h
+++ b/polyp/polyplib-introspect.h
@@ -42,7 +42,7 @@ struct pa_sink_info {
     pa_volume_t volume;
     uint32_t monitor_source;
     const char *monitor_source_name;
-    uint32_t latency;
+    pa_usec_t latency;
 };
 
 struct pa_operation* pa_context_get_sink_info_by_name(struct pa_context *c, const char *name, void (*cb)(struct pa_context *c, const struct pa_sink_info *i, int is_last, void *userdata), void *userdata);
@@ -101,7 +101,7 @@ struct pa_sink_input_info {
     uint32_t sink;
     struct pa_sample_spec sample_spec;
     pa_volume_t volume;
-    uint32_t latency;
+    pa_usec_t latency;
 };
 
 struct pa_operation* pa_context_get_sink_input_info(struct pa_context *c, uint32_t index, void (*cb)(struct pa_context *c, const struct pa_sink_input_info*i, int is_last, void *userdata), void *userdata);
diff --git a/polyp/polyplib-stream.c b/polyp/polyplib-stream.c
index c0ec9e7eb..f45e1e7c0 100644
--- a/polyp/polyplib-stream.c
+++ b/polyp/polyplib-stream.c
@@ -168,6 +168,8 @@ void pa_command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag,
     
     s->requested_bytes += bytes;
 
+    fprintf(stderr, "total req: %u (%u)\n", s->requested_bytes, bytes);
+
     if (s->requested_bytes && s->write_callback)
         s->write_callback(s, s->requested_bytes, s->write_userdata);
 
@@ -320,22 +322,26 @@ struct pa_operation * pa_stream_drain(struct pa_stream *s, void (*cb) (struct pa
 
 static void stream_get_latency_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
     struct pa_operation *o = userdata;
-    uint32_t latency;
+    struct pa_latency_info i, *p = NULL;
     assert(pd && o && o->stream && o->context);
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
 
-        latency = (uint32_t) -1;
-    } else if (pa_tagstruct_getu32(t, &latency) < 0 || !pa_tagstruct_eof(t)) {
+    } else if (pa_tagstruct_getu32(t, &i.buffer_usec) < 0 ||
+               pa_tagstruct_getu32(t, &i.sink_usec) < 0 ||
+               pa_tagstruct_getu32(t, &i.playing) < 0 ||
+               pa_tagstruct_getu32(t, &i.queue_length) < 0 ||
+               !pa_tagstruct_eof(t)) {
         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
         goto finish;
-    }
+    } else
+        p = &i;
 
     if (o->callback) {
-        void (*cb)(struct pa_stream *s, uint32_t latency, void *userdata) = o->callback;
-        cb(o->stream, latency, o->userdata);
+        void (*cb)(struct pa_stream *s, const struct pa_latency_info *i, void *userdata) = o->callback;
+        cb(o->stream, p, o->userdata);
     }
 
 finish:
@@ -343,7 +349,7 @@ finish:
     pa_operation_unref(o);
 }
 
-struct pa_operation* pa_stream_get_latency(struct pa_stream *s, void (*cb)(struct pa_stream *p, uint32_t latency, void *userdata), void *userdata) {
+struct pa_operation* pa_stream_get_latency(struct pa_stream *s, void (*cb)(struct pa_stream *p, const struct pa_latency_info*i, void *userdata), void *userdata) {
     uint32_t tag;
     struct pa_operation *o;
     struct pa_tagstruct *t;
@@ -476,24 +482,31 @@ struct pa_operation* pa_stream_cork(struct pa_stream *s, int b, void (*cb) (stru
     return pa_operation_ref(o);
 }
 
-struct pa_operation* pa_stream_flush(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
-    struct pa_operation *o;
+struct pa_operation* pa_stream_send_simple_command(struct pa_stream *s, uint32_t command, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
     struct pa_tagstruct *t;
+    struct pa_operation *o;
     uint32_t tag;
     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
-
+    
     o = pa_operation_new(s->context, s);
-    assert(o);
     o->callback = cb;
     o->userdata = userdata;
 
     t = pa_tagstruct_new(NULL, 0);
-    assert(t);
-    pa_tagstruct_putu32(t, PA_COMMAND_FLUSH_PLAYBACK_STREAM);
+    pa_tagstruct_putu32(t, command);
     pa_tagstruct_putu32(t, tag = s->context->ctag++);
     pa_tagstruct_putu32(t, s->channel);
     pa_pstream_send_tagstruct(s->context->pstream, t);
     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 
-    return pa_operation_ref(o);   
+    return pa_operation_ref(o);
+}
+
+
+struct pa_operation* pa_stream_flush(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
+    return pa_stream_send_simple_command(s, PA_COMMAND_FLUSH_PLAYBACK_STREAM, cb, userdata);
+}
+
+struct pa_operation* pa_stream_trigger(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
+    return pa_stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata);
 }
diff --git a/polyp/polyplib-stream.h b/polyp/polyplib-stream.h
index a0cbe521c..d74c3cb21 100644
--- a/polyp/polyplib-stream.h
+++ b/polyp/polyplib-stream.h
@@ -102,7 +102,7 @@ size_t pa_stream_writable_size(struct pa_stream *p);
 struct pa_operation* pa_stream_drain(struct pa_stream *s, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata);
 
 /** Get the playback latency of a stream */
-struct pa_operation* pa_stream_get_latency(struct pa_stream *p, void (*cb)(struct pa_stream *p, uint32_t latency, void *userdata), void *userdata);
+struct pa_operation* pa_stream_get_latency(struct pa_stream *p, void (*cb)(struct pa_stream *p, const struct pa_latency_info *i, void *userdata), void *userdata);
 
 /** Set the callback function that is called whenever the state of the stream changes */
 void pa_stream_set_state_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata);
@@ -121,6 +121,11 @@ struct pa_operation* pa_stream_cork(struct pa_stream *s, int b, void (*cb) (stru
  * function. \since 0.3 */
 struct pa_operation* pa_stream_flush(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata);
 
+/** Request immediate start of playback on this stream. This disables
+ * prebuffering as specified in the pa_buffer_attr structure. \since
+ * 0.3 */
+struct pa_operation* pa_stream_trigger(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata);
+
 PA_C_DECL_END
 
 #endif
diff --git a/polyp/protocol-native.c b/polyp/protocol-native.c
index 8b39482ca..7b7dfef63 100644
--- a/polyp/protocol-native.c
+++ b/polyp/protocol-native.c
@@ -136,7 +136,7 @@ static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, u
 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
-static void command_flush_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
+static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 
 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
     [PA_COMMAND_ERROR] = { NULL },
@@ -179,7 +179,8 @@ static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
     [PA_COMMAND_SET_SINK_VOLUME] = { command_set_volume },
     [PA_COMMAND_SET_SINK_INPUT_VOLUME] = { command_set_volume },
     [PA_COMMAND_CORK_PLAYBACK_STREAM] = { command_cork_playback_stream },
-    [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = { command_flush_playback_stream },
+    [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
+    [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 };
 
 /* structure management */
@@ -438,6 +439,8 @@ static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk
         pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
         s->drain_request = 0;
     }
+
+    /*fprintf(stderr, "after_drop: %u\n", pa_memblockq_get_length(s->memblockq));*/
 }
 
 static void sink_input_kill_cb(struct pa_sink_input *i) {
@@ -451,6 +454,8 @@ static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i) {
     assert(i && i->userdata);
     s = i->userdata;
 
+    /*fprintf(stderr, "get_latency: %u\n", pa_memblockq_get_length(s->memblockq));*/
+    
     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 }
 
@@ -797,7 +802,7 @@ static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t comma
     struct connection *c = userdata;
     struct pa_tagstruct *reply;
     struct playback_stream *s;
-    uint32_t index, latency;
+    uint32_t index;
     assert(c && t);
 
     if (pa_tagstruct_getu32(t, &index) < 0 ||
@@ -816,12 +821,14 @@ static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t comma
         return;
     }
 
-    latency = pa_sink_input_get_latency(s->sink_input);
     reply = pa_tagstruct_new(NULL, 0);
     assert(reply);
     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
     pa_tagstruct_putu32(reply, tag);
-    pa_tagstruct_putu32(reply, latency);
+    pa_tagstruct_putu32(reply, pa_sink_input_get_latency(s->sink_input));
+    pa_tagstruct_putu32(reply, pa_sink_get_latency(s->sink_input->sink));
+    pa_tagstruct_putu32(reply, pa_memblockq_is_readable(s->memblockq));
+    pa_tagstruct_putu32(reply, pa_memblockq_get_length(s->memblockq));
     pa_pstream_send_tagstruct(c->pstream, reply);
 }
 
@@ -1325,7 +1332,7 @@ static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t comma
     pa_pstream_send_simple_ack(c->pstream, tag);
 }
 
-static void command_flush_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
+static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
     struct connection *c = userdata;
     uint32_t index;
     struct playback_stream *s;
@@ -1347,7 +1354,14 @@ static void command_flush_playback_stream(struct pa_pdispatch *pd, uint32_t comm
         return;
     }
 
-    pa_memblockq_flush(s->memblockq);
+    if (command == PA_COMMAND_TRIGGER_PLAYBACK_STREAM)
+        pa_memblockq_prebuf_disable(s->memblockq);
+    else {
+        assert(command == PA_COMMAND_FLUSH_PLAYBACK_STREAM);
+        pa_memblockq_flush(s->memblockq);
+    }
+
+    pa_sink_notify(s->sink_input->sink);
     pa_pstream_send_simple_ack(c->pstream, tag);
 }
 
@@ -1383,8 +1397,11 @@ static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, ui
         
         pa_memblockq_push_align(p->memblockq, chunk, delta);
         assert(p->sink_input);
+        /*fprintf(stderr, "after_recv: %u\n", pa_memblockq_get_length(p->memblockq));*/
+
         pa_sink_notify(p->sink_input->sink);
         /*fprintf(stderr, "Recieved %u bytes.\n", chunk->length);*/
+
     } else {
         struct upload_stream *u = (struct upload_stream*) stream;
         size_t l;
diff --git a/polyp/pstream.c b/polyp/pstream.c
index ad3dd0e07..81ee0b439 100644
--- a/polyp/pstream.c
+++ b/polyp/pstream.c
@@ -211,6 +211,8 @@ void pa_pstream_send_packet(struct pa_pstream*p, struct pa_packet *packet) {
     struct item_info *i;
     assert(p && packet);
 
+    /*fprintf(stderr, "push-packet %p\n", packet);*/
+    
     i = pa_xmalloc(sizeof(struct item_info));
     i->type = PA_PSTREAM_ITEM_PACKET;
     i->packet = pa_packet_ref(packet);
@@ -258,6 +260,8 @@ static void prepare_next_write_item(struct pa_pstream *p) {
     p->write.index = 0;
     
     if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
+        /*fprintf(stderr, "pop-packet %p\n", p->write.current->packet);*/
+        
         assert(p->write.current->packet);
         p->write.data = p->write.current->packet->data;
         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length);
diff --git a/polyp/sample.c b/polyp/sample.c
index f4a80861c..edfe19594 100644
--- a/polyp/sample.c
+++ b/polyp/sample.c
@@ -61,7 +61,7 @@ size_t pa_bytes_per_second(const struct pa_sample_spec *spec) {
 uint32_t pa_bytes_to_usec(size_t length, const struct pa_sample_spec *spec) {
     assert(spec);
 
-    return (uint32_t) (((double) length /pa_frame_size(spec))/spec->rate*1000000);
+    return (uint32_t) (((double) length/pa_frame_size(spec)*1000000)/spec->rate);
 }
 
 int pa_sample_spec_valid(const struct pa_sample_spec *spec) {
diff --git a/polyp/sample.h b/polyp/sample.h
index 459f8a302..28ae51eab 100644
--- a/polyp/sample.h
+++ b/polyp/sample.h
@@ -66,6 +66,9 @@ struct pa_sample_spec {
     uint8_t channels;              /**< Audio channels. (1 for mono, 2 for stereo, ...) */
 };
 
+/** Type for usec specifications */
+typedef uint32_t pa_usec_t;
+
 /** Return the amount of bytes playback of a second of audio with the speicified sample type takes */
 size_t pa_bytes_per_second(const struct pa_sample_spec *spec);
 
@@ -73,7 +76,7 @@ size_t pa_bytes_per_second(const struct pa_sample_spec *spec);
 size_t pa_frame_size(const struct pa_sample_spec *spec);
 
 /** Calculate the time the specified bytes take to play with the specified sample type */
-uint32_t pa_bytes_to_usec(size_t length, const struct pa_sample_spec *spec);
+pa_usec_t pa_bytes_to_usec(size_t length, const struct pa_sample_spec *spec);
 
 /** Return non-zero when the sample type specification is valid */
 int pa_sample_spec_valid(const struct pa_sample_spec *spec);
diff --git a/polyp/sink-input.c b/polyp/sink-input.c
index 5009033f2..9238fac0f 100644
--- a/polyp/sink-input.c
+++ b/polyp/sink-input.c
@@ -105,23 +105,19 @@ void pa_sink_input_kill(struct pa_sink_input*i) {
         i->kill(i);
 }
 
-uint32_t pa_sink_input_get_latency(struct pa_sink_input *i) {
-    uint32_t l = 0;
-    
+pa_usec_t pa_sink_input_get_latency(struct pa_sink_input *i) {
     assert(i);
+    
     if (i->get_latency)
-        l += i->get_latency(i);
-
-    assert(i->sink);
-    l += pa_sink_get_latency(i->sink);
+        return i->get_latency(i);
 
-    return l;
+    return 0;
 }
 
 int pa_sink_input_peek(struct pa_sink_input *i, struct pa_memchunk *chunk) {
     assert(i && chunk && i->peek && i->drop);
 
-    if (i->corked == 0)
+    if (i->corked)
         return -1;
     
     if (!i->resampler)
@@ -139,11 +135,11 @@ int pa_sink_input_peek(struct pa_sink_input *i, struct pa_memchunk *chunk) {
         
         l = pa_resampler_request(i->resampler, CONVERT_BUFFER_LENGTH);
 
-        i->drop(i, &tchunk, l);
-
         if (tchunk.length > l)
             tchunk.length = l;
 
+        i->drop(i, &tchunk, tchunk.length);
+
         pa_resampler_run(i->resampler, &tchunk, &i->resampled_chunk);
         pa_memblock_unref(tchunk.memblock);
     }
diff --git a/polyp/sink-input.h b/polyp/sink-input.h
index b06445409..df6ead6be 100644
--- a/polyp/sink-input.h
+++ b/polyp/sink-input.h
@@ -61,7 +61,7 @@ void pa_sink_input_free(struct pa_sink_input* i);
  * request destruction of it */
 void pa_sink_input_kill(struct pa_sink_input *i);
 
-uint32_t pa_sink_input_get_latency(struct pa_sink_input *i);
+pa_usec_t pa_sink_input_get_latency(struct pa_sink_input *i);
 
 int pa_sink_input_peek(struct pa_sink_input *i, struct pa_memchunk *chunk);
 void pa_sink_input_drop(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length);
diff --git a/polyp/sink.c b/polyp/sink.c
index 43fd351cf..b520dd8aa 100644
--- a/polyp/sink.c
+++ b/polyp/sink.c
@@ -142,7 +142,7 @@ static unsigned fill_mix_info(struct pa_sink *s, struct pa_mix_info *info, unsig
 
 static void inputs_drop(struct pa_sink *s, struct pa_mix_info *info, unsigned maxinfo, size_t length) {
     assert(s && info);
-    
+
     for (; maxinfo > 0; maxinfo--, info++) {
         struct pa_sink_input *i = info->userdata;
         assert(i && info->chunk.memblock);
@@ -267,7 +267,7 @@ void pa_sink_render_into_full(struct pa_sink *s, struct pa_memchunk *target) {
     }
 }
 
-uint32_t pa_sink_get_latency(struct pa_sink *s) {
+pa_usec_t pa_sink_get_latency(struct pa_sink *s) {
     assert(s);
 
     if (!s->get_latency)
diff --git a/polyp/sink.h b/polyp/sink.h
index 8248d00cc..9c91692e1 100644
--- a/polyp/sink.h
+++ b/polyp/sink.h
@@ -56,7 +56,7 @@ int pa_sink_render(struct pa_sink*s, size_t length, struct pa_memchunk *result);
 int pa_sink_render_into(struct pa_sink*s, struct pa_memchunk *target);
 void pa_sink_render_into_full(struct pa_sink *s, struct pa_memchunk *target);
 
-uint32_t pa_sink_get_latency(struct pa_sink *s);
+pa_usec_t pa_sink_get_latency(struct pa_sink *s);
 
 void pa_sink_notify(struct pa_sink*s);