oysteikt/mpd
0
0
Fork 0
Code Issues Pull Requests Releases Activity

util/FifoBuffer: C++ version of the fifo_buffer library

Browse Source
This commit is contained in:
Max Kellermann
2013-10-15 10:28:52 +02:00
parent 0c13703da3
commit 84d20d9e43
10 changed files with 294 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile.am
View File

@@ -256,6 +256,8 @@ libutil_a_SOURCES = \
src/util/Manual.hxx \
src/util/RefCount.hxx \
src/util/fifo_buffer.c src/util/fifo_buffer.h \
src/util/FifoBuffer.hxx \
src/util/WritableBuffer.hxx \
src/util/growing_fifo.c src/util/growing_fifo.h \
src/util/LazyRandomEngine.cxx src/util/LazyRandomEngine.hxx \
src/util/SliceBuffer.hxx \

30
src/InotifySource.cxx
View File

@@ -20,7 +20,6 @@
#include "config.h"
#include "InotifySource.hxx"
#include "InotifyDomain.hxx"
#include "util/fifo_buffer.h"
#include "util/Error.hxx"
#include "system/fd_util.h"
#include "system/FatalError.hxx"
@@ -34,30 +33,27 @@
bool
InotifySource::OnSocketReady(gcc_unused unsigned flags)
{
void *dest;
size_t length;
ssize_t nbytes;
dest = fifo_buffer_write(buffer, &length);
if (dest == NULL)
const auto dest = buffer.Write();
if (dest.IsEmpty())
FatalError("buffer full");
nbytes = read(Get(), dest, length);
ssize_t nbytes = read(Get(), dest.data, dest.size);
if (nbytes < 0)
FatalSystemError("Failed to read from inotify");
if (nbytes == 0)
FatalError("end of file from inotify");
fifo_buffer_append(buffer, nbytes);
buffer.Append(nbytes);
while (true) {
const char *name;
auto range = buffer.Read();
const struct inotify_event *event =
(const struct inotify_event *)
fifo_buffer_read(buffer, &length);
if (event == NULL || length < sizeof(*event) ||
length < sizeof(*event) + event->len)
range.data;
if (range.size < sizeof(*event) ||
range.size < sizeof(*event) + event->len)
break;
if (event->len > 0 && event->name[event->len - 1] == 0)
@@ -66,7 +62,7 @@ InotifySource::OnSocketReady(gcc_unused unsigned flags)
name = NULL;
callback(event->wd, event->mask, name, callback_ctx);
fifo_buffer_consume(buffer, sizeof(*event) + event->len);
buffer.Consume(sizeof(*event) + event->len);
}
return true;
@@ -77,8 +73,7 @@ InotifySource::InotifySource(EventLoop &_loop,
mpd_inotify_callback_t _callback, void *_ctx,
int _fd)
:SocketMonitor(_fd, _loop),
callback(_callback), callback_ctx(_ctx),
buffer(fifo_buffer_new(4096))
callback(_callback), callback_ctx(_ctx)
{
ScheduleRead();
@@ -98,11 +93,6 @@ InotifySource::Create(EventLoop &loop,
return new InotifySource(loop, callback, callback_ctx, fd);
}
InotifySource::~InotifySource()
{
fifo_buffer_free(buffer);
}
int
InotifySource::Add(const char *path_fs, unsigned mask, Error &error)
{

6
src/InotifySource.hxx
View File

@@ -21,6 +21,7 @@
#define MPD_INOTIFY_SOURCE_HXX
#include "event/SocketMonitor.hxx"
#include "util/FifoBuffer.hxx"
#include "Compiler.h"
class Error;
@@ -32,7 +33,7 @@ class InotifySource final : private SocketMonitor {
mpd_inotify_callback_t callback;
void *callback_ctx;
struct fifo_buffer *buffer;
FifoBuffer<uint8_t, 4096> buffer;
InotifySource(EventLoop &_loop,
mpd_inotify_callback_t callback, void *ctx, int fd);
@@ -49,9 +50,6 @@ public:
void *ctx,
Error &error);
~InotifySource();
/**
* Adds a path to the notify list.
*

46
src/TextInputStream.cxx
View File

@@ -29,35 +29,23 @@
#include <assert.h>
#include <string.h>
TextInputStream::TextInputStream(struct input_stream *_is)
: is(_is),
buffer(fifo_buffer_new(4096))
{
}
TextInputStream::~TextInputStream()
{
fifo_buffer_free(buffer);
}
bool TextInputStream::ReadLine(std::string &line)
{
void *dest;
const char *src, *p;
size_t length, nbytes;
do {
dest = fifo_buffer_write(buffer, &length);
if (dest != nullptr && length >= 2) {
size_t nbytes;
auto dest = buffer.Write();
if (dest.size >= 2) {
/* reserve one byte for the null terminator if
the last line is not terminated by a
newline character */
--length;
--dest.size;
Error error;
nbytes = is->LockRead(dest, length, error);
nbytes = is->LockRead(dest.data, dest.size, error);
if (nbytes > 0)
fifo_buffer_append(buffer, nbytes);
buffer.Append(nbytes);
else if (error.IsDefined()) {
LogError(error);
return false;
@@ -65,28 +53,28 @@ bool TextInputStream::ReadLine(std::string &line)
} else
nbytes = 0;
auto src_p = fifo_buffer_read(buffer, &length);
src = reinterpret_cast<const char *>(src_p);
if (src == nullptr)
auto src_p = buffer.Read();
if (src_p.IsEmpty())
return false;
p = reinterpret_cast<const char*>(memchr(src, '\n', length));
src = src_p.data;
p = reinterpret_cast<const char*>(memchr(src, '\n', src_p.size));
if (p == nullptr && nbytes == 0) {
/* end of file (or line too long): terminate
the current line */
dest = fifo_buffer_write(buffer, &nbytes);
assert(dest != nullptr);
*(char *)dest = '\n';
fifo_buffer_append(buffer, 1);
dest = buffer.Write();
assert(!dest.IsEmpty());
dest.data[0] = '\n';
buffer.Append(1);
}
} while (p == nullptr);
length = p - src + 1;
size_t length = p - src + 1;
while (p > src && g_ascii_isspace(p[-1]))
--p;
line = std::string(src, p - src);
fifo_buffer_consume(buffer, length);
buffer.Consume(length);
return true;
}

14
src/TextInputStream.hxx
View File

@@ -20,6 +20,8 @@
#ifndef MPD_TEXT_INPUT_STREAM_HXX
#define MPD_TEXT_INPUT_STREAM_HXX
#include "util/FifoBuffer.hxx"
#include <string>
struct input_stream;
@@ -27,7 +29,8 @@ struct fifo_buffer;
class TextInputStream {
struct input_stream *is;
struct fifo_buffer *buffer;
FifoBuffer<char, 4096> buffer;
public:
/**
* Wraps an existing #input_stream object into a #TextInputStream,
@@ -35,13 +38,8 @@ public:
*
* @param _is an open #input_stream object
*/
explicit TextInputStream(struct input_stream *_is);
/**
* Frees the #TextInputStream object. Does not close or free the
* underlying #input_stream.
*/
~TextInputStream();
explicit TextInputStream(struct input_stream *_is)
:is(_is) {}
TextInputStream(const TextInputStream &) = delete;
TextInputStream& operator=(const TextInputStream &) = delete;

49
src/event/BufferedSocket.cxx
View File

@@ -20,20 +20,9 @@
#include "config.h"
#include "BufferedSocket.hxx"
#include "system/SocketError.hxx"
#include "util/fifo_buffer.h"
#include "util/Error.hxx"
#include "util/Domain.hxx"
#include <assert.h>
#include <stdint.h>
#include <string.h>
BufferedSocket::~BufferedSocket()
{
if (input != nullptr)
fifo_buffer_free(input);
}
BufferedSocket::ssize_t
BufferedSocket::DirectRead(void *data, size_t length)
{
@@ -62,16 +51,12 @@ BufferedSocket::ReadToBuffer()
{
assert(IsDefined());
if (input == nullptr)
input = fifo_buffer_new(8192);
const auto buffer = input.Write();
assert(!buffer.IsEmpty());
size_t length;
void *buffer = fifo_buffer_write(input, &length);
assert(buffer != nullptr);
const auto nbytes = DirectRead(buffer, length);
const auto nbytes = DirectRead(buffer.data, buffer.size);
if (nbytes > 0)
fifo_buffer_append(input, nbytes);
input.Append(nbytes);
return nbytes >= 0;
}
@@ -81,23 +66,17 @@ BufferedSocket::ResumeInput()
{
assert(IsDefined());
if (input == nullptr) {
ScheduleRead();
return true;
}
while (true) {
size_t length;
const void *data = fifo_buffer_read(input, &length);
if (data == nullptr) {
const auto buffer = input.Read();
if (buffer.IsEmpty()) {
ScheduleRead();
return true;
}
const auto result = OnSocketInput(data, length);
const auto result = OnSocketInput(buffer.data, buffer.size);
switch (result) {
case InputResult::MORE:
if (fifo_buffer_is_full(input)) {
if (input.IsFull()) {
// TODO
static constexpr Domain buffered_socket_domain("buffered_socket");
Error error;
@@ -123,14 +102,6 @@ BufferedSocket::ResumeInput()
}
}
void
BufferedSocket::ConsumeInput(size_t nbytes)
{
assert(IsDefined());
fifo_buffer_consume(input, nbytes);
}
bool
BufferedSocket::OnSocketReady(unsigned flags)
{
@@ -142,12 +113,12 @@ BufferedSocket::OnSocketReady(unsigned flags)
}
if (flags & READ) {
assert(input == nullptr || !fifo_buffer_is_full(input));
assert(!input.IsFull());
if (!ReadToBuffer() || !ResumeInput())
return false;
if (input == nullptr || !fifo_buffer_is_full(input))
if (input.IsFull())
ScheduleRead();
}

16
src/event/BufferedSocket.hxx
View File

@@ -22,8 +22,12 @@
#include "check.h"
#include "SocketMonitor.hxx"
#include "util/FifoBuffer.hxx"
#include "Compiler.h"
#include <assert.h>
#include <stdint.h>
struct fifo_buffer;
class Error;
@@ -31,16 +35,14 @@ class Error;
* A #SocketMonitor specialization that adds an input buffer.
*/
class BufferedSocket : protected SocketMonitor {
fifo_buffer *input;
FifoBuffer<uint8_t, 8192> input;
public:
BufferedSocket(int _fd, EventLoop &_loop)
:SocketMonitor(_fd, _loop), input(nullptr) {
:SocketMonitor(_fd, _loop) {
ScheduleRead();
}
~BufferedSocket();
using SocketMonitor::IsDefined;
using SocketMonitor::Close;
using SocketMonitor::Write;
@@ -67,7 +69,11 @@ protected:
* does not invalidate the pointer passed to OnSocketInput()
* yet.
*/
void ConsumeInput(size_t nbytes);
void ConsumeInput(size_t nbytes) {
assert(IsDefined());
input.Consume(nbytes);
}
enum class InputResult {
/**

132
src/util/FifoBuffer.hxx Normal file
View File

@@ -0,0 +1,132 @@
/*
* Copyright (C) 2003-2010 Max Kellermann <max@duempel.org>
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* - Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* - Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the
* distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
* FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
* FOUNDATION OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
* INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
* SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
* STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
* OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef FIFO_BUFFER_HPP
#define FIFO_BUFFER_HPP
#include "WritableBuffer.hxx"
#include <utility>
#include <algorithm>
#include <assert.h>
#include <stddef.h>
/**
* A first-in-first-out buffer: you can append data at the end, and
* read data from the beginning. This class automatically shifts the
* buffer as needed. It is not thread safe.
*/
template<class T, size_t size>
class FifoBuffer {
public:
typedef size_t size_type;
public:
typedef WritableBuffer<T> Range;
protected:
size_type head, tail;
T data[size];
public:
constexpr
FifoBuffer():head(0), tail(0) {}
protected:
void Shift() {
if (head == 0)
return;
assert(head <= size);
assert(tail <= size);
assert(tail >= head);
std::move(data + head, data + tail, data);
tail -= head;
head = 0;
}
public:
void Clear() {
head = tail = 0;
}
bool IsEmpty() const {
return head == tail;
}
bool IsFull() const {
return head == 0 && tail == size;
}
/**
* Prepares writing. Returns a buffer range which may be written.
* When you are finished, call append().
*/
Range Write() {
Shift();
return Range(data + tail, size - tail);
}
/**
* Expands the tail of the buffer, after data has been written to
* the buffer returned by write().
*/
void Append(size_type n) {
assert(tail <= size);
assert(n <= size);
assert(tail + n <= size);
tail += n;
}
/**
* Return a buffer range which may be read. The buffer pointer is
* writable, to allow modifications while parsing.
*/
Range Read() {
return Range(data + head, tail - head);
}
/**
* Marks a chunk as consumed.
*/
void Consume(size_type n) {
assert(tail <= size);
assert(head <= tail);
assert(n <= tail);
assert(head + n <= tail);
head += n;
}
};
#endif

87
src/util/WritableBuffer.hxx Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright (C) 2013 Max Kellermann <max@duempel.org>
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* - Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* - Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the
* distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
* FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
* FOUNDATION OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
* INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
* SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
* STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
* OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef WRITABLE_BUFFER_HPP
#define WRITABLE_BUFFER_HPP
#include "Compiler.h"
#include <stddef.h>
/**
* A reference to a memory area that is writable.
*
* @see ConstBuffer
*/
template<typename T>
struct WritableBuffer {
typedef size_t size_type;
typedef T *pointer_type;
typedef const T *const_pointer_type;
typedef pointer_type iterator;
typedef const_pointer_type const_iterator;
pointer_type data;
size_type size;
WritableBuffer() = default;
constexpr WritableBuffer(pointer_type _data, size_type _size)
:data(_data), size(_size) {}
constexpr static WritableBuffer Null() {
return { nullptr, 0 };
}
constexpr bool IsNull() const {
return data == nullptr;
}
constexpr bool IsEmpty() const {
return size == 0;
}
constexpr iterator begin() const {
return data;
}
constexpr iterator end() const {
return data + size;
}
constexpr const_iterator cbegin() const {
return data;
}
constexpr const_iterator cend() const {
return data + size;
}
};
#endif

33
test/run_convert.cxx
View File

@@ -28,7 +28,7 @@
#include "AudioFormat.hxx"
#include "pcm/PcmConvert.hxx"
#include "ConfigGlobal.hxx"
#include "util/fifo_buffer.h"
#include "util/FifoBuffer.hxx"
#include "util/Error.hxx"
#include "stdbin.h"
@@ -59,8 +59,6 @@ int main(int argc, char **argv)
{
AudioFormat in_audio_format, out_audio_format;
const void *output;
ssize_t nbytes;
size_t length;
if (argc != 3) {
g_printerr("Usage: run_convert IN_FORMAT OUT_FORMAT <IN >OUT\n");
@@ -92,28 +90,31 @@ int main(int argc, char **argv)
PcmConvert state;
struct fifo_buffer *buffer = fifo_buffer_new(4096);
FifoBuffer<uint8_t, 4096> buffer;
while (true) {
void *p = fifo_buffer_write(buffer, &length);
assert(p != NULL);
{
const auto dest = buffer.Write();
assert(!dest.IsEmpty());
nbytes = read(0, p, length);
if (nbytes <= 0)
break;
ssize_t nbytes = read(0, dest.data, dest.size);
if (nbytes <= 0)
break;
fifo_buffer_append(buffer, nbytes);
buffer.Append(nbytes);
}
const void *src = fifo_buffer_read(buffer, &length);
assert(src != NULL);
auto src = buffer.Read();
assert(!src.IsEmpty());
length -= length % in_frame_size;
if (length == 0)
src.size -= src.size % in_frame_size;
if (src.IsEmpty())
continue;
fifo_buffer_consume(buffer, length);
buffer.Consume(src.size);
output = state.Convert(in_audio_format, src, length,
size_t length;
output = state.Convert(in_audio_format, src.data, src.size,
out_audio_format, &length, error);
if (output == NULL) {
g_printerr("Failed to convert: %s\n", error.GetMessage());