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

input/Curl: move code to class MultiSocketMonitor

Browse Source
This commit is contained in:
Max Kellermann 2013-01-15 07:14:36 +01:00
parent 4cfc4a5a94
commit 083065f433
4 changed files with 298 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Makefile.am
View File

@ -359,6 +359,7 @@ libevent_a_SOURCES = \
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/MultiSocketMonitor.cxx src/event/MultiSocketMonitor.hxx \
src/event/Loop.hxx
# PCM library
@ -1070,6 +1071,7 @@ test_DumpDatabase_SOURCES = test/DumpDatabase.cxx \
test_run_input_LDADD = \
$(INPUT_LIBS) \
$(ARCHIVE_LIBS) \
libevent.a \
$(GLIB_LIBS)
test_run_input_SOURCES = test/run_input.c \
test/stdbin.h \
@ -1080,9 +1082,10 @@ test_run_input_SOURCES = test/run_input.c \
src/fd_util.c
test_dump_text_file_LDADD = \
libutil.a \
$(INPUT_LIBS) \
$(ARCHIVE_LIBS) \
libevent.a \
libutil.a \
$(GLIB_LIBS)
test_dump_text_file_SOURCES = test/dump_text_file.cxx \
test/stdbin.h \
@ -1100,6 +1103,7 @@ test_dump_playlist_LDADD = \
$(ARCHIVE_LIBS) \
$(DECODER_LIBS) \
$(TAG_LIBS) \
libevent.a \
libutil.a \
$(GLIB_LIBS)
test_dump_playlist_SOURCES = test/dump_playlist.cxx \
@ -1126,6 +1130,7 @@ test_run_decoder_LDADD = \
$(INPUT_LIBS) \
$(ARCHIVE_LIBS) \
$(TAG_LIBS) \
libevent.a \
libutil.a \
$(GLIB_LIBS)
test_run_decoder_SOURCES = test/run_decoder.cxx \
@ -1149,6 +1154,7 @@ test_read_tags_LDADD = \
$(INPUT_LIBS) \
$(ARCHIVE_LIBS) \
$(TAG_LIBS) \
libevent.a \
libutil.a \
$(GLIB_LIBS)
test_read_tags_SOURCES = test/read_tags.cxx \

107
src/event/MultiSocketMonitor.cxx Normal file
View File

@ -0,0 +1,107 @@
/*
* 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 "MultiSocketMonitor.hxx"
#include "Loop.hxx"
#include "fd_util.h"
#include "gcc.h"
#include <assert.h>
/**
* 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 multi_socket_monitor_source_funcs = {
MultiSocketMonitor::Prepare,
MultiSocketMonitor::Check,
MultiSocketMonitor::Dispatch,
nullptr,
nullptr,
nullptr,
};
MultiSocketMonitor::MultiSocketMonitor(EventLoop &_loop)
:loop(_loop),
source((Source *)g_source_new(&multi_socket_monitor_source_funcs,
sizeof(*source))) {
source->monitor = this;
g_source_attach(&source->base, loop.GetContext());
}
MultiSocketMonitor::~MultiSocketMonitor()
{
g_source_destroy(&source->base);
g_source_unref(&source->base);
source = nullptr;
}
bool
MultiSocketMonitor::Check() const
{
if (CheckSockets())
return true;
for (const auto &i : fds)
if (i.revents != 0)
return true;
return false;
}
/*
* GSource methods
*
*/
gboolean
MultiSocketMonitor::Prepare(GSource *_source, gint *timeout_r)
{
Source &source = *(Source *)_source;
MultiSocketMonitor &monitor = *source.monitor;
assert(_source == &monitor.source->base);
return monitor.Prepare(timeout_r);
}
gboolean
MultiSocketMonitor::Check(GSource *_source)
{
const Source &source = *(const Source *)_source;
const MultiSocketMonitor &monitor = *source.monitor;
assert(_source == &monitor.source->base);
return monitor.Check();
}
gboolean
MultiSocketMonitor::Dispatch(GSource *_source,
gcc_unused GSourceFunc callback,
gcc_unused gpointer user_data)
{
Source &source = *(Source *)_source;
MultiSocketMonitor &monitor = *source.monitor;
assert(_source == &monitor.source->base);
monitor.Dispatch();
return true;
}

124
src/event/MultiSocketMonitor.hxx Normal file
View File

@ -0,0 +1,124 @@
/*
* 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_MULTI_SOCKET_MONITOR_HXX
#define MPD_MULTI_SOCKET_MONITOR_HXX
#include "check.h"
#include "gcc.h"
#include <glib.h>
#include <forward_list>
#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;
/**
* Monitor multiple sockets.
*/
class MultiSocketMonitor {
struct Source {
GSource base;
MultiSocketMonitor *monitor;
};
EventLoop &loop;
Source *source;
std::forward_list<GPollFD> fds;
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;
MultiSocketMonitor(EventLoop &_loop);
~MultiSocketMonitor();
public:
gcc_pure
gint64 GetTime() const {
return g_source_get_time(&source->base);
}
void InvalidateSockets() {
/* no-op because GLib always calls the GSource's
"prepare" method before each poll() anyway */
}
void AddSocket(int fd, unsigned events) {
fds.push_front({fd, gushort(events), 0});
g_source_add_poll(&source->base, &fds.front());
}
template<typename E>
void UpdateSocketList(E &&e) {
for (auto prev = fds.before_begin(), end = fds.end(),
i = std::next(prev);
i != end; i = std::next(prev)) {
assert(i->events != 0);
unsigned events = e(i->fd);
if (events != 0) {
i->events = events;
prev = i;
} else {
g_source_remove_poll(&source->base, &*i);
fds.erase_after(prev);
}
}
}
protected:
virtual void PrepareSockets(gcc_unused gint *timeout_r) {}
virtual bool CheckSockets() const { return false; }
virtual void DispatchSockets() = 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 Prepare(gint *timeout_r) {
PrepareSockets(timeout_r);
return false;
}
bool Check() const;
void Dispatch() {
DispatchSockets();
}
};
#endif

176
src/input/CurlInputPlugin.cxx
View File

@ -23,6 +23,7 @@
#include "conf.h"
#include "tag.h"
#include "IcyMetaDataParser.hxx"
#include "event/MultiSocketMonitor.hxx"
extern "C" {
#include "input_internal.h"
@ -182,6 +183,36 @@ struct input_curl {
input_curl &operator=(const input_curl &) = delete;
};
/**
* This class monitors all CURL file descriptors.
*/
class CurlSockets final : private MultiSocketMonitor {
/**
* Did CURL give us a timeout? If yes, then we need to call
* curl_multi_perform(), even if there was no event on any
* file descriptor.
*/
bool have_timeout;
/**
* The absolute time stamp when the timeout expires.
*/
gint64 absolute_timeout;
public:
CurlSockets(EventLoop &_loop)
:MultiSocketMonitor(_loop) {}
using MultiSocketMonitor::InvalidateSockets;
private:
void UpdateSockets();
virtual void PrepareSockets(gcc_unused gint *timeout_r) override;
virtual bool CheckSockets() const override;
virtual void DispatchSockets() override;
};
/** libcurl should accept "ICY 200 OK" */
static struct curl_slist *http_200_aliases;
@ -198,32 +229,7 @@ static struct {
*/
std::forward_list<input_curl *> requests;
/**
* The GMainLoop source used to poll all CURL file
* descriptors.
*/
GSource *source;
/**
* The source id of #source.
*/
guint source_id;
/** a linked list of all registered GPollFD objects */
std::forward_list<GPollFD> fds;
/**
* Did CURL give us a timeout? If yes, then we need to call
* curl_multi_perform(), even if there was no event on any
* file descriptor.
*/
bool timeout;
/**
* The absolute time stamp when the timeout expires. This is
* used in the GSource method check().
*/
gint64 absolute_timeout;
CurlSockets *sockets;
} curl;
static inline GQuark
@ -268,7 +274,7 @@ input_curl_resume(gpointer data)
* Calculates the GLib event bit mask for one file descriptor,
* obtained from three #fd_set objects filled by curl_multi_fdset().
*/
static gushort
static unsigned
input_curl_fd_events(int fd, fd_set *rfds, fd_set *wfds, fd_set *efds)
{
gushort events = 0;
@ -297,8 +303,8 @@ input_curl_fd_events(int fd, fd_set *rfds, fd_set *wfds, fd_set *efds)
*
* Runs in the I/O thread. No lock needed.
*/
static void
curl_update_fds(void)
void
CurlSockets::UpdateSockets()
{
assert(io_thread_inside());
@ -317,29 +323,15 @@ curl_update_fds(void)
return;
}
for (auto prev = curl.fds.before_begin(), end = curl.fds.end(),
i = std::next(prev);
i != end; i = std::next(prev)) {
const auto poll_fd = &*i;
assert(poll_fd->events != 0);
gushort events = input_curl_fd_events(poll_fd->fd, &rfds,
&wfds, &efds);
if (events != 0) {
poll_fd->events = events;
prev = i;
} else {
g_source_remove_poll(curl.source, poll_fd);
curl.fds.erase_after(prev);
}
}
UpdateSocketList([&rfds, &wfds, &efds](int fd){
return input_curl_fd_events(fd, &rfds,
&wfds, &efds);
});
for (int fd = 0; fd <= max_fd; ++fd) {
gushort events = input_curl_fd_events(fd, &rfds, &wfds, &efds);
if (events != 0) {
curl.fds.push_front({fd, events, 0});
g_source_add_poll(curl.source, &curl.fds.front());
}
unsigned events = input_curl_fd_events(fd, &rfds, &wfds, &efds);
if (events != 0)
AddSocket(fd, events);
}
}
@ -364,7 +356,7 @@ input_curl_easy_add(struct input_curl *c, GError **error_r)
return false;
}
curl_update_fds();
curl.sockets->InvalidateSockets();
return true;
}
@ -438,7 +430,7 @@ input_curl_easy_free_callback(gpointer data)
struct input_curl *c = (struct input_curl *)data;
input_curl_easy_free(c);
curl_update_fds();
curl.sockets->InvalidateSockets();
return NULL;
}
@ -575,27 +567,18 @@ input_curl_perform(void)
return true;
}
/*
* GSource methods
*
*/
/**
* The GSource prepare() method implementation.
*/
static gboolean
input_curl_source_prepare(G_GNUC_UNUSED GSource *source, gint *timeout_r)
void
CurlSockets::PrepareSockets(gint *timeout_r)
{
curl_update_fds();
UpdateSockets();
curl.timeout = false;
have_timeout = false;
long timeout2;
CURLMcode mcode = curl_multi_timeout(curl.multi, &timeout2);
if (mcode == CURLM_OK) {
if (timeout2 >= 0)
curl.absolute_timeout = g_source_get_time(source)
+ timeout2 * 1000;
absolute_timeout = GetTime() + timeout2 * 1000;
if (timeout2 >= 0 && timeout2 < 10)
/* CURL 7.21.1 likes to report "timeout=0",
@ -606,65 +589,28 @@ input_curl_source_prepare(G_GNUC_UNUSED GSource *source, gint *timeout_r)
*timeout_r = timeout2;
curl.timeout = timeout2 >= 0;
have_timeout = timeout2 >= 0;
} else
g_warning("curl_multi_timeout() failed: %s\n",
curl_multi_strerror(mcode));
return false;
}
/**
* The GSource check() method implementation.
*/
static gboolean
input_curl_source_check(G_GNUC_UNUSED GSource *source)
bool
CurlSockets::CheckSockets() const
{
if (curl.timeout) {
/* when a timeout has expired, we need to call
curl_multi_perform(), even if there was no file
descriptor event */
if (g_source_get_time(source) >= curl.absolute_timeout)
return true;
}
for (const auto &i : curl.fds)
if (i.revents != 0)
return true;
return false;
/* when a timeout has expired, we need to call
curl_multi_perform(), even if there was no file descriptor
event */
return have_timeout && GetTime() >= absolute_timeout;
}
/**
* The GSource dispatch() method implementation. The callback isn't
* used, because we're handling all events directly.
*/
static gboolean
input_curl_source_dispatch(G_GNUC_UNUSED GSource *source,
G_GNUC_UNUSED GSourceFunc callback,
G_GNUC_UNUSED gpointer user_data)
void
CurlSockets::DispatchSockets()
{
if (input_curl_perform())
input_curl_info_read();
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 curl_source_funcs = {
input_curl_source_prepare,
input_curl_source_check,
input_curl_source_dispatch,
nullptr,
nullptr,
nullptr,
};
/*
* input_plugin methods
*
@ -706,9 +652,7 @@ input_curl_init(const struct config_param *param,
return false;
}
curl.source = g_source_new(&curl_source_funcs, sizeof(*curl.source));
curl.source_id = g_source_attach(curl.source,
io_thread_get().GetContext());
curl.sockets = new CurlSockets(io_thread_get());
return true;
}
@ -716,7 +660,7 @@ input_curl_init(const struct config_param *param,
static gpointer
curl_destroy_sources(G_GNUC_UNUSED gpointer data)
{
g_source_destroy(curl.source);
delete curl.sockets;
return NULL;
}