daniel/MontaukOS
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement new IPC layer

Browse Source
This commit is contained in:
Daniel Hammer
2026-04-01 13:11:03 +02:00
parent 6276f8b162
commit 346977a0f9
34 changed files with 3210 additions and 897 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kernel/src/Net/Socket.cpp
+22 -286
View File
@@ -5,324 +5,60 @@
*/
#include "Socket.hpp"
#include <Net/Tcp.hpp>
#include <Net/Udp.hpp>
#include <Net/NetConfig.hpp>
#include <Ipc/Ipc.hpp>
#include <Terminal/Terminal.hpp>
#include <CppLib/Stream.hpp>
#include <Libraries/Memory.hpp>
using namespace Kt;
namespace Net::Socket {
// ---- UDP socket state ----
static constexpr uint32_t UDP_RING_SIZE = 4096;
static constexpr int MAX_UDP_SOCKETS = 16;
struct UdpDgramHeader {
uint32_t SrcIp;
uint16_t SrcPort;
uint16_t DataLen;
};
struct UdpSocketState {
uint8_t Ring[UDP_RING_SIZE];
uint32_t Head;
uint32_t Tail;
uint32_t Count;
uint16_t LocalPort;
bool Active;
};
static UdpSocketState g_udpSockets[MAX_UDP_SOCKETS] = {};
static SocketEntry g_sockets[MAX_SOCKETS] = {};
static uint16_t g_nextEphemeralPort = 49152;
static uint16_t AllocEphemeralPort() {
uint16_t port = g_nextEphemeralPort++;
if (g_nextEphemeralPort == 0) g_nextEphemeralPort = 49152;
return port;
}
static bool ValidFd(int fd, int pid) {
if (fd < 0 || fd >= MAX_SOCKETS) return false;
if (!g_sockets[fd].Active) return false;
if (g_sockets[fd].OwnerPid != pid) return false;
return true;
}
static UdpSocketState* AllocUdpState() {
for (int i = 0; i < MAX_UDP_SOCKETS; i++) {
if (!g_udpSockets[i].Active) {
g_udpSockets[i].Active = true;
g_udpSockets[i].Head = 0;
g_udpSockets[i].Tail = 0;
g_udpSockets[i].Count = 0;
g_udpSockets[i].LocalPort = 0;
return &g_udpSockets[i];
}
}
return nullptr;
}
static void FreeUdpState(UdpSocketState* state) {
if (state) {
if (state->LocalPort != 0) {
Udp::Unbind(state->LocalPort);
}
state->Active = false;
}
}
static void UdpSocketDispatcher(uint32_t srcIp, uint16_t srcPort,
uint16_t dstPort,
const uint8_t* data, uint16_t length) {
for (int i = 0; i < MAX_UDP_SOCKETS; i++) {
if (g_udpSockets[i].Active && g_udpSockets[i].LocalPort == dstPort) {
UdpSocketState* st = &g_udpSockets[i];
uint32_t needed = sizeof(UdpDgramHeader) + length;
if (st->Count + needed > UDP_RING_SIZE) {
return; // drop if buffer full
}
// Enqueue header
UdpDgramHeader hdr;
hdr.SrcIp = srcIp;
hdr.SrcPort = srcPort;
hdr.DataLen = length;
const uint8_t* hdrBytes = (const uint8_t*)&hdr;
for (uint32_t j = 0; j < sizeof(UdpDgramHeader); j++) {
st->Ring[st->Tail] = hdrBytes[j];
st->Tail = (st->Tail + 1) % UDP_RING_SIZE;
}
// Enqueue payload
for (uint16_t j = 0; j < length; j++) {
st->Ring[st->Tail] = data[j];
st->Tail = (st->Tail + 1) % UDP_RING_SIZE;
}
st->Count += needed;
return;
}
}
}
// ---- Public API ----
void Initialize() {
for (int i = 0; i < MAX_SOCKETS; i++) {
g_sockets[i].Active = false;
}
for (int i = 0; i < MAX_UDP_SOCKETS; i++) {
g_udpSockets[i].Active = false;
}
KernelLogStream(OK, "Net") << "Socket table initialized";
KernelLogStream(OK, "Net") << "Socket handles initialized";
}
int Create(int type, int pid) {
if (type != SOCK_TCP && type != SOCK_UDP) return -1;
for (int i = 0; i < MAX_SOCKETS; i++) {
if (!g_sockets[i].Active) {
g_sockets[i].Active = true;
g_sockets[i].Type = type;
g_sockets[i].OwnerPid = pid;
g_sockets[i].TcpConn = nullptr;
g_sockets[i].UdpState = nullptr;
g_sockets[i].LocalPort = 0;
if (type == SOCK_UDP) {
UdpSocketState* us = AllocUdpState();
if (!us) {
g_sockets[i].Active = false;
return -1;
}
g_sockets[i].UdpState = us;
}
return i;
}
}
return -1;
return Ipc::CreateSocketHandleForSlot(Ipc::SlotForPid(pid), type);
}
int Connect(int fd, uint32_t ip, uint16_t port, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_TCP) return -1;
if (g_sockets[fd].TcpConn != nullptr) return -1;
uint16_t srcPort = AllocEphemeralPort();
g_sockets[fd].LocalPort = srcPort;
Tcp::Connection* conn = Tcp::Connect(ip, port, srcPort);
if (conn == nullptr) return -1;
g_sockets[fd].TcpConn = conn;
return 0;
int Connect(int fd, uint32_t ip, uint16_t port, int /*pid*/) {
return Ipc::SocketConnectHandle(fd, ip, port);
}
int Bind(int fd, uint16_t port, int pid) {
if (!ValidFd(fd, pid)) return -1;
g_sockets[fd].LocalPort = port;
if (g_sockets[fd].Type == SOCK_UDP) {
UdpSocketState* us = g_sockets[fd].UdpState;
if (!us) return -1;
us->LocalPort = port;
if (!Udp::Bind(port, UdpSocketDispatcher)) return -1;
}
return 0;
int Bind(int fd, uint16_t port, int /*pid*/) {
return Ipc::SocketBindHandle(fd, port);
}
int Listen(int fd, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_TCP) return -1;
if (g_sockets[fd].LocalPort == 0) return -1;
if (g_sockets[fd].TcpConn != nullptr) return -1;
Tcp::Connection* conn = Tcp::Listen(g_sockets[fd].LocalPort);
if (conn == nullptr) return -1;
g_sockets[fd].TcpConn = conn;
return 0;
int Listen(int fd, int /*pid*/) {
return Ipc::SocketListenHandle(fd);
}
int Accept(int fd, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_TCP) return -1;
if (g_sockets[fd].TcpConn == nullptr) return -1;
Tcp::Connection* clientConn = Tcp::Accept(g_sockets[fd].TcpConn);
if (clientConn == nullptr) return -1;
// Allocate a new socket entry for the accepted connection
for (int i = 0; i < MAX_SOCKETS; i++) {
if (!g_sockets[i].Active) {
g_sockets[i].Active = true;
g_sockets[i].Type = SOCK_TCP;
g_sockets[i].OwnerPid = pid;
g_sockets[i].TcpConn = clientConn;
g_sockets[i].UdpState = nullptr;
g_sockets[i].LocalPort = g_sockets[fd].LocalPort;
return i;
}
}
// No free socket slot — close the accepted connection
Tcp::Close(clientConn);
return -1;
int Accept(int fd, int /*pid*/) {
return Ipc::SocketAcceptHandle(fd);
}
int Send(int fd, const uint8_t* data, uint32_t len, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_TCP) return -1;
if (g_sockets[fd].TcpConn == nullptr) return -1;
return Tcp::Send(g_sockets[fd].TcpConn, data, (uint16_t)len);
int Send(int fd, const uint8_t* data, uint32_t len, int /*pid*/) {
return Ipc::SocketSendHandle(fd, data, len);
}
int Recv(int fd, uint8_t* buf, uint32_t maxLen, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_TCP) return -1;
if (g_sockets[fd].TcpConn == nullptr) return -1;
return Tcp::ReceiveNonBlocking(g_sockets[fd].TcpConn, buf, (uint16_t)maxLen);
int Recv(int fd, uint8_t* buf, uint32_t maxLen, int /*pid*/) {
return Ipc::SocketRecvHandle(fd, buf, maxLen);
}
int SendTo(int fd, const uint8_t* data, uint32_t len,
uint32_t destIp, uint16_t destPort, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_UDP) return -1;
UdpSocketState* us = g_sockets[fd].UdpState;
if (!us) return -1;
// Auto-bind ephemeral port if not already bound
if (us->LocalPort == 0) {
uint16_t ep = AllocEphemeralPort();
us->LocalPort = ep;
g_sockets[fd].LocalPort = ep;
if (!Udp::Bind(ep, UdpSocketDispatcher)) return -1;
}
if (!Udp::Send(destIp, us->LocalPort, destPort, data, (uint16_t)len)) {
return -1;
}
return (int)len;
uint32_t destIp, uint16_t destPort, int /*pid*/) {
return Ipc::SocketSendToHandle(fd, data, len, destIp, destPort);
}
int RecvFrom(int fd, uint8_t* buf, uint32_t maxLen,
uint32_t* srcIp, uint16_t* srcPort, int pid) {
if (!ValidFd(fd, pid)) return -1;
if (g_sockets[fd].Type != SOCK_UDP) return -1;
UdpSocketState* us = g_sockets[fd].UdpState;
if (!us) return -1;
if (us->Count < sizeof(UdpDgramHeader)) {
return -1; // no data available
}
// Dequeue header
UdpDgramHeader hdr;
uint8_t* hdrBytes = (uint8_t*)&hdr;
for (uint32_t j = 0; j < sizeof(UdpDgramHeader); j++) {
hdrBytes[j] = us->Ring[us->Head];
us->Head = (us->Head + 1) % UDP_RING_SIZE;
}
us->Count -= sizeof(UdpDgramHeader);
// Dequeue payload
uint16_t copyLen = hdr.DataLen;
if (copyLen > maxLen) copyLen = (uint16_t)maxLen;
for (uint16_t j = 0; j < copyLen; j++) {
buf[j] = us->Ring[us->Head];
us->Head = (us->Head + 1) % UDP_RING_SIZE;
}
// Skip remaining data if buffer was too small
for (uint16_t j = copyLen; j < hdr.DataLen; j++) {
us->Head = (us->Head + 1) % UDP_RING_SIZE;
}
us->Count -= hdr.DataLen;
if (srcIp) *srcIp = hdr.SrcIp;
if (srcPort) *srcPort = hdr.SrcPort;
return (int)copyLen;
uint32_t* srcIp, uint16_t* srcPort, int /*pid*/) {
return Ipc::SocketRecvFromHandle(fd, buf, maxLen, srcIp, srcPort);
}
void Close(int fd, int pid) {
if (!ValidFd(fd, pid)) return;
if (g_sockets[fd].TcpConn != nullptr) {
Tcp::Close(g_sockets[fd].TcpConn);
g_sockets[fd].TcpConn = nullptr;
}
if (g_sockets[fd].UdpState != nullptr) {
FreeUdpState(g_sockets[fd].UdpState);
g_sockets[fd].UdpState = nullptr;
}
g_sockets[fd].Active = false;
void Close(int fd, int /*pid*/) {
Ipc::CloseHandle(fd);
}
void CleanupProcess(int pid) {
for (int i = 0; i < MAX_SOCKETS; i++) {
if (g_sockets[i].Active && g_sockets[i].OwnerPid == pid) {
if (g_sockets[i].TcpConn != nullptr) {
Tcp::Close(g_sockets[i].TcpConn);
g_sockets[i].TcpConn = nullptr;
}
if (g_sockets[i].UdpState != nullptr) {
FreeUdpState(g_sockets[i].UdpState);
g_sockets[i].UdpState = nullptr;
}
g_sockets[i].Active = false;
}
}
void CleanupProcess(int /*pid*/) {
}
}
kernel/src/Net/Socket.hpp
+5 -18
View File
@@ -6,40 +6,27 @@
#pragma once
#include <cstdint>
#include <Net/Tcp.hpp>
namespace Net::Socket {
static constexpr int SOCK_TCP = 1;
static constexpr int SOCK_UDP = 2;
static constexpr int MAX_SOCKETS = 64;
struct UdpSocketState;
struct SocketEntry {
bool Active;
int Type;
int OwnerPid;
Tcp::Connection* TcpConn;
UdpSocketState* UdpState;
uint16_t LocalPort;
};
void Initialize();
// Create a socket of the given type. Returns fd or -1.
// Create a socket of the given type. Returns an IPC handle or -1.
int Create(int type, int pid);
// Connect socket fd to remote ip:port. Returns 0 or -1.
// Connect socket handle to remote ip:port. Returns 0 or -1.
int Connect(int fd, uint32_t ip, uint16_t port, int pid);
// Bind socket fd to a local port. Returns 0 or -1.
// Bind socket handle to a local port. Returns 0 or -1.
int Bind(int fd, uint16_t port, int pid);
// Start listening on a bound socket. Returns 0 or -1.
int Listen(int fd, int pid);
// Accept an incoming connection. Returns new fd or -1.
// Accept an incoming connection. Returns a new IPC handle or -1.
int Accept(int fd, int pid);
// Send data on a connected socket. Returns bytes sent or -1.
@@ -59,7 +46,7 @@ namespace Net::Socket {
// Close a socket.
void Close(int fd, int pid);
// Close all sockets owned by a process (called on process exit).
// Per-process cleanup is handled by IPC handle teardown.
void CleanupProcess(int pid);
}
kernel/src/Net/Tcp.cpp
+95 -10
View File
@@ -8,11 +8,13 @@
#include <Net/Ipv4.hpp>
#include <Net/ByteOrder.hpp>
#include <Net/NetConfig.hpp>
#include <Ipc/Ipc.hpp>
#include <Libraries/Memory.hpp>
#include <Terminal/Terminal.hpp>
#include <CppLib/Stream.hpp>
#include <CppLib/Spinlock.hpp>
#include <Timekeeping/ApicTimer.hpp>
#include <Sched/Scheduler.hpp>
using namespace Kt;
@@ -216,6 +218,8 @@ namespace Net::Tcp {
listener->PendingRemotePort = srcPort;
listener->PendingSeq = seqNum;
listener->Lock.Release();
Sched::WakeObjectWaiters(listener);
Ipc::NotifyTcpConnectionChanged(listener);
return;
}
}
@@ -235,12 +239,15 @@ namespace Net::Tcp {
}
conn->Lock.Acquire();
bool notify = false;
// RST handling
if (flags & FLAG_RST) {
conn->CurrentState = State::Closed;
conn->Active = false;
conn->Lock.Release();
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return;
}
@@ -255,6 +262,7 @@ namespace Net::Tcp {
// Send ACK
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
}
}
break;
@@ -266,6 +274,7 @@ namespace Net::Tcp {
if (ackNum == conn->SendNext) {
conn->SendUnack = ackNum;
conn->CurrentState = State::Established;
notify = true;
}
}
break;
@@ -275,6 +284,7 @@ namespace Net::Tcp {
// Handle incoming data
if (flags & FLAG_ACK) {
conn->SendUnack = ackNum;
notify = true;
}
if (payloadLen > 0 && seqNum == conn->RecvNext) {
@@ -283,6 +293,7 @@ namespace Net::Tcp {
// Send ACK
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
}
if (flags & FLAG_FIN) {
@@ -291,6 +302,7 @@ namespace Net::Tcp {
// Send ACK for the FIN
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
}
break;
}
@@ -302,13 +314,16 @@ namespace Net::Tcp {
conn->RecvNext = seqNum + 1;
conn->CurrentState = State::TimeWait;
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
} else {
conn->CurrentState = State::FinWait2;
notify = true;
}
} else if (flags & FLAG_FIN) {
conn->RecvNext = seqNum + 1;
conn->CurrentState = State::TimeWait;
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
}
break;
}
@@ -318,6 +333,7 @@ namespace Net::Tcp {
conn->RecvNext = seqNum + 1;
conn->CurrentState = State::TimeWait;
SendSegment(conn, FLAG_ACK, nullptr, 0);
notify = true;
}
break;
}
@@ -326,6 +342,7 @@ namespace Net::Tcp {
if (flags & FLAG_ACK) {
conn->CurrentState = State::Closed;
conn->Active = false;
notify = true;
}
break;
}
@@ -340,6 +357,10 @@ namespace Net::Tcp {
}
conn->Lock.Release();
if (notify) {
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
}
}
Connection* Listen(uint16_t port) {
@@ -421,19 +442,25 @@ namespace Net::Tcp {
}
// Wait for ACK to complete the handshake
for (int i = 0; i < 100; i++) {
uint64_t deadline = Timekeeping::GetMilliseconds() + 5000;
while (Timekeeping::GetMilliseconds() < deadline) {
if (conn->CurrentState == State::Established) {
return conn;
}
Timekeeping::Sleep(50);
uint64_t now = Timekeeping::GetMilliseconds();
uint64_t waitMs = (deadline > now) ? (deadline - now) : 0;
if (waitMs == 0) break;
Sched::BlockOnObject(conn, waitMs);
}
// Timed out waiting for ACK
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return nullptr;
}
listener->Lock.Release();
Timekeeping::Sleep(10);
Sched::BlockOnObject(listener, 0);
}
}
@@ -480,11 +507,15 @@ namespace Net::Tcp {
// Wait for SYN-ACK
for (int attempt = 0; attempt < MAX_RETRANSMITS; attempt++) {
for (int i = 0; i < 20; i++) {
uint64_t deadline = Timekeeping::GetMilliseconds() + 1000;
while (Timekeeping::GetMilliseconds() < deadline) {
if (conn->CurrentState == State::Established) {
return conn;
}
Timekeeping::Sleep(50);
uint64_t now = Timekeeping::GetMilliseconds();
uint64_t waitMs = (deadline > now) ? (deadline - now) : 0;
if (waitMs == 0) break;
Sched::BlockOnObject(conn, waitMs);
}
if (conn->CurrentState == State::SynSent) {
@@ -512,6 +543,8 @@ namespace Net::Tcp {
// Failed to connect
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return nullptr;
}
@@ -580,8 +613,15 @@ namespace Net::Tcp {
conn->RetransmitTime = Timekeeping::GetMilliseconds();
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
continue;
}
Timekeeping::Sleep(10);
uint64_t waitMs = 10;
if (conn->RetransmitLen > 0 && now <= conn->RetransmitTime + RETRANSMIT_TIMEOUT_MS) {
waitMs = (conn->RetransmitTime + RETRANSMIT_TIMEOUT_MS) - now;
if (waitMs == 0) waitMs = 1;
}
Sched::BlockOnObject(conn, waitMs);
}
return sent;
@@ -625,7 +665,7 @@ namespace Net::Tcp {
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
Timekeeping::Sleep(10);
Sched::BlockOnObject(conn, 0);
}
}
@@ -691,9 +731,11 @@ namespace Net::Tcp {
conn->CurrentState == State::Closed) {
break;
}
Timekeeping::Sleep(50);
Sched::BlockOnObject(conn, 50);
}
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return;
}
@@ -709,9 +751,11 @@ namespace Net::Tcp {
if (conn->CurrentState == State::Closed) {
break;
}
Timekeeping::Sleep(50);
Sched::BlockOnObject(conn, 50);
}
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return;
}
@@ -721,6 +765,8 @@ namespace Net::Tcp {
conn->Active = false;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return;
}
@@ -728,6 +774,8 @@ namespace Net::Tcp {
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
return;
}
}
@@ -736,7 +784,44 @@ namespace Net::Tcp {
if (conn == nullptr) {
return State::Closed;
}
return conn->CurrentState;
conn->Lock.Acquire();
State state = conn->CurrentState;
conn->Lock.Release();
return state;
}
bool HasPendingAccept(Connection* conn) {
if (conn == nullptr) return false;
conn->Lock.Acquire();
bool pending = conn->PendingAccept;
conn->Lock.Release();
return pending;
}
bool HasReceiveData(Connection* conn) {
if (conn == nullptr) return false;
conn->Lock.Acquire();
bool hasData = conn->RecvCount > 0;
conn->Lock.Release();
return hasData;
}
bool CanSend(Connection* conn) {
if (conn == nullptr) return false;
conn->Lock.Acquire();
bool writable = conn->CurrentState == State::Established;
conn->Lock.Release();
return writable;
}
bool IsClosedForIo(Connection* conn) {
if (conn == nullptr) return true;
conn->Lock.Acquire();
bool closed = conn->CurrentState == State::CloseWait ||
conn->CurrentState == State::Closed ||
conn->CurrentState == State::TimeWait;
conn->Lock.Release();
return closed;
}
}
kernel/src/Net/Tcp.hpp
+6
View File
@@ -80,4 +80,10 @@ namespace Net::Tcp {
// Get the state of a connection
State GetState(Connection* conn);
// Readiness helpers used by the IPC layer.
bool HasPendingAccept(Connection* conn);
bool HasReceiveData(Connection* conn);
bool CanSend(Connection* conn);
bool IsClosedForIo(Connection* conn);
}