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

event/SocketMonitor: refactor to SocketEvent

Browse Source 
Similar to commits 1686f4e857 and
30a5dd267b
This commit is contained in:
Max Kellermann
2020-10-14 14:24:16 +02:00
parent 4d68a12f03
commit 5a4055fb08
26 changed files with 282 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/event/BufferedSocket.cxx
+9 -11
View File
@@ -69,7 +69,7 @@ BufferedSocket::ResumeInput() noexcept
while (true) {
const auto buffer = input.Read();
if (buffer.empty()) {
ScheduleRead();
event.ScheduleRead();
return true;
}
@@ -81,11 +81,11 @@ BufferedSocket::ResumeInput() noexcept
return false;
}
ScheduleRead();
event.ScheduleRead();
return true;
case InputResult::PAUSE:
CancelRead();
event.CancelRead();
return true;
case InputResult::AGAIN:
@@ -97,25 +97,23 @@ BufferedSocket::ResumeInput() noexcept
}
}
bool
void
BufferedSocket::OnSocketReady(unsigned flags) noexcept
{
assert(IsDefined());
if (gcc_unlikely(flags & (ERROR|HANGUP))) {
if (gcc_unlikely(flags & (SocketEvent::ERROR|SocketEvent::HANGUP))) {
OnSocketClosed();
return false;
return;
}
if (flags & READ) {
if (flags & SocketEvent::READ) {
assert(!input.IsFull());
if (!ReadToBuffer() || !ResumeInput())
return false;
return;
if (!input.IsFull())
ScheduleRead();
event.ScheduleRead();
}
return true;
}
src/event/BufferedSocket.hxx
+26 -10
View File
@@ -20,7 +20,7 @@
#ifndef MPD_BUFFERED_SOCKET_HXX
#define MPD_BUFFERED_SOCKET_HXX
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "util/StaticFifoBuffer.hxx"
#include <cassert>
@@ -30,20 +30,37 @@
class EventLoop;
/**
* A #SocketMonitor specialization that adds an input buffer.
* A #SocketEvent specialization that adds an input buffer.
*/
class BufferedSocket : protected SocketMonitor {
class BufferedSocket {
StaticFifoBuffer<uint8_t, 8192> input;
protected:
SocketEvent event;
public:
using ssize_t = SocketEvent::ssize_t;
BufferedSocket(SocketDescriptor _fd, EventLoop &_loop) noexcept
:SocketMonitor(_fd, _loop) {
ScheduleRead();
:event(_loop, BIND_THIS_METHOD(OnSocketReady), _fd) {
event.ScheduleRead();
}
using SocketMonitor::GetEventLoop;
using SocketMonitor::IsDefined;
using SocketMonitor::Close;
auto &GetEventLoop() const noexcept {
return event.GetEventLoop();
}
bool IsDefined() const noexcept {
return event.IsDefined();
}
auto GetSocket() const noexcept {
return event.GetSocket();
}
void Close() noexcept {
event.Close();
}
private:
/**
@@ -116,8 +133,7 @@ protected:
virtual void OnSocketError(std::exception_ptr ep) noexcept = 0;
virtual void OnSocketClosed() noexcept = 0;
/* virtual methods from class SocketMonitor */
bool OnSocketReady(unsigned flags) noexcept override;
virtual void OnSocketReady(unsigned flags) noexcept;
};
#endif
src/event/FullyBufferedSocket.cxx
+8 -8
View File
@@ -35,7 +35,7 @@ FullyBufferedSocket::DirectWrite(const void *data, size_t length) noexcept
return 0;
idle_event.Cancel();
BufferedSocket::Cancel();
event.Cancel();
if (IsSocketErrorClosed(code))
OnSocketClosed();
@@ -54,7 +54,7 @@ FullyBufferedSocket::Flush() noexcept
const auto data = output.Read();
if (data.empty()) {
idle_event.Cancel();
CancelWrite();
event.CancelWrite();
return true;
}
@@ -66,7 +66,7 @@ FullyBufferedSocket::Flush() noexcept
if (output.empty()) {
idle_event.Cancel();
CancelWrite();
event.CancelWrite();
}
return true;
@@ -92,23 +92,23 @@ FullyBufferedSocket::Write(const void *data, size_t length) noexcept
return true;
}
bool
void
FullyBufferedSocket::OnSocketReady(unsigned flags) noexcept
{
if (flags & WRITE) {
if (flags & SocketEvent::WRITE) {
assert(!output.empty());
assert(!idle_event.IsActive());
if (!Flush())
return false;
return;
}
return BufferedSocket::OnSocketReady(flags);
BufferedSocket::OnSocketReady(flags);
}
void
FullyBufferedSocket::OnIdle() noexcept
{
if (Flush() && !output.empty())
ScheduleWrite();
event.ScheduleWrite();
}
src/event/FullyBufferedSocket.hxx
+3 -3
View File
@@ -69,10 +69,10 @@ protected:
*/
bool Write(const void *data, size_t length) noexcept;
/* virtual methods from class SocketMonitor */
bool OnSocketReady(unsigned flags) noexcept override;
void OnIdle() noexcept;
/* virtual methods from class BufferedSocket */
void OnSocketReady(unsigned flags) noexcept override;
};
#endif
src/event/Loop.cxx
+9 -11
View File
@@ -19,7 +19,7 @@
#include "Loop.hxx"
#include "TimerEvent.hxx"
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "IdleEvent.hxx"
#include "util/ScopeExit.hxx"
@@ -47,7 +47,7 @@ EventLoop::EventLoop(
)
:
#ifdef HAVE_THREADED_EVENT_LOOP
SocketMonitor(*this),
wake_event(*this, BIND_THIS_METHOD(OnSocketReady)),
thread(_thread),
/* if this instance is hosted by an EventThread (no ThreadId
known yet) then we're not yet alive until the thread is
@@ -59,7 +59,7 @@ EventLoop::EventLoop(
quit(false)
{
#ifdef HAVE_THREADED_EVENT_LOOP
SocketMonitor::Open(SocketDescriptor(wake_fd.Get()));
wake_event.Open(SocketDescriptor(wake_fd.Get()));
#endif
}
@@ -198,7 +198,7 @@ EventLoop::Run() noexcept
assert(alive);
assert(busy);
SocketMonitor::Schedule(SocketMonitor::READ);
wake_event.Schedule(SocketEvent::READ);
#endif
#ifdef HAVE_URING
@@ -214,7 +214,7 @@ EventLoop::Run() noexcept
#ifdef HAVE_THREADED_EVENT_LOOP
AtScopeExit(this) {
SocketMonitor::Cancel();
wake_event.Cancel();
};
#endif
@@ -262,9 +262,9 @@ EventLoop::Run() noexcept
ready_sockets.clear();
for (size_t i = 0; i < poll_result.GetSize(); ++i) {
auto &sm = *(SocketMonitor *)poll_result.GetObject(i);
sm.SetReadyFlags(poll_result.GetEvents(i));
ready_sockets.push_back(sm);
auto &s = *(SocketEvent *)poll_result.GetObject(i);
s.SetReadyFlags(poll_result.GetEvents(i));
ready_sockets.push_back(s);
}
now = std::chrono::steady_clock::now();
@@ -339,7 +339,7 @@ EventLoop::HandleDeferred() noexcept
}
}
bool
void
EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
{
assert(IsInside());
@@ -348,8 +348,6 @@ EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
const std::lock_guard<Mutex> lock(mutex);
HandleDeferred();
return true;
}
#endif
src/event/Loop.hxx
+12 -14
View File
@@ -22,7 +22,7 @@
#include "Chrono.hxx"
#include "PollGroup.hxx"
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "event/Features.h"
#include "util/Compiler.h"
@@ -56,15 +56,13 @@ class DeferEvent;
* thread that runs it, except where explicitly documented as
* thread-safe.
*
* @see SocketMonitor, MultiSocketMonitor, TimerEvent, IdleEvent
* @see SocketEvent, MultiSocketMonitor, TimerEvent, IdleEvent
*/
class EventLoop final
#ifdef HAVE_THREADED_EVENT_LOOP
: SocketMonitor
#endif
{
#ifdef HAVE_THREADED_EVENT_LOOP
WakeFD wake_fd;
SocketEvent wake_event;
#endif
struct TimerCompare {
@@ -96,14 +94,14 @@ class EventLoop final
#endif
using ReadySocketList =
boost::intrusive::list<SocketMonitor,
boost::intrusive::member_hook<SocketMonitor,
SocketMonitor::ReadyListHook,
&SocketMonitor::ready_siblings>,
boost::intrusive::list<SocketEvent,
boost::intrusive::member_hook<SocketEvent,
SocketEvent::ReadyListHook,
&SocketEvent::ready_siblings>,
boost::intrusive::constant_time_size<false>>;
/**
* A linked list of #SocketMonitor instances which have a
* A linked list of #SocketEvent instances which have a
* non-zero "ready_flags" field, and need to be dispatched.
*/
ReadySocketList ready_sockets;
@@ -190,7 +188,7 @@ public:
*/
void Break() noexcept;
bool AddFD(int _fd, unsigned flags, SocketMonitor &m) noexcept {
bool AddFD(int _fd, unsigned flags, SocketEvent &m) noexcept {
#ifdef HAVE_THREADED_EVENT_LOOP
assert(!IsAlive() || IsInside());
#endif
@@ -198,7 +196,7 @@ public:
return poll_group.Add(_fd, flags, &m);
}
bool ModifyFD(int _fd, unsigned flags, SocketMonitor &m) noexcept {
bool ModifyFD(int _fd, unsigned flags, SocketEvent &m) noexcept {
#ifdef HAVE_THREADED_EVENT_LOOP
assert(!IsAlive() || IsInside());
#endif
@@ -207,7 +205,7 @@ public:
}
/**
* Remove the given #SocketMonitor after the file descriptor
* Remove the given #SocketEvent after the file descriptor
* has been closed. This is like RemoveFD(), but does not
* attempt to use #EPOLL_CTL_DEL.
*/
@@ -261,7 +259,7 @@ private:
Event::Duration HandleTimers() noexcept;
#ifdef HAVE_THREADED_EVENT_LOOP
bool OnSocketReady(unsigned flags) noexcept override;
void OnSocketReady(unsigned flags) noexcept;
#endif
public:
src/event/MultiSocketMonitor.hxx
+18 -16
View File
@@ -22,7 +22,7 @@
#include "IdleEvent.hxx"
#include "TimerEvent.hxx"
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "event/Features.h"
#include <cassert>
@@ -36,35 +36,43 @@ struct pollfd;
class EventLoop;
/**
* Similar to #SocketMonitor, but monitors multiple sockets. To use
* Similar to #SocketEvent, but monitors multiple sockets. To use
* it, implement the methods PrepareSockets() and DispatchSockets().
* In PrepareSockets(), use UpdateSocketList() and AddSocket().
* DispatchSockets() will be called if at least one socket is ready.
*/
class MultiSocketMonitor
{
class SingleFD final : public SocketMonitor {
class SingleFD final {
MultiSocketMonitor &multi;
SocketEvent event;
unsigned revents;
public:
SingleFD(MultiSocketMonitor &_multi,
SocketDescriptor _fd) noexcept
:SocketMonitor(_fd, _multi.GetEventLoop()),
multi(_multi), revents(0) {}
:multi(_multi),
event(multi.GetEventLoop(),
BIND_THIS_METHOD(OnSocketReady), _fd),
revents(0) {}
SocketDescriptor GetSocket() const noexcept {
return SocketMonitor::GetSocket();
return event.GetSocket();
}
unsigned GetEvents() const noexcept {
return SocketMonitor::GetScheduledFlags();
return event.GetScheduledFlags();
}
void SetEvents(unsigned _events) noexcept {
revents &= _events;
SocketMonitor::Schedule(_events);
event.Schedule(_events);
}
bool Schedule(unsigned events) noexcept {
return event.Schedule(events);
}
unsigned GetReturnedEvents() const noexcept {
@@ -75,11 +83,10 @@ class MultiSocketMonitor
revents = 0;
}
protected:
bool OnSocketReady(unsigned flags) noexcept override {
private:
void OnSocketReady(unsigned flags) noexcept {
revents = flags;
multi.SetReady();
return true;
}
};
@@ -119,11 +126,6 @@ class MultiSocketMonitor
#endif
public:
static constexpr unsigned READ = SocketMonitor::READ;
static constexpr unsigned WRITE = SocketMonitor::WRITE;
static constexpr unsigned ERROR = SocketMonitor::ERROR;
static constexpr unsigned HANGUP = SocketMonitor::HANGUP;
MultiSocketMonitor(EventLoop &_loop) noexcept;
EventLoop &GetEventLoop() const noexcept {
src/event/ServerSocket.cxx
+19 -12
View File
@@ -29,7 +29,7 @@
#include "net/Resolver.hxx"
#include "net/AddressInfo.hxx"
#include "net/ToString.hxx"
#include "event/SocketMonitor.hxx"
#include "event/SocketEvent.hxx"
#include "fs/AllocatedPath.hxx"
#include "util/RuntimeError.hxx"
#include "util/Domain.hxx"
@@ -43,9 +43,11 @@
#include <sys/stat.h>
#endif
class ServerSocket::OneServerSocket final : private SocketMonitor {
class ServerSocket::OneServerSocket final {
ServerSocket &parent;
SocketEvent event;
const unsigned serial;
#ifdef HAVE_UN
@@ -59,8 +61,9 @@ public:
OneServerSocket(EventLoop &_loop, ServerSocket &_parent,
unsigned _serial,
A &&_address) noexcept
:SocketMonitor(_loop),
parent(_parent), serial(_serial),
:parent(_parent),
event(_loop, BIND_THIS_METHOD(OnSocketReady)),
serial(_serial),
#ifdef HAVE_UN
path(nullptr),
#endif
@@ -88,10 +91,15 @@ public:
}
#endif
bool IsDefined() const noexcept {
return event.IsDefined();
}
void Open();
using SocketMonitor::IsDefined;
using SocketMonitor::Close;
void Close() noexcept {
event.Close();
}
[[nodiscard]] gcc_pure
std::string ToString() const noexcept {
@@ -99,14 +107,14 @@ public:
}
void SetFD(UniqueSocketDescriptor _fd) noexcept {
SocketMonitor::Open(_fd.Release());
SocketMonitor::ScheduleRead();
event.Open(_fd.Release());
event.ScheduleRead();
}
void Accept() noexcept;
private:
bool OnSocketReady(unsigned flags) noexcept override;
void OnSocketReady(unsigned flags) noexcept;
};
static constexpr Domain server_socket_domain("server_socket");
@@ -140,7 +148,7 @@ inline void
ServerSocket::OneServerSocket::Accept() noexcept
{
StaticSocketAddress peer_address;
UniqueSocketDescriptor peer_fd(GetSocket().AcceptNonBlock(peer_address));
UniqueSocketDescriptor peer_fd(event.GetSocket().AcceptNonBlock(peer_address));
if (!peer_fd.IsDefined()) {
const SocketErrorMessage msg;
FormatError(server_socket_domain,
@@ -160,11 +168,10 @@ ServerSocket::OneServerSocket::Accept() noexcept
parent.OnAccept(std::move(peer_fd), peer_address, uid);
}
bool
void
ServerSocket::OneServerSocket::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
{
Accept();
return true;
}
inline void
src/event/SignalMonitor.cxx
+15 -13
View File
@@ -22,7 +22,7 @@
#ifndef _WIN32
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "util/Manual.hxx"
#include "system/Error.hxx"
@@ -44,33 +44,37 @@
#include <pthread.h>
#endif
class SignalMonitor final : private SocketMonitor {
class SignalMonitor final {
#ifdef USE_SIGNALFD
SignalFD fd;
#else
WakeFD fd;
#endif
SocketEvent event;
public:
explicit SignalMonitor(EventLoop &_loop)
:SocketMonitor(_loop) {
:event(_loop, BIND_THIS_METHOD(OnSocketReady)) {
#ifndef USE_SIGNALFD
SocketMonitor::Open(SocketDescriptor(fd.Get()));
SocketMonitor::ScheduleRead();
event.Open(SocketDescriptor(fd.Get()));
event.ScheduleRead();
#endif
}
using SocketMonitor::GetEventLoop;
auto &GetEventLoop() const noexcept {
return event.GetEventLoop();
}
#ifdef USE_SIGNALFD
void Update(sigset_t &mask) noexcept {
const bool was_open = SocketMonitor::IsDefined();
const bool was_open = event.IsDefined();
fd.Create(mask);
if (!was_open) {
SocketMonitor::Open(SocketDescriptor(fd.Get()));
SocketMonitor::ScheduleRead();
event.Open(SocketDescriptor(fd.Get()));
event.ScheduleRead();
}
}
#else
@@ -80,7 +84,7 @@ public:
#endif
private:
bool OnSocketReady(unsigned flags) noexcept override;
void OnSocketReady(unsigned flags) noexcept;
};
/* this should be enough - is it? */
@@ -195,7 +199,7 @@ SignalMonitorRegister(int signo, SignalHandler handler)
#endif
}
bool
void
SignalMonitor::OnSocketReady(unsigned) noexcept
{
#ifdef USE_SIGNALFD
@@ -213,8 +217,6 @@ SignalMonitor::OnSocketReady(unsigned) noexcept
if (signal_pending[i].exchange(false))
signal_handlers[i]();
#endif
return true;
}
#endif
src/event/SocketMonitor.cxx → src/event/SocketEvent.cxx
+8 -8
View File
@@ -17,7 +17,7 @@
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "Loop.hxx"
#include "event/Features.h"
@@ -29,23 +29,23 @@
#endif
void
SocketMonitor::Dispatch() noexcept
SocketEvent::Dispatch() noexcept
{
const unsigned flags = std::exchange(ready_flags, 0) &
(GetScheduledFlags() | IMPLICIT_FLAGS);
if (flags != 0)
OnSocketReady(flags);
callback(flags);
}
SocketMonitor::~SocketMonitor() noexcept
SocketEvent::~SocketEvent() noexcept
{
if (IsDefined())
Cancel();
}
void
SocketMonitor::Open(SocketDescriptor _fd) noexcept
SocketEvent::Open(SocketDescriptor _fd) noexcept
{
assert(!fd.IsDefined());
assert(_fd.IsDefined());
@@ -54,7 +54,7 @@ SocketMonitor::Open(SocketDescriptor _fd) noexcept
}
SocketDescriptor
SocketMonitor::Steal() noexcept
SocketEvent::Steal() noexcept
{
assert(IsDefined());
@@ -64,13 +64,13 @@ SocketMonitor::Steal() noexcept
}
void
SocketMonitor::Close() noexcept
SocketEvent::Close() noexcept
{
Steal().Close();
}
bool
SocketMonitor::Schedule(unsigned flags) noexcept
SocketEvent::Schedule(unsigned flags) noexcept
{
assert(IsDefined());
src/event/SocketMonitor.hxx → src/event/SocketEvent.hxx
+17 -18
View File
@@ -17,11 +17,12 @@
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#ifndef MPD_SOCKET_MONITOR_HXX
#define MPD_SOCKET_MONITOR_HXX
#ifndef MPD_SOCKET_EVENT_HXX
#define MPD_SOCKET_EVENT_HXX
#include "PollGroup.hxx"
#include "net/SocketDescriptor.hxx"
#include "util/BindMethod.hxx"
#include <boost/intrusive/list_hook.hpp>
@@ -34,8 +35,8 @@ class EventLoop;
/**
* Monitor events on a socket. Call Schedule() to announce events
* you're interested in, or Cancel() to cancel your subscription. The
* #EventLoop will invoke virtual method OnSocketReady() as soon as
* any of the subscribed events are ready.
* #EventLoop will invoke the callback as soon as any of the
* subscribed events are ready.
*
* This class does not feel responsible for closing the socket. Call
* Close() to do it manually.
@@ -44,14 +45,18 @@ class EventLoop;
* thread that runs the #EventLoop, except where explicitly documented
* as thread-safe.
*/
class SocketMonitor {
class SocketEvent {
friend class EventLoop;
EventLoop &loop;
using ReadyListHook = boost::intrusive::list_member_hook<boost::intrusive::link_mode<boost::intrusive::auto_unlink>>;
ReadyListHook ready_siblings;
using Callback = BoundMethod<void(unsigned events) noexcept>;
const Callback callback;
SocketDescriptor fd = SocketDescriptor::Undefined();
EventLoop &loop;
/**
* A bit mask of events that is currently registered in the
@@ -80,13 +85,13 @@ public:
typedef std::make_signed<size_t>::type ssize_t;
explicit SocketMonitor(EventLoop &_loop) noexcept
:loop(_loop) {}
SocketEvent(EventLoop &_loop, Callback _callback,
SocketDescriptor _fd=SocketDescriptor::Undefined()) noexcept
:loop(_loop),
callback(_callback),
fd(_fd) {}
SocketMonitor(SocketDescriptor _fd, EventLoop &_loop) noexcept
:fd(_fd), loop(_loop) {}
~SocketMonitor() noexcept;
~SocketEvent() noexcept;
auto &GetEventLoop() const noexcept {
return loop;
@@ -148,12 +153,6 @@ public:
Schedule(GetScheduledFlags() & ~WRITE);
}
protected:
/**
* @return false if the socket has been closed
*/
virtual bool OnSocketReady(unsigned flags) noexcept = 0;
public:
void Dispatch() noexcept;
};
src/event/UringManager.cxx
+1 -3
View File
@@ -22,15 +22,13 @@
namespace Uring {
bool
void
Manager::OnSocketReady(unsigned) noexcept
{
try {
DispatchCompletions();
return true;
} catch (...) {
PrintException(std::current_exception());
return false;
}
}
src/event/UringManager.hxx
+7 -6
View File
@@ -19,23 +19,24 @@
#pragma once
#include "SocketMonitor.hxx"
#include "SocketEvent.hxx"
#include "IdleEvent.hxx"
#include "io/uring/Queue.hxx"
namespace Uring {
class Manager final : public Queue, SocketMonitor {
class Manager final : public Queue {
SocketEvent event;
IdleEvent idle_event;
public:
explicit Manager(EventLoop &event_loop)
:Queue(1024, 0),
SocketMonitor(SocketDescriptor::FromFileDescriptor(GetFileDescriptor()),
event_loop),
event(event_loop, BIND_THIS_METHOD(OnSocketReady),
SocketDescriptor::FromFileDescriptor(GetFileDescriptor())),
idle_event(event_loop, BIND_THIS_METHOD(OnIdle))
{
SocketMonitor::ScheduleRead();
event.ScheduleRead();
}
void Push(struct io_uring_sqe &sqe,
@@ -45,7 +46,7 @@ public:
}
private:
bool OnSocketReady(unsigned flags) noexcept override;
void OnSocketReady(unsigned flags) noexcept;
void OnIdle() noexcept;
};
src/event/meson.build
+1 -1
View File
@@ -26,7 +26,7 @@ event = static_library(
'IdleEvent.cxx',
'DeferEvent.cxx',
'MaskMonitor.cxx',
'SocketMonitor.cxx',
'SocketEvent.cxx',
'BufferedSocket.cxx',
'FullyBufferedSocket.cxx',
'MultiSocketMonitor.cxx',