nt-lite/code/kernel/ke_port.c

// Kernel Port IPC mechanism. Ports provide copied messaging, synchronous calls, waitability, and capability transfer.

#include <stdint.h>
#include <stdbool.h>
#include <string.h>
#include <intrin.h>
#include <kernel/ke_port.h>
#include <kernel/ke_system_service.h>
#include <kernel/ke_handle.h>
#include <kernel/ke_process.h>
#include <kernel/ke_thread.h>
#include <kernel/ke_list.h>
#include <kernel/ke_spinlock.h>
#include <kernel/ke_time.h>
#include <kernel/ke_id.h>
#include <kernel/mm_virt.h>
#include <kernel/mm_pool.h>
#include <kernel/ke_debug.h>
#include <kernel/se_security.h>

#define PORT_KIND_LISTENER 1u
#define PORT_KIND_ENDPOINT 2u

#define PORT_CALL_PENDING   1u
#define PORT_CALL_DELIVERED 2u
#define PORT_CALL_REPLYING  3u
#define PORT_CALL_COMPLETED 4u
#define PORT_CALL_CANCELLED 5u

#define PORT_PROCESS_MESSAGE_LIMIT 256u
#define PORT_PROCESS_BYTE_LIMIT    (256u * 1024u)
#define PORT_PROCESS_HANDLE_LIMIT  512u
#define PORT_PROCESS_CALL_LIMIT    128u

typedef struct PORT_CONNECTION PORT_CONNECTION;
typedef struct PORT_MESSAGE PORT_MESSAGE;
typedef struct PORT_CALL PORT_CALL;
typedef struct PORT_RECEIVE_WAIT PORT_RECEIVE_WAIT;

typedef struct PORT_PENDING_CONNECTION {
    LIST_ENTRY Link;
    struct PORT* ServerEndpoint;
    SYS_PORT_CONNECTION_INFO Info;
} PORT_PENDING_CONNECTION;

struct PORT_MESSAGE {
    LIST_ENTRY Link;
    SYS_PORT_MESSAGE_INFO Info;
    void* Payload;
    struct PROCESS* AccountingProcess;
    uint32_t AccountedBytes;
    uint32_t AccountedHandles;
    struct KE_CAPTURED_HANDLE* Captured[SYS_PORT_MAX_TRANSFER_HANDLES];
    uint32_t CapturedCount;
    PORT_CALL* Call;
};

struct PORT_RECEIVE_WAIT {
    LIST_ENTRY Link;
    struct PROCESS* Process;
    void* Payload;
    uint32_t PayloadCapacity;
    uint32_t HandleCapacity;
    MM_LOCKED_USER_BUFFER* LockedPayload;
    EVENT* CompletionEvent;
    SYS_PORT_MESSAGE_INFO Info;
    SYS_HANDLE Handles[SYS_PORT_MAX_TRANSFER_HANDLES];
    uint32_t HandleCount;
    KSTATUS Status;
    volatile long Claimed;
    volatile long Completed;
};

struct PORT_CALL {
    LIST_ENTRY Link;
    volatile long RefCount;
    PORT_CONNECTION* Connection;
    uint32_t MessageId;
    uint32_t CallerSide;
    volatile long State;
    struct PROCESS* CallerProcess;
    EVENT* CompletionEvent;
    MM_LOCKED_USER_BUFFER* LockedReplyPayload;
    void* ReplyPayload;
    uint32_t ReplyCapacity;
    uint32_t ReplyHandleCapacity;
    SYS_PORT_MESSAGE_INFO ReplyInfo;
    SYS_HANDLE ReplyHandles[SYS_PORT_MAX_TRANSFER_HANDLES];
    uint32_t ReplyHandleCount;
    KSTATUS Status;
};

struct PORT_CONNECTION {
    volatile long RefCount;
    SPINLOCK Lock;
    struct PORT* Endpoints[2];
    uint8_t Disconnected[2];
    LIST_ENTRY Calls;
};

typedef struct PORT_LISTENER_STATE {
    SPINLOCK Lock;
    LIST_ENTRY Pending;
    uint32_t PendingCount;
    uint32_t Backlog;
    EVENT* WaitEvent;
} PORT_LISTENER_STATE;

typedef struct PORT_ENDPOINT_STATE {
    PORT_CONNECTION* Connection;
    uint32_t Side;
    EVENT* WaitEvent;
    EVENT* SpaceEvent;
    LIST_ENTRY Incoming;
    uint32_t IncomingCount;
    uint32_t IncomingBytes;
    uint32_t IncomingHandles;
    LIST_ENTRY WaitingReceivers;
} PORT_ENDPOINT_STATE;

struct PORT {
    OBJECT Header;
    uint32_t Kind;
    union {
        PORT_LISTENER_STATE Listener;
        PORT_ENDPOINT_STATE Endpoint;
    } u;
};

static KE_ID_ALLOCATOR gPortMessageIds = KE_ID_ALLOCATOR_INIT(1u);

typedef struct PORT_COUNTERS {
    volatile long long ConnectionsCreated;
    volatile long long DirectDeliveries;
    volatile long long DirectBytes;
    volatile long long QueuedMessages;
    volatile long long QueuedBytes;
    volatile long long QueuedDeliveries;
    volatile long long RepliesCompleted;
    volatile long long Disconnects;
    volatile long long ProcessQuotaRejects;
} PORT_COUNTERS;

static PORT_COUNTERS gPortCounters;

// Reads one 64-bit diagnostic counter atomically without changing it.
static uint64_t KiPortReadCounter(volatile long long* counter) {
    return (uint64_t)_InterlockedCompareExchange64(counter, 0, 0);
}

// Copies current Port transport diagnostics for the kernel stats syscall.
void KePortQueryStats(KE_PORT_STATS* statsOut) {
    if (!statsOut)
        return;
    statsOut->ConnectionsCreated = KiPortReadCounter(&gPortCounters.ConnectionsCreated);
    statsOut->DirectDeliveries = KiPortReadCounter(&gPortCounters.DirectDeliveries);
    statsOut->DirectBytes = KiPortReadCounter(&gPortCounters.DirectBytes);
    statsOut->QueuedMessages = KiPortReadCounter(&gPortCounters.QueuedMessages);
    statsOut->QueuedBytes = KiPortReadCounter(&gPortCounters.QueuedBytes);
    statsOut->QueuedDeliveries = KiPortReadCounter(&gPortCounters.QueuedDeliveries);
    statsOut->RepliesCompleted = KiPortReadCounter(&gPortCounters.RepliesCompleted);
    statsOut->Disconnects = KiPortReadCounter(&gPortCounters.Disconnects);
    statsOut->ProcessQuotaRejects = KiPortReadCounter(&gPortCounters.ProcessQuotaRejects);
}

// Returns the complete capability rights granted to a newly created endpoint handle.
static SYS_RIGHTS KiPortEndpointRights(void) {
    return SYS_RIGHT_MANAGE | SYS_RIGHT_QUERY | SYS_RIGHT_WAIT |
           SYS_PORT_RIGHT_SEND | SYS_PORT_RIGHT_RECEIVE |
           SYS_PORT_RIGHT_CALL | SYS_PORT_RIGHT_REPLY |
           SYS_PORT_RIGHT_DISCONNECT;
}

// Returns one process security identity for kernel-stamped IPC metadata.
static uint32_t KiPortSecurityId(const PROCESS* process) {
    return process && process->SecurityContext ? process->SecurityContext->SecurityId : 0u;
}

// Charges one process for the kernel resources retained by a submitted Port message.
static KSTATUS KiPortChargeProcess(PROCESS* process, uint32_t payloadBytes, uint32_t handleCount) {
    if (!process || payloadBytes > SYS_PORT_MAX_INLINE_BYTES || handleCount > SYS_PORT_MAX_TRANSFER_HANDLES)
        return NTLSTATUS_INVALID_PARAMETER;
    if (!ObTryReferenceObject(&process->Header))
        return NTLSTATUS_OBJECT_DEAD;

    uintptr_t flags = KeAcquireSpinlock(&process->PortQuotaLock);
    if (process->PortOutstandingMessages >= PORT_PROCESS_MESSAGE_LIMIT ||
        payloadBytes > PORT_PROCESS_BYTE_LIMIT - process->PortOutstandingBytes ||
        handleCount > PORT_PROCESS_HANDLE_LIMIT - process->PortOutstandingHandles) {
        KeReleaseSpinlock(&process->PortQuotaLock, flags);
        ObDereferenceObject(&process->Header);
        _InterlockedIncrement64(&gPortCounters.ProcessQuotaRejects);
        return NTLSTATUS_LIMIT_REACHED;
    }
    process->PortOutstandingMessages++;
    process->PortOutstandingBytes += payloadBytes;
    process->PortOutstandingHandles += handleCount;
    KeReleaseSpinlock(&process->PortQuotaLock, flags);
    return NTLSTATUS_SUCCESS;
}

// Releases one process Port quota charge after delivery, cancellation, or disconnect.
static void KiPortReleaseProcessCharge(PROCESS* process, uint32_t payloadBytes, uint32_t handleCount) {
    if (!process)
        return;

    uintptr_t flags = KeAcquireSpinlock(&process->PortQuotaLock);
    if (process->PortOutstandingMessages == 0 ||
        process->PortOutstandingBytes < payloadBytes ||
        process->PortOutstandingHandles < handleCount) {
        KeReleaseSpinlock(&process->PortQuotaLock, flags);
        KePanic(KE_PANIC_CODE_GENERIC,
                "port: process quota accounting underflow pid=%u\n",
                process->ProcessId);
    }
    process->PortOutstandingMessages--;
    process->PortOutstandingBytes -= payloadBytes;
    process->PortOutstandingHandles -= handleCount;
    KeReleaseSpinlock(&process->PortQuotaLock, flags);
    ObDereferenceObject(&process->Header);
}

// Charges one caller for a retained synchronous call context and locked reply buffer.
static KSTATUS KiPortChargeCall(PROCESS* process) {
    if (!process)
        return NTLSTATUS_INVALID_PARAMETER;
    if (!ObTryReferenceObject(&process->Header))
        return NTLSTATUS_OBJECT_DEAD;

    uintptr_t flags = KeAcquireSpinlock(&process->PortQuotaLock);
    if (process->PortOutstandingCalls >= PORT_PROCESS_CALL_LIMIT) {
        KeReleaseSpinlock(&process->PortQuotaLock, flags);
        ObDereferenceObject(&process->Header);
        _InterlockedIncrement64(&gPortCounters.ProcessQuotaRejects);
        return NTLSTATUS_LIMIT_REACHED;
    }
    process->PortOutstandingCalls++;
    KeReleaseSpinlock(&process->PortQuotaLock, flags);
    return NTLSTATUS_SUCCESS;
}

// Releases a retained synchronous call quota charge and its process reference.
static void KiPortReleaseCallCharge(PROCESS* process) {
    if (!process)
        return;
    uintptr_t flags = KeAcquireSpinlock(&process->PortQuotaLock);
    if (process->PortOutstandingCalls == 0) {
        KeReleaseSpinlock(&process->PortQuotaLock, flags);
        KePanic(KE_PANIC_CODE_GENERIC,
                "port: process call quota underflow pid=%u\n",
                process->ProcessId);
    }
    process->PortOutstandingCalls--;
    KeReleaseSpinlock(&process->PortQuotaLock, flags);
    ObDereferenceObject(&process->Header);
}

// Drops one shared connection reference after an endpoint object releases it.
static void KiPortDereferenceConnection(PORT_CONNECTION* connection) {
    if (!connection)
        return;
    long refs = _InterlockedDecrement(&connection->RefCount);
    if (refs > 0)
        return;
    if (refs < 0) {
        KePanic(KE_PANIC_CODE_GENERIC,
                "port: connection ref underflow connection=0x%016llX refs=%ld\n",
                (uint64_t)(uintptr_t)connection,
                refs);
    }
    if (!LIST_IS_EMPTY(&connection->Calls)) {
        KePanic(KE_PANIC_CODE_GENERIC,
                "port: destroying connection with pending calls connection=0x%016llX\n",
                (uint64_t)(uintptr_t)connection);
    }
    MmFreePool(connection);
}

// Releases one synchronous-call reference and destroys the context after the last owner leaves.
static void KiPortReleaseCall(PORT_CALL* call);

// Releases every capability capture and payload buffer owned by one queued message.
static void KiPortFreeMessage(PORT_MESSAGE* message) {
    if (!message)
        return;
    for (uint32_t i = 0; i < message->CapturedCount; ++i) {
        if (message->Captured[i]) {
            KeReleaseCapturedHandle(message->Captured[i]);
            message->Captured[i] = (struct KE_CAPTURED_HANDLE*)0;
        }
    }
    if (message->Payload)
        MmFreePool(message->Payload);
    if (message->Call) {
        KiPortReleaseCall(message->Call);
        message->Call = (PORT_CALL*)0;
    }
    if (message->AccountingProcess) {
        KiPortReleaseProcessCharge(message->AccountingProcess,
                                   message->AccountedBytes,
                                   message->AccountedHandles);
        message->AccountingProcess = (PROCESS*)0;
    }
    MmFreePool(message);
}

// Closes a batch of handles that were materialized but could not be returned to the receiver.
static void KiPortCloseHandles(PROCESS* process, SYS_HANDLE* handles, uint32_t count) {
    if (!process || !handles)
        return;
    for (uint32_t i = 0; i < count; ++i) {
        if (handles[i] != SYS_HANDLE_INVALID) {
            (void)KeCloseHandle(process, handles[i]);
            handles[i] = SYS_HANDLE_INVALID;
        }
    }
}

// Captures user-specified handles as kernel capability grants for later delivery.
static KSTATUS KiPortCaptureHandles(PROCESS* sourceProcess,
                                    const SYS_PORT_HANDLE_TRANSFER* transfers,
                                    uint32_t count,
                                    struct KE_CAPTURED_HANDLE** captured) {
    if (count == 0)
        return NTLSTATUS_SUCCESS;
    if (!sourceProcess || !transfers || !captured || count > SYS_PORT_MAX_TRANSFER_HANDLES)
        return NTLSTATUS_INVALID_PARAMETER;

    for (uint32_t i = 0; i < count; ++i)
        captured[i] = (struct KE_CAPTURED_HANDLE*)0;

    for (uint32_t i = 0; i < count; ++i) {
        uint32_t flags = 0;
        if (transfers[i].Flags & SYS_PORT_TRANSFER_FLAG_SAME_RIGHTS)
            flags |= KE_HANDLE_TRANSFER_FLAG_SAME_RIGHTS;
        if (transfers[i].Flags & SYS_PORT_TRANSFER_FLAG_DERIVE)
            flags |= KE_HANDLE_TRANSFER_FLAG_DERIVE;
        if (transfers[i].Flags & ~(SYS_PORT_TRANSFER_FLAG_SAME_RIGHTS |
                                   SYS_PORT_TRANSFER_FLAG_DERIVE)) {
            for (uint32_t j = 0; j < i; ++j)
                KeReleaseCapturedHandle(captured[j]);
            return NTLSTATUS_INVALID_PARAMETER;
        }

        KSTATUS status = KeCaptureHandleForTransfer(sourceProcess,
                                                    transfers[i].Handle,
                                                    transfers[i].DesiredRights,
                                                    flags,
                                                    &captured[i]);
        if (!KSUCCESS(status)) {
            for (uint32_t j = 0; j < i; ++j)
                KeReleaseCapturedHandle(captured[j]);
            return status;
        }
    }
    return NTLSTATUS_SUCCESS;
}

// Allocates one message header and captures its monotonic identity and sender metadata.
static KSTATUS KiPortCreateMessage(PROCESS* senderProcess,
                                   THREAD* senderThread,
                                   uint32_t type,
                                   uint32_t messageId,
                                   uint32_t payloadLength,
                                   const SYS_PORT_HANDLE_TRANSFER* transfers,
                                   uint32_t transferCount,
                                   PORT_CALL* call,
                                   PORT_MESSAGE** messageOut) {
    if (!senderProcess || !senderThread || !messageOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *messageOut = (PORT_MESSAGE*)0;
    if (payloadLength > SYS_PORT_MAX_INLINE_BYTES || transferCount > SYS_PORT_MAX_TRANSFER_HANDLES)
        return NTLSTATUS_INVALID_PARAMETER;

    KSTATUS quotaStatus = KiPortChargeProcess(senderProcess, payloadLength, transferCount);
    if (!KSUCCESS(quotaStatus))
        return quotaStatus;

    PORT_MESSAGE* message = (PORT_MESSAGE*)MmAllocatePoolTag(sizeof(PORT_MESSAGE),
                                                             MM_POOL_TAG('p', 'm', 's', 'g'));
    if (!message) {
        KiPortReleaseProcessCharge(senderProcess, payloadLength, transferCount);
        return NTLSTATUS_NO_MEMORY;
    }
    memset(message, 0, sizeof(*message));
    LIST_INIT(&message->Link);
    message->AccountingProcess = senderProcess;
    message->AccountedBytes = payloadLength;
    message->AccountedHandles = transferCount;

    if (messageId == 0) {
        KSTATUS status = KeAllocateId32(&gPortMessageIds, &messageId);
        if (!KSUCCESS(status)) {
            KiPortFreeMessage(message);
            return status;
        }
    }

    message->Info.Size = sizeof(message->Info);
    message->Info.MessageId = messageId;
    message->Info.Type = type;
    message->Info.PayloadLength = payloadLength;
    message->Info.SenderProcessId = senderProcess->ProcessId;
    message->Info.SenderThreadId = senderThread->ThreadId;
    message->Info.SenderSecurityId = KiPortSecurityId(senderProcess);
    message->CapturedCount = transferCount;
    message->Info.HandleCount = transferCount;
    message->Call = call;
    if (call)
        _InterlockedIncrement(&call->RefCount);

    KSTATUS status = KiPortCaptureHandles(senderProcess,
                                          transfers,
                                          transferCount,
                                          message->Captured);
    if (!KSUCCESS(status)) {
        KiPortFreeMessage(message);
        return status;
    }

    *messageOut = message;
    return NTLSTATUS_SUCCESS;
}

// Copies the sender payload into kernel-owned storage when immediate rendezvous is unavailable.
static KSTATUS KiPortSnapshotPayload(PORT_MESSAGE* message, const void* payload, uint32_t payloadLength) {
    if (!message)
        return NTLSTATUS_INVALID_PARAMETER;
    if (message->Payload)
        return NTLSTATUS_SUCCESS;
    if (payloadLength == 0) {
        message->Info.PayloadLength = 0;
        return NTLSTATUS_SUCCESS;
    }
    if (!payload || payloadLength > SYS_PORT_MAX_INLINE_BYTES)
        return NTLSTATUS_INVALID_PARAMETER;

    void* copy = MmAllocatePoolTag(payloadLength, MM_POOL_TAG('p', 'd', 'a', 't'));
    if (!copy)
        return NTLSTATUS_NO_MEMORY;
    KSTATUS status = CopyFromUser(copy, payload, payloadLength);
    if (!KSUCCESS(status)) {
        MmFreePool(copy);
        return status;
    }
    message->Payload = copy;
    message->Info.PayloadLength = payloadLength;
    return NTLSTATUS_SUCCESS;
}

// Updates endpoint queue events after one message is removed.
static void KiPortNoteDequeuedLocked(PORT* endpoint, PORT_MESSAGE* message) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !message)
        return;
    PORT_ENDPOINT_STATE* state = &endpoint->u.Endpoint;
    if (state->IncomingCount == 0 || state->IncomingBytes < message->Info.PayloadLength ||
        state->IncomingHandles < message->CapturedCount) {
        KePanic(KE_PANIC_CODE_GENERIC, "port: queue accounting underflow\n");
    }
    state->IncomingCount--;
    state->IncomingBytes -= message->Info.PayloadLength;
    state->IncomingHandles -= message->CapturedCount;
    KeSignalEvent(state->SpaceEvent);
    if (state->IncomingCount == 0 &&
        !state->Connection->Disconnected[1u - state->Side])
        KeClearEvent(state->WaitEvent);
}

// Returns whether one queued message can fit within the peer endpoint quotas.
static bool KiPortMessageFitsLocked(const PORT* endpoint, const PORT_MESSAGE* message) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !message)
        return false;
    const PORT_ENDPOINT_STATE* state = &endpoint->u.Endpoint;
    if (state->IncomingCount >= SYS_PORT_QUEUE_MESSAGE_LIMIT)
        return false;
    if (message->Info.PayloadLength > SYS_PORT_QUEUE_BYTE_LIMIT - state->IncomingBytes)
        return false;
    if (message->CapturedCount > SYS_PORT_QUEUE_HANDLE_LIMIT - state->IncomingHandles)
        return false;
    return true;
}

// Completes a direct receive rendezvous after the connection lock has claimed the waiting receiver.
static KSTATUS KiPortDeliverDirect(PORT* targetEndpoint,
                                   PORT_RECEIVE_WAIT* receiver,
                                   PORT_MESSAGE* message,
                                   PROCESS* senderProcess,
                                   const void* sourcePayload) {
    if (!targetEndpoint || !receiver || !message || !senderProcess)
        return NTLSTATUS_INVALID_PARAMETER;

    KSTATUS status = NTLSTATUS_SUCCESS;
    SYS_HANDLE handles[SYS_PORT_MAX_TRANSFER_HANDLES] = {0};
    uint32_t handleCount = message->CapturedCount;

    if (handleCount) {
        status = KeMaterializeCapturedHandles(receiver->Process,
                                              message->Captured,
                                              handleCount,
                                              handles);
    }
    if (KSUCCESS(status) && message->Info.PayloadLength) {
        status = MmCopyUserToUser(senderProcess,
                                  sourcePayload,
                                  receiver->Process,
                                  receiver->Payload,
                                  message->Info.PayloadLength);
    }

    if (!KSUCCESS(status)) {
        KiPortCloseHandles(receiver->Process, handles, handleCount);
        receiver->HandleCount = 0;
    } else {
        memcpy(receiver->Handles, handles, sizeof(SYS_HANDLE) * handleCount);
        receiver->HandleCount = handleCount;
    }

    receiver->Info = message->Info;
    receiver->Status = status;
    if (KSUCCESS(status)) {
        _InterlockedIncrement64(&gPortCounters.DirectDeliveries);
        _InterlockedExchangeAdd64(&gPortCounters.DirectBytes, (long long)message->Info.PayloadLength);
    }
    _InterlockedExchange(&receiver->Completed, 1);
    KeSignalEvent(receiver->CompletionEvent);
    return status;
}

// Completes one synchronous call with an error that does not include a reply payload.
static void KiPortCompleteCallError(PORT_CALL* call, KSTATUS status) {
    if (!call)
        return;
    call->Status = status;
    call->ReplyHandleCount = 0;
    _InterlockedExchange(&call->State, PORT_CALL_COMPLETED);
    KeSignalEvent(call->CompletionEvent);
}

// Removes one queued request belonging to a call while the connection lock is held.
static PORT_MESSAGE* KiPortRemoveQueuedCallLocked(PORT_CONNECTION* connection, PORT_CALL* call) {
    if (!connection || !call)
        return (PORT_MESSAGE*)0;
    uint32_t targetSide = 1u - call->CallerSide;
    PORT* endpoint = connection->Endpoints[targetSide];
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT)
        return (PORT_MESSAGE*)0;

    LIST_ENTRY* head = &endpoint->u.Endpoint.Incoming;
    for (LIST_ENTRY* link = head->Fwd; link != head; link = link->Fwd) {
        PORT_MESSAGE* message = CONTAINING_RECORD(link, PORT_MESSAGE, Link);
        if (message->Call == call) {
            KeListRemoveChecked(&message->Link);
            KiPortNoteDequeuedLocked(endpoint, message);
            return message;
        }
    }
    return (PORT_MESSAGE*)0;
}

// Disconnects one endpoint side, drains undelivered messages, and completes pending calls.
static void KiPortMarkDisconnected(PORT* endpoint, bool removeEndpointPointer) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT)
        return;

    PORT_CONNECTION* connection = endpoint->u.Endpoint.Connection;
    if (!connection)
        return;

    LIST_ENTRY abandonedMessages;
    LIST_ENTRY completedCalls;
    LIST_INIT(&abandonedMessages);
    LIST_INIT(&completedCalls);

    uint32_t side = endpoint->u.Endpoint.Side;
    uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
    if (!connection->Disconnected[side]) {
        connection->Disconnected[side] = 1;
        _InterlockedIncrement64(&gPortCounters.Disconnects);
    }
    if (removeEndpointPointer && connection->Endpoints[side] == endpoint)
        connection->Endpoints[side] = (PORT*)0;

    while (!LIST_IS_EMPTY(&endpoint->u.Endpoint.WaitingReceivers)) {
        LIST_ENTRY* link = KeListPopHeadChecked(&endpoint->u.Endpoint.WaitingReceivers);
        PORT_RECEIVE_WAIT* waiter = CONTAINING_RECORD(link, PORT_RECEIVE_WAIT, Link);
        if (_InterlockedCompareExchange(&waiter->Claimed, 1, 0) == 0) {
            waiter->Status = NTLSTATUS_PORT_DISCONNECTED;
            _InterlockedExchange(&waiter->Completed, 1);
            KeSignalEvent(waiter->CompletionEvent);
        }
    }

    while (!LIST_IS_EMPTY(&endpoint->u.Endpoint.Incoming)) {
        LIST_ENTRY* link = KeListPopHeadChecked(&endpoint->u.Endpoint.Incoming);
        PORT_MESSAGE* message = CONTAINING_RECORD(link, PORT_MESSAGE, Link);
        if (endpoint->u.Endpoint.IncomingCount)
            endpoint->u.Endpoint.IncomingCount--;
        if (endpoint->u.Endpoint.IncomingBytes >= message->Info.PayloadLength)
            endpoint->u.Endpoint.IncomingBytes -= message->Info.PayloadLength;
        if (endpoint->u.Endpoint.IncomingHandles >= message->CapturedCount)
            endpoint->u.Endpoint.IncomingHandles -= message->CapturedCount;
        KeListInsertTailChecked(&abandonedMessages, &message->Link);
    }

    LIST_ENTRY* callHead = &connection->Calls;
    LIST_ENTRY* callLink = callHead->Fwd;
    while (callLink != callHead) {
        PORT_CALL* call = CONTAINING_RECORD(callLink, PORT_CALL, Link);
        callLink = callLink->Fwd;
        long state = _InterlockedCompareExchange(&call->State, 0, 0);
        if (state == PORT_CALL_REPLYING || state == PORT_CALL_COMPLETED || state == PORT_CALL_CANCELLED)
            continue;
        PORT_MESSAGE* queuedMessage = KiPortRemoveQueuedCallLocked(connection, call);
        if (queuedMessage)
            KeListInsertTailChecked(&abandonedMessages, &queuedMessage->Link);
        KeListRemoveChecked(&call->Link);
        call->Status = NTLSTATUS_PORT_DISCONNECTED;
        _InterlockedExchange(&call->State, PORT_CALL_COMPLETED);
        KeListInsertTailChecked(&completedCalls, &call->Link);
    }

    PORT* peer = connection->Endpoints[1u - side];
    if (peer) {
        KeSignalEvent(peer->u.Endpoint.WaitEvent);
        KeSignalEvent(peer->u.Endpoint.SpaceEvent);
    }
    KeSignalEvent(endpoint->u.Endpoint.WaitEvent);
    KeSignalEvent(endpoint->u.Endpoint.SpaceEvent);
    KeReleaseSpinlock(&connection->Lock, flags);

    while (!LIST_IS_EMPTY(&abandonedMessages)) {
        LIST_ENTRY* link = KeListPopHeadChecked(&abandonedMessages);
        PORT_MESSAGE* message = CONTAINING_RECORD(link, PORT_MESSAGE, Link);
        KiPortFreeMessage(message);
    }
    while (!LIST_IS_EMPTY(&completedCalls)) {
        LIST_ENTRY* link = KeListPopHeadChecked(&completedCalls);
        PORT_CALL* call = CONTAINING_RECORD(link, PORT_CALL, Link);
        KeSignalEvent(call->CompletionEvent);
    }
}

// Destroys either a listener or endpoint Port object.
static void KiPortDestroy(OBJECT* object) {
    PORT* port = (PORT*)object;
    if (!port)
        return;

    if (port->Kind == PORT_KIND_LISTENER) {
        KeSystemServiceUnregisterListener(port);
        while (true) {
            PORT_PENDING_CONNECTION* pending = (PORT_PENDING_CONNECTION*)0;
            uintptr_t flags = KeAcquireSpinlock(&port->u.Listener.Lock);
            LIST_ENTRY* link = KeListPopHeadChecked(&port->u.Listener.Pending);
            if (link) {
                pending = CONTAINING_RECORD(link, PORT_PENDING_CONNECTION, Link);
                if (port->u.Listener.PendingCount)
                    port->u.Listener.PendingCount--;
            }
            KeReleaseSpinlock(&port->u.Listener.Lock, flags);
            if (!pending)
                break;
            ObDereferenceObject((OBJECT*)pending->ServerEndpoint);
            MmFreePool(pending);
        }
        if (port->u.Listener.WaitEvent)
            ObDereferenceObject(&port->u.Listener.WaitEvent->Header);
        return;
    }

    if (port->Kind == PORT_KIND_ENDPOINT) {
        KiPortMarkDisconnected(port, true);
        if (port->u.Endpoint.WaitEvent)
            ObDereferenceObject(&port->u.Endpoint.WaitEvent->Header);
        if (port->u.Endpoint.SpaceEvent)
            ObDereferenceObject(&port->u.Endpoint.SpaceEvent->Header);
        KiPortDereferenceConnection(port->u.Endpoint.Connection);
        port->u.Endpoint.Connection = (PORT_CONNECTION*)0;
    }
}

// Allocates one endpoint object attached to a shared connection side.
static KSTATUS KiPortCreateEndpoint(PORT_CONNECTION* connection, uint32_t side, PORT** endpointOut) {
    if (!connection || !endpointOut || side > 1u)
        return NTLSTATUS_INVALID_PARAMETER;
    *endpointOut = (PORT*)0;

    PORT* endpoint = (PORT*)ObAllocateObject(ObjectTypePort, sizeof(PORT));
    if (!endpoint)
        return NTLSTATUS_NO_MEMORY;
    endpoint->Header.Destroy = KiPortDestroy;
    endpoint->Kind = PORT_KIND_ENDPOINT;
    endpoint->u.Endpoint.Connection = connection;
    endpoint->u.Endpoint.Side = side;
    LIST_INIT(&endpoint->u.Endpoint.Incoming);
    LIST_INIT(&endpoint->u.Endpoint.WaitingReceivers);

    KSTATUS status = KeEventCreate(EventManualReset, 0, &endpoint->u.Endpoint.WaitEvent);
    if (!KSUCCESS(status)) {
        endpoint->u.Endpoint.Connection = (PORT_CONNECTION*)0;
        ObDereferenceObject(&endpoint->Header);
        return status;
    }
    status = KeEventCreate(EventManualReset, 1, &endpoint->u.Endpoint.SpaceEvent);
    if (!KSUCCESS(status)) {
        endpoint->u.Endpoint.Connection = (PORT_CONNECTION*)0;
        ObDereferenceObject(&endpoint->Header);
        return status;
    }

    *endpointOut = endpoint;
    return NTLSTATUS_SUCCESS;
}

// Allocates one paired connection and its two endpoint objects.
static KSTATUS KiPortCreateConnection(PORT** clientOut, PORT** serverOut) {
    if (!clientOut || !serverOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *clientOut = (PORT*)0;
    *serverOut = (PORT*)0;

    PORT_CONNECTION* connection = (PORT_CONNECTION*)MmAllocatePoolTag(sizeof(PORT_CONNECTION),
                                                                       MM_POOL_TAG('p', 'c', 'o', 'n'));
    if (!connection)
        return NTLSTATUS_NO_MEMORY;
    memset(connection, 0, sizeof(*connection));
    connection->RefCount = 2;
    connection->Lock = (SPINLOCK)SPINLOCK_INIT;
    LIST_INIT(&connection->Calls);

    PORT* client = (PORT*)0;
    PORT* server = (PORT*)0;
    KSTATUS status = KiPortCreateEndpoint(connection, 0, &client);
    if (!KSUCCESS(status)) {
        connection->RefCount = 1;
        KiPortDereferenceConnection(connection);
        return status;
    }
    status = KiPortCreateEndpoint(connection, 1, &server);
    if (!KSUCCESS(status)) {
        connection->RefCount = 1;
        ObDereferenceObject(&client->Header);
        return status;
    }

    uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
    connection->Endpoints[0] = client;
    connection->Endpoints[1] = server;
    KeReleaseSpinlock(&connection->Lock, flags);

    *clientOut = client;
    *serverOut = server;
    return NTLSTATUS_SUCCESS;
}

// Creates one bounded listener Port with a waitable pending-connection event.
KSTATUS KePortCreate(uint32_t backlog, PORT** portOut) {
    if (!portOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *portOut = (PORT*)0;
    if (backlog == 0)
        backlog = SYS_PORT_DEFAULT_BACKLOG;
    if (backlog > SYS_PORT_MAX_BACKLOG)
        return NTLSTATUS_INVALID_PARAMETER;

    PORT* port = (PORT*)ObAllocateObject(ObjectTypePort, sizeof(PORT));
    if (!port)
        return NTLSTATUS_NO_MEMORY;
    port->Header.Destroy = KiPortDestroy;
    port->Kind = PORT_KIND_LISTENER;
    port->u.Listener.Lock = (SPINLOCK)SPINLOCK_INIT;
    port->u.Listener.Backlog = backlog;
    LIST_INIT(&port->u.Listener.Pending);

    KSTATUS status = KeEventCreate(EventManualReset, 0, &port->u.Listener.WaitEvent);
    if (!KSUCCESS(status)) {
        ObDereferenceObject(&port->Header);
        return status;
    }

    *portOut = port;
    return NTLSTATUS_SUCCESS;
}

// Returns whether one Port object is a listener.
bool KePortIsListener(const PORT* port) {
    return port && port->Kind == PORT_KIND_LISTENER;
}

// Returns the event whose signaled state represents listener or endpoint readiness.
EVENT* KePortWaitEvent(PORT* port) {
    if (!port)
        return (EVENT*)0;
    if (port->Kind == PORT_KIND_LISTENER)
        return port->u.Listener.WaitEvent;
    if (port->Kind == PORT_KIND_ENDPOINT)
        return port->u.Endpoint.WaitEvent;
    return (EVENT*)0;
}

// Connects one client process to a listener by queueing the server endpoint for later acceptance.
KSTATUS KePortConnect(PORT* listener,
                      PROCESS* clientProcess,
                      THREAD* clientThread,
                      SYS_HANDLE* endpointHandleOut) {
    if (!listener || listener->Kind != PORT_KIND_LISTENER || !clientProcess || !clientThread || !endpointHandleOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *endpointHandleOut = SYS_HANDLE_INVALID;

    uintptr_t flags = KeAcquireSpinlock(&listener->u.Listener.Lock);
    if (listener->u.Listener.PendingCount >= listener->u.Listener.Backlog) {
        KeReleaseSpinlock(&listener->u.Listener.Lock, flags);
        return NTLSTATUS_LIMIT_REACHED;
    }
    KeReleaseSpinlock(&listener->u.Listener.Lock, flags);

    PORT* clientEndpoint = (PORT*)0;
    PORT* serverEndpoint = (PORT*)0;
    KSTATUS status = KiPortCreateConnection(&clientEndpoint, &serverEndpoint);
    if (!KSUCCESS(status))
        return status;

    SYS_HANDLE clientHandle = SYS_HANDLE_INVALID;
    status = KeCreateHandle(clientProcess,
                            &clientEndpoint->Header,
                            KiPortEndpointRights(),
                            &clientHandle);
    if (!KSUCCESS(status)) {
        ObDereferenceObject(&serverEndpoint->Header);
        ObDereferenceObject(&clientEndpoint->Header);
        return status;
    }
    ObDereferenceObject(&clientEndpoint->Header);

    PORT_PENDING_CONNECTION* pending = (PORT_PENDING_CONNECTION*)MmAllocatePoolTag(sizeof(*pending),
                                                                                    MM_POOL_TAG('p', 'c', 'n', 'q'));
    if (!pending) {
        (void)KeCloseHandle(clientProcess, clientHandle);
        ObDereferenceObject(&serverEndpoint->Header);
        return NTLSTATUS_NO_MEMORY;
    }
    memset(pending, 0, sizeof(*pending));
    LIST_INIT(&pending->Link);
    pending->ServerEndpoint = serverEndpoint;
    pending->Info.Size = sizeof(pending->Info);
    pending->Info.ClientProcessId = clientProcess->ProcessId;
    pending->Info.ClientThreadId = clientThread->ThreadId;
    pending->Info.ClientSecurityId = KiPortSecurityId(clientProcess);

    flags = KeAcquireSpinlock(&listener->u.Listener.Lock);
    if (listener->u.Listener.PendingCount >= listener->u.Listener.Backlog) {
        KeReleaseSpinlock(&listener->u.Listener.Lock, flags);
        MmFreePool(pending);
        ObDereferenceObject(&serverEndpoint->Header);
        (void)KeCloseHandle(clientProcess, clientHandle);
        return NTLSTATUS_LIMIT_REACHED;
    }
    KeListInsertTailChecked(&listener->u.Listener.Pending, &pending->Link);
    listener->u.Listener.PendingCount++;
    _InterlockedIncrement64(&gPortCounters.ConnectionsCreated);
    KeSignalEvent(listener->u.Listener.WaitEvent);
    KeReleaseSpinlock(&listener->u.Listener.Lock, flags);

    *endpointHandleOut = clientHandle;
    return NTLSTATUS_SUCCESS;
}

// Accepts the oldest queued connection, waiting on the listener event when requested.
KSTATUS KePortAccept(PORT* listener,
                     PROCESS* serverProcess,
                     uint64_t timeoutMilliseconds,
                     SYS_HANDLE* endpointHandleOut,
                     SYS_PORT_CONNECTION_INFO* infoOut) {
    if (!listener || listener->Kind != PORT_KIND_LISTENER || !serverProcess || !endpointHandleOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *endpointHandleOut = SYS_HANDLE_INVALID;

    uint64_t deadline = 0;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        deadline = KeMonotonicDeadlineAfterMilliseconds(timeoutMilliseconds);

    while (true) {
        PORT_PENDING_CONNECTION* pending = (PORT_PENDING_CONNECTION*)0;
        uintptr_t flags = KeAcquireSpinlock(&listener->u.Listener.Lock);
        LIST_ENTRY* link = KeListPopHeadChecked(&listener->u.Listener.Pending);
        if (link) {
            pending = CONTAINING_RECORD(link, PORT_PENDING_CONNECTION, Link);
            if (listener->u.Listener.PendingCount == 0) {
                KeReleaseSpinlock(&listener->u.Listener.Lock, flags);
                KePanic(KE_PANIC_CODE_GENERIC, "port: listener pending count underflow\n");
            }
            listener->u.Listener.PendingCount--;
        }
        if (listener->u.Listener.PendingCount == 0)
            KeClearEvent(listener->u.Listener.WaitEvent);
        KeReleaseSpinlock(&listener->u.Listener.Lock, flags);

        if (pending) {
            SYS_HANDLE handle = SYS_HANDLE_INVALID;
            KSTATUS status = KeCreateHandle(serverProcess,
                                            &pending->ServerEndpoint->Header,
                                            KiPortEndpointRights(),
                                            &handle);
            if (KSUCCESS(status)) {
                if (infoOut)
                    *infoOut = pending->Info;
                *endpointHandleOut = handle;
            }
            ObDereferenceObject(&pending->ServerEndpoint->Header);
            MmFreePool(pending);
            return status;
        }

        if (timeoutMilliseconds == 0)
            return NTLSTATUS_WOULD_BLOCK;

        uint64_t wait = timeoutMilliseconds;
        if (timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE) {
            wait = KeMonotonicMillisecondsUntil(deadline);
            if (wait == 0)
                return NTLSTATUS_TIMEOUT;
        }
        EVENT* events[1] = { listener->u.Listener.WaitEvent };
        KSTATUS status = KeWaitForMultipleEvents(events, 1, 0, wait, (uint32_t*)0);
        if (!KSUCCESS(status))
            return status;
    }
}

// Submits a prepared message to the peer, using direct rendezvous when a receiver is already waiting.
// This function consumes the message on every return path; queued success transfers ownership to the peer queue.
static KSTATUS KiPortSubmitMessage(PORT* endpoint,
                                   PROCESS* senderProcess,
                                   const void* sourcePayload,
                                   uint32_t payloadLength,
                                   PORT_MESSAGE* message,
                                   uint64_t timeoutMilliseconds) {
    if (!message)
        return NTLSTATUS_INVALID_PARAMETER;
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !senderProcess ||
        payloadLength > SYS_PORT_MAX_INLINE_BYTES || (payloadLength && !sourcePayload)) {
        KiPortFreeMessage(message);
        return NTLSTATUS_INVALID_PARAMETER;
    }
    message->Info.PayloadLength = payloadLength;

    PORT_CONNECTION* connection = endpoint->u.Endpoint.Connection;
    uint32_t peerSide = 1u - endpoint->u.Endpoint.Side;
    uint64_t deadline = 0;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        deadline = KeMonotonicDeadlineAfterMilliseconds(timeoutMilliseconds);

    bool snapshotted = false;
    while (true) {
        PORT* peer = (PORT*)0;
        PORT_RECEIVE_WAIT* receiver = (PORT_RECEIVE_WAIT*)0;
        uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
        if (connection->Disconnected[endpoint->u.Endpoint.Side] ||
            connection->Disconnected[peerSide] ||
            !(peer = connection->Endpoints[peerSide])) {
            KeReleaseSpinlock(&connection->Lock, flags);
            KiPortFreeMessage(message);
            return NTLSTATUS_PORT_DISCONNECTED;
        }

        if (!snapshotted && peer->u.Endpoint.IncomingCount == 0) {
            LIST_ENTRY* waitHead = &peer->u.Endpoint.WaitingReceivers;
            for (LIST_ENTRY* link = waitHead->Fwd; link != waitHead; link = link->Fwd) {
                PORT_RECEIVE_WAIT* candidate = CONTAINING_RECORD(link, PORT_RECEIVE_WAIT, Link);
                if (payloadLength > candidate->PayloadCapacity ||
                    message->CapturedCount > candidate->HandleCapacity)
                    continue;
                if (_InterlockedCompareExchange(&candidate->Claimed, 1, 0) != 0)
                    continue;
                receiver = candidate;
                KeListRemoveChecked(&receiver->Link);
                break;
            }
            if (receiver) {
                if (message->Call)
                    _InterlockedExchange(&message->Call->State, PORT_CALL_DELIVERED);
                if (!ObTryReferenceObject(&peer->Header)) {
                    receiver->Claimed = 0;
                    KeListInsertHeadChecked(&peer->u.Endpoint.WaitingReceivers, &receiver->Link);
                    KeReleaseSpinlock(&connection->Lock, flags);
                    KiPortFreeMessage(message);
                    return NTLSTATUS_OBJECT_DEAD;
                }
                KeReleaseSpinlock(&connection->Lock, flags);

                KSTATUS status = KiPortDeliverDirect(peer,
                                                     receiver,
                                                     message,
                                                     senderProcess,
                                                     sourcePayload);
                ObDereferenceObject(&peer->Header);
                KiPortFreeMessage(message);
                return status;
            }
        }
        KeReleaseSpinlock(&connection->Lock, flags);

        if (!snapshotted) {
            KSTATUS status = KiPortSnapshotPayload(message, sourcePayload, payloadLength);
            if (!KSUCCESS(status)) {
                KiPortFreeMessage(message);
                return status;
            }
            snapshotted = true;
        }

        flags = KeAcquireSpinlock(&connection->Lock);
        if (connection->Disconnected[endpoint->u.Endpoint.Side] ||
            connection->Disconnected[peerSide] ||
            !(peer = connection->Endpoints[peerSide])) {
            KeReleaseSpinlock(&connection->Lock, flags);
            KiPortFreeMessage(message);
            return NTLSTATUS_PORT_DISCONNECTED;
        }

        if (KiPortMessageFitsLocked(peer, message)) {
            KeListInsertTailChecked(&peer->u.Endpoint.Incoming, &message->Link);
            peer->u.Endpoint.IncomingCount++;
            peer->u.Endpoint.IncomingBytes += message->Info.PayloadLength;
            peer->u.Endpoint.IncomingHandles += message->CapturedCount;
            _InterlockedIncrement64(&gPortCounters.QueuedMessages);
            _InterlockedExchangeAdd64(&gPortCounters.QueuedBytes, (long long)message->Info.PayloadLength);
            if (message->Call)
                _InterlockedExchange(&message->Call->State, PORT_CALL_PENDING);
            KeSignalEvent(peer->u.Endpoint.WaitEvent);
            if (peer->u.Endpoint.IncomingCount >= SYS_PORT_QUEUE_MESSAGE_LIMIT ||
                peer->u.Endpoint.IncomingBytes >= SYS_PORT_QUEUE_BYTE_LIMIT ||
                peer->u.Endpoint.IncomingHandles >= SYS_PORT_QUEUE_HANDLE_LIMIT)
                KeClearEvent(peer->u.Endpoint.SpaceEvent);
            KeReleaseSpinlock(&connection->Lock, flags);
            return NTLSTATUS_SUCCESS;
        }

        KeClearEvent(peer->u.Endpoint.SpaceEvent);
        EVENT* spaceEvent = peer->u.Endpoint.SpaceEvent;
        if (!ObTryReferenceObject(&spaceEvent->Header)) {
            KeReleaseSpinlock(&connection->Lock, flags);
            KiPortFreeMessage(message);
            return NTLSTATUS_OBJECT_DEAD;
        }
        KeReleaseSpinlock(&connection->Lock, flags);

        if (timeoutMilliseconds == 0) {
            ObDereferenceObject(&spaceEvent->Header);
            KiPortFreeMessage(message);
            return NTLSTATUS_WOULD_BLOCK;
        }

        uint64_t wait = timeoutMilliseconds;
        if (timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE) {
            wait = KeMonotonicMillisecondsUntil(deadline);
            if (wait == 0) {
                ObDereferenceObject(&spaceEvent->Header);
                KiPortFreeMessage(message);
                return NTLSTATUS_TIMEOUT;
            }
        }
        EVENT* events[1] = { spaceEvent };
        KSTATUS status = KeWaitForMultipleEvents(events, 1, 0, wait, (uint32_t*)0);
        ObDereferenceObject(&spaceEvent->Header);
        if (!KSUCCESS(status)) {
            KiPortFreeMessage(message);
            return status;
        }
    }
}

// Sends one one-way copied notification to the peer endpoint.
KSTATUS KePortSend(PORT* endpoint,
                   PROCESS* senderProcess,
                   THREAD* senderThread,
                   const void* payload,
                   uint32_t payloadLength,
                   const SYS_PORT_HANDLE_TRANSFER* transfers,
                   uint32_t transferCount,
                   uint64_t timeoutMilliseconds) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !senderProcess || !senderThread)
        return NTLSTATUS_INVALID_PARAMETER;

    PORT_MESSAGE* message = (PORT_MESSAGE*)0;
    KSTATUS status = KiPortCreateMessage(senderProcess,
                                         senderThread,
                                         SYS_PORT_MESSAGE_NOTIFICATION,
                                         0,
                                         payloadLength,
                                         transfers,
                                         transferCount,
                                         (PORT_CALL*)0,
                                         &message);
    if (!KSUCCESS(status))
        return status;

    return KiPortSubmitMessage(endpoint,
                               senderProcess,
                               payload,
                               payloadLength,
                               message,
                               timeoutMilliseconds);
}

// Delivers one already-dequeued kernel-buffered message to the current receiver.
static KSTATUS KiPortDeliverQueued(PORT* endpoint,
                                   PROCESS* receiverProcess,
                                   PORT_MESSAGE* message,
                                   void* payload,
                                   SYS_HANDLE* handlesOut,
                                   SYS_PORT_MESSAGE_INFO* infoOut) {
    if (!endpoint || !receiverProcess || !message || !infoOut)
        return NTLSTATUS_INVALID_PARAMETER;

    SYS_HANDLE handles[SYS_PORT_MAX_TRANSFER_HANDLES] = {0};
    KSTATUS status = NTLSTATUS_SUCCESS;
    if (message->CapturedCount) {
        status = KeMaterializeCapturedHandles(receiverProcess,
                                              message->Captured,
                                              message->CapturedCount,
                                              handles);
    }
    if (KSUCCESS(status) && message->Info.PayloadLength)
        status = CopyToUser(payload, message->Payload, message->Info.PayloadLength);

    if (!KSUCCESS(status)) {
        KiPortCloseHandles(receiverProcess, handles, message->CapturedCount);
        if (message->Call) {
            PORT_CALL* call = message->Call;
            PORT_CONNECTION* connection = endpoint->u.Endpoint.Connection;
            bool completeCall = false;
            uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
            long callState = _InterlockedCompareExchange(&call->State, 0, 0);
            if (callState != PORT_CALL_CANCELLED && callState != PORT_CALL_COMPLETED) {
                if (call->Link.Fwd != &call->Link && call->Link.Bwd != &call->Link)
                    KeListRemoveChecked(&call->Link);
                completeCall = true;
            }
            KeReleaseSpinlock(&connection->Lock, flags);
            if (completeCall)
                KiPortCompleteCallError(call, status);
        }
        return status;
    }

    if (message->CapturedCount)
        memcpy(handlesOut, handles, sizeof(SYS_HANDLE) * message->CapturedCount);
    *infoOut = message->Info;
    _InterlockedIncrement64(&gPortCounters.QueuedDeliveries);
    return NTLSTATUS_SUCCESS;
}

// Receives one copied message or registers one direct-delivery rendezvous while blocked.
KSTATUS KePortReceive(PORT* endpoint,
                      PROCESS* receiverProcess,
                      void* payload,
                      uint32_t payloadCapacity,
                      SYS_HANDLE* handlesOut,
                      uint32_t handleCapacity,
                      SYS_PORT_MESSAGE_INFO* infoOut,
                      uint64_t timeoutMilliseconds) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !receiverProcess || !infoOut)
        return NTLSTATUS_INVALID_PARAMETER;
    if (payloadCapacity > SYS_PORT_MAX_INLINE_BYTES ||
        handleCapacity > SYS_PORT_MAX_TRANSFER_HANDLES ||
        (payloadCapacity && !payload) || (handleCapacity && !handlesOut))
        return NTLSTATUS_INVALID_PARAMETER;

    PORT_RECEIVE_WAIT wait;
    memset(&wait, 0, sizeof(wait));
    LIST_INIT(&wait.Link);
    wait.Process = receiverProcess;
    wait.Payload = payload;
    wait.PayloadCapacity = payloadCapacity;
    wait.HandleCapacity = handleCapacity;
    wait.Status = NTLSTATUS_PENDING;

    if (payloadCapacity) {
        KSTATUS status = MmLockUserBuffer(receiverProcess,
                                          payload,
                                          payloadCapacity,
                                          1,
                                          &wait.LockedPayload);
        if (!KSUCCESS(status))
            return status;
    }
    if (timeoutMilliseconds != 0) {
        KSTATUS status = KeEventCreate(EventAutoReset, 0, &wait.CompletionEvent);
        if (!KSUCCESS(status)) {
            if (wait.LockedPayload)
                MmUnlockUserBuffer(wait.LockedPayload);
            return status;
        }
    }

    PORT_CONNECTION* connection = endpoint->u.Endpoint.Connection;
    uint64_t deadline = 0;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        deadline = KeMonotonicDeadlineAfterMilliseconds(timeoutMilliseconds);

    KSTATUS result = NTLSTATUS_FAILURE;
    while (true) {
        PORT_MESSAGE* message = (PORT_MESSAGE*)0;
        uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
        if (!LIST_IS_EMPTY(&endpoint->u.Endpoint.Incoming)) {
            message = CONTAINING_RECORD(endpoint->u.Endpoint.Incoming.Fwd,
                                        PORT_MESSAGE,
                                        Link);
            if (message->Info.PayloadLength > payloadCapacity ||
                message->CapturedCount > handleCapacity) {
                *infoOut = message->Info;
                KeReleaseSpinlock(&connection->Lock, flags);
                result = NTLSTATUS_BUFFER_TOO_SMALL;
                break;
            }
            KeListRemoveChecked(&message->Link);
            KiPortNoteDequeuedLocked(endpoint, message);
            if (message->Call)
                _InterlockedExchange(&message->Call->State, PORT_CALL_DELIVERED);
            KeReleaseSpinlock(&connection->Lock, flags);

            result = KiPortDeliverQueued(endpoint,
                                         receiverProcess,
                                         message,
                                         payload,
                                         handlesOut,
                                         infoOut);
            KiPortFreeMessage(message);
            break;
        }

        uint32_t peerSide = 1u - endpoint->u.Endpoint.Side;
        if (connection->Disconnected[endpoint->u.Endpoint.Side] ||
            connection->Disconnected[peerSide]) {
            KeReleaseSpinlock(&connection->Lock, flags);
            result = NTLSTATUS_PORT_DISCONNECTED;
            break;
        }

        if (timeoutMilliseconds == 0) {
            KeReleaseSpinlock(&connection->Lock, flags);
            result = NTLSTATUS_WOULD_BLOCK;
            break;
        }

        if (wait.Link.Fwd == &wait.Link && wait.Link.Bwd == &wait.Link)
            KeListInsertTailChecked(&endpoint->u.Endpoint.WaitingReceivers, &wait.Link);
        KeClearEvent(endpoint->u.Endpoint.WaitEvent);
        KeReleaseSpinlock(&connection->Lock, flags);

        uint64_t waitMilliseconds = timeoutMilliseconds;
        if (timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE) {
            waitMilliseconds = KeMonotonicMillisecondsUntil(deadline);
            if (waitMilliseconds == 0)
                waitMilliseconds = 1;
        }
        EVENT* events[2] = { wait.CompletionEvent, endpoint->u.Endpoint.WaitEvent };
        KSTATUS waitStatus = KeWaitForMultipleEvents(events,
                                                     2,
                                                     0,
                                                     waitMilliseconds,
                                                     (uint32_t*)0);

        flags = KeAcquireSpinlock(&connection->Lock);
        if (_InterlockedCompareExchange(&wait.Completed, 0, 0) != 0) {
            if (wait.Link.Fwd != &wait.Link || wait.Link.Bwd != &wait.Link)
                KeListRemoveChecked(&wait.Link);
            if (endpoint->u.Endpoint.IncomingCount == 0 &&
                !connection->Disconnected[1u - endpoint->u.Endpoint.Side])
                KeClearEvent(endpoint->u.Endpoint.WaitEvent);
            KeReleaseSpinlock(&connection->Lock, flags);
            result = wait.Status;
            if (KSUCCESS(result)) {
                *infoOut = wait.Info;
                if (wait.HandleCount)
                    memcpy(handlesOut, wait.Handles, sizeof(SYS_HANDLE) * wait.HandleCount);
            }
            break;
        }

        if (_InterlockedCompareExchange(&wait.Claimed, 0, 0) != 0) {
            KeReleaseSpinlock(&connection->Lock, flags);
            EVENT* claimedEvents[1] = { wait.CompletionEvent };
            (void)KeWaitForMultipleEvents(claimedEvents,
                                          1,
                                          0,
                                          SYS_PORT_TIMEOUT_INFINITE,
                                          (uint32_t*)0);
            continue;
        }

        if (wait.Link.Fwd != &wait.Link || wait.Link.Bwd != &wait.Link)
            KeListRemoveChecked(&wait.Link);
        KeReleaseSpinlock(&connection->Lock, flags);

        if (!KSUCCESS(waitStatus)) {
            result = waitStatus;
            break;
        }
        if (timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE &&
            KeMonotonicMillisecondsUntil(deadline) == 0) {
            result = NTLSTATUS_TIMEOUT;
            break;
        }
    }

    if (wait.CompletionEvent)
        ObDereferenceObject(&wait.CompletionEvent->Header);
    if (wait.LockedPayload)
        MmUnlockUserBuffer(wait.LockedPayload);
    return result;
}

// Creates one synchronous call context and holds caller resources until reply, timeout, or disconnect.
static KSTATUS KiPortCreateCall(PORT* endpoint,
                                PROCESS* callerProcess,
                                void* replyPayload,
                                uint32_t replyCapacity,
                                uint32_t replyHandleCapacity,
                                PORT_CALL** callOut) {
    if (!endpoint || !callerProcess || !callOut)
        return NTLSTATUS_INVALID_PARAMETER;
    *callOut = (PORT_CALL*)0;

    KSTATUS quotaStatus = KiPortChargeCall(callerProcess);
    if (!KSUCCESS(quotaStatus))
        return quotaStatus;

    PORT_CALL* call = (PORT_CALL*)MmAllocatePoolTag(sizeof(PORT_CALL), MM_POOL_TAG('p', 'c', 'a', 'l'));
    if (!call) {
        KiPortReleaseCallCharge(callerProcess);
        return NTLSTATUS_NO_MEMORY;
    }
    memset(call, 0, sizeof(*call));
    LIST_INIT(&call->Link);
    call->RefCount = 1;
    call->Connection = endpoint->u.Endpoint.Connection;
    call->CallerSide = endpoint->u.Endpoint.Side;
    call->State = PORT_CALL_PENDING;
    call->CallerProcess = callerProcess;
    call->ReplyPayload = replyPayload;
    call->ReplyCapacity = replyCapacity;
    call->ReplyHandleCapacity = replyHandleCapacity;
    call->Status = NTLSTATUS_PENDING;

    KSTATUS status = KeAllocateId32(&gPortMessageIds, &call->MessageId);
    if (!KSUCCESS(status)) {
        KiPortReleaseCallCharge(callerProcess);
        MmFreePool(call);
        return status;
    }

    status = KeEventCreate(EventManualReset, 0, &call->CompletionEvent);
    if (!KSUCCESS(status)) {
        KiPortReleaseCallCharge(callerProcess);
        MmFreePool(call);
        return status;
    }

    if (replyCapacity) {
        status = MmLockUserBuffer(callerProcess,
                                  replyPayload,
                                  replyCapacity,
                                  1,
                                  &call->LockedReplyPayload);
        if (!KSUCCESS(status)) {
            ObDereferenceObject(&call->CompletionEvent->Header);
            KiPortReleaseCallCharge(callerProcess);
            MmFreePool(call);
            return status;
        }
    }

    uintptr_t flags = KeAcquireSpinlock(&call->Connection->Lock);
    if (call->Connection->Disconnected[call->CallerSide] ||
        call->Connection->Disconnected[1u - call->CallerSide]) {
        KeReleaseSpinlock(&call->Connection->Lock, flags);
        if (call->LockedReplyPayload) MmUnlockUserBuffer(call->LockedReplyPayload);
        ObDereferenceObject(&call->CompletionEvent->Header);
        KiPortReleaseCallCharge(callerProcess);
        MmFreePool(call);
        return NTLSTATUS_PORT_DISCONNECTED;
    }
    KeListInsertTailChecked(&call->Connection->Calls, &call->Link);
    KeReleaseSpinlock(&call->Connection->Lock, flags);

    *callOut = call;
    return NTLSTATUS_SUCCESS;
}

// Releases one synchronous-call reference and frees its retained resources after the final owner leaves.
static void KiPortReleaseCall(PORT_CALL* call) {
    if (!call)
        return;
    long refs = _InterlockedDecrement(&call->RefCount);
    if (refs > 0)
        return;
    if (refs < 0) {
        KePanic(KE_PANIC_CODE_GENERIC,
                "port: call ref underflow message=%u refs=%ld\n",
                call->MessageId,
                refs);
    }
    if (call->LockedReplyPayload)
        MmUnlockUserBuffer(call->LockedReplyPayload);
    if (call->CompletionEvent)
        ObDereferenceObject(&call->CompletionEvent->Header);
    if (call->CallerProcess)
        KiPortReleaseCallCharge(call->CallerProcess);
    MmFreePool(call);
}

// Performs one synchronous Port request and waits for exactly one matching reply.
KSTATUS KePortCall(PORT* endpoint,
                   PROCESS* callerProcess,
                   THREAD* callerThread,
                   const void* requestPayload,
                   uint32_t requestLength,
                   const SYS_PORT_HANDLE_TRANSFER* requestTransfers,
                   uint32_t requestTransferCount,
                   void* replyPayload,
                   uint32_t replyCapacity,
                   SYS_HANDLE* replyHandlesOut,
                   uint32_t replyHandleCapacity,
                   SYS_PORT_MESSAGE_INFO* replyInfoOut,
                   uint64_t timeoutMilliseconds) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !callerProcess || !callerThread)
        return NTLSTATUS_INVALID_PARAMETER;
    if (timeoutMilliseconds == 0)
        return NTLSTATUS_WOULD_BLOCK;
    if (requestLength > SYS_PORT_MAX_INLINE_BYTES || replyCapacity > SYS_PORT_MAX_INLINE_BYTES ||
        requestTransferCount > SYS_PORT_MAX_TRANSFER_HANDLES ||
        replyHandleCapacity > SYS_PORT_MAX_TRANSFER_HANDLES ||
        (requestLength && !requestPayload) || (replyCapacity && !replyPayload) ||
        (replyHandleCapacity && !replyHandlesOut))
        return NTLSTATUS_INVALID_PARAMETER;

    uint64_t deadline = 0;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        deadline = KeMonotonicDeadlineAfterMilliseconds(timeoutMilliseconds);

    PORT_CALL* call = (PORT_CALL*)0;
    KSTATUS status = KiPortCreateCall(endpoint,
                                      callerProcess,
                                      replyPayload,
                                      replyCapacity,
                                      replyHandleCapacity,
                                      &call);
    if (!KSUCCESS(status))
        return status;

    PORT_MESSAGE* message = (PORT_MESSAGE*)0;
    status = KiPortCreateMessage(callerProcess,
                                 callerThread,
                                 SYS_PORT_MESSAGE_REQUEST,
                                 call->MessageId,
                                 requestLength,
                                 requestTransfers,
                                 requestTransferCount,
                                 call,
                                 &message);
    if (!KSUCCESS(status)) {
        uintptr_t flags = KeAcquireSpinlock(&call->Connection->Lock);
        if (call->Link.Fwd != &call->Link && call->Link.Bwd != &call->Link)
            KeListRemoveChecked(&call->Link);
        KeReleaseSpinlock(&call->Connection->Lock, flags);
        KiPortReleaseCall(call);
        return status;
    }

    uint64_t submitTimeout = timeoutMilliseconds;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        submitTimeout = KeMonotonicMillisecondsUntil(deadline);

    status = KiPortSubmitMessage(endpoint,
                                 callerProcess,
                                 requestPayload,
                                 requestLength,
                                 message,
                                 submitTimeout);
    if (!KSUCCESS(status)) {
        uintptr_t flags = KeAcquireSpinlock(&call->Connection->Lock);
        if (call->Link.Fwd != &call->Link && call->Link.Bwd != &call->Link)
            KeListRemoveChecked(&call->Link);
        KeReleaseSpinlock(&call->Connection->Lock, flags);
        KiPortReleaseCall(call);
        return status;
    }

    uint64_t replyWait = timeoutMilliseconds;
    if (timeoutMilliseconds != 0 && timeoutMilliseconds != SYS_PORT_TIMEOUT_INFINITE)
        replyWait = KeMonotonicMillisecondsUntil(deadline);

    EVENT* events[1] = { call->CompletionEvent };
    KSTATUS waitStatus = KeWaitForMultipleEvents(events,
                                                 1,
                                                 0,
                                                 replyWait,
                                                 (uint32_t*)0);
    if (!KSUCCESS(waitStatus)) {
        PORT_MESSAGE* queuedMessage = (PORT_MESSAGE*)0;
        uintptr_t flags = KeAcquireSpinlock(&call->Connection->Lock);
        long state = _InterlockedCompareExchange(&call->State, 0, 0);
        if (state == PORT_CALL_REPLYING) {
            KeReleaseSpinlock(&call->Connection->Lock, flags);
            (void)KeWaitForMultipleEvents(events, 1, 0, SYS_PORT_TIMEOUT_INFINITE, (uint32_t*)0);
        } else if (state == PORT_CALL_COMPLETED) {
            KeReleaseSpinlock(&call->Connection->Lock, flags);
        } else {
            if (call->Link.Fwd != &call->Link && call->Link.Bwd != &call->Link)
                KeListRemoveChecked(&call->Link);
            _InterlockedExchange(&call->State, PORT_CALL_CANCELLED);
            queuedMessage = KiPortRemoveQueuedCallLocked(call->Connection, call);
            KeReleaseSpinlock(&call->Connection->Lock, flags);
            if (queuedMessage)
                KiPortFreeMessage(queuedMessage);
            KiPortReleaseCall(call);
            return waitStatus;
        }
    }

    status = call->Status;
    if (KSUCCESS(status)) {
        if (replyInfoOut)
            *replyInfoOut = call->ReplyInfo;
        if (call->ReplyHandleCount && replyHandlesOut)
            memcpy(replyHandlesOut,
                   call->ReplyHandles,
                   sizeof(SYS_HANDLE) * call->ReplyHandleCount);
    } else if (call->ReplyHandleCount) {
        KiPortCloseHandles(callerProcess, call->ReplyHandles, call->ReplyHandleCount);
        call->ReplyHandleCount = 0;
    }

    KiPortReleaseCall(call);
    return status;
}

// Replies to one delivered synchronous request and directly copies the reply into the blocked caller buffer.
KSTATUS KePortReply(PORT* endpoint,
                    PROCESS* senderProcess,
                    THREAD* senderThread,
                    uint32_t messageId,
                    const void* payload,
                    uint32_t payloadLength,
                    const SYS_PORT_HANDLE_TRANSFER* transfers,
                    uint32_t transferCount) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT || !senderProcess || !senderThread || messageId == 0)
        return NTLSTATUS_INVALID_PARAMETER;
    if (payloadLength > SYS_PORT_MAX_INLINE_BYTES || transferCount > SYS_PORT_MAX_TRANSFER_HANDLES ||
        (payloadLength && !payload))
        return NTLSTATUS_INVALID_PARAMETER;

    struct KE_CAPTURED_HANDLE* captured[SYS_PORT_MAX_TRANSFER_HANDLES] = {0};
    KSTATUS status = KiPortCaptureHandles(senderProcess,
                                          transfers,
                                          transferCount,
                                          captured);
    if (!KSUCCESS(status))
        return status;

    PORT_CONNECTION* connection = endpoint->u.Endpoint.Connection;
    PORT_CALL* call = (PORT_CALL*)0;
    uintptr_t flags = KeAcquireSpinlock(&connection->Lock);
    LIST_ENTRY* head = &connection->Calls;
    for (LIST_ENTRY* link = head->Fwd; link != head; link = link->Fwd) {
        PORT_CALL* candidate = CONTAINING_RECORD(link, PORT_CALL, Link);
        if (candidate->MessageId == messageId &&
            candidate->CallerSide != endpoint->u.Endpoint.Side) {
            call = candidate;
            break;
        }
    }

    if (!call || _InterlockedCompareExchange(&call->State, 0, 0) != PORT_CALL_DELIVERED) {
        KeReleaseSpinlock(&connection->Lock, flags);
        for (uint32_t i = 0; i < transferCount; ++i)
            KeReleaseCapturedHandle(captured[i]);
        return NTLSTATUS_NOT_FOUND;
    }
    if (payloadLength > call->ReplyCapacity || transferCount > call->ReplyHandleCapacity) {
        KeReleaseSpinlock(&connection->Lock, flags);
        for (uint32_t i = 0; i < transferCount; ++i)
            KeReleaseCapturedHandle(captured[i]);
        return NTLSTATUS_BUFFER_TOO_SMALL;
    }

    _InterlockedExchange(&call->State, PORT_CALL_REPLYING);
    KeListRemoveChecked(&call->Link);
    KeReleaseSpinlock(&connection->Lock, flags);

    SYS_HANDLE handles[SYS_PORT_MAX_TRANSFER_HANDLES] = {0};
    if (transferCount)
        status = KeMaterializeCapturedHandles(call->CallerProcess, captured, transferCount, handles);
    if (KSUCCESS(status) && payloadLength) {
        status = MmCopyUserToUser(senderProcess,
                                  payload,
                                  call->CallerProcess,
                                  call->ReplyPayload,
                                  payloadLength);
    }

    if (!KSUCCESS(status)) {
        KiPortCloseHandles(call->CallerProcess, handles, transferCount);
        for (uint32_t i = 0; i < transferCount; ++i)
            if (captured[i]) KeReleaseCapturedHandle(captured[i]);
        call->ReplyHandleCount = 0;
    } else {
        memcpy(call->ReplyHandles, handles, sizeof(SYS_HANDLE) * transferCount);
        call->ReplyHandleCount = transferCount;
        call->ReplyInfo.Size = sizeof(call->ReplyInfo);
        call->ReplyInfo.MessageId = messageId;
        call->ReplyInfo.Type = SYS_PORT_MESSAGE_REPLY;
        call->ReplyInfo.PayloadLength = payloadLength;
        call->ReplyInfo.HandleCount = transferCount;
        call->ReplyInfo.SenderProcessId = senderProcess->ProcessId;
        call->ReplyInfo.SenderThreadId = senderThread->ThreadId;
        call->ReplyInfo.SenderSecurityId = KiPortSecurityId(senderProcess);
    }

    if (KSUCCESS(status))
        _InterlockedIncrement64(&gPortCounters.RepliesCompleted);
    call->Status = status;
    _InterlockedExchange(&call->State, PORT_CALL_COMPLETED);
    KeSignalEvent(call->CompletionEvent);
    return status;
}

// Explicitly disconnects one endpoint capability without requiring its last handle to close.
KSTATUS KePortDisconnect(PORT* endpoint) {
    if (!endpoint || endpoint->Kind != PORT_KIND_ENDPOINT)
        return NTLSTATUS_INVALID_PARAMETER;
    KiPortMarkDisconnected(endpoint, false);
    return NTLSTATUS_SUCCESS;
}