Increasing lag when transferring audio between streams with PulseAudio

I've been working on a personal project in C++ using the PulseAudio library and I've noticed some strange behavior where I'm not sure what's causing it.

My setup so far is fairly simple:

I create two audio streams (one record stream and one playback stream)
The audio is read from the record stream and written to a buffer inside the read callback
It is then read from that buffer and written to the playback stream inside the write callback

This setup does work (I can hear the audio just fine), but I've noticed that over time the buffer size seems to be increasing ever so slightly, thus also increasing the latency, eventually leading to noticeable audio "lag".

This issue can be reproduced with some fairly barebones code (Ignore the fact that the amount of allocated memory for the buffer keeps growing in the program, I'm only worried about the buffer_length increasing):

#include <iostream>
#include <pulse/pulseaudio.h>
#include <stdio.h>
#include <ctype.h>
#include <stdlib.h>
#include <unistd.h>
#include <cstring>

void contextStateChanged(pa_context *ctx, void *userdata);
void sinkCreated(pa_context *context, uint32_t idx, void *userdata);
void writeToStream(pa_stream *stream, size_t nbytes, void *userdata);
void readFromStream(pa_stream *stream, size_t nbytes, void *userdata);
void streamStateChanged(pa_stream *p, void *userdata);

pa_context *context;

void *buffer;
size_t buffer_index, buffer_length;

int32_t bytesRead = 0;
int32_t bytesWritten = 0;

pa_mainloop *mainloop;

pa_sample_spec spec = {
    .format = PA_SAMPLE_S16BE,
    .rate = 48000,
    .channels = 2
};

int main(int argc, char **argv) {
    mainloop = pa_mainloop_new();
    assert(mainloop);

    pa_mainloop_api *mainloopAPI = pa_mainloop_get_api(mainloop);
    assert(mainloopAPI);

    pa_proplist *props = pa_proplist_new();
    pa_proplist_sets(props, PA_PROP_APPLICATION_NAME, "PulseTest");
    pa_proplist_sets(props, PA_PROP_APPLICATION_ID, "me.mrletsplay.pulsetest");
    pa_proplist_sets(props, PA_PROP_APPLICATION_VERSION, "1.0");
    pa_proplist_sets(props, PA_PROP_APPLICATION_ICON_NAME, "audio-card");

    context = pa_context_new_with_proplist(mainloopAPI, "PulseTest", props);
    assert(context);

    pa_context_set_state_callback(context, contextStateChanged, NULL);

    pa_context_connect(context, NULL, (pa_context_flags_t) 0, NULL);

    std::cout << "Waiting for Pulseaudio" << std::endl;

    return pa_mainloop_run(mainloop, 0);
}

void initStreams() {
    pa_stream *stream = pa_stream_new(context, "playback", &spec, NULL);

    pa_buffer_attr bufferAttr;
    bufferAttr.maxlength = (uint32_t) 4096;
    bufferAttr.tlength = (uint32_t) 256;
    bufferAttr.prebuf = (uint32_t) -1;
    bufferAttr.minreq = (uint32_t) 64;
    assert(pa_stream_connect_playback(stream, NULL, &bufferAttr, PA_STREAM_ADJUST_LATENCY, NULL, NULL) == 0);
    pa_stream_set_state_callback(stream, streamStateChanged, NULL);
    pa_stream_set_write_callback(stream, writeToStream, NULL);

    pa_stream *in = pa_stream_new(context, "record", &spec, NULL);

    pa_buffer_attr inBuffer;
    inBuffer.maxlength = (uint32_t) 1024;
    inBuffer.fragsize = (uint32_t) 512;
    assert(pa_stream_connect_record(in, NULL, &inBuffer, PA_STREAM_ADJUST_LATENCY) == 0);
    pa_stream_set_state_callback(in, streamStateChanged, NULL);
    pa_stream_set_read_callback(in, readFromStream, NULL);
}

void contextStateChanged(pa_context *ctx, void *userdata) {
    if(pa_context_get_state(ctx) == PA_CONTEXT_READY) {
        std::cout << "Connected to Pulseaudio" << std::endl;
        initStreams();
    }
}

void writeToStream(pa_stream *stream, size_t nbytes, void *userdata) {
    bytesWritten += nbytes;

    // Output the difference between how many bytes we've read and how many bytes we've written
    std::cout << (bytesRead - bytesWritten) << std::endl;

    size_t write = nbytes;
    if(write > buffer_length) {
        write = buffer_length;
    }

    void *data;
    if(pa_stream_begin_write(stream, &data, &nbytes) < 0) {
        std::cout << "ERROR writing data: " << pa_strerror(pa_context_errno(context)) << std::endl;
        exit(1);
        return;
    }

    memcpy(data, (uint8_t *) buffer + buffer_index, write);
    buffer_length -= write;
    buffer_index += write;

    if(pa_stream_write(stream, data, nbytes, NULL, 0, PA_SEEK_RELATIVE) < 0) {
        std::cout << "ERROR writing data: " << pa_strerror(pa_context_errno(context)) << std::endl;
        exit(1);
        return;
    }
}

void readFromStream(pa_stream *stream, size_t nbytes, void *userdata) {
    bytesRead += nbytes;

    const void *data;
    if(pa_stream_peek(stream, &data, &nbytes) < 0) {
        std::cout << "ERROR reading data: " << pa_strerror(pa_context_errno(context)) << std::endl;
        exit(1);
        return;
    }

    if(buffer) {
        buffer = pa_xrealloc(buffer, buffer_index + buffer_length + nbytes);
        memcpy((uint8_t *) buffer + buffer_index + buffer_length, data, nbytes);
        buffer_length += nbytes;
    }else {
        buffer = pa_xmalloc(nbytes);
        memcpy(buffer, data, nbytes);
        buffer_length = nbytes;
        buffer_index = 0;
    }

    pa_stream_drop(stream);
}

void streamStateChanged(pa_stream *p, void *userdata) {
    std::cout << "State changed for stream: " << pa_stream_get_state(p) << std::endl;

    if(pa_stream_get_state(p) == PA_STREAM_READY) {
        std::cout << "Stream is ready" << std::endl;
    }
}

In the code I keep track of how many bytes I've read and how many bytes I've written. bytesRead seems to be growing more than bytesWritten, leading to the buffer growing over time.

I've tried writing more bytes than PulseAudio requests, but that just seems to cause PulseAudio to hang and not play any audio at all.

You can see the problem pretty easily in this chart generated from the output of the program over roughly 10 minutes: Program output chart

Solution

It's common trouble of sound application (with pulse audio, Direct Sound, etc.) with simultaneous input and output of the same data. The reason of lag may be something from list:

different devices for input and output;
USB sound devices;
rarely lags in your processing thread; ... etc.

The trouble is the result of lacking output data and sound device (rarely) has to add crack/silence/anysound.

Common solution of the trouble is:

reads amount of data have been outputted (It can be gotten reading play position for output device).
calculates amount of you data (sent plus available into buffer)
controls the "tail" how many sound data you sent and they haven't played.
And you must feed sound data to output according "tail": if tail is short - adds more data; if tail is long - drop some data.

Of course, you should generate fake sound data (sometimes) and drop extradata (sometimes).

Reasonable length of tail is about 100 - 300 milliseconds.