aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--Makefile.am4
-rw-r--r--src/event/BufferedSocket.cxx256
-rw-r--r--src/event/BufferedSocket.hxx125
-rw-r--r--src/event/SocketMonitor.cxx108
-rw-r--r--src/event/SocketMonitor.hxx118
5 files changed, 610 insertions, 1 deletions
diff --git a/Makefile.am b/Makefile.am
index 5f6ce85c4..eca6d7d7c 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -39,8 +39,8 @@ src_mpd_LDADD = \
$(FILTER_LIBS) \
$(ENCODER_LIBS) \
$(MIXER_LIBS) \
- libutil.a \
libevent.a \
+ libutil.a \
$(SYSTEMD_DAEMON_LIBS) \
$(GLIB_LIBS)
@@ -357,6 +357,8 @@ libutil_a_SOURCES = \
libevent_a_SOURCES = \
src/event/WakeFD.cxx src/event/WakeFD.hxx \
src/event/TimeoutMonitor.hxx src/event/TimeoutMonitor.cxx \
+ src/event/SocketMonitor.cxx src/event/SocketMonitor.hxx \
+ src/event/BufferedSocket.cxx src/event/BufferedSocket.hxx \
src/event/Loop.hxx
# PCM library
diff --git a/src/event/BufferedSocket.cxx b/src/event/BufferedSocket.cxx
new file mode 100644
index 000000000..f84fe808b
--- /dev/null
+++ b/src/event/BufferedSocket.cxx
@@ -0,0 +1,256 @@
+/*
+ * Copyright (C) 2003-2013 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include "config.h"
+#include "BufferedSocket.hxx"
+#include "SocketError.hxx"
+#include "util/fifo_buffer.h"
+
+#include <assert.h>
+#include <stdint.h>
+#include <string.h>
+
+#ifndef WIN32
+#include <sys/types.h>
+#include <sys/socket.h>
+#endif
+
+BufferedSocket::~BufferedSocket()
+{
+ if (input != nullptr)
+ fifo_buffer_free(input);
+}
+
+BufferedSocket::ssize_t
+BufferedSocket::DirectWrite(const void *data, size_t length)
+{
+ int flags = 0;
+#ifdef MSG_NOSIGNAL
+ flags |= MSG_NOSIGNAL;
+#endif
+#ifdef MSG_DONTWAIT
+ flags |= MSG_DONTWAIT;
+#endif
+
+ const auto nbytes = send(Get(), (const char *)data, length, flags);
+ if (gcc_unlikely(nbytes < 0)) {
+ const auto code = GetSocketError();
+ if (IsSocketErrorAgain(code))
+ return 0;
+
+ Cancel();
+
+ if (IsSocketErrorClosed(code))
+ OnSocketClosed();
+ else
+ OnSocketError(NewSocketError(code));
+ }
+
+ return nbytes;
+}
+
+ssize_t
+BufferedSocket::DirectRead(void *data, size_t length)
+{
+ int flags = 0;
+#ifdef MSG_DONTWAIT
+ flags |= MSG_DONTWAIT;
+#endif
+
+ const auto nbytes = recv(Get(), (char *)data, length, flags);
+ if (gcc_likely(nbytes > 0))
+ return nbytes;
+
+ if (nbytes == 0) {
+ OnSocketClosed();
+ return -1;
+ }
+
+ const auto code = GetSocketError();
+ if (IsSocketErrorAgain(code))
+ return 0;
+
+ if (IsSocketErrorClosed(code))
+ OnSocketClosed();
+ else
+ OnSocketError(NewSocketError(code));
+ return -1;
+}
+
+bool
+BufferedSocket::WriteFromBuffer()
+{
+ assert(IsDefined());
+
+ size_t length;
+ const void *data = output.Read(&length);
+ if (data == nullptr) {
+ CancelWrite();
+ return true;
+ }
+
+ auto nbytes = DirectWrite(data, length);
+ if (gcc_unlikely(nbytes <= 0))
+ return nbytes == 0;
+
+ output.Consume(nbytes);
+
+ if (output.IsEmpty())
+ CancelWrite();
+
+ return true;
+}
+
+bool
+BufferedSocket::ReadToBuffer()
+{
+ assert(IsDefined());
+
+ if (input == nullptr)
+ input = fifo_buffer_new(8192);
+
+ size_t length;
+ void *buffer = fifo_buffer_write(input, &length);
+ assert(buffer != nullptr);
+
+ const auto nbytes = DirectRead(buffer, length);
+ if (nbytes > 0)
+ fifo_buffer_append(input, nbytes);
+
+ return nbytes >= 0;
+}
+
+bool
+BufferedSocket::Write(const void *data, size_t length)
+{
+ assert(IsDefined());
+
+#if 0
+ /* TODO: disabled because this would add overhead on some callers (the ones that often), but it may be useful */
+
+ if (output.IsEmpty()) {
+ /* try to write it directly first */
+ const auto nbytes = DirectWrite(data, length);
+ if (gcc_likely(nbytes > 0)) {
+ data = (const uint8_t *)data + nbytes;
+ length -= nbytes;
+ if (length == 0)
+ return true;
+ } else if (nbytes < 0)
+ return false;
+ }
+#endif
+
+ if (!output.Append(data, length)) {
+ // TODO
+ OnSocketError(g_error_new_literal(g_quark_from_static_string("buffered_socket"),
+ 0, "Output buffer is full"));
+ return false;
+ }
+
+ ScheduleWrite();
+ return true;
+}
+
+bool
+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) {
+ ScheduleRead();
+ return true;
+ }
+
+ const auto result = OnSocketInput(data, length);
+ switch (result) {
+ case InputResult::MORE:
+ if (fifo_buffer_is_full(input)) {
+ // TODO
+ OnSocketError(g_error_new_literal(g_quark_from_static_string("buffered_socket"),
+ 0, "Input buffer is full"));
+ return false;
+ }
+
+ ScheduleRead();
+ return true;
+
+ case InputResult::PAUSE:
+ CancelRead();
+ return true;
+
+ case InputResult::AGAIN:
+ continue;
+
+ case InputResult::CLOSED:
+ return false;
+ }
+ }
+}
+
+void
+BufferedSocket::ConsumeInput(size_t nbytes)
+{
+ assert(IsDefined());
+
+ fifo_buffer_consume(input, nbytes);
+}
+
+void
+BufferedSocket::OnSocketReady(unsigned flags)
+{
+ assert(IsDefined());
+
+ if (gcc_unlikely(flags & (ERROR|HANGUP))) {
+ OnSocketClosed();
+ return;
+ }
+
+ if (flags & READ) {
+ assert(input == nullptr || !fifo_buffer_is_full(input));
+
+ if (!ReadToBuffer() || !ResumeInput())
+ return;
+
+ if (input == nullptr || !fifo_buffer_is_full(input))
+ ScheduleRead();
+
+ /* just in case the OnSocketInput() method has added
+ data to the output buffer: try to send it now
+ instead of waiting for the next event loop
+ iteration */
+ if (!output.IsEmpty())
+ flags |= WRITE;
+ }
+
+ if (flags & WRITE) {
+ assert(!output.IsEmpty());
+
+ if (!WriteFromBuffer())
+ return;
+ }
+}
diff --git a/src/event/BufferedSocket.hxx b/src/event/BufferedSocket.hxx
new file mode 100644
index 000000000..49b17c86f
--- /dev/null
+++ b/src/event/BufferedSocket.hxx
@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2003-2013 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#ifndef MPD_BUFFERED_SOCKET_HXX
+#define MPD_BUFFERED_SOCKET_HXX
+
+#include "check.h"
+#include "SocketMonitor.hxx"
+#include "util/PeakBuffer.hxx"
+#include "gcc.h"
+
+#include <type_traits>
+
+#include <stddef.h>
+
+struct fifo_buffer;
+class EventLoop;
+
+class BufferedSocket : private SocketMonitor {
+ typedef std::make_signed<size_t>::type ssize_t;
+
+ fifo_buffer *input;
+ PeakBuffer output;
+
+public:
+ BufferedSocket(int _fd, EventLoop &_loop,
+ size_t normal_size, size_t peak_size=0)
+ :SocketMonitor(_fd, _loop), input(nullptr),
+ output(normal_size, peak_size) {
+ ScheduleRead();
+ }
+
+ ~BufferedSocket();
+
+ using SocketMonitor::IsDefined;
+ using SocketMonitor::Close;
+
+private:
+ ssize_t DirectWrite(const void *data, size_t length);
+ ssize_t DirectRead(void *data, size_t length);
+
+ /**
+ * Send data from the output buffer to the socket.
+ *
+ * @return false if the socket has been closed
+ */
+ bool WriteFromBuffer();
+
+ /**
+ * Receive data from the socket to the input buffer.
+ *
+ * @return false if the socket has been closed
+ */
+ bool ReadToBuffer();
+
+protected:
+ /**
+ * @return false if the socket has been closed
+ */
+ bool Write(const void *data, size_t length);
+
+ /**
+ * @return false if the socket has been closed
+ */
+ bool ResumeInput();
+
+ /**
+ * Mark a portion of the input buffer "consumed". Only
+ * allowed to be called from OnSocketInput(). This method
+ * does not invalidate the pointer passed to OnSocketInput()
+ * yet.
+ */
+ void ConsumeInput(size_t nbytes);
+
+ enum class InputResult {
+ /**
+ * The method was successful, and it is ready to
+ * read more data.
+ */
+ MORE,
+
+ /**
+ * The method does not want to get more data for now.
+ * It will call ResumeInput() when it's ready for
+ * more.
+ */
+ PAUSE,
+
+ /**
+ * The method wants to be called again immediately, if
+ * there's more data in the buffer.
+ */
+ AGAIN,
+
+ /**
+ * The method has closed the socket.
+ */
+ CLOSED,
+ };
+
+ virtual InputResult OnSocketInput(const void *data, size_t length) = 0;
+ virtual void OnSocketError(GError *error) = 0;
+ virtual void OnSocketClosed() = 0;
+
+private:
+ virtual void OnSocketReady(unsigned flags) override;
+};
+
+#endif
diff --git a/src/event/SocketMonitor.cxx b/src/event/SocketMonitor.cxx
new file mode 100644
index 000000000..ec31647f2
--- /dev/null
+++ b/src/event/SocketMonitor.cxx
@@ -0,0 +1,108 @@
+/*
+ * Copyright (C) 2003-2013 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include "config.h"
+#include "SocketMonitor.hxx"
+#include "Loop.hxx"
+#include "fd_util.h"
+#include "gcc.h"
+
+#include <assert.h>
+
+/*
+ * GSource methods
+ *
+ */
+
+gboolean
+SocketMonitor::Prepare(gcc_unused GSource *source, gcc_unused gint *timeout_r)
+{
+ return false;
+}
+
+gboolean
+SocketMonitor::Check(GSource *_source)
+{
+ const Source &source = *(const Source *)_source;
+ const SocketMonitor &monitor = *source.monitor;
+ assert(_source == &monitor.source->base);
+
+ return monitor.Check();
+}
+
+gboolean
+SocketMonitor::Dispatch(GSource *_source,
+ gcc_unused GSourceFunc callback,
+ gcc_unused gpointer user_data)
+{
+ Source &source = *(Source *)_source;
+ SocketMonitor &monitor = *source.monitor;
+ assert(_source == &monitor.source->base);
+
+ monitor.Dispatch();
+ return true;
+}
+
+/**
+ * The vtable for our GSource implementation. Unfortunately, we
+ * cannot declare it "const", because g_source_new() takes a non-const
+ * pointer, for whatever reason.
+ */
+static GSourceFuncs socket_monitor_source_funcs = {
+ SocketMonitor::Prepare,
+ SocketMonitor::Check,
+ SocketMonitor::Dispatch,
+ nullptr,
+ nullptr,
+ nullptr,
+};
+
+SocketMonitor::SocketMonitor(int _fd, EventLoop &_loop)
+ :fd(_fd), loop(_loop),
+ source((Source *)g_source_new(&socket_monitor_source_funcs,
+ sizeof(*source))),
+ poll{fd, 0, 0} {
+ assert(fd >= 0);
+
+ source->monitor = this;
+
+ g_source_attach(&source->base, loop.GetContext());
+ g_source_add_poll(&source->base, &poll);
+}
+
+SocketMonitor::~SocketMonitor()
+{
+ if (IsDefined())
+ Close();
+}
+
+void
+SocketMonitor::Close()
+{
+ assert(IsDefined());
+
+ Cancel();
+
+ close_socket(fd);
+ fd = -1;
+
+ g_source_destroy(&source->base);
+ g_source_unref(&source->base);
+ source = nullptr;
+}
diff --git a/src/event/SocketMonitor.hxx b/src/event/SocketMonitor.hxx
new file mode 100644
index 000000000..ca3c5dcc4
--- /dev/null
+++ b/src/event/SocketMonitor.hxx
@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2003-2013 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#ifndef MPD_SOCKET_MONITOR_HXX
+#define MPD_SOCKET_MONITOR_HXX
+
+#include "check.h"
+
+#include <glib.h>
+
+#include <assert.h>
+
+#ifdef WIN32
+/* ERRORis a WIN32 macro that poisons our namespace; this is a
+ kludge to allow us to use it anyway */
+#ifdef ERROR
+#undef ERROR
+#endif
+#endif
+
+class EventLoop;
+
+class SocketMonitor {
+ struct Source {
+ GSource base;
+
+ SocketMonitor *monitor;
+ };
+
+ int fd;
+ EventLoop &loop;
+ Source *source;
+ GPollFD poll;
+
+public:
+ static constexpr unsigned READ = G_IO_IN;
+ static constexpr unsigned WRITE = G_IO_OUT;
+ static constexpr unsigned ERROR = G_IO_ERR;
+ static constexpr unsigned HANGUP = G_IO_HUP;
+
+ SocketMonitor(int _fd, EventLoop &_loop);
+
+ ~SocketMonitor();
+
+ bool IsDefined() const {
+ return fd >= 0;
+ }
+
+ int Get() const {
+ assert(IsDefined());
+
+ return fd;
+ }
+
+ void Close();
+
+ void Schedule(unsigned flags) {
+ poll.events = flags;
+ poll.revents &= flags;
+ }
+
+ void Cancel() {
+ poll.events = 0;
+ }
+
+ void ScheduleRead() {
+ poll.events |= READ|HANGUP|ERROR;
+ }
+
+ void ScheduleWrite() {
+ poll.events |= WRITE;
+ }
+
+ void CancelRead() {
+ poll.events &= ~(READ|HANGUP|ERROR);
+ }
+
+ void CancelWrite() {
+ poll.events &= ~WRITE;
+ }
+
+protected:
+ virtual void OnSocketReady(unsigned flags) = 0;
+
+public:
+ /* GSource callbacks */
+ static gboolean Prepare(GSource *source, gint *timeout_r);
+ static gboolean Check(GSource *source);
+ static gboolean Dispatch(GSource *source, GSourceFunc callback,
+ gpointer user_data);
+
+private:
+ bool Check() const {
+ return (poll.revents & poll.events) != 0;
+ }
+
+ void Dispatch() {
+ OnSocketReady(poll.revents & poll.events);
+ }
+};
+
+#endif