mirror of
https://github.com/SerenityOS/serenity.git
synced 2026-11-07 12:38:38 -10:00
Sönke Holz
f5cbee83cc
The ThreadPool destructor joins all threads. Therefore, this destructor should be called before the destructors of all static variables that are used in those threads.
440 lines
15 KiB
C++
440 lines
15 KiB
C++
/*
|
|
* Copyright (c) 2018-2020, Andreas Kling <[email protected]>
|
|
*
|
|
* SPDX-License-Identifier: BSD-2-Clause
|
|
*/
|
|
|
|
#include <AK/Badge.h>
|
|
#include <AK/IDAllocator.h>
|
|
#include <AK/NonnullOwnPtr.h>
|
|
#include <AK/RefCounted.h>
|
|
#include <AK/Weakable.h>
|
|
#include <LibCore/Proxy.h>
|
|
#include <LibCore/Socket.h>
|
|
#include <LibWebSocket/ConnectionInfo.h>
|
|
#include <LibWebSocket/Message.h>
|
|
#include <RequestServer/ConnectionFromClient.h>
|
|
#include <RequestServer/Protocol.h>
|
|
#include <RequestServer/Request.h>
|
|
#include <RequestServer/RequestClientEndpoint.h>
|
|
#include <netdb.h>
|
|
|
|
namespace RequestServer {
|
|
|
|
struct ThreadPipeFds {
|
|
ThreadPipeFds()
|
|
: fds(MUST(Core::System::pipe2(O_CLOEXEC | O_NONBLOCK)))
|
|
{
|
|
}
|
|
|
|
~ThreadPipeFds()
|
|
{
|
|
close(fds[0]);
|
|
close(fds[1]);
|
|
}
|
|
|
|
auto& operator[](size_t x) { return fds[x]; }
|
|
auto const& operator[](size_t x) const { return fds[x]; }
|
|
|
|
Array<int, 2> fds { -1, -1 };
|
|
};
|
|
|
|
struct ThreadPoolEntry {
|
|
NonnullRefPtr<ConnectionFromClient> client;
|
|
ConnectionFromClient::Work work;
|
|
};
|
|
|
|
static HashMap<int, RefPtr<ConnectionFromClient>> s_connections;
|
|
static IDAllocator s_client_ids;
|
|
static ThreadPipeFds s_thread_pipe_fds {};
|
|
static OwnPtr<Threading::ThreadPool<ThreadPoolEntry, ConnectionFromClient::Looper>> s_thread_pool;
|
|
|
|
ConnectionFromClient::ConnectionFromClient(NonnullOwnPtr<Core::LocalSocket> socket)
|
|
: IPC::ConnectionFromClient<RequestClientEndpoint, RequestServerEndpoint>(*this, move(socket), s_client_ids.allocate())
|
|
{
|
|
s_connections.set(client_id(), *this);
|
|
}
|
|
|
|
ConnectionFromClient::~ConnectionFromClient() = default;
|
|
|
|
class Job : public RefCounted<Job>
|
|
, public Weakable<Job> {
|
|
public:
|
|
static NonnullRefPtr<Job> ensure(URL::URL const& url)
|
|
{
|
|
RefPtr<Job> job;
|
|
if (auto it = s_jobs.find(url); it != s_jobs.end())
|
|
job = it->value.strong_ref();
|
|
if (job == nullptr) {
|
|
job = adopt_ref(*new Job(url));
|
|
s_jobs.set(url, job);
|
|
}
|
|
return *job;
|
|
}
|
|
|
|
void start(Core::Socket& socket)
|
|
{
|
|
auto is_connected = socket.is_open();
|
|
VERIFY(is_connected);
|
|
ConnectionCache::request_did_finish(m_url, &socket);
|
|
}
|
|
|
|
void fail(Core::NetworkJob::Error error)
|
|
{
|
|
dbgln("Pre-connect to {} failed: {}", m_url, Core::to_string(error));
|
|
}
|
|
|
|
void will_be_destroyed() const
|
|
{
|
|
s_jobs.remove(m_url);
|
|
}
|
|
|
|
URL::URL const& url() const { return m_url; }
|
|
|
|
private:
|
|
explicit Job(URL::URL url)
|
|
: m_url(move(url))
|
|
{
|
|
}
|
|
|
|
URL::URL m_url;
|
|
inline static HashMap<URL::URL, WeakPtr<Job>> s_jobs {};
|
|
};
|
|
|
|
template<typename Pool>
|
|
IterationDecision ConnectionFromClient::Looper<Pool>::next(Pool& pool, bool wait)
|
|
{
|
|
if (done)
|
|
return IterationDecision::Break;
|
|
|
|
auto should_quit = false;
|
|
|
|
auto exit_timer = Core::Timer::create_repeating(100, [&] {
|
|
if (pool.was_exit_requested()) {
|
|
done = true;
|
|
should_quit = true;
|
|
event_loop.quit(0);
|
|
}
|
|
});
|
|
|
|
exit_timer->start();
|
|
|
|
notifier->on_activation = [&] {
|
|
char buffer[1];
|
|
auto nread = read(notifier->fd(), buffer, 1);
|
|
if (nread == 1) {
|
|
if (pool.was_exit_requested()) {
|
|
done = true;
|
|
should_quit = true;
|
|
} else {
|
|
should_quit = Threading::ThreadPoolLooper<Pool>::next(pool, true) == IterationDecision::Break;
|
|
}
|
|
|
|
if (should_quit)
|
|
event_loop.quit(0);
|
|
}
|
|
};
|
|
|
|
if (!wait) {
|
|
event_loop.deferred_invoke([&] {
|
|
event_loop.quit(0);
|
|
});
|
|
}
|
|
|
|
event_loop.exec();
|
|
|
|
exit_timer->stop();
|
|
notifier->on_activation = nullptr;
|
|
|
|
if (should_quit)
|
|
return IterationDecision::Break;
|
|
return IterationDecision::Continue;
|
|
}
|
|
|
|
void ConnectionFromClient::worker_do_work(Work work)
|
|
{
|
|
work.visit(
|
|
[&](StartRequest& start_request) {
|
|
auto* protocol = Protocol::find_by_name(start_request.url.scheme().to_byte_string());
|
|
if (!protocol) {
|
|
dbgln("StartRequest: No protocol handler for URL: '{}'", start_request.url);
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
(void)post_message(Messages::RequestClient::RequestFinished(start_request.request_id, false, 0));
|
|
return;
|
|
}
|
|
auto request = protocol->start_request(start_request.request_id, *this, start_request.method, start_request.url, start_request.request_headers, start_request.request_body, start_request.proxy_data);
|
|
if (!request) {
|
|
dbgln("StartRequest: Protocol handler failed to start request: '{}'", start_request.url);
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
(void)post_message(Messages::RequestClient::RequestFinished(start_request.request_id, false, 0));
|
|
return;
|
|
}
|
|
auto id = request->id();
|
|
auto fd = request->request_fd();
|
|
m_requests.with_locked([&](auto& map) { map.set(id, move(request)); });
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
(void)post_message(Messages::RequestClient::RequestStarted(start_request.request_id, IPC::File::adopt_fd(fd)));
|
|
},
|
|
[&](EnsureConnection& ensure_connection) {
|
|
auto& url = ensure_connection.url;
|
|
auto& cache_level = ensure_connection.cache_level;
|
|
|
|
if (cache_level == CacheLevel::ResolveOnly) {
|
|
Core::deferred_invoke([host = url.serialized_host().release_value_but_fixme_should_propagate_errors().to_byte_string()] {
|
|
dbgln("EnsureConnection: DNS-preload for {}", host);
|
|
auto resolved_host = Core::Socket::resolve_host(host, Core::Socket::SocketType::Stream);
|
|
if (resolved_host.is_error())
|
|
dbgln("EnsureConnection: DNS-preload failed for {}", host);
|
|
});
|
|
dbgln("EnsureConnection: DNS-preload for {} done", url);
|
|
return;
|
|
}
|
|
|
|
auto job = Job::ensure(url);
|
|
dbgln("EnsureConnection: Pre-connect to {}", url);
|
|
auto do_preconnect = [&](auto& cache) {
|
|
ConnectionCache::ensure_connection(cache, url, job);
|
|
};
|
|
|
|
if (url.scheme() == "http"sv)
|
|
do_preconnect(ConnectionCache::g_tcp_connection_cache);
|
|
else if (url.scheme() == "https"sv)
|
|
do_preconnect(ConnectionCache::g_tls_connection_cache);
|
|
else
|
|
dbgln("EnsureConnection: Invalid URL scheme: '{}'", url.scheme());
|
|
},
|
|
[&](Empty) {});
|
|
}
|
|
|
|
void ConnectionFromClient::die()
|
|
{
|
|
auto client_id = this->client_id();
|
|
s_connections.remove(client_id);
|
|
s_client_ids.deallocate(client_id);
|
|
|
|
if (s_connections.is_empty())
|
|
Core::EventLoop::current().quit(0);
|
|
}
|
|
|
|
Messages::RequestServer::ConnectNewClientResponse ConnectionFromClient::connect_new_client()
|
|
{
|
|
int socket_fds[2] {};
|
|
if (auto err = Core::System::socketpair(AF_LOCAL, SOCK_STREAM, 0, socket_fds); err.is_error()) {
|
|
dbgln("Failed to create client socketpair: {}", err.error());
|
|
return IPC::File {};
|
|
}
|
|
|
|
auto client_socket_or_error = Core::LocalSocket::adopt_fd(socket_fds[0]);
|
|
if (client_socket_or_error.is_error()) {
|
|
close(socket_fds[0]);
|
|
close(socket_fds[1]);
|
|
dbgln("Failed to adopt client socket: {}", client_socket_or_error.error());
|
|
return IPC::File {};
|
|
}
|
|
auto client_socket = client_socket_or_error.release_value();
|
|
// Note: A ref is stored in the static s_connections map
|
|
auto client = adopt_ref(*new ConnectionFromClient(move(client_socket)));
|
|
|
|
return IPC::File::adopt_fd(socket_fds[1]);
|
|
}
|
|
|
|
void ConnectionFromClient::enqueue(Work work)
|
|
{
|
|
if (!s_thread_pool) {
|
|
s_thread_pool = adopt_own(*new (nothrow) Threading::ThreadPool<ThreadPoolEntry, ConnectionFromClient::Looper>(
|
|
[](ThreadPoolEntry entry) {
|
|
entry.client->worker_do_work(move(entry.work));
|
|
},
|
|
{}, s_thread_pipe_fds.fds[0]));
|
|
}
|
|
s_thread_pool->submit({ *this, move(work) });
|
|
auto nwritten = write(s_thread_pipe_fds[1], "x", 1); // notify the worker threads
|
|
if (nwritten < 0) {
|
|
VERIFY_NOT_REACHED();
|
|
}
|
|
}
|
|
|
|
Messages::RequestServer::IsSupportedProtocolResponse ConnectionFromClient::is_supported_protocol(ByteString const& protocol)
|
|
{
|
|
bool supported = Protocol::find_by_name(protocol.to_lowercase());
|
|
return supported;
|
|
}
|
|
|
|
void ConnectionFromClient::start_request(i32 request_id, ByteString const& method, URL::URL const& url, HTTP::HeaderMap const& request_headers, ByteBuffer const& request_body, Core::ProxyData const& proxy_data)
|
|
{
|
|
if (!url.is_valid()) {
|
|
dbgln("StartRequest: Invalid URL requested: '{}'", url);
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
(void)post_message(Messages::RequestClient::RequestFinished(request_id, false, 0));
|
|
return;
|
|
}
|
|
|
|
auto headers = request_headers;
|
|
if (!headers.contains("Accept-Encoding"))
|
|
headers.set("Accept-Encoding", "gzip, deflate, br");
|
|
|
|
enqueue(StartRequest {
|
|
.request_id = request_id,
|
|
.method = method,
|
|
.url = url,
|
|
.request_headers = move(headers),
|
|
.request_body = request_body,
|
|
.proxy_data = proxy_data,
|
|
});
|
|
}
|
|
|
|
Messages::RequestServer::StopRequestResponse ConnectionFromClient::stop_request(i32 request_id)
|
|
{
|
|
return m_requests.with_locked([&](auto& map) {
|
|
auto* request = const_cast<Request*>(map.get(request_id).value_or(nullptr));
|
|
bool success = false;
|
|
if (request) {
|
|
request->stop();
|
|
map.remove(request_id);
|
|
success = true;
|
|
}
|
|
return success;
|
|
});
|
|
}
|
|
|
|
void ConnectionFromClient::did_receive_headers(Badge<Request>, Request& request)
|
|
{
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_headers_became_available(request.id(), request.response_headers(), request.status_code());
|
|
}
|
|
|
|
void ConnectionFromClient::did_finish_request(Badge<Request>, Request& request, bool success)
|
|
{
|
|
if (request.total_size().has_value()) {
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_request_finished(request.id(), success, request.total_size().value());
|
|
}
|
|
|
|
m_requests.with_locked([&](auto& map) { map.remove(request.id()); });
|
|
}
|
|
|
|
void ConnectionFromClient::did_progress_request(Badge<Request>, Request& request)
|
|
{
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_request_progress(request.id(), request.total_size(), request.downloaded_size());
|
|
}
|
|
|
|
void ConnectionFromClient::did_request_certificates(Badge<Request>, Request& request)
|
|
{
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_certificate_requested(request.id());
|
|
}
|
|
|
|
Messages::RequestServer::SetCertificateResponse ConnectionFromClient::set_certificate(i32 request_id, ByteString const& certificate, ByteString const& key)
|
|
{
|
|
return m_requests.with_locked([&](auto& map) {
|
|
auto* request = const_cast<Request*>(map.get(request_id).value_or(nullptr));
|
|
bool success = false;
|
|
if (request) {
|
|
request->set_certificate(certificate, key);
|
|
success = true;
|
|
}
|
|
return success;
|
|
});
|
|
}
|
|
|
|
void ConnectionFromClient::ensure_connection(URL::URL const& url, ::RequestServer::CacheLevel const& cache_level)
|
|
{
|
|
if (!url.is_valid()) {
|
|
dbgln("EnsureConnection: Invalid URL requested: '{}'", url);
|
|
return;
|
|
}
|
|
|
|
enqueue(EnsureConnection {
|
|
.url = url,
|
|
.cache_level = cache_level,
|
|
});
|
|
}
|
|
|
|
static i32 s_next_websocket_id = 1;
|
|
Messages::RequestServer::WebsocketConnectResponse ConnectionFromClient::websocket_connect(URL::URL const& url, ByteString const& origin, Vector<ByteString> const& protocols, Vector<ByteString> const& extensions, HTTP::HeaderMap const& additional_request_headers)
|
|
{
|
|
if (!url.is_valid()) {
|
|
dbgln("WebSocket::Connect: Invalid URL requested: '{}'", url);
|
|
return -1;
|
|
}
|
|
|
|
WebSocket::ConnectionInfo connection_info(url);
|
|
connection_info.set_origin(origin);
|
|
connection_info.set_protocols(protocols);
|
|
connection_info.set_extensions(extensions);
|
|
connection_info.set_headers(additional_request_headers);
|
|
|
|
auto id = ++s_next_websocket_id;
|
|
auto connection = WebSocket::WebSocket::create(move(connection_info));
|
|
connection->on_open = [this, id]() {
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_websocket_connected(id);
|
|
};
|
|
connection->on_message = [this, id](auto message) {
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_websocket_received(id, message.is_text(), message.data());
|
|
};
|
|
connection->on_error = [this, id](auto message) {
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_websocket_errored(id, (i32)message);
|
|
};
|
|
connection->on_close = [this, id](u16 code, ByteString reason, bool was_clean) {
|
|
auto lock = Threading::MutexLocker(m_ipc_mutex);
|
|
async_websocket_closed(id, code, move(reason), was_clean);
|
|
};
|
|
|
|
connection->start();
|
|
m_websockets.set(id, move(connection));
|
|
return id;
|
|
}
|
|
|
|
Messages::RequestServer::WebsocketReadyStateResponse ConnectionFromClient::websocket_ready_state(i32 connection_id)
|
|
{
|
|
if (auto connection = m_websockets.get(connection_id).value_or({}))
|
|
return (u32)connection->ready_state();
|
|
return (u32)WebSocket::ReadyState::Closed;
|
|
}
|
|
|
|
Messages::RequestServer::WebsocketSubprotocolInUseResponse ConnectionFromClient::websocket_subprotocol_in_use(i32 connection_id)
|
|
{
|
|
if (auto connection = m_websockets.get(connection_id).value_or({}))
|
|
return connection->subprotocol_in_use();
|
|
return ByteString::empty();
|
|
}
|
|
|
|
void ConnectionFromClient::websocket_send(i32 connection_id, bool is_text, ByteBuffer const& data)
|
|
{
|
|
if (auto connection = m_websockets.get(connection_id).value_or({}); connection && connection->ready_state() == WebSocket::ReadyState::Open)
|
|
connection->send(WebSocket::Message { data, is_text });
|
|
}
|
|
|
|
void ConnectionFromClient::websocket_close(i32 connection_id, u16 code, ByteString const& reason)
|
|
{
|
|
if (auto connection = m_websockets.get(connection_id).value_or({}); connection && connection->ready_state() == WebSocket::ReadyState::Open)
|
|
connection->close(code, reason);
|
|
}
|
|
|
|
Messages::RequestServer::WebsocketSetCertificateResponse ConnectionFromClient::websocket_set_certificate(i32 connection_id, ByteString const&, ByteString const&)
|
|
{
|
|
auto success = false;
|
|
if (auto connection = m_websockets.get(connection_id).value_or({}); connection) {
|
|
// NO OP here
|
|
// connection->set_certificate(certificate, key);
|
|
success = true;
|
|
}
|
|
return success;
|
|
}
|
|
|
|
void ConnectionFromClient::dump_connection_info()
|
|
{
|
|
ConnectionCache::dump_jobs();
|
|
}
|
|
|
|
void ConnectionFromClient::destroy_thread_pool()
|
|
{
|
|
s_thread_pool.clear();
|
|
}
|
|
|
|
}
|