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

refactor: network - harden TCP/IP and unify HTTP clients

Browse Source
This commit is contained in:
daniel
2026-07-29 15:35:17 +01:00
parent 75ae7ede56
commit a288dee7df
28 changed files with 1852 additions and 1361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kernel/src/Api/BuildNo.hpp
+1 -1
View File
@@ -12,4 +12,4 @@
#pragma once
#define MONTAUK_BUILD_NUMBER 7
#define MONTAUK_BUILD_NUMBER 8
kernel/src/Ipc/Ipc.cpp
+5 -2
View File
@@ -1382,7 +1382,8 @@ namespace Ipc {
Net::Tcp::Connection* conn = socket->tcpConn;
socket->socketLock.Release();
if (conn == nullptr) return -1;
return Net::Tcp::Send(conn, data, (uint16_t)len);
if (len > 0x7FFFFFFFu) return -1;
return Net::Tcp::Send(conn, data, len);
}
int SocketRecvHandle(int handle, uint8_t* buffer, uint32_t maxLen) {
@@ -1399,13 +1400,15 @@ namespace Ipc {
socket->socketLock.Release();
if (conn == nullptr) return -1;
int result = Net::Tcp::ReceiveNonBlocking(conn, buffer, (uint16_t)maxLen);
uint16_t cappedLen = maxLen > 0xFFFFu ? 0xFFFFu : (uint16_t)maxLen;
int result = Net::Tcp::ReceiveNonBlocking(conn, buffer, cappedLen);
if (result != 0) NotifyObjectChanged((Object*)socket);
return result;
}
int SocketSendToHandle(int handle, const uint8_t* data, uint32_t len, uint32_t destIp, uint16_t destPort) {
if (data == nullptr) return -1;
if (len > 1472) return -1;
if (len > 0 && !montauk::abi::UserMemory::Range((uint64_t)data, len, false)) return -1;
Socket* socket = nullptr;
kernel/src/Net/Arp.cpp
+72 -11
View File
@@ -15,6 +15,7 @@
#include <Terminal/Terminal.hpp>
#include <CppLib/Stream.hpp>
#include <Timekeeping/ApicTimer.hpp>
#include <CppLib/Spinlock.hpp>
using namespace Kt;
@@ -38,6 +39,16 @@ namespace Net::Arp {
static constexpr uint64_t ARP_CACHE_TIMEOUT_MS = 60000; // 60 seconds
static CacheEntry g_cache[ARP_CACHE_SIZE] = {};
static kcp::Spinlock g_cacheLock;
struct PendingRequest {
uint32_t Ip;
uint64_t Timestamp;
bool Valid;
};
static constexpr uint32_t PENDING_REQUEST_SIZE = 16;
static constexpr uint64_t REQUEST_RETRY_MS = 1000;
static PendingRequest g_pendingRequests[PENDING_REQUEST_SIZE] = {};
void Initialize() {
for (uint32_t i = 0; i < ARP_CACHE_SIZE; i++) {
@@ -47,43 +58,91 @@ namespace Net::Arp {
}
static void CacheInsert(uint32_t ip, const uint8_t* mac) {
g_cacheLock.Acquire();
// Look for existing entry or empty slot
uint32_t emptySlot = ARP_CACHE_SIZE;
uint32_t oldestSlot = 0;
uint64_t oldestTimestamp = ~0ULL;
for (uint32_t i = 0; i < ARP_CACHE_SIZE; i++) {
if (g_cache[i].Valid && g_cache[i].Ip == ip) {
// Update existing entry
memcpy(g_cache[i].Mac, mac, 6);
g_cache[i].Timestamp = Timekeeping::GetMilliseconds();
for (uint32_t j = 0; j < PENDING_REQUEST_SIZE; ++j)
if (g_pendingRequests[j].Valid && g_pendingRequests[j].Ip == ip)
g_pendingRequests[j].Valid = false;
g_cacheLock.Release();
return;
}
if (!g_cache[i].Valid && emptySlot == ARP_CACHE_SIZE) {
emptySlot = i;
}
if (g_cache[i].Valid && g_cache[i].Timestamp < oldestTimestamp) {
oldestTimestamp = g_cache[i].Timestamp;
oldestSlot = i;
}
}
if (emptySlot < ARP_CACHE_SIZE) {
g_cache[emptySlot].Ip = ip;
memcpy(g_cache[emptySlot].Mac, mac, 6);
g_cache[emptySlot].Timestamp = Timekeeping::GetMilliseconds();
g_cache[emptySlot].Valid = true;
}
uint32_t slot = emptySlot < ARP_CACHE_SIZE ? emptySlot : oldestSlot;
g_cache[slot].Ip = ip;
memcpy(g_cache[slot].Mac, mac, 6);
g_cache[slot].Timestamp = Timekeeping::GetMilliseconds();
g_cache[slot].Valid = true;
for (uint32_t j = 0; j < PENDING_REQUEST_SIZE; ++j)
if (g_pendingRequests[j].Valid && g_pendingRequests[j].Ip == ip)
g_pendingRequests[j].Valid = false;
g_cacheLock.Release();
}
static bool CacheLookup(uint32_t ip, uint8_t* outMac) {
g_cacheLock.Acquire();
uint64_t now = Timekeeping::GetMilliseconds();
for (uint32_t i = 0; i < ARP_CACHE_SIZE; i++) {
if (g_cache[i].Valid && g_cache[i].Ip == ip) {
if ((now - g_cache[i].Timestamp) > ARP_CACHE_TIMEOUT_MS) {
g_cache[i].Valid = false;
g_cacheLock.Release();
return false;
}
memcpy(outMac, g_cache[i].Mac, 6);
g_cacheLock.Release();
return true;
}
}
g_cacheLock.Release();
return false;
}
static bool ShouldSendRequest(uint32_t ip) {
g_cacheLock.Acquire();
uint64_t now = Timekeeping::GetMilliseconds();
uint32_t slot = PENDING_REQUEST_SIZE;
uint32_t oldest = 0;
uint64_t oldestTimestamp = ~0ULL;
for (uint32_t i = 0; i < PENDING_REQUEST_SIZE; ++i) {
if (g_pendingRequests[i].Valid && g_pendingRequests[i].Ip == ip) {
if (now - g_pendingRequests[i].Timestamp < REQUEST_RETRY_MS) {
g_cacheLock.Release();
return false;
}
slot = i;
break;
}
if (!g_pendingRequests[i].Valid && slot == PENDING_REQUEST_SIZE) slot = i;
if (g_pendingRequests[i].Valid &&
g_pendingRequests[i].Timestamp < oldestTimestamp) {
oldestTimestamp = g_pendingRequests[i].Timestamp;
oldest = i;
}
}
if (slot == PENDING_REQUEST_SIZE) slot = oldest;
g_pendingRequests[slot].Ip = ip;
g_pendingRequests[slot].Timestamp = now;
g_pendingRequests[slot].Valid = true;
g_cacheLock.Release();
return true;
}
void OnPacketReceived(const uint8_t* data, uint16_t length) {
if (length < sizeof(Packet)) {
return;
@@ -92,10 +151,14 @@ namespace Net::Arp {
const Packet* pkt = (const Packet*)data;
if (Ntohs(pkt->HardwareType) != HW_TYPE_ETHERNET ||
Ntohs(pkt->ProtocolType) != PROTO_TYPE_IPV4) {
Ntohs(pkt->ProtocolType) != PROTO_TYPE_IPV4 ||
pkt->HardwareAddrLen != 6 || pkt->ProtocolAddrLen != 4) {
return;
}
uint16_t op = Ntohs(pkt->Operation);
if (op != OP_REQUEST && op != OP_REPLY) return;
uint32_t senderIp = pkt->SenderIp; // Already in network byte order in struct
uint32_t targetIp = pkt->TargetIp;
@@ -103,8 +166,6 @@ namespace Net::Arp {
CacheInsert(senderIp, pkt->SenderMac);
Ipv4::FlushPending();
uint16_t op = Ntohs(pkt->Operation);
if (op == OP_REQUEST && targetIp == GetIpAddress()) {
// Someone is asking for our MAC address -- send a reply
Packet reply;
@@ -135,8 +196,8 @@ namespace Net::Arp {
return true;
}
// Not in cache, send a request
SendRequest(ip);
// Coalesce bursts of packets awaiting the same next hop.
if (ShouldSendRequest(ip)) SendRequest(ip);
return false;
}
kernel/src/Net/Dns.cpp
+113 -54
View File
@@ -13,6 +13,7 @@
#include <Timekeeping/ApicTimer.hpp>
#include <Sched/Scheduler.hpp>
#include <Terminal/Terminal.hpp>
#include <CppLib/Spinlock.hpp>
namespace Net::Dns {
@@ -36,16 +37,22 @@ namespace Net::Dns {
};
static CacheEntry g_cache[CACHE_SIZE] = {};
static kcp::Spinlock g_cacheLock;
static bool streq(const char* a, const char* b) {
while (*a && *b) {
if (*a != *b) return false;
char ca = *a;
char cb = *b;
if (ca >= 'A' && ca <= 'Z') ca += 'a' - 'A';
if (cb >= 'A' && cb <= 'Z') cb += 'a' - 'A';
if (ca != cb) return false;
a++; b++;
}
return *a == *b;
}
static uint32_t CacheLookup(const char* hostname) {
g_cacheLock.Acquire();
uint64_t now = Timekeeping::GetMilliseconds();
for (int i = 0; i < CACHE_SIZE; i++) {
if (!g_cache[i].valid) continue;
@@ -53,17 +60,22 @@ namespace Net::Dns {
// Check TTL
uint64_t elapsed = (now - g_cache[i].timestamp) / 1000;
if (elapsed < g_cache[i].ttl) {
return g_cache[i].ip;
uint32_t ip = g_cache[i].ip;
g_cacheLock.Release();
return ip;
}
// Expired
g_cache[i].valid = false;
g_cacheLock.Release();
return 0;
}
g_cacheLock.Release();
return 0;
}
static void CacheStore(const char* hostname, uint32_t ip, uint32_t ttl) {
if (ttl == 0) ttl = 60; // Minimum 60s TTL
g_cacheLock.Acquire();
// Find free or oldest slot
int slot = 0;
@@ -85,6 +97,7 @@ namespace Net::Dns {
e.ttl = ttl;
e.timestamp = Timekeeping::GetMilliseconds();
e.valid = true;
g_cacheLock.Release();
}
// ---- DNS query building ----
@@ -178,8 +191,10 @@ namespace Net::Dns {
maxJumps--;
continue;
}
if ((len & 0xC0) != 0) return -1;
// Regular label
if (offset + 1 + len > packetLen) return -1;
offset += 1 + len;
maxJumps--;
}
@@ -209,6 +224,7 @@ namespace Net::Dns {
// Check RCODE (must be 0 = no error)
uint8_t rcode = packet[3] & 0x0F;
if (rcode != 0) return result;
if (packet[2] & 0x02) return result; // Truncated UDP response
uint16_t qdcount = ((uint16_t)packet[4] << 8) | packet[5];
uint16_t ancount = ((uint16_t)packet[6] << 8) | packet[7];
@@ -228,7 +244,7 @@ namespace Net::Dns {
if (offset < 0 || offset + 10 > packetLen) return result;
uint16_t atype = ((uint16_t)packet[offset] << 8) | packet[offset + 1];
// uint16_t aclass = ((uint16_t)packet[offset + 2] << 8) | packet[offset + 3];
uint16_t aclass = ((uint16_t)packet[offset + 2] << 8) | packet[offset + 3];
uint32_t attl = ((uint32_t)packet[offset + 4] << 24) |
((uint32_t)packet[offset + 5] << 16) |
((uint32_t)packet[offset + 6] << 8) |
@@ -238,7 +254,7 @@ namespace Net::Dns {
if (offset + rdlen > packetLen) return result;
if (atype == DNS_QTYPE_A && rdlen == 4) {
if (atype == DNS_QTYPE_A && aclass == DNS_QCLASS_IN && rdlen == 4) {
// A record: 4-byte IPv4 address (already in network byte order)
result.ip = ((uint32_t)packet[offset])
| ((uint32_t)packet[offset + 1] << 8)
@@ -255,26 +271,38 @@ namespace Net::Dns {
return result;
}
// ---- Resolve state (shared with UDP callback) ----
// ---- Concurrent resolve state (shared with UDP callback) ----
static volatile bool g_gotResponse = false;
static volatile uint16_t g_currentId = 0;
static uint8_t g_responseBuffer[512];
static volatile int g_responseLen = 0;
static constexpr int MAX_QUERIES = 8;
struct Query {
volatile bool active;
volatile bool gotResponse;
uint16_t id;
uint16_t localPort;
uint32_t serverIp;
uint8_t response[512];
volatile int responseLen;
};
static Query g_queries[MAX_QUERIES] = {};
static kcp::Spinlock g_queriesLock;
static void DnsRecvCallback(uint32_t srcIp, uint16_t srcPort,
uint16_t dstPort,
const uint8_t* data, uint16_t length) {
(void)srcIp;
(void)srcPort;
(void)dstPort;
if (g_gotResponse) return; // Already got a response
if (length > sizeof(g_responseBuffer)) length = sizeof(g_responseBuffer);
memcpy(g_responseBuffer, data, length);
g_responseLen = length;
g_gotResponse = true;
if (srcPort != DNS_PORT || length < 2) return;
uint16_t responseId = ((uint16_t)data[0] << 8) | data[1];
for (int i = 0; i < MAX_QUERIES; ++i) {
Query& query = g_queries[i];
if (!query.active || query.gotResponse ||
query.localPort != dstPort || query.serverIp != srcIp ||
query.id != responseId) continue;
if (length > sizeof(query.response)) length = sizeof(query.response);
memcpy(query.response, data, length);
query.responseLen = length;
asm volatile("" ::: "memory");
query.gotResponse = true;
return;
}
}
// ---- Simple PRNG for transaction IDs ----
@@ -288,21 +316,26 @@ namespace Net::Dns {
// ---- Check if string is already an IP address ----
static bool IsIpAddress(const char* s) {
int dotCount = 0;
bool hasDigit = false;
for (int i = 0; s[i]; i++) {
if (s[i] >= '0' && s[i] <= '9') {
hasDigit = true;
} else if (s[i] == '.') {
if (!hasDigit) return false;
dotCount++;
hasDigit = false;
} else {
static bool ParseIpAddress(const char* s, uint32_t* out) {
uint32_t ip = 0;
for (int octet = 0; octet < 4; ++octet) {
if (*s < '0' || *s > '9') return false;
uint32_t value = 0;
int digits = 0;
while (*s >= '0' && *s <= '9') {
value = value * 10 + (uint32_t)(*s - '0');
if (value > 255 || ++digits > 3) return false;
++s;
}
ip |= value << (octet * 8);
if (octet < 3) {
if (*s++ != '.') return false;
} else if (*s != '\0') {
return false;
}
}
return hasDigit && dotCount == 3;
*out = ip;
return true;
}
// ---- Public API ----
@@ -310,8 +343,8 @@ namespace Net::Dns {
uint32_t Resolve(const char* hostname, uint32_t timeoutMs) {
if (hostname == nullptr || hostname[0] == '\0') return 0;
// Don't try to resolve IP addresses
if (IsIpAddress(hostname)) return 0;
uint32_t literalIp = 0;
if (ParseIpAddress(hostname, &literalIp)) return literalIp;
// Check cache first
uint32_t cached = CacheLookup(hostname);
@@ -321,51 +354,77 @@ namespace Net::Dns {
uint32_t dnsServer = Net::GetDnsServer();
if (dnsServer == 0) return 0;
// Pick a local port for receiving the response (ephemeral range)
uint16_t localPort = 10000 + (NextId() % 50000);
uint16_t txId = NextId();
g_queriesLock.Acquire();
Query* query = nullptr;
for (int i = 0; i < MAX_QUERIES; ++i) {
if (!g_queries[i].active) {
query = &g_queries[i];
query->active = true;
break;
}
}
if (!query) {
g_queriesLock.Release();
return 0;
}
query->gotResponse = false;
query->responseLen = 0;
query->serverIp = dnsServer;
query->id = NextId();
uint16_t txId = query->id;
g_queriesLock.Release();
// Build DNS query
uint8_t queryPacket[512];
int queryLen = BuildQuery(txId, hostname, queryPacket, sizeof(queryPacket));
if (queryLen == 0) return 0;
if (queryLen == 0) {
query->active = false;
return 0;
}
// Reset response state
g_gotResponse = false;
g_responseLen = 0;
g_currentId = txId;
// Bind our receive port
if (!Net::Udp::Bind(localPort, DnsRecvCallback)) {
// Port might be in use, try another
localPort = 10000 + (NextId() % 50000);
if (!Net::Udp::Bind(localPort, DnsRecvCallback)) {
return 0;
bool bound = false;
for (int attempt = 0; attempt < 16; ++attempt) {
g_queriesLock.Acquire();
uint16_t localPort = (uint16_t)(10000 + (NextId() % 50000));
g_queriesLock.Release();
if (Net::Udp::Bind(localPort, DnsRecvCallback)) {
query->localPort = localPort;
bound = true;
break;
}
}
if (!bound) {
query->active = false;
return 0;
}
// Send the query to DNS server port 53
bool sent = Net::Udp::Send(dnsServer, localPort, DNS_PORT, queryPacket, (uint16_t)queryLen);
bool sent = Net::Udp::Send(dnsServer, query->localPort, DNS_PORT,
queryPacket, (uint16_t)queryLen);
if (!sent) {
Net::Udp::Unbind(localPort);
Net::Udp::Unbind(query->localPort);
query->active = false;
return 0;
}
// Wait for response with timeout
uint64_t start = Timekeeping::GetMilliseconds();
while (!g_gotResponse) {
while (!query->gotResponse) {
if (Timekeeping::GetMilliseconds() - start >= timeoutMs) {
Net::Udp::Unbind(localPort);
Net::Udp::Unbind(query->localPort);
query->active = false;
return 0;
}
Sched::Schedule();
}
// Unbind the port
Net::Udp::Unbind(localPort);
Net::Udp::Unbind(query->localPort);
// Parse the response
DnsAnswer answer = ParseResponse(txId, g_responseBuffer, g_responseLen);
asm volatile("" ::: "memory");
DnsAnswer answer = ParseResponse(txId, query->response, query->responseLen);
query->active = false;
if (!answer.found) return 0;
// Cache the result
kernel/src/Net/Ipv4.cpp
+53 -6
View File
@@ -15,6 +15,8 @@
#include <Libraries/Memory.hpp>
#include <Terminal/Terminal.hpp>
#include <CppLib/Stream.hpp>
#include <CppLib/Spinlock.hpp>
#include <Timekeeping/ApicTimer.hpp>
using namespace Kt;
@@ -28,11 +30,15 @@ namespace Net::Ipv4 {
uint8_t Protocol;
uint8_t Data[Ethernet::MAX_PAYLOAD_SIZE - HEADER_SIZE];
uint16_t Length;
uint64_t Timestamp;
bool Active;
bool Processing;
};
static constexpr uint32_t PENDING_QUEUE_SIZE = 8;
static constexpr uint32_t PENDING_QUEUE_SIZE = 64;
static constexpr uint64_t PENDING_TIMEOUT_MS = 30000;
static PendingPacket g_pendingQueue[PENDING_QUEUE_SIZE] = {};
static kcp::Spinlock g_pendingLock;
void Initialize() {
g_identification = 0;
@@ -125,6 +131,11 @@ namespace Net::Ipv4 {
return;
}
// Fragment reassembly is not implemented yet. Never pass a fragment
// to TCP/UDP as though it were a complete transport packet.
uint16_t fragment = Ntohs(hdr->FlagsFragment);
if (fragment & 0x3FFFu) return;
// Check destination: accept packets addressed to us or broadcast
uint32_t ourIp = GetIpAddress();
if (hdr->DstIp != ourIp && hdr->DstIp != 0xFFFFFFFF) {
@@ -158,7 +169,9 @@ namespace Net::Ipv4 {
hdr->VersionIhl = (4 << 4) | 5; // IPv4, 5 dwords (20 bytes)
hdr->Tos = 0;
hdr->TotalLength = Htons(HEADER_SIZE + payloadLen);
hdr->Identification = Htons(g_identification++);
uint16_t identification =
__atomic_fetch_add(&g_identification, 1, __ATOMIC_RELAXED);
hdr->Identification = Htons(identification);
hdr->FlagsFragment = 0;
hdr->Ttl = DEFAULT_TTL;
hdr->Protocol = protocol;
@@ -187,16 +200,25 @@ namespace Net::Ipv4 {
}
// ARP request already sent by Resolve(), queue the packet for later
g_pendingLock.Acquire();
uint64_t now = Timekeeping::GetMilliseconds();
for (uint32_t i = 0; i < PENDING_QUEUE_SIZE; i++) {
if (g_pendingQueue[i].Active && !g_pendingQueue[i].Processing &&
now - g_pendingQueue[i].Timestamp >= PENDING_TIMEOUT_MS)
g_pendingQueue[i].Active = false;
if (!g_pendingQueue[i].Active) {
g_pendingQueue[i].DestIp = destIp;
g_pendingQueue[i].Protocol = protocol;
g_pendingQueue[i].Length = payloadLen;
g_pendingQueue[i].Timestamp = now;
memcpy(g_pendingQueue[i].Data, payload, payloadLen);
g_pendingQueue[i].Active = true;
g_pendingQueue[i].Processing = false;
g_pendingLock.Release();
return true;
}
}
g_pendingLock.Release();
// Queue full, drop the packet
return false;
@@ -204,17 +226,42 @@ namespace Net::Ipv4 {
void FlushPending() {
for (uint32_t i = 0; i < PENDING_QUEUE_SIZE; i++) {
if (!g_pendingQueue[i].Active) {
uint32_t destIp;
uint8_t protocol;
uint16_t length;
uint8_t data[Ethernet::MAX_PAYLOAD_SIZE - HEADER_SIZE];
g_pendingLock.Acquire();
if (!g_pendingQueue[i].Active || g_pendingQueue[i].Processing) {
g_pendingLock.Release();
continue;
}
if (Timekeeping::GetMilliseconds() - g_pendingQueue[i].Timestamp >=
PENDING_TIMEOUT_MS) {
g_pendingQueue[i].Active = false;
g_pendingLock.Release();
continue;
}
g_pendingQueue[i].Processing = true;
destIp = g_pendingQueue[i].DestIp;
protocol = g_pendingQueue[i].Protocol;
length = g_pendingQueue[i].Length;
memcpy(data, g_pendingQueue[i].Data, length);
g_pendingLock.Release();
uint32_t nextHop = GetNextHop(g_pendingQueue[i].DestIp);
uint32_t nextHop = GetNextHop(destIp);
uint8_t destMac[6];
if (Arp::Resolve(nextHop, destMac)) {
SendDirect(g_pendingQueue[i].DestIp, g_pendingQueue[i].Protocol,
destMac, g_pendingQueue[i].Data, g_pendingQueue[i].Length);
g_pendingLock.Acquire();
g_pendingQueue[i].Active = false;
g_pendingQueue[i].Processing = false;
g_pendingLock.Release();
SendDirect(destIp, protocol, destMac, data, length);
} else {
g_pendingLock.Acquire();
g_pendingQueue[i].Processing = false;
g_pendingLock.Release();
}
}
}
kernel/src/Net/Tcp.cpp
+141 -45
View File
@@ -23,7 +23,8 @@ namespace Net::Tcp {
// Receive buffer size per connection
static constexpr uint16_t RECV_BUFFER_SIZE = 32768;
static constexpr uint16_t WINDOW_SIZE = RECV_BUFFER_SIZE;
static constexpr uint32_t MAX_CONNECTIONS = 16;
static constexpr uint32_t MAX_CONNECTIONS = 64;
static constexpr uint8_t ACCEPT_BACKLOG_SIZE = 8;
static constexpr uint64_t RETRANSMIT_TIMEOUT_MS = 1000;
static constexpr int MAX_RETRANSMITS = 5;
static constexpr uint64_t TIME_WAIT_MS = 2000;
@@ -51,12 +52,18 @@ namespace Net::Tcp {
uint16_t RetransmitLen;
uint64_t RetransmitTime;
int RetransmitCount;
bool SendBusy;
// For Listen/Accept
bool PendingAccept;
uint32_t PendingRemoteIp;
uint16_t PendingRemotePort;
uint32_t PendingSeq;
struct PendingSyn {
uint32_t RemoteIp;
uint16_t RemotePort;
uint32_t Seq;
};
PendingSyn Pending[ACCEPT_BACKLOG_SIZE];
uint8_t PendingHead;
uint8_t PendingTail;
uint8_t PendingCount;
bool Active;
@@ -71,6 +78,17 @@ namespace Net::Tcp {
return (uint32_t)(Timekeeping::GetMilliseconds() * 2654435761u);
}
// TCP sequence numbers are compared modulo 2^32. This is valid for
// ranges smaller than 2^31 bytes, which all of our send windows are.
static bool SeqLessOrEqual(uint32_t a, uint32_t b) {
return (int32_t)(a - b) <= 0;
}
static bool AckIsValid(const Connection* conn, uint32_t ack) {
return SeqLessOrEqual(conn->SendUnack, ack) &&
SeqLessOrEqual(ack, conn->SendNext);
}
static Connection* FindConnection(uint32_t remoteIp, uint16_t remotePort,
uint16_t localPort) {
for (uint32_t i = 0; i < MAX_CONNECTIONS; i++) {
@@ -109,6 +127,13 @@ namespace Net::Tcp {
return nullptr;
}
static void ReleaseConnection(Connection* conn) {
if (!conn) return;
g_connectionsLock.Acquire();
conn->Active = false;
g_connectionsLock.Release();
}
static bool SendSegment(Connection* conn, uint8_t flags,
const uint8_t* payload, uint16_t payloadLen) {
uint8_t packet[1500];
@@ -213,15 +238,35 @@ namespace Net::Tcp {
if (flags & FLAG_SYN) {
Connection* listener = FindListener(dstPort);
if (listener != nullptr) {
// Signal the listener about this incoming connection
// Queue the SYN for accept(). Retransmitted SYNs refresh
// their sequence number instead of consuming backlog.
listener->Lock.Acquire();
listener->PendingAccept = true;
listener->PendingRemoteIp = srcIp;
listener->PendingRemotePort = srcPort;
listener->PendingSeq = seqNum;
bool duplicate = false;
for (uint8_t i = 0; i < listener->PendingCount; ++i) {
uint8_t index = (uint8_t)((listener->PendingHead + i) %
ACCEPT_BACKLOG_SIZE);
Connection::PendingSyn& pending = listener->Pending[index];
if (pending.RemoteIp == srcIp && pending.RemotePort == srcPort) {
pending.Seq = seqNum;
duplicate = true;
break;
}
}
if (!duplicate && listener->PendingCount < ACCEPT_BACKLOG_SIZE) {
Connection::PendingSyn& pending =
listener->Pending[listener->PendingTail];
pending.RemoteIp = srcIp;
pending.RemotePort = srcPort;
pending.Seq = seqNum;
listener->PendingTail =
(uint8_t)((listener->PendingTail + 1) % ACCEPT_BACKLOG_SIZE);
listener->PendingCount++;
}
listener->Lock.Release();
Sched::WakeObjectWaiters(listener);
Ipc::NotifyTcpConnectionChanged(listener);
if (duplicate || listener->PendingCount > 0) {
Sched::WakeObjectWaiters(listener);
Ipc::NotifyTcpConnectionChanged(listener);
}
return;
}
}
@@ -246,7 +291,9 @@ namespace Net::Tcp {
// RST handling
if (flags & FLAG_RST) {
conn->CurrentState = State::Closed;
conn->Active = false;
// Keep the slot reserved until its socket handle is closed.
// Reusing it here would leave that handle pointing at an
// unrelated future connection.
conn->Lock.Release();
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
@@ -285,8 +332,10 @@ namespace Net::Tcp {
case State::Established: {
// Handle incoming data
if (flags & FLAG_ACK) {
conn->SendUnack = ackNum;
notify = true;
if (AckIsValid(conn, ackNum)) {
conn->SendUnack = ackNum;
notify = true;
}
}
uint16_t accepted = 0;
@@ -365,9 +414,9 @@ namespace Net::Tcp {
}
case State::LastAck: {
if (flags & FLAG_ACK) {
if ((flags & FLAG_ACK) && ackNum == conn->SendNext) {
conn->SendUnack = ackNum;
conn->CurrentState = State::Closed;
conn->Active = false;
notify = true;
}
break;
@@ -401,7 +450,9 @@ namespace Net::Tcp {
conn->LocalIp = Net::GetIpAddress();
conn->LocalPort = port;
conn->CurrentState = State::Listen;
conn->PendingAccept = false;
conn->PendingHead = 0;
conn->PendingTail = 0;
conn->PendingCount = 0;
KernelLogStream(INFO, "Net") << "TCP listening on port " << base::dec << (uint64_t)port;
return conn;
@@ -415,12 +466,15 @@ namespace Net::Tcp {
// Block until a SYN arrives
while (true) {
listener->Lock.Acquire();
if (listener->PendingAccept) {
listener->PendingAccept = false;
if (listener->PendingCount > 0) {
Connection::PendingSyn pending = listener->Pending[listener->PendingHead];
listener->PendingHead =
(uint8_t)((listener->PendingHead + 1) % ACCEPT_BACKLOG_SIZE);
listener->PendingCount--;
uint32_t remoteIp = listener->PendingRemoteIp;
uint16_t remotePort = listener->PendingRemotePort;
uint32_t remoteSeq = listener->PendingSeq;
uint32_t remoteIp = pending.RemoteIp;
uint16_t remotePort = pending.RemotePort;
uint32_t remoteSeq = pending.Seq;
listener->Lock.Release();
// Allocate a new connection for this client
@@ -480,9 +534,9 @@ namespace Net::Tcp {
}
// Timed out waiting for ACK
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return nullptr;
}
listener->Lock.Release();
@@ -568,25 +622,56 @@ namespace Net::Tcp {
}
// Failed to connect
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return nullptr;
}
int Send(Connection* conn, const uint8_t* data, uint16_t length) {
if (conn == nullptr || conn->CurrentState != State::Established) {
return -1;
int Send(Connection* conn, const uint8_t* data, uint32_t length) {
if (conn == nullptr || data == nullptr) return -1;
if (length == 0) return 0;
// A connection has one send sequence/retransmission stream. Serialize
// writers so concurrent threads cannot overwrite its in-flight
// retransmit buffer or allocate overlapping sequence numbers.
while (true) {
uint64_t flags;
asm volatile("pushfq; pop %0; cli" : "=r"(flags) :: "memory");
conn->Lock.Acquire();
if (conn->CurrentState != State::Established) {
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return -1;
}
if (!conn->SendBusy) {
conn->SendBusy = true;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
break;
}
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
Sched::BlockOnObject(conn, 50);
}
auto finishSend = [&]() {
uint64_t flags;
asm volatile("pushfq; pop %0; cli" : "=r"(flags) :: "memory");
conn->Lock.Acquire();
conn->SendBusy = false;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
};
constexpr uint16_t MSS = 1460;
uint16_t sent = 0;
uint32_t sent = 0;
while (sent < length) {
uint16_t segLen = length - sent;
if (segLen > MSS) {
segLen = MSS;
}
uint32_t remaining = length - sent;
uint16_t segLen = remaining > MSS ? MSS : (uint16_t)remaining;
uint32_t segSeq = 0;
uint32_t expectedAck = 0;
@@ -597,7 +682,9 @@ namespace Net::Tcp {
if (conn->CurrentState != State::Established) {
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return sent > 0 ? sent : -1;
int result = sent > 0 ? (int)sent : -1;
finishSend();
return result;
}
segSeq = conn->SendNext;
@@ -607,7 +694,9 @@ namespace Net::Tcp {
if (!ok) {
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return sent > 0 ? sent : -1;
int result = sent > 0 ? (int)sent : -1;
finishSend();
return result;
}
conn->SendNext = expectedAck;
@@ -626,7 +715,7 @@ namespace Net::Tcp {
asm volatile("pushfq; pop %0; cli" : "=r"(flags) :: "memory");
conn->Lock.Acquire();
if (conn->SendUnack >= expectedAck) {
if (SeqLessOrEqual(expectedAck, conn->SendUnack)) {
conn->RetransmitLen = 0;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
@@ -639,7 +728,9 @@ namespace Net::Tcp {
conn->RetransmitLen = 0;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return sent > 0 ? sent : -1;
int result = sent > 0 ? (int)sent : -1;
finishSend();
return result;
}
uint64_t now = Timekeeping::GetMilliseconds();
@@ -653,7 +744,9 @@ namespace Net::Tcp {
conn->RetransmitLen = 0;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return sent > 0 ? sent : -1;
int result = sent > 0 ? (int)sent : -1;
finishSend();
return result;
}
uint32_t savedNext = conn->SendNext;
@@ -667,7 +760,9 @@ namespace Net::Tcp {
conn->RetransmitLen = 0;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
return sent > 0 ? sent : -1;
int result = sent > 0 ? (int)sent : -1;
finishSend();
return result;
}
}
@@ -684,7 +779,8 @@ namespace Net::Tcp {
}
}
return sent;
finishSend();
return (int)sent;
}
int Receive(Connection* conn, uint8_t* buffer, uint16_t bufferSize) {
@@ -815,9 +911,9 @@ namespace Net::Tcp {
}
Sched::BlockOnObject(conn, 50);
}
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return;
}
@@ -835,29 +931,29 @@ namespace Net::Tcp {
}
Sched::BlockOnObject(conn, 50);
}
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return;
}
case State::Listen:
case State::SynSent: {
conn->CurrentState = State::Closed;
conn->Active = false;
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return;
}
default:
conn->Lock.Release();
asm volatile("push %0; popfq" :: "r"(flags) : "memory");
conn->Active = false;
Sched::WakeObjectWaiters(conn);
Ipc::NotifyTcpConnectionChanged(conn);
ReleaseConnection(conn);
return;
}
}
@@ -875,7 +971,7 @@ namespace Net::Tcp {
bool HasPendingAccept(Connection* conn) {
if (conn == nullptr) return false;
conn->Lock.Acquire();
bool pending = conn->PendingAccept;
bool pending = conn->PendingCount > 0;
conn->Lock.Release();
return pending;
}
kernel/src/Net/Tcp.hpp
+1 -1
View File
@@ -65,7 +65,7 @@ namespace Net::Tcp {
Connection* Connect(uint32_t destIp, uint16_t destPort, uint16_t srcPort);
// Send data on an established connection. Returns number of bytes sent.
int Send(Connection* conn, const uint8_t* data, uint16_t length);
int Send(Connection* conn, const uint8_t* data, uint32_t length);
// Receive data from an established connection. Returns number of bytes received.
// Blocks until data is available or connection is closed.
kernel/src/Net/Udp.cpp
+20 -4
View File
@@ -11,6 +11,7 @@
#include <Libraries/Memory.hpp>
#include <Terminal/Terminal.hpp>
#include <CppLib/Stream.hpp>
#include <CppLib/Spinlock.hpp>
using namespace Kt;
@@ -22,8 +23,9 @@ namespace Net::Udp {
bool Active;
};
static constexpr uint32_t MAX_BINDINGS = 16;
static constexpr uint32_t MAX_BINDINGS = 64;
static PortBinding g_bindings[MAX_BINDINGS] = {};
static kcp::Spinlock g_bindingsLock;
void Initialize() {
for (uint32_t i = 0; i < MAX_BINDINGS; i++) {
@@ -58,13 +60,18 @@ namespace Net::Udp {
const uint8_t* payload = data + HEADER_SIZE;
uint16_t payloadLen = udpLen - HEADER_SIZE;
// Dispatch to bound callback
// Snapshot the callback under the binding lock, then invoke it after
// releasing the lock so callbacks may safely bind/unbind other ports.
RecvCallback callback = nullptr;
g_bindingsLock.Acquire();
for (uint32_t i = 0; i < MAX_BINDINGS; i++) {
if (g_bindings[i].Active && g_bindings[i].Port == dstPort) {
g_bindings[i].Callback(srcIp, srcPort, dstPort, payload, payloadLen);
return;
callback = g_bindings[i].Callback;
break;
}
}
g_bindingsLock.Release();
if (callback) callback(srcIp, srcPort, dstPort, payload, payloadLen);
}
bool Send(uint32_t destIp, uint16_t srcPort, uint16_t destPort,
@@ -96,9 +103,12 @@ namespace Net::Udp {
}
bool Bind(uint16_t port, RecvCallback callback) {
if (port == 0 || callback == nullptr) return false;
g_bindingsLock.Acquire();
// Check for duplicate
for (uint32_t i = 0; i < MAX_BINDINGS; i++) {
if (g_bindings[i].Active && g_bindings[i].Port == port) {
g_bindingsLock.Release();
return false;
}
}
@@ -109,19 +119,25 @@ namespace Net::Udp {
g_bindings[i].Port = port;
g_bindings[i].Callback = callback;
g_bindings[i].Active = true;
g_bindingsLock.Release();
return true;
}
}
g_bindingsLock.Release();
return false;
}
void Unbind(uint16_t port) {
g_bindingsLock.Acquire();
for (uint32_t i = 0; i < MAX_BINDINGS; i++) {
if (g_bindings[i].Active && g_bindings[i].Port == port) {
g_bindings[i].Active = false;
g_bindings[i].Callback = nullptr;
g_bindingsLock.Release();
return;
}
}
g_bindingsLock.Release();
}
}