Add ISink::flush()

[gavv]

Implement explicit flushing of buffered samples from sink to file.

• add virtual void flush() = 0 to sndio::ISink
• add this method to all ISink implementations
• for most implementations, flush() should be no-op, as they don't use buffering
• for SoxSink, flush() should send buffered samples to disk
• sndio::Pump should invoke flush() when it exits from run()

[jeshwanthreddy13]

Hi @gavv, I would like to work on this. Can I get it assigned?

[gavv]

@jeshwanthreddy13 Sure, thanks!

[gavv]

Sorry, I forgot to update task with one detail. We've recently introduced status codes, so the method should actually be virtual ROC_ATTR_NODISCARD status::StatusCode flush(), to be able to report I/O error.

[jeshwanthreddy13]

Hi @gavv, I implemented flush() in SoxSink,

status::StatusCode SoxSink::flush() {
    sox_sample_t* buffer_data = buffer_.data();
    const status::StatusCode code = write_(buffer_data, buffer_size_);
    if (code != status::StatusOK) {
        return code;
    }

    return status::StatusOK;
}

and returning sink_.flush() inside Pump::run()

But when I execute roc-test-sndio the following test cases fail,

src/tests/roc_sndio/test_pump.cpp:113: error: Failure in TEST(pump, write_overwrite_read)
src/tests/roc_sndio/test_helpers/mock_sink.h:78: error:
	expected <5632 (0x1600)>
	but was  <5120 (0x1400)>

src/tests/roc_sndio/test_pump.cpp:64: error: Failure in TEST(pump, write_read)
src/tests/roc_sndio/test_helpers/mock_sink.h:78: error:
	expected <5632 (0x1600)>
	but was  <5120 (0x1400)>

src/tests/roc_sndio/test_backend_source.cpp:196: error: Failure in TEST(backend_source, rewind_after_eof)
src/tests/roc_sndio/test_backend_source.cpp:75: error:
	LONGS_EQUAL(expected_code, code) failed
	expected <4 (0x4)>
	but was  <1 (0x1)>

Errors (3 failures, 11 tests, 11 ran, 26976 checks, 0 ignored, 0 filtered out, 66 ms)

I'm finding it tricky pinpointing where I might be going wrong. Could you please help me identify it?

[gavv]

I took a closer look at SoxSink::write:

status::StatusCode SoxSink::write(audio::Frame& frame) {
    ...

    while (frame_size > 0) {
        ...

        if (buffer_pos == buffer_size_) {
            const status::StatusCode code = write_(buffer_data, buffer_pos);
            ...
        }
    }

    const status::StatusCode code = write_(buffer_data, buffer_pos);
    ...

    return status::StatusOK;
}

We can see that the second call to write_ (after the loop) already flushes samples from buffer_ to sox.

Which means that the problem that we've encountered in #660 (when we created this task) is not because we don't flush from buffer_ to sox, but is because we don't flush from sox to disk. In other words, sox has its own buffering and we need to trigger a flush of it.

Seems that your implementation:

status::StatusCode SoxSink::flush() {
    sox_sample_t* buffer_data = buffer_.data();
    const status::StatusCode code = write_(buffer_data, buffer_size_);
    if (code != status::Statua closer look to SoxSink::write:

status::StatusCode SoxSink::write(audio::Frame& frame) {
    ...

    while (frame_size > 0sOK) {
        return code;
    }

    return status::StatusOK;
}

..actually writes the same buffer to sox again, i.e. it duplicates last frame. Hence the failures in tests. Furthermore, if frame passed to write() was shorter than buffer_size_ (which is almost always true), this flush() implementation also writes to the file some garbage in the end of the buffer.

So what we need instead is to tell sox to flush its buffer. I suspect that this buffering is actually done by stdio, because I see that it uses setvbuf(_IOFBF) internally:

https://github.com/chirlu/sox/blob/42b3557e13e0fe01a83465b672d89faddbe65f49/src/formats.c#L531

If my assumption is true, we could implement flush something like this:

fflush((FILE*)output_->fp);

as it's done here:

https://github.com/mansr/sox/blob/0be259eaa9ce3f3fa587a3ef0cf2c0b9c73167a2/src/formats_i.c#L149

[jeshwanthreddy13]

I took a closer look at SoxSink::write:

status::StatusCode SoxSink::write(audio::Frame& frame) {
    ...

    while (frame_size > 0) {
        ...

        if (buffer_pos == buffer_size_) {
            const status::StatusCode code = write_(buffer_data, buffer_pos);
            ...
        }
    }

    const status::StatusCode code = write_(buffer_data, buffer_pos);
    ...

    return status::StatusOK;
}

We can see that the second call to write_ (after the loop) already flushes samples from buffer_ to sox.

Which means that the problem that we've encountered in #660 (when we created this task) is not because we don't flush from buffer_ to sox, but is because we don't flush from sox to disk. In other words, sox has its own buffering and we need to trigger a flush of it.

Seems that your implementation:

status::StatusCode SoxSink::flush() {
    sox_sample_t* buffer_data = buffer_.data();
    const status::StatusCode code = write_(buffer_data, buffer_size_);
    if (code != status::Statua closer look to SoxSink::write:

status::StatusCode SoxSink::write(audio::Frame& frame) {
    ...

    while (frame_size > 0sOK) {
        return code;
    }

    return status::StatusOK;
}

..actually writes the same buffer to sox again, i.e. it duplicates last frame. Hence the failures in tests. Furthermore, if frame passed to write() was shorter than buffer_size_ (which is almost always true), this flush() implementation also writes to the file some garbage in the end of the buffer.

So what we need instead is to tell sox to flush its buffer. I suspect that this buffering is actually done by stdio, because I see that it uses setvbuf(_IOFBF) internally:

https://github.com/chirlu/sox/blob/42b3557e13e0fe01a83465b672d89faddbe65f49/src/formats.c#L531

If my assumption is true, we could implement flush something like this:

fflush((FILE*)output_->fp);

as it's done here:

https://github.com/mansr/sox/blob/0be259eaa9ce3f3fa587a3ef0cf2c0b9c73167a2/src/formats_i.c#L149

Did not realise that write_ flushes buffer to sox, makes sense why the test cases fail. As you told, the buffering for sox is taken care by stdio.
Thanks for pointing them out.

[gavv]

Landed