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

event/SocketMonitor: wrapper class for GSource + GPollFD

Browse Source
This commit is contained in:
Max Kellermann 2013-01-10 19:05:47 +01:00
parent 3e03527930
commit 396480cf94
5 changed files with 610 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile.am
View File

@ -39,8 +39,8 @@ src_mpd_LDADD = \
$(FILTER_LIBS) \ $(FILTER_LIBS) \
$(ENCODER_LIBS) \ $(ENCODER_LIBS) \
$(MIXER_LIBS) \ $(MIXER_LIBS) \
libutil.a \
libevent.a \ libevent.a \
libutil.a \
$(SYSTEMD_DAEMON_LIBS) \ $(SYSTEMD_DAEMON_LIBS) \
$(GLIB_LIBS) $(GLIB_LIBS)
@ -357,6 +357,8 @@ libutil_a_SOURCES = \
libevent_a_SOURCES = \ libevent_a_SOURCES = \
src/event/WakeFD.cxx src/event/WakeFD.hxx \ src/event/WakeFD.cxx src/event/WakeFD.hxx \
src/event/TimeoutMonitor.hxx src/event/TimeoutMonitor.cxx \ 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 src/event/Loop.hxx
# PCM library # PCM library

256
src/event/BufferedSocket.cxx Normal file
View File

@ -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;
}
}

125
src/event/BufferedSocket.hxx Normal file
View File

@ -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

108
src/event/SocketMonitor.cxx Normal file
View File

@ -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;
}

118
src/event/SocketMonitor.hxx Normal file
View File

@ -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