LadybirdBrowser/ladybird
1
1
Fork 0
mirror of https://github.com/LadybirdBrowser/ladybird.git synced 2024-11-10 13:00:29 +03:00
Code Issues Packages Projects Releases Wiki Activity

LibIPC: Rename base classes to IClientConnection and IServerConnection

Browse Source 
This matches what we're already calling the server-side subclasses
better, though we'll probably want to find some better names for the
client-side classes eventually.
This commit is contained in:
Andreas Kling 2019-12-02 09:58:25 +01:00
parent 5d4ee0f58a
commit 4a37bec27c
Notes: sideshowbarker 2024-07-19 10:59:47 +09:00 
Author: https://github.com/awesomekling
Commit: https://github.com/SerenityOS/serenity/commit/4a37bec27c4
16 changed files with 254 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Libraries/LibAudio/AClientConnection.cpp
View File

@ -3,7 +3,7 @@
#include <SharedBuffer.h>
AClientConnection::AClientConnection()
: ConnectionNG(*this, "/tmp/portal/audio")
: IServerConnection(*this, "/tmp/portal/audio")
{
}

2
Libraries/LibAudio/AClientConnection.h
View File

@ -6,7 +6,7 @@
class ABuffer;
class AClientConnection : public IPC::Client::ConnectionNG<AudioClientEndpoint, AudioServerEndpoint>
class AClientConnection : public IServerConnection<AudioClientEndpoint, AudioServerEndpoint>
, public AudioClientEndpoint {
C_OBJECT(AClientConnection)
public:

4
Libraries/LibGUI/GWindowServerConnection.h
View File

@ -5,12 +5,12 @@
#include <WindowServer/WindowServerEndpoint.h>
class GWindowServerConnection
: public IPC::Client::ConnectionNG<WindowClientEndpoint, WindowServerEndpoint>
: public IServerConnection<WindowClientEndpoint, WindowServerEndpoint>
, public WindowClientEndpoint {
C_OBJECT(GWindowServerConnection)
public:
GWindowServerConnection()
: ConnectionNG(*this, "/tmp/portal/window")
: IServerConnection(*this, "/tmp/portal/window")
{
handshake();
}

259
Libraries/LibIPC/IClientConnection.h
View File

@ -1,6 +1,5 @@
#pragma once
#include <AK/Queue.h>
#include <LibCore/CEvent.h>
#include <LibCore/CEventLoop.h>
#include <LibCore/CIODevice.h>
@ -10,163 +9,153 @@
#include <LibIPC/IEndpoint.h>
#include <LibIPC/IMessage.h>
#include <errno.h>
#include <sched.h>
#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <unistd.h>
//#define CIPC_DEBUG
namespace IPC {
namespace Server {
class Event : public CEvent {
public:
enum Type {
Invalid = 2000,
Disconnected,
};
Event() {}
explicit Event(Type type)
: CEvent(type)
{
}
class IEvent : public CEvent {
public:
enum Type {
Invalid = 2000,
Disconnected,
};
class DisconnectedEvent : public Event {
public:
explicit DisconnectedEvent(int client_id)
: Event(Disconnected)
, m_client_id(client_id)
{
}
int client_id() const { return m_client_id; }
private:
int m_client_id { 0 };
};
template<typename T, class... Args>
NonnullRefPtr<T> new_connection_ng_for_client(Args&&... args)
IEvent() {}
explicit IEvent(Type type)
: CEvent(type)
{
}
};
class IDisconnectedEvent : public IEvent {
public:
explicit IDisconnectedEvent(int client_id)
: IEvent(Disconnected)
, m_client_id(client_id)
{
return T::construct(forward<Args>(args)...) /* arghs */;
}
template<typename Endpoint>
class ConnectionNG : public CObject {
public:
ConnectionNG(Endpoint& endpoint, CLocalSocket& socket, int client_id)
: m_endpoint(endpoint)
, m_socket(socket)
, m_client_id(client_id)
{
add_child(socket);
m_socket->on_ready_to_read = [this] { drain_messages_from_client(); };
}
int client_id() const { return m_client_id; }
virtual ~ConnectionNG() override
{
}
private:
int m_client_id { 0 };
};
void post_message(const IMessage& message)
{
auto buffer = message.encode();
template<typename T, class... Args>
NonnullRefPtr<T> new_client_connection(Args&&... args)
{
return T::construct(forward<Args>(args)...) /* arghs */;
}
int nwritten = write(m_socket->fd(), buffer.data(), (size_t)buffer.size());
if (nwritten < 0) {
switch (errno) {
case EPIPE:
dbg() << "Connection::post_message: Disconnected from peer";
shutdown();
return;
case EAGAIN:
dbg() << "Connection::post_message: Client buffer overflowed.";
did_misbehave();
return;
default:
perror("Connection::post_message write");
ASSERT_NOT_REACHED();
}
}
template<typename Endpoint>
class IClientConnection : public CObject {
public:
IClientConnection(Endpoint& endpoint, CLocalSocket& socket, int client_id)
: m_endpoint(endpoint)
, m_socket(socket)
, m_client_id(client_id)
{
add_child(socket);
m_socket->on_ready_to_read = [this] { drain_messages_from_client(); };
}
ASSERT(nwritten == buffer.size());
}
virtual ~IClientConnection() override
{
}
void drain_messages_from_client()
{
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()) {
CEventLoop::current().post_event(*this, make<DisconnectedEvent>(client_id()));
return;
}
break;
}
if (nread < 0) {
perror("recv");
ASSERT_NOT_REACHED();
}
bytes.append(buffer, nread);
}
void post_message(const IMessage& message)
{
auto buffer = message.encode();
size_t decoded_bytes = 0;
for (size_t index = 0; index < (size_t)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);
int nwritten = write(m_socket->fd(), buffer.data(), (size_t)buffer.size());
if (nwritten < 0) {
switch (errno) {
case EPIPE:
dbg() << "Connection::post_message: Disconnected from peer";
shutdown();
return;
case EAGAIN:
dbg() << "Connection::post_message: Client buffer overflowed.";
did_misbehave();
return;
default:
perror("Connection::post_message write");
ASSERT_NOT_REACHED();
}
}
void did_misbehave()
{
dbg() << "Connection{" << this << "} (id=" << m_client_id << ", pid=" << m_client_pid << ") misbehaved, disconnecting.";
shutdown();
ASSERT(nwritten == buffer.size());
}
void drain_messages_from_client()
{
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()) {
CEventLoop::current().post_event(*this, make<IDisconnectedEvent>(client_id()));
return;
}
break;
}
if (nread < 0) {
perror("recv");
ASSERT_NOT_REACHED();
}
bytes.append(buffer, nread);
}
void shutdown()
{
m_socket->close();
die();
}
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; }
virtual void die() = 0;
protected:
void event(CEvent& event) override
{
if (event.type() == Event::Disconnected) {
int client_id = static_cast<const DisconnectedEvent&>(event).client_id();
dbgprintf("Connection: Client disconnected: %d\n", client_id);
die();
size_t decoded_bytes = 0;
for (size_t index = 0; index < (size_t)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);
}
}
CObject::event(event);
void did_misbehave()
{
dbg() << "Connection{" << this << "} (id=" << m_client_id << ", pid=" << m_client_pid << ") misbehaved, disconnecting.";
shutdown();
}
void shutdown()
{
m_socket->close();
die();
}
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; }
virtual void die() = 0;
protected:
void event(CEvent& event) override
{
if (event.type() == IEvent::Disconnected) {
int client_id = static_cast<const IDisconnectedEvent&>(event).client_id();
dbgprintf("Connection: Client disconnected: %d\n", client_id);
die();
return;
}
private:
Endpoint& m_endpoint;
RefPtr<CLocalSocket> m_socket;
int m_client_id { -1 };
int m_client_pid { -1 };
};
CObject::event(event);
}
} // Server
} // IPC
private:
Endpoint& m_endpoint;
RefPtr<CLocalSocket> m_socket;
int m_client_id { -1 };
int m_client_pid { -1 };
};

240
Libraries/LibIPC/IServerConnection.h
View File

@ -6,59 +6,71 @@
#include <LibCore/CNotifier.h>
#include <LibCore/CSyscallUtils.h>
#include <LibIPC/IMessage.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <unistd.h>
//#define CIPC_DEBUG
template<typename LocalEndpoint, typename PeerEndpoint>
class IServerConnection : public CObject {
public:
IServerConnection(LocalEndpoint& local_endpoint, const StringView& address)
: m_local_endpoint(local_endpoint)
, m_connection(CLocalSocket::construct(this))
, m_notifier(CNotifier::construct(m_connection->fd(), CNotifier::Read, this))
{
// We want to rate-limit our clients
m_connection->set_blocking(true);
m_notifier->on_ready_to_read = [this] {
drain_messages_from_server();
};
namespace IPC {
namespace Client {
template<typename LocalEndpoint, typename PeerEndpoint>
class ConnectionNG : public CObject {
public:
ConnectionNG(LocalEndpoint& local_endpoint, const StringView& address)
: m_local_endpoint(local_endpoint)
, m_connection(CLocalSocket::construct(this))
, m_notifier(CNotifier::construct(m_connection->fd(), CNotifier::Read, this))
{
// We want to rate-limit our clients
m_connection->set_blocking(true);
m_notifier->on_ready_to_read = [this] {
drain_messages_from_server();
};
int retries = 100000;
while (retries) {
if (m_connection->connect(CSocketAddress::local(address))) {
break;
}
dbgprintf("Client::Connection: connect failed: %d, %s\n", errno, strerror(errno));
usleep(10000);
--retries;
int retries = 100000;
while (retries) {
if (m_connection->connect(CSocketAddress::local(address))) {
break;
}
ASSERT(m_connection->is_connected());
dbgprintf("Client::Connection: connect failed: %d, %s\n", errno, strerror(errno));
usleep(10000);
--retries;
}
ASSERT(m_connection->is_connected());
}
virtual void handshake() = 0;
virtual void handshake() = 0;
void set_server_pid(pid_t pid) { m_server_pid = pid; }
pid_t server_pid() const { return m_server_pid; }
void set_my_client_id(int id) { m_my_client_id = id; }
int my_client_id() const { return m_my_client_id; }
void set_server_pid(pid_t pid) { m_server_pid = pid; }
pid_t server_pid() const { return m_server_pid; }
void set_my_client_id(int id) { m_my_client_id = id; }
int my_client_id() const { return m_my_client_id; }
template<typename MessageType>
OwnPtr<MessageType> wait_for_specific_message()
{
// Double check we don't already have the event waiting for us.
// Otherwise we might end up blocked for a while for no reason.
template<typename MessageType>
OwnPtr<MessageType> wait_for_specific_message()
{
// Double check we don't already have the event waiting for us.
// Otherwise we might end up blocked for a while for no reason.
for (ssize_t i = 0; i < m_unprocessed_messages.size(); ++i) {
if (m_unprocessed_messages[i]->id() == MessageType::static_message_id()) {
auto message = move(m_unprocessed_messages[i]);
m_unprocessed_messages.remove(i);
return message;
}
}
for (;;) {
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(m_connection->fd(), &rfds);
int rc = CSyscallUtils::safe_syscall(select, m_connection->fd() + 1, &rfds, nullptr, nullptr, nullptr);
if (rc < 0) {
perror("select");
}
ASSERT(rc > 0);
ASSERT(FD_ISSET(m_connection->fd(), &rfds));
if (!drain_messages_from_server())
return nullptr;
for (ssize_t i = 0; i < m_unprocessed_messages.size(); ++i) {
if (m_unprocessed_messages[i]->id() == MessageType::static_message_id()) {
auto message = move(m_unprocessed_messages[i]);
@ -66,96 +78,74 @@ namespace Client {
return message;
}
}
for (;;) {
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(m_connection->fd(), &rfds);
int rc = CSyscallUtils::safe_syscall(select, m_connection->fd() + 1, &rfds, nullptr, nullptr, nullptr);
if (rc < 0) {
perror("select");
}
ASSERT(rc > 0);
ASSERT(FD_ISSET(m_connection->fd(), &rfds));
if (!drain_messages_from_server())
return nullptr;
for (ssize_t i = 0; i < m_unprocessed_messages.size(); ++i) {
if (m_unprocessed_messages[i]->id() == MessageType::static_message_id()) {
auto message = move(m_unprocessed_messages[i]);
m_unprocessed_messages.remove(i);
return message;
}
}
}
}
}
bool post_message(const IMessage& message)
{
auto buffer = message.encode();
int nwritten = write(m_connection->fd(), buffer.data(), (size_t)buffer.size());
if (nwritten < 0) {
perror("write");
ASSERT_NOT_REACHED();
bool post_message(const IMessage& message)
{
auto buffer = message.encode();
int nwritten = write(m_connection->fd(), buffer.data(), (size_t)buffer.size());
if (nwritten < 0) {
perror("write");
ASSERT_NOT_REACHED();
return false;
}
ASSERT(nwritten == buffer.size());
return true;
}
template<typename RequestType, typename... Args>
OwnPtr<typename RequestType::ResponseType> send_sync(Args&&... args)
{
bool success = post_message(RequestType(forward<Args>(args)...));
ASSERT(success);
auto response = wait_for_specific_message<typename RequestType::ResponseType>();
ASSERT(response);
return response;
}
private:
bool drain_messages_from_server()
{
Vector<u8> bytes;
for (;;) {
u8 buffer[4096];
ssize_t nread = recv(m_connection->fd(), buffer, sizeof(buffer), MSG_DONTWAIT);
if (nread < 0) {
if (errno == EAGAIN)
break;
perror("read");
exit(1);
return false;
}
ASSERT(nwritten == buffer.size());
return true;
}
template<typename RequestType, typename... Args>
OwnPtr<typename RequestType::ResponseType> send_sync(Args&&... args)
{
bool success = post_message(RequestType(forward<Args>(args)...));
ASSERT(success);
auto response = wait_for_specific_message<typename RequestType::ResponseType>();
ASSERT(response);
return response;
}
private:
bool drain_messages_from_server()
{
Vector<u8> bytes;
for (;;) {
u8 buffer[4096];
ssize_t nread = recv(m_connection->fd(), buffer, sizeof(buffer), MSG_DONTWAIT);
if (nread < 0) {
if (errno == EAGAIN)
break;
perror("read");
exit(1);
return false;
}
if (nread == 0) {
dbg() << "EOF on IPC fd";
// FIXME: Dying is definitely not always appropriate!
exit(1);
return false;
}
bytes.append(buffer, nread);
if (nread == 0) {
dbg() << "EOF on IPC fd";
// FIXME: Dying is definitely not always appropriate!
exit(1);
return false;
}
size_t decoded_bytes = 0;
for (size_t index = 0; index < (size_t)bytes.size(); index += decoded_bytes) {
auto remaining_bytes = ByteBuffer::wrap(bytes.data() + index, bytes.size() - index);
if (auto message = LocalEndpoint::decode_message(remaining_bytes, decoded_bytes)) {
m_local_endpoint.handle(*message);
} else if (auto message = PeerEndpoint::decode_message(remaining_bytes, decoded_bytes)) {
m_unprocessed_messages.append(move(message));
} else {
ASSERT_NOT_REACHED();
}
ASSERT(decoded_bytes);
}
return true;
bytes.append(buffer, nread);
}
LocalEndpoint& m_local_endpoint;
RefPtr<CLocalSocket> m_connection;
RefPtr<CNotifier> m_notifier;
Vector<OwnPtr<IMessage>> m_unprocessed_messages;
int m_server_pid { -1 };
int m_my_client_id { -1 };
};
size_t decoded_bytes = 0;
for (size_t index = 0; index < (size_t)bytes.size(); index += decoded_bytes) {
auto remaining_bytes = ByteBuffer::wrap(bytes.data() + index, bytes.size() - index);
if (auto message = LocalEndpoint::decode_message(remaining_bytes, decoded_bytes)) {
m_local_endpoint.handle(*message);
} else if (auto message = PeerEndpoint::decode_message(remaining_bytes, decoded_bytes)) {
m_unprocessed_messages.append(move(message));
} else {
ASSERT_NOT_REACHED();
}
ASSERT(decoded_bytes);
}
return true;
}
} // Client
} // IPC
LocalEndpoint& m_local_endpoint;
RefPtr<CLocalSocket> m_connection;
RefPtr<CNotifier> m_notifier;
Vector<OwnPtr<IMessage>> m_unprocessed_messages;
int m_server_pid { -1 };
int m_my_client_id { -1 };
};

2
Libraries/LibProtocol/Client.cpp
View File

@ -5,7 +5,7 @@
namespace LibProtocol {
Client::Client()
: ConnectionNG(*this, "/tmp/portal/protocol")
: IServerConnection(*this, "/tmp/portal/protocol")
{
handshake();
}

2
Libraries/LibProtocol/Client.h
View File

@ -8,7 +8,7 @@ namespace LibProtocol {
class Download;
class Client : public IPC::Client::ConnectionNG<ProtocolClientEndpoint, ProtocolServerEndpoint>
class Client : public IServerConnection<ProtocolClientEndpoint, ProtocolServerEndpoint>
, public ProtocolClientEndpoint {
C_OBJECT(Client)
public:

2
Servers/AudioServer/ASClientConnection.cpp
View File

@ -23,7 +23,7 @@ void ASClientConnection::for_each(Function<void(ASClientConnection&)> callback)
}
ASClientConnection::ASClientConnection(CLocalSocket& client_socket, int client_id, ASMixer& mixer)
: ConnectionNG(*this, client_socket, client_id)
: IClientConnection(*this, client_socket, client_id)
, m_mixer(mixer)
{
s_connections.set(client_id, *this);

2
Servers/AudioServer/ASClientConnection.h
View File

@ -7,7 +7,7 @@ class ABuffer;
class ASBufferQueue;
class ASMixer;
class ASClientConnection final : public IPC::Server::ConnectionNG<AudioServerEndpoint>
class ASClientConnection final : public IClientConnection<AudioServerEndpoint>
, public AudioServerEndpoint {
C_OBJECT(ASClientConnection)
public:

2
Servers/AudioServer/ASEventLoop.cpp
View File

@ -17,6 +17,6 @@ ASEventLoop::ASEventLoop()
}
static int s_next_client_id = 0;
int client_id = ++s_next_client_id;
IPC::Server::new_connection_ng_for_client<ASClientConnection>(*client_socket, client_id, m_mixer);
new_client_connection<ASClientConnection>(*client_socket, client_id, m_mixer);
};
}

2
Servers/ProtocolServer/PSClientConnection.cpp
View File

@ -7,7 +7,7 @@
static HashMap<int, RefPtr<PSClientConnection>> s_connections;
PSClientConnection::PSClientConnection(CLocalSocket& socket, int client_id)
: ConnectionNG(*this, socket, client_id)
: IClientConnection(*this, socket, client_id)
{
s_connections.set(client_id, *this);
}

2
Servers/ProtocolServer/PSClientConnection.h
View File

@ -7,7 +7,7 @@
class Download;
class SharedBuffer;
class PSClientConnection final : public IPC::Server::ConnectionNG<ProtocolServerEndpoint>
class PSClientConnection final : public IClientConnection<ProtocolServerEndpoint>
, public ProtocolServerEndpoint {
C_OBJECT(PSClientConnection)
public:

2
Servers/ProtocolServer/main.cpp
View File

@ -19,7 +19,7 @@ int main(int, char**)
}
static int s_next_client_id = 0;
int client_id = ++s_next_client_id;
IPC::Server::new_connection_ng_for_client<PSClientConnection>(*client_socket, client_id);
new_client_connection<PSClientConnection>(*client_socket, client_id);
};
return event_loop.exec();
}

2
Servers/WindowServer/WSClientConnection.cpp
View File

@ -42,7 +42,7 @@ WSClientConnection* WSClientConnection::from_client_id(int client_id)
}
WSClientConnection::WSClientConnection(CLocalSocket& client_socket, int client_id)
: ConnectionNG(*this, client_socket, client_id)
: IClientConnection(*this, client_socket, client_id)
{
if (!s_connections)
s_connections = new HashMap<int, NonnullRefPtr<WSClientConnection>>;

2
Servers/WindowServer/WSClientConnection.h
View File

@ -15,7 +15,7 @@ class WSMenu;
class WSMenuBar;
class WSClientConnection final
: public IPC::Server::ConnectionNG<WindowServerEndpoint>
: public IClientConnection<WindowServerEndpoint>
, public WindowServerEndpoint {
C_OBJECT(WSClientConnection)
public:

2
Servers/WindowServer/WSEventLoop.cpp
View File

@ -37,7 +37,7 @@ WSEventLoop::WSEventLoop()
}
static int s_next_client_id = 0;
int client_id = ++s_next_client_id;
IPC::Server::new_connection_ng_for_client<WSClientConnection>(*client_socket, client_id);
new_client_connection<WSClientConnection>(*client_socket, client_id);
};
ASSERT(m_keyboard_fd >= 0);