PipeWire improvements

* Use given latency
* Fix (non)blocking microphone
* Mute microphone on startuo
* Reset buffer when the microphone stops
* Delete client-info handler
* Refactor the naming conventions, styling

audio/common/pipewire.c

@@ -31,7 +31,6 @@ static void core_error_cb(void *data, uint32_t id, int seq, int res, const char
    RARCH_ERR("[PipeWire]: error id:%u seq:%d res:%d (%s): %s\n",
              id, seq, res, spa_strerror(res), message);
 
-   /* stop and exit the thread loop */
    pw_thread_loop_stop(pw->thread_loop);
 }
 
@@ -42,11 +41,9 @@ static void core_done_cb(void *data, uint32_t id, int seq)
    retro_assert(id == PW_ID_CORE);
 
    pw->last_seq = seq;
+
    if (pw->pending_seq == seq)
-   {
-      /* stop and exit the thread loop */
       pw_thread_loop_signal(pw->thread_loop, false);
-   }
 }
 
 static const struct pw_core_events core_events = {
@@ -55,7 +52,7 @@ static const struct pw_core_events core_events = {
       .error = core_error_cb,
 };
 
-size_t calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels)
+size_t pipewire_calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels)
 {
    uint32_t sample_size = 1;
    switch (fmt)
@@ -85,7 +82,7 @@ size_t calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels)
    return sample_size * nchannels;
 }
 
-void set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS])
+void pipewire_set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS])
 {
    memcpy(position, (uint32_t[SPA_AUDIO_MAX_CHANNELS]) { SPA_AUDIO_CHANNEL_UNKNOWN, },
          sizeof(uint32_t) * SPA_AUDIO_MAX_CHANNELS);
@@ -114,7 +111,7 @@ void set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS])
    }
 }
 
-void pipewire_wait_resync(pipewire_core_t *pw)
+void pipewire_core_wait_resync(pipewire_core_t *pw)
 {
    retro_assert(pw);
    pw->pending_seq = pw_core_sync(pw->core, PW_ID_CORE, pw->pending_seq);
@@ -127,7 +124,7 @@ void pipewire_wait_resync(pipewire_core_t *pw)
    }
 }
 
-bool pipewire_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active)
+bool pipewire_stream_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active)
 {
    enum pw_stream_state st;
    const char       *error;

audio/common/pipewire.h

@@ -38,27 +38,26 @@ typedef struct pipewire_core
 {
    struct pw_thread_loop *thread_loop;
    struct pw_context *ctx;
+
    struct pw_core *core;
    struct spa_hook core_listener;
    int last_seq, pending_seq;
 
    struct pw_registry *registry;
    struct spa_hook registry_listener;
-   struct pw_client *client;
-   struct spa_hook client_listener;
 
-   bool nonblock;
    struct string_list *devicelist;
+   bool nonblock;
 } pipewire_core_t;
 
-size_t calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels);
+size_t pipewire_calc_frame_size(enum spa_audio_format fmt, uint32_t nchannels);
 
-void set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS]);
+void pipewire_set_position(uint32_t channels, uint32_t position[SPA_AUDIO_MAX_CHANNELS]);
-
-void pipewire_wait_resync(pipewire_core_t *pipewire);
-
-bool pipewire_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active);
 
 bool pipewire_core_init(pipewire_core_t *pipewire, const char *loop_name);
 
+void pipewire_core_wait_resync(pipewire_core_t *pipewire);
+
+bool pipewire_stream_set_active(struct pw_thread_loop *loop, struct pw_stream *stream, bool active);
+
 #endif  /* _RETROARCH_PIPEWIRE */

audio/drivers/pipewire.c

@@ -36,25 +36,21 @@
 
 
 #define DEFAULT_CHANNELS   2
-#define QUANTUM            1024  /* TODO: detect */
 #define RINGBUFFER_SIZE    (1u << 22)
 #define RINGBUFFER_MASK    (RINGBUFFER_SIZE - 1)
 
 typedef struct pipewire_audio
 {
    pipewire_core_t *pw;
-
    struct pw_stream *stream;
    struct spa_hook stream_listener;
    struct spa_audio_info_raw info;
    uint32_t highwater_mark;
    uint32_t frame_size;
-   uint32_t req;
    struct spa_ringbuffer ring;
    uint8_t buffer[RINGBUFFER_SIZE];
 } pipewire_audio_t;
 
-
 static void stream_destroy_cb(void *data)
 {
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
@@ -68,7 +64,7 @@ static void playback_process_cb(void *data)
    void *p;
    struct pw_buffer *b;
    struct spa_buffer *buf;
-   uint32_t req, index, n_bytes;
+   uint32_t req, idx, n_bytes;
    int32_t avail;
 
    retro_assert(audio->stream);
@@ -80,20 +76,15 @@ static void playback_process_cb(void *data)
    }
 
    buf = b->buffer;
-   p = buf->datas[0].data;
+   if ((p = buf->datas[0].data) == NULL)
-   if (p == NULL)
       goto done;
 
    /* calculate the total no of bytes to read data from buffer */
-   req = b->requested * audio->frame_size;
+   n_bytes = buf->datas[0].maxsize;
+   if (b->requested)
+      n_bytes = SPA_MIN(b->requested * audio->frame_size, n_bytes);
 
-   if (req == 0)
+   avail = spa_ringbuffer_get_read_index(&audio->ring, &idx);
-      req = audio->req;
-
-   n_bytes = SPA_MIN(req, buf->datas[0].maxsize);
-
-   /* get no of available bytes to read data from buffer */
-   avail = spa_ringbuffer_get_read_index(&audio->ring, &index);
 
    if (avail <= 0)
       /* fill rest buffer with silence */
@@ -105,10 +96,10 @@ static void playback_process_cb(void *data)
 
       spa_ringbuffer_read_data(&audio->ring,
                                audio->buffer, RINGBUFFER_SIZE,
-                               index & RINGBUFFER_MASK, p, n_bytes);
+                               idx & RINGBUFFER_MASK, p, n_bytes);
 
-      index += n_bytes;
+      idx += n_bytes;
-      spa_ringbuffer_read_update(&audio->ring, index);
+      spa_ringbuffer_read_update(&audio->ring, idx);
    }
 
    buf->datas[0].chunk->offset = 0;
@@ -116,7 +107,6 @@ static void playback_process_cb(void *data)
    buf->datas[0].chunk->size   = n_bytes;
 
 done:
-   /* queue the buffer for playback */
    pw_stream_queue_buffer(audio->stream, b);
    pw_thread_loop_signal(audio->pw->thread_loop, false);
 }
@@ -158,42 +148,17 @@ static const struct pw_stream_events playback_stream_events = {
       .state_changed = stream_state_changed_cb,
 };
 
-static void client_info_cb(void *data, const struct pw_client_info *info)
-{
-   const struct spa_dict_item *item;
-   pipewire_core_t *pw = (pipewire_core_t*)data;
-
-   RARCH_DBG("[PipeWire]: client: id:%u\n", info->id);
-   RARCH_DBG("[PipeWire]: \tprops:\n");
-   spa_dict_for_each(item, info->props)
-      RARCH_DBG("[PipeWire]: \t\t%s: \"%s\"\n", item->key, item->value);
-
-   pw_thread_loop_signal(pw->thread_loop, false);
-}
-
-static const struct pw_client_events client_events = {
-      PW_VERSION_CLIENT_EVENTS,
-      .info = client_info_cb,
-};
-
 static void registry_event_global(void *data, uint32_t id,
                 uint32_t permissions, const char *type, uint32_t version,
                 const struct spa_dict *props)
 {
    union string_list_elem_attr attr;
+   const struct spa_dict_item *item;
    pipewire_core_t              *pw = (pipewire_core_t*)data;
    const char                *media = NULL;
    const char                 *sink = NULL;
 
-   if (!pw->client && spa_streq(type, PW_TYPE_INTERFACE_Client))
+   if (spa_streq(type, PW_TYPE_INTERFACE_Node))
-   {
-      pw->client = pw_registry_bind(pw->registry,
-                                          id, type, PW_VERSION_CLIENT, 0);
-      pw_client_add_listener(pw->client,
-                             &pw->client_listener,
-                             &client_events, pw);
-   }
-   else if (spa_streq(type, PW_TYPE_INTERFACE_Node))
    {
       media = spa_dict_lookup(props, PW_KEY_MEDIA_CLASS);
       if (media && strcmp(media, "Audio/Sink") == 0)
@@ -205,13 +170,12 @@ static void registry_event_global(void *data, uint32_t id,
             string_list_append(pw->devicelist, sink, attr);
             RARCH_LOG("[PipeWire]: Found Sink Node: %s\n", sink);
          }
+
+         RARCH_DBG("[PipeWire]: Object: id:%u Type:%s/%d\n", id, type, version);
+         spa_dict_for_each(item, props)
+            RARCH_DBG("[PipeWire]: \t\t%s: \"%s\"\n", item->key, item->value);
       }
    }
-
-   const struct spa_dict_item *item;
-   RARCH_DBG("[PipeWire]: Object: id:%u Type:%s/%d\n", id, type, version);
-   spa_dict_for_each(item, props)
-      RARCH_DBG("[PipeWire]: \t\t%s: \"%s\"\n", item->key, item->value);
 }
 
 static const struct pw_registry_events registry_events = {
@@ -245,12 +209,19 @@ static void *pipewire_init(const char *device, unsigned rate,
    if (!pipewire_core_init(pw, "audio_driver"))
       goto error;
 
+   pw->registry = pw_core_get_registry(pw->core, PW_VERSION_REGISTRY, 0);
+
+   spa_zero(pw->registry_listener);
+   pw_registry_add_listener(pw->registry, &pw->registry_listener, &registry_events, pw);
+
+   /* unlock, run the loop and wait, this will trigger the callbacks */
+   pipewire_core_wait_resync(pw);
+
    audio->info.format = is_little_endian() ? SPA_AUDIO_FORMAT_F32_LE : SPA_AUDIO_FORMAT_F32_BE;
    audio->info.channels = DEFAULT_CHANNELS;
-   set_position(DEFAULT_CHANNELS, audio->info.position);
+   pipewire_set_position(DEFAULT_CHANNELS, audio->info.position);
    audio->info.rate = rate;
-   audio->frame_size = calc_frame_size(audio->info.format, DEFAULT_CHANNELS);
+   audio->frame_size = pipewire_calc_frame_size(audio->info.format, DEFAULT_CHANNELS);
-   audio->req = QUANTUM * rate * 1 / 2 / 100000 * audio->frame_size;
 
    props = pw_properties_new(PW_KEY_MEDIA_TYPE,          PW_RARCH_MEDIA_TYPE_AUDIO,
                              PW_KEY_MEDIA_CATEGORY,      PW_RARCH_MEDIA_CATEGORY_PLAYBACK,
@@ -262,22 +233,18 @@ static void *pipewire_init(const char *device, unsigned rate,
                              PW_KEY_APP_ICON_NAME,       PW_RARCH_APPNAME,
                              NULL);
    if (!props)
-      goto error;
+      goto unlock_error;
 
    if (device)
       pw_properties_set(props, PW_KEY_TARGET_OBJECT, device);
 
-   buf_samples = QUANTUM * rate * 3 / 4 / 100000;
+   buf_samples = latency * rate / 1000;
-
+   pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%" PRIu64 "/%u", buf_samples, rate);
-   pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%" PRIu64 "/%u",
-                      buf_samples, rate);
-
    pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", rate);
-
    audio->stream = pw_stream_new(pw->core, PW_RARCH_APPNAME, props);
 
    if (!audio->stream)
-      goto error;
+      goto unlock_error;
 
    pw_stream_add_listener(audio->stream, &audio->stream_listener, &playback_stream_events, audio);
 
@@ -294,23 +261,17 @@ static void *pipewire_init(const char *device, unsigned rate,
                            params, 1);
 
    if (res < 0)
-      goto error;
+      goto unlock_error;
 
    audio->highwater_mark = MIN(RINGBUFFER_SIZE,
-                               latency? (latency * 1000): 46440 * (uint64_t)rate / 1000000 * audio->frame_size);
+                               latency * (uint64_t)rate / 1000 * audio->frame_size);
-   RARCH_DBG("[PipeWire]: Bufer size: %u, RingBuffer size: %u\n", audio->highwater_mark, RINGBUFFER_SIZE);
 
-   pw->registry = pw_core_get_registry(pw->core, PW_VERSION_REGISTRY, 0);
-
-   spa_zero(pw->registry_listener);
-   pw_registry_add_listener(pw->registry, &pw->registry_listener, &registry_events, pw);
-
-   /* unlock, run the loop and wait, this will trigger the callbacks */
-   pipewire_wait_resync(pw);
    pw_thread_loop_unlock(pw->thread_loop);
 
    return audio;
 
+unlock_error:
+   pw_thread_loop_unlock(audio->pw->thread_loop);
 error:
    RARCH_ERR("[PipeWire]: Failed to initialize audio\n");
    pipewire_free(audio);
@@ -320,28 +281,33 @@ error:
 static ssize_t pipewire_write(void *data, const void *buf_, size_t size)
 {
    int32_t   filled, avail;
-   uint32_t          index;
+   uint32_t            idx;
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
    const char       *error = NULL;
 
    if (pw_stream_get_state(audio->stream, &error) != PW_STREAM_STATE_STREAMING)
       return 0;  /* wait for stream to become ready */
 
+   if (size > audio->highwater_mark)
+   {
+      RARCH_ERR("[PipeWire]: Buffer too small! Please try increasing the latency.\n");
+      return 0;
+   }
+
    pw_thread_loop_lock(audio->pw->thread_loop);
 
    while (size)
    {
-      filled = spa_ringbuffer_get_write_index(&audio->ring, &index);
+      filled = spa_ringbuffer_get_write_index(&audio->ring, &idx);
       avail = audio->highwater_mark - filled;
 
 #if 0  /* Useful for tracing */
       RARCH_DBG("[PipeWire]: Ringbuffer utilization: filled %d, avail %d, index %d, size %d\n",
-                filled, avail, index, size);
+                filled, avail, idx, size);
 #endif
 
       /* in non-blocking mode we play as much as we can
-       * in blocking mode we expect a freed buffer of at least the given size
+       * in blocking mode we expect a freed buffer of at least the given size */
-       */
       if (size > (size_t)avail)
       {
          if (audio->pw->nonblock)
@@ -349,31 +315,29 @@ static ssize_t pipewire_write(void *data, const void *buf_, size_t size)
             size = avail;
             break;
          }
-         else
+
-         {
+         pw_thread_loop_wait(audio->pw->thread_loop);
-            pw_thread_loop_wait(audio->pw->thread_loop);
-         }
       }
       else
          break;
    }
 
    if (filled < 0)
-      RARCH_ERR("[Pipewire]: %p: underrun write:%u filled:%d\n", audio, index, filled);
+      RARCH_ERR("[Pipewire]: %p: underrun write:%u filled:%d\n", audio, idx, filled);
    else
    {
       if ((uint32_t) filled + size > RINGBUFFER_SIZE)
       {
          RARCH_ERR("[PipeWire]: %p: overrun write:%u filled:%d + size:%zu > max:%u\n",
-         audio, index, filled, size, RINGBUFFER_SIZE);
+         audio, idx, filled, size, RINGBUFFER_SIZE);
       }
    }
 
    spa_ringbuffer_write_data(&audio->ring,
                              audio->buffer, RINGBUFFER_SIZE,
-                             index & RINGBUFFER_MASK, buf_, size);
+                             idx & RINGBUFFER_MASK, buf_, size);
-   index += size;
+   idx += size;
-   spa_ringbuffer_write_update(&audio->ring, index);
+   spa_ringbuffer_write_update(&audio->ring, idx);
 
    pw_thread_loop_unlock(audio->pw->thread_loop);
    return size;
@@ -383,26 +347,33 @@ static bool pipewire_stop(void *data)
 {
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
    const char       *error = NULL;
+
+   if (!audio || !audio->pw)
+      return false;
    if (pw_stream_get_state(audio->stream, &error) == PW_STREAM_STATE_PAUSED)
       return true;
 
-   return pipewire_set_active(audio->pw->thread_loop, audio->stream, false);
+   return pipewire_stream_set_active(audio->pw->thread_loop, audio->stream, false);
 }
 
 static bool pipewire_start(void *data, bool is_shutdown)
 {
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
    const char       *error = NULL;
+
+   if (!audio || !audio->pw)
+      return false;
    if (pw_stream_get_state(audio->stream, &error) == PW_STREAM_STATE_STREAMING)
       return true;
 
-   return pipewire_set_active(audio->pw->thread_loop, audio->stream, true);
+   return pipewire_stream_set_active(audio->pw->thread_loop, audio->stream, true);
 }
 
 static bool pipewire_alive(void *data)
 {
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
    const char       *error = NULL;
+
    if (!audio)
       return false;
 
@@ -432,9 +403,6 @@ static void pipewire_free(void *data)
       audio->stream = NULL;
    }
 
-   if (audio->pw->client)
-      pw_proxy_destroy((struct pw_proxy *)audio->pw->client);
-
    if (audio->pw->registry)
       pw_proxy_destroy((struct pw_proxy*)audio->pw->registry);
 
@@ -484,7 +452,7 @@ static void pipewire_device_list_free(void *data, void *array_list_data)
 
 static size_t pipewire_write_avail(void *data)
 {
-   uint32_t index, written, length;
+   uint32_t idx, written, length;
    pipewire_audio_t *audio = (pipewire_audio_t*)data;
    const char       *error = NULL;
 
@@ -495,7 +463,7 @@ static size_t pipewire_write_avail(void *data)
       return  0;  /* wait for stream to become ready */
 
    pw_thread_loop_lock(audio->pw->thread_loop);
-   written = spa_ringbuffer_get_write_index(&audio->ring, &index);
+   written = spa_ringbuffer_get_write_index(&audio->ring, &idx);
    length = audio->highwater_mark - written;
    pw_thread_loop_unlock(audio->pw->thread_loop);
 

audio/drivers_microphone/pipewire.c

@@ -33,21 +33,18 @@
 
 
 #define DEFAULT_CHANNELS   1
-#define QUANTUM            1024  /* TODO: detect */
 #define RINGBUFFER_SIZE    (1u << 22)
 #define RINGBUFFER_MASK    (RINGBUFFER_SIZE - 1)
 
 typedef struct pipewire_microphone
 {
    pipewire_core_t *pw;
-
    struct pw_stream *stream;
    struct spa_hook stream_listener;
    struct spa_audio_info_raw info;
    uint32_t frame_size;
    struct spa_ringbuffer ring;
    uint8_t buffer[RINGBUFFER_SIZE];
-
    bool is_ready;
 } pipewire_microphone_t;
 
@@ -90,7 +87,7 @@ static void capture_process_cb(void *data)
    struct pw_buffer *b;
    struct spa_buffer *buf;
    int32_t filled;
-   uint32_t index, offs, n_bytes;
+   uint32_t idx, offs, n_bytes;
 
    assert(microphone->stream);
 
@@ -108,24 +105,25 @@ static void capture_process_cb(void *data)
    offs = SPA_MIN(buf->datas[0].chunk->offset, buf->datas[0].maxsize);
    n_bytes = SPA_MIN(buf->datas[0].chunk->size, buf->datas[0].maxsize - offs);
 
-   filled = spa_ringbuffer_get_write_index(&microphone->ring, &index);
+   filled = spa_ringbuffer_get_write_index(&microphone->ring, &idx);
    if (filled < 0)
-      RARCH_ERR("[PipeWire]: %p: underrun write:%u filled:%d\n", p, index, filled);
+      RARCH_ERR("[PipeWire]: %p: underrun write:%u filled:%d\n", p, idx, filled);
    else
    {
       if ((uint32_t)filled + n_bytes > RINGBUFFER_SIZE)
          RARCH_ERR("[PipeWire]: %p: overrun write:%u filled:%d + size:%u > max:%u\n",
-                   p, index, filled, n_bytes, RINGBUFFER_SIZE);
+                   p, idx, filled, n_bytes, RINGBUFFER_SIZE);
    }
    spa_ringbuffer_write_data(&microphone->ring,
                              microphone->buffer, RINGBUFFER_SIZE,
-                             index & RINGBUFFER_MASK,
+                             idx & RINGBUFFER_MASK,
                              SPA_PTROFF(p, offs, void), n_bytes);
-   index += n_bytes;
+   idx += n_bytes;
-   spa_ringbuffer_write_update(&microphone->ring, index);
+   spa_ringbuffer_write_update(&microphone->ring, idx);
 
 done:
    pw_stream_queue_buffer(microphone->stream, b);
+   pw_thread_loop_signal(microphone->pw->thread_loop, false);
 }
 
 static const struct pw_stream_events capture_stream_events = {
@@ -140,6 +138,7 @@ static void registry_event_global(void *data, uint32_t id,
                 const struct spa_dict *props)
 {
    union string_list_elem_attr attr;
+   const struct spa_dict_item *item;
    pipewire_core_t              *pw = (pipewire_core_t*)data;
    const char                *media = NULL;
    const char                 *sink = NULL;
@@ -158,6 +157,10 @@ static void registry_event_global(void *data, uint32_t id,
             string_list_append(pw->devicelist, sink, attr);
             RARCH_LOG("[PipeWire]: Found Source Node: %s\n", sink);
          }
+
+         RARCH_DBG("[PipeWire]: Object: id:%u Type:%s/%d\n", id, type, version);
+         spa_dict_for_each(item, props)
+            RARCH_DBG("[PipeWire]: \t\t%s: \"%s\"\n", item->key, item->value);
       }
    }
 }
@@ -197,7 +200,7 @@ static void *pipewire_microphone_init(void)
    spa_zero(pw->registry_listener);
    pw_registry_add_listener(pw->registry, &pw->registry_listener, &registry_events, pw);
 
-   pipewire_wait_resync(pw);
+   pipewire_core_wait_resync(pw);
    pw_thread_loop_unlock(pw->thread_loop);
 
    return pw;
@@ -219,9 +222,6 @@ static void pipewire_microphone_free(void *driver_context)
    if (pw->thread_loop)
       pw_thread_loop_stop(pw->thread_loop);
 
-   if (pw->client)
-      pw_proxy_destroy((struct pw_proxy *)pw->client);
-
    if (pw->registry)
       pw_proxy_destroy((struct pw_proxy*)pw->registry);
 
@@ -247,26 +247,41 @@ static void pipewire_microphone_free(void *driver_context)
 static int pipewire_microphone_read(void *driver_context, void *microphone_context, void *buf_, size_t size_)
 {
    int32_t                  readable;
-   uint32_t                    index;
+   uint32_t                      idx;
    const char                 *error = NULL;
    pipewire_core_t               *pw = (pipewire_core_t*)driver_context;
    pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context;
 
-   if (!microphone->is_ready || pw_stream_get_state(microphone->stream, &error) != PW_STREAM_STATE_STREAMING)
+   if (  !microphone->is_ready
+       || pw_stream_get_state(microphone->stream, &error) != PW_STREAM_STATE_STREAMING)
       return -1;
 
    pw_thread_loop_lock(pw->thread_loop);
-   /* get no of available bytes to read data from buffer */
-   readable = spa_ringbuffer_get_read_index(&microphone->ring, &index);
 
-   if (readable < (int32_t)size_)
+   while (size_)
-      size_ = readable;
+   {
+      /* get no of available bytes to read data from buffer */
+      readable = spa_ringbuffer_get_read_index(&microphone->ring, &idx);
+
+      if (readable < (int32_t)size_)
+      {
+         if (pw->nonblock)
+         {
+            size_ = readable;
+            break;
+         }
+
+         pw_thread_loop_wait(pw->thread_loop);
+      }
+      else
+         break;
+   }
 
    spa_ringbuffer_read_data(&microphone->ring,
                             microphone->buffer, RINGBUFFER_SIZE,
-                            index & RINGBUFFER_MASK, buf_, size_);
+                            idx & RINGBUFFER_MASK, buf_, size_);
-   index += size_;
+   idx += size_;
-   spa_ringbuffer_read_update(&microphone->ring, index);
+   spa_ringbuffer_read_update(&microphone->ring, idx);
    pw_thread_loop_unlock(pw->thread_loop);
 
    return size_;
@@ -328,15 +343,16 @@ static void *pipewire_microphone_open_mic(void *driver_context,
    if (!microphone)
       goto error;
 
+   microphone->is_ready = false;
    microphone->pw = (pipewire_core_t*)driver_context;
 
    pw_thread_loop_lock(microphone->pw->thread_loop);
 
    microphone->info.format = is_little_endian() ? SPA_AUDIO_FORMAT_F32_LE : SPA_AUDIO_FORMAT_F32_BE;
    microphone->info.channels = DEFAULT_CHANNELS;
-   set_position(DEFAULT_CHANNELS, microphone->info.position);
+   pipewire_set_position(DEFAULT_CHANNELS, microphone->info.position);
    microphone->info.rate = rate;
-   microphone->frame_size = calc_frame_size(microphone->info.format, DEFAULT_CHANNELS);
+   microphone->frame_size = pipewire_calc_frame_size(microphone->info.format, DEFAULT_CHANNELS);
 
    props = pw_properties_new(PW_KEY_MEDIA_TYPE,          PW_RARCH_MEDIA_TYPE_AUDIO,
                              PW_KEY_MEDIA_CATEGORY,      PW_RARCH_MEDIA_CATEGORY_RECORD,
@@ -353,13 +369,9 @@ static void *pipewire_microphone_open_mic(void *driver_context,
    if (device)
       pw_properties_set(props, PW_KEY_TARGET_OBJECT, device);
 
-   buf_samples = QUANTUM * rate * 3 / 4 / 100000;
+   buf_samples = latency * rate / 1000;
-
+   pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%" PRIu64 "/%u", buf_samples, rate);
-   pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%" PRIu64 "/%u",
-                      buf_samples, rate);
-
    pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", rate);
-
    microphone->stream = pw_stream_new(microphone->pw->core, PW_RARCH_APPNAME, props);
 
    if (!microphone->stream)
@@ -375,6 +387,7 @@ static void *pipewire_microphone_open_mic(void *driver_context,
                            PW_DIRECTION_INPUT,
                            PW_ID_ANY,
                            PW_STREAM_FLAG_AUTOCONNECT |
+                           PW_STREAM_FLAG_INACTIVE |
                            PW_STREAM_FLAG_MAP_BUFFERS |
                            PW_STREAM_FLAG_RT_PROCESS,
                            params, 1);
@@ -383,11 +396,11 @@ static void *pipewire_microphone_open_mic(void *driver_context,
       goto unlock_error;
 
    pw_thread_loop_wait(microphone->pw->thread_loop);
-
    pw_thread_loop_unlock(microphone->pw->thread_loop);
-   *new_rate = microphone->info.rate;
-   microphone->is_ready = true;
 
+   *new_rate = microphone->info.rate;
+
+   microphone->is_ready = true;
    return microphone;
 
 unlock_error:
@@ -419,12 +432,12 @@ static bool pipewire_microphone_start_mic(void *driver_context, void *microphone
    pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context;
    const char                 *error = NULL;
 
-   if (!microphone->is_ready)
+   if (!pw || !microphone || !microphone->is_ready)
       return false;
    if (pw_stream_get_state(microphone->stream, &error) == PW_STREAM_STATE_STREAMING)
       return true;
 
-   return pipewire_set_active(pw->thread_loop, microphone->stream, true);
+   return pipewire_stream_set_active(pw->thread_loop, microphone->stream, true);
 }
 
 static bool pipewire_microphone_stop_mic(void *driver_context, void *microphone_context)
@@ -432,13 +445,18 @@ static bool pipewire_microphone_stop_mic(void *driver_context, void *microphone_
    pipewire_core_t               *pw = (pipewire_core_t*)driver_context;
    pipewire_microphone_t *microphone = (pipewire_microphone_t*)microphone_context;
    const char                 *error = NULL;
+   bool                          res = false;
 
-   if (!microphone->is_ready)
+   if (!pw || !microphone || !microphone->is_ready)
       return false;
    if (pw_stream_get_state(microphone->stream, &error) == PW_STREAM_STATE_PAUSED)
       return true;
 
-   return pipewire_set_active(pw->thread_loop, microphone->stream, false);
+   res = pipewire_stream_set_active(pw->thread_loop, microphone->stream, false);
+   spa_ringbuffer_read_update(&microphone->ring, 0);
+   spa_ringbuffer_write_update(&microphone->ring, 0);
+
+   return res;
 }
 
 static bool pipewire_microphone_mic_use_float(const void *driver_context, const void *microphone_context)