RGBCube/serenity
1
Fork 0
mirror of https://github.com/RGBCube/serenity synced 2025-07-26 16:07:45 +00:00
Code Issues Activity Actions

LibIPC: Share most of the code between {Client,Server}Connection

Browse source 
This patch introduces IPC::Connection which becomes the new base class
of ClientConnection and ServerConnection. Most of the functionality
has been hoisted up to the base class since almost all of it is useful
on both sides.

This gives us the ability to send synchronous messages in both
directions, which is needed for the WebContent server process.
Unlike other servers, WebContent does not mind blocking on a response
from its client.
This commit is contained in:
Andreas Kling 2020-09-12 11:44:00 +02:00
parent 633e0bc944
commit aba793fb3e
19 changed files with 293 additions and 260 deletions
Download patch file Download diff file Expand all files Collapse all files

128
Libraries/LibIPC/ClientConnection.h
View file

@ -32,6 +32,7 @@
#include <LibCore/LocalSocket.h>
#include <LibCore/Object.h>
#include <LibCore/Timer.h>
#include <LibIPC/Connection.h>
#include <LibIPC/Endpoint.h>
#include <LibIPC/Message.h>
#include <errno.h>
@ -75,133 +76,38 @@ NonnullRefPtr<T> new_client_connection(Args&&... args)
return T::construct(forward<Args>(args)...) /* arghs */;
}
template<typename Endpoint>
class ClientConnection : public Core::Object {
template<typename ClientEndpoint, typename ServerEndpoint>
class ClientConnection : public Connection<ServerEndpoint, ClientEndpoint> {
public:
ClientConnection(Endpoint& endpoint, NonnullRefPtr<Core::LocalSocket> socket, int client_id)
: m_endpoint(endpoint)
, m_socket(move(socket))
ClientConnection(ServerEndpoint& endpoint, NonnullRefPtr<Core::LocalSocket> socket, int client_id)
: IPC::Connection<ServerEndpoint, ClientEndpoint>(endpoint, move(socket))
, m_client_id(client_id)
{
ASSERT(m_socket->is_connected());
ucred creds;
socklen_t creds_size = sizeof(creds);
if (getsockopt(m_socket->fd(), SOL_SOCKET, SO_PEERCRED, &creds, &creds_size) < 0) {
ASSERT_NOT_REACHED();
}
m_client_pid = creds.pid;
m_socket->on_ready_to_read = [this] { drain_messages_from_client(); };
m_responsiveness_timer = Core::Timer::create_single_shot(3000, [this] { may_have_become_unresponsive(); });
ASSERT(this->socket().is_connected());
this->socket().on_ready_to_read = [this] { this->drain_messages_from_peer(); };
this->initialize_peer_info();
}
virtual ~ClientConnection() override
{
}
virtual void may_have_become_unresponsive() { }
virtual void did_become_responsive() { }
void post_message(const Message& message)
{
// NOTE: If this connection is being shut down, but has not yet been destroyed,
// the socket will be closed. Don't try to send more messages.
if (!m_socket->is_open())
return;
auto buffer = message.encode();
auto bytes_remaining = buffer.size();
while (bytes_remaining) {
auto nwritten = write(m_socket->fd(), buffer.data(), buffer.size());
if (nwritten < 0) {
switch (errno) {
case EPIPE:
dbg() << *this << "::post_message: Disconnected from peer";
shutdown();
return;
case EAGAIN:
dbg() << *this << "::post_message: Client buffer overflowed.";
did_misbehave();
return;
default:
perror("Connection::post_message write");
shutdown();
return;
}
}
bytes_remaining -= nwritten;
}
m_responsiveness_timer->start();
}
void drain_messages_from_client()
{
if (!m_socket->is_open())
return;
Vector<u8> bytes;
for (;;) {
u8 buffer[4096];
ssize_t nread = recv(m_socket->fd(), buffer, sizeof(buffer), MSG_DONTWAIT);
if (nread == 0 || (nread == -1 && errno == EAGAIN)) {
if (bytes.is_empty()) {
Core::EventLoop::current().post_event(*this, make<DisconnectedEvent>(client_id()));
return;
}
break;
}
if (nread < 0) {
perror("recv");
shutdown();
return;
}
bytes.append(buffer, nread);
}
if (!bytes.is_empty()) {
m_responsiveness_timer->stop();
did_become_responsive();
}
size_t decoded_bytes = 0;
for (size_t index = 0; index < bytes.size(); index += decoded_bytes) {
auto remaining_bytes = ByteBuffer::wrap(bytes.data() + index, bytes.size() - index);
auto message = Endpoint::decode_message(remaining_bytes, decoded_bytes);
if (!message) {
dbg() << "drain_messages_from_client: Endpoint didn't recognize message";
did_misbehave();
return;
}
if (auto response = m_endpoint.handle(*message))
post_message(*response);
ASSERT(decoded_bytes);
}
}
void did_misbehave()
{
dbg() << *this << " (id=" << m_client_id << ", pid=" << m_client_pid << ") misbehaved, disconnecting.";
shutdown();
dbg() << *this << " (id=" << m_client_id << ", pid=" << client_pid() << ") misbehaved, disconnecting.";
this->shutdown();
}
void did_misbehave(const char* message)
{
dbg() << *this << " (id=" << m_client_id << ", pid=" << m_client_pid << ") misbehaved (" << message << "), disconnecting.";
shutdown();
}
void shutdown()
{
m_socket->close();
die();
dbg() << *this << " (id=" << m_client_id << ", pid=" << client_pid() << ") misbehaved (" << message << "), disconnecting.";
this->shutdown();
}
int client_id() const { return m_client_id; }
pid_t client_pid() const { return m_client_pid; }
void set_client_pid(pid_t pid) { m_client_pid = pid; }
pid_t client_pid() const { return this->peer_pid(); }
void set_client_pid(pid_t pid) { this->set_peer_pid(pid); }
virtual void die() = 0;
@ -213,7 +119,7 @@ protected:
int client_id = static_cast<const DisconnectedEvent&>(event).client_id();
dbg() << *this << ": Client disconnected: " << client_id;
#endif
die();
this->die();
return;
}
@ -221,11 +127,7 @@ protected:
}
private:
Endpoint& m_endpoint;
NonnullRefPtr<Core::LocalSocket> m_socket;
RefPtr<Core::Timer> m_responsiveness_timer;
int m_client_id { -1 };
int m_client_pid { -1 };
};
}