client: allow zero-copy writing to the stream

src/pulse/internal.h

@@ -151,6 +151,11 @@ struct pa_stream {
     uint32_t device_index;
     char *device_name;
 
+    /* playback */
+    pa_memblock *write_memblock;
+    void *write_data;
+
+    /* recording */
     pa_memchunk peek_memchunk;
     void *peek_data;
     pa_memblockq *record_memblockq;

src/pulse/stream.c

@@ -144,12 +144,13 @@ pa_stream *pa_stream_new_with_proplist(
     s->suspended = FALSE;
     s->corked = FALSE;
 
+    s->write_memblock = NULL;
+    s->write_data = NULL;
+
     pa_memchunk_reset(&s->peek_memchunk);
     s->peek_data = NULL;
-
     s->record_memblockq = NULL;
 
-
     memset(&s->timing_info, 0, sizeof(s->timing_info));
     s->timing_info_valid = FALSE;
 
@@ -221,6 +222,11 @@ static void stream_free(pa_stream *s) {
 
     stream_unlink(s);
 
+    if (s->write_memblock) {
+        pa_memblock_release(s->write_memblock);
+        pa_memblock_unref(s->write_data);
+    }
+
     if (s->peek_memchunk.memblock) {
         if (s->peek_data)
             pa_memblock_release(s->peek_memchunk.memblock);
@@ -1187,20 +1193,60 @@ int pa_stream_connect_record(
     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
 }
 
+int pa_stream_begin_write(
+        pa_stream *s,
+        void **data,
+        size_t *nbytes) {
+
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, data, PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context, nbytes && *nbytes != 0, PA_ERR_INVALID);
+
+    if (!s->write_memblock) {
+        s->write_memblock = pa_memblock_new(s->context->mempool, *nbytes);
+        s->write_data = pa_memblock_acquire(s->write_memblock);
+    }
+
+    *data = s->write_data;
+    *nbytes = pa_memblock_get_length(s->write_memblock);
+
+    return 0;
+}
+
+int pa_stream_cancel_write(
+        pa_stream *s) {
+
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->write_memblock, PA_ERR_BADSTATE);
+
+    pa_assert(s->write_data);
+
+    pa_memblock_release(s->write_memblock);
+    pa_memblock_unref(s->write_memblock);
+    s->write_memblock = NULL;
+    s->write_data = NULL;
+
+    return 0;
+}
+
 int pa_stream_write(
         pa_stream *s,
         const void *data,
         size_t length,
-        void (*free_cb)(void *p),
+        pa_free_cb_t free_cb,
         int64_t offset,
         pa_seek_mode_t seek) {
 
-    pa_memchunk chunk;
-    pa_seek_mode_t t_seek;
-    int64_t t_offset;
-    size_t t_length;
-    const void *t_data;
-
     pa_assert(s);
     pa_assert(PA_REFCNT_VALUE(s) >= 1);
     pa_assert(data);
@@ -1210,47 +1256,72 @@ int pa_stream_write(
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context,
+                      !s->write_memblock ||
+                      ((data >= s->write_data) &&
+                       ((const char*) data + length <= (const char*) s->write_data + pa_memblock_get_length(s->write_memblock))),
+                      PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context, !free_cb || !s->write_memblock, PA_ERR_INVALID);
 
-    if (length <= 0)
+    if (s->write_memblock) {
-        return 0;
+        pa_memchunk chunk;
 
-    t_seek = seek;
+        /* pa_stream_write_begin() was called before */
-    t_offset = offset;
-    t_length = length;
-    t_data = data;
 
-    while (t_length > 0) {
+        pa_memblock_release(s->write_memblock);
 
-        chunk.index = 0;
+        chunk.memblock = s->write_memblock;
+        chunk.index = (const char *) data - (const char *) s->write_data;
+        chunk.length = length;
 
-        if (free_cb && !pa_pstream_get_shm(s->context->pstream)) {
+        s->write_memblock = NULL;
-            chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) t_data, t_length, free_cb, 1);
+        s->write_data = NULL;
-            chunk.length = t_length;
-        } else {
-            void *d;
 
-            chunk.length = PA_MIN(t_length, pa_mempool_block_size_max(s->context->mempool));
+        pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
-            chunk.memblock = pa_memblock_new(s->context->mempool, chunk.length);
+        pa_memblock_unref(chunk.memblock);
 
-            d = pa_memblock_acquire(chunk.memblock);
+    } else {
-            memcpy(d, t_data, chunk.length);
+        pa_seek_mode_t t_seek = seek;
-            pa_memblock_release(chunk.memblock);
+        int64_t t_offset = offset;
+        size_t t_length = length;
+        const void *t_data = data;
+
+        /* pa_stream_write_begin() was not called before */
+
+        while (t_length > 0) {
+            pa_memchunk chunk;
+
+            chunk.index = 0;
+
+            if (free_cb && !pa_pstream_get_shm(s->context->pstream)) {
+                chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) t_data, t_length, free_cb, 1);
+                chunk.length = t_length;
+            } else {
+                void *d;
+
+                chunk.length = PA_MIN(t_length, pa_mempool_block_size_max(s->context->mempool));
+                chunk.memblock = pa_memblock_new(s->context->mempool, chunk.length);
+
+                d = pa_memblock_acquire(chunk.memblock);
+                memcpy(d, t_data, chunk.length);
+                pa_memblock_release(chunk.memblock);
+            }
+
+            pa_pstream_send_memblock(s->context->pstream, s->channel, t_offset, t_seek, &chunk);
+
+            t_offset = 0;
+            t_seek = PA_SEEK_RELATIVE;
+
+            t_data = (const uint8_t*) t_data + chunk.length;
+            t_length -= chunk.length;
+
+            pa_memblock_unref(chunk.memblock);
         }
 
-        pa_pstream_send_memblock(s->context->pstream, s->channel, t_offset, t_seek, &chunk);
+        if (free_cb && pa_pstream_get_shm(s->context->pstream))
-
+            free_cb((void*) data);
-        t_offset = 0;
-        t_seek = PA_SEEK_RELATIVE;
-
-        t_data = (const uint8_t*) t_data + chunk.length;
-        t_length -= chunk.length;
-
-        pa_memblock_unref(chunk.memblock);
     }
 
-    if (free_cb && pa_pstream_get_shm(s->context->pstream))
-        free_cb((void*) data);
-
     /* This is obviously wrong since we ignore the seeking index . But
      * that's OK, the server side applies the same error */
     s->requested_bytes -= (seek == PA_SEEK_RELATIVE ? offset : 0) + (int64_t) length;

src/pulse/stream.h

@@ -418,15 +418,71 @@ int pa_stream_connect_record(
 /** Disconnect a stream from a source/sink */
 int pa_stream_disconnect(pa_stream *s);
 
-/** Write some data to the server (for playback sinks), if free_cb is
+/** Prepare writing data to the server (for playback streams). This
- * non-NULL this routine is called when all data has been written out
+ * function may be used to optimize the number of memory copies when
- * and an internal reference to the specified data is kept, the data
+ * doing playback ("zero-copy"). It is recommended to call this
- * is not copied. If NULL, the data is copied into an internal
+ * function before each call to pa_stream_write(). Pass in the address
- * buffer. The client my freely seek around in the output buffer. For
+ * to a pointer and an address of the number of bytes you want to
+ * write. On return the two values will contain a pointer where you
+ * can place the data to write and the maximum number of bytes you can
+ * write. On return *nbytes can be larger or have the same value as
+ * you passed in. You need to be able to handle both cases. Accessing
+ * memory beyond the returned *nbytes value is invalid. Acessing the
+ * memory returned after the following pa_stream_write() or
+ * pa_stream_cancel_write() is invalid. On invocation only *nbytes
+ * needs to be initialized, on return both *data and *nbytes will be
+ * valid. If you place (size_t) -1 in *nbytes on invocation the memory
+ * size will be chosen automatically (which is recommended to
+ * do). After placing your data in the memory area returned call
+ * pa_stream_write() with data set to an address within this memory
+ * area and an nbytes value that is smaller or equal to what was
+ * returned by this function to actually execute the write. An
+ * invocation of pa_stream_write() should follow "quickly" on
+ * pa_stream_begin_write(). It is not recommended letting an unbounded
+ * amount of time pass after calling pa_stream_begin_write() and
+ * before calling pa_stream_write(). If you want to cancel a
+ * previously called pa_stream_begin_write() without calling
+ * pa_stream_write() use pa_stream_cancel_write() instead. Calling
+ * pa_stream_begin_write() twice without calling pa_stream_write() or
+ * pa_stream_cancel_write() in between will return exactly the same
+ * pointer/nbytes values.\since 0.9.16 */
+int pa_stream_begin_write(
+        pa_stream *p,
+        void **data,
+        size_t *nbytes);
+
+/** Reverses the effect of pa_stream_begin_write() dropping all data
+ * that has already been placed in the memory area returned by
+ * pa_stream_begin_write(). Only valid to call if
+ * pa_stream_begin_write() was called before and neither
+ * pa_stream_cancel_write() nor pa_stream_write() have been called
+ * yet. Accessing the memory previously returned by
+ * pa_stream_begin_write() after this call is invalid. Any further
+ * explicit freeing of the memory area is not necessary. \since
+ * 0.9.16 */
+int pa_stream_cancel_write(
+        pa_stream *p);
+
+/** Write some data to the server (for playback streams), if free_cb
+ * is non-NULL this routine is called when all data has been written
+ * out and an internal reference to the specified data is kept, the
+ * data is not copied. If NULL, the data is copied into an internal
+ * buffer. The client may freely seek around in the output buffer. For
  * most applications passing 0 and PA_SEEK_RELATIVE as arguments for
  * offset and seek should be useful. Afte ther write call succeeded
  * the write index will be a the position after where this chunk of
- * data has been written to. */
+ * data has been written to.
+ *
+ * As an optimization for avoiding needless memory copies you may call
+ * pa_stream_begin_write() before this call and then place your audio
+ * data directly in the memory area returned by that call. Then, pass
+ * a pointer to that memory area to pa_stream_write(). After the
+ * invocation of pa_stream_write() the memory area may no longer be
+ * accessed. Any further explicit freeing of the memory area is not
+ * necessary. It is OK to write the memory area returned by
+ * pa_stream_begin_write() only partially with this call, skipping
+ * bytes both at the end and at the beginning of the reserved memory
+ * area.*/
 int pa_stream_write(
         pa_stream *p             /**< The stream to use */,
         const void *data         /**< The data to write */,
@@ -435,7 +491,7 @@ int pa_stream_write(
         int64_t offset,          /**< Offset for seeking, must be 0 for upload streams */
         pa_seek_mode_t seek      /**< Seek mode, must be PA_SEEK_RELATIVE for upload streams */);
 
-/** Read the next fragment from the buffer (for recording).
+/** Read the next fragment from the buffer (for recording streams).
  * data will point to the actual data and length will contain the size
  * of the data in bytes (which can be less than a complete framgnet).
  * Use pa_stream_drop() to actually remove the data from the