EventQueueLifetimeTests.cpp

#include "kernel/eventQueue.h"
#include "libs/errno.h"

#include <algorithm>
#include <atomic>
#include <cstdio>
#include <cstdlib>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>

namespace {

namespace EventQueue = Libs::LibKernel::EventQueue;
using Libs::LibKernel::KERNEL_ERROR_EBADF;
using Libs::LibKernel::KERNEL_ERROR_ENOENT;

void Check(bool value, const char* text) {
	if (!value) {
		std::fprintf(stderr, "EventQueueLifetimeTests: failed: %s\n", text);
		std::abort();
	}
}

void CheckConcurrentResult(int result, const char* text) {
	Check(result == OK || result == KERNEL_ERROR_EBADF || result == KERNEL_ERROR_ENOENT, text);
}

void CountDeletedEvent(EventQueue::KernelEqueue, EventQueue::KernelEqueueEvent* event) {
	auto* count = static_cast<std::atomic_uint32_t*>(event->filter.data);
	count->fetch_add(1, std::memory_order_relaxed);
}

struct DuplicateEventOwner {
	std::atomic_uint32_t delete_count {0};
};

void QueueDuplicateEvent(EventQueue::KernelEqueueEvent* event, void* trigger_data) {
	auto next = event->event;
	next.data = reinterpret_cast<intptr_t>(trigger_data);
	if (event->triggered) {
		event->pending_events.push_back(next);
	} else {
		event->event     = next;
		event->triggered = true;
	}
}

void ResetDuplicateEvent(EventQueue::KernelEqueueEvent* event) {
	event->triggered  = false;
	event->event.data = 0;
}

void DeleteDuplicateEvent(EventQueue::KernelEqueue, EventQueue::KernelEqueueEvent* event) {
	auto* owner = static_cast<DuplicateEventOwner*>(event->filter.data);
	owner->delete_count.fetch_add(1, std::memory_order_relaxed);
}

void PoisonDuplicateEvent(EventQueue::KernelEqueueEvent*, void*) {
	Check(false, "duplicate add replaced trigger callback");
}

void TestDuplicateAddPreservesEventState() {
	EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&queue, "duplicate-add") == OK,
	      "create duplicate add queue");

	auto                               original_owner = std::make_shared<DuplicateEventOwner>();
	std::weak_ptr<DuplicateEventOwner> weak_original  = original_owner;
	EventQueue::KernelEqueueEvent      original {};
	original.event.ident              = 17;
	original.event.filter             = EventQueue::KERNEL_EVFILT_VIDEO_OUT;
	original.event.udata              = reinterpret_cast<void*>(0x1111);
	original.filter.data              = original_owner.get();
	original.filter.owner             = original_owner;
	original.filter.trigger_func      = QueueDuplicateEvent;
	original.filter.reset_func        = ResetDuplicateEvent;
	original.filter.delete_event_func = DeleteDuplicateEvent;
	Check(EventQueue::KernelAddEvent(queue, original) == OK, "add original duplicate event");
	Check(EventQueue::KernelTriggerEvent(queue, 17, EventQueue::KERNEL_EVFILT_VIDEO_OUT,
	                                     reinterpret_cast<void*>(0x1234)) == OK,
	      "queue first trigger");
	Check(EventQueue::KernelTriggerEvent(queue, 17, EventQueue::KERNEL_EVFILT_VIDEO_OUT,
	                                     reinterpret_cast<void*>(0x5678)) == OK,
	      "queue pending trigger");

	auto                               replacement_owner = std::make_shared<DuplicateEventOwner>();
	std::weak_ptr<DuplicateEventOwner> weak_replacement  = replacement_owner;
	EventQueue::KernelEqueueEvent      duplicate {};
	duplicate.triggered           = false;
	duplicate.deadline_ns         = 1;
	duplicate.event.ident         = 17;
	duplicate.event.filter        = EventQueue::KERNEL_EVFILT_VIDEO_OUT;
	duplicate.event.data          = 0x7fffffff;
	duplicate.event.udata         = reinterpret_cast<void*>(0x2222);
	duplicate.filter.data         = replacement_owner.get();
	duplicate.filter.owner        = replacement_owner;
	duplicate.filter.trigger_func = PoisonDuplicateEvent;
	Check(EventQueue::KernelAddEvent(queue, duplicate) == OK, "update duplicate event");

	duplicate.filter.owner.reset();
	replacement_owner.reset();
	Check(weak_replacement.expired(), "duplicate owner is not retained");
	original.filter.owner.reset();
	original_owner.reset();
	Check(!weak_original.expired(), "original event owner remains retained");

	EventQueue::KernelEvent         events[2] {};
	int                             out     = 0;
	Libs::LibKernel::KernelUseconds timeout = 0;
	Check(EventQueue::KernelWaitEqueue(queue, events, 2, &out, &timeout) == OK,
	      "read queued duplicate triggers");
	Check(out == 2, "duplicate add preserves pending event count");
	Check(events[0].data == 0x1234 && events[1].data == 0x5678,
	      "duplicate add preserves current and pending event data");
	Check(events[0].udata == reinterpret_cast<void*>(0x2222),
	      "duplicate add updates current user data");
	Check(events[1].udata == reinterpret_cast<void*>(0x2222),
	      "duplicate add updates pending user data");

	EventQueue::KernelEvent timer_event {};
	Check(EventQueue::KernelWaitEqueue(queue, &timer_event, 1, &out, &timeout) == OK && out == 1,
	      "duplicate add updates deadline metadata");
	Check(timer_event.data == 0 && timer_event.udata == reinterpret_cast<void*>(0x2222),
	      "deadline trigger retains updated duplicate metadata");

	auto retained_owner = weak_original.lock();
	Check(retained_owner != nullptr, "original owner alive before delete");
	Check(EventQueue::KernelDeleteEvent(queue, 17, EventQueue::KERNEL_EVFILT_VIDEO_OUT) == OK,
	      "delete duplicate event");
	Check(retained_owner->delete_count.load(std::memory_order_relaxed) == 1,
	      "duplicate add preserves delete callback");
	retained_owner.reset();
	Check(weak_original.expired(), "original owner released on delete");
	Check(EventQueue::KernelDeleteEqueue(queue) == OK, "delete duplicate add queue");
}

struct SimulatedVideoOutEventState;

struct SimulatedVideoOutRegistration {
	EventQueue::KernelEqueue                     handle = EventQueue::KERNEL_EQUEUE_INVALID;
	std::shared_ptr<SimulatedVideoOutEventState> state;
	uint64_t                                     marker = 0x123456789abcdef0ull;
};

struct SimulatedVideoOutEventState {
	SimulatedVideoOutEventState(std::atomic_uint32_t& stage, std::atomic_uint32_t& destroy_count)
	    : stage(stage), destroy_count(destroy_count) {}

	~SimulatedVideoOutEventState() { destroy_count.fetch_add(1, std::memory_order_relaxed); }

	std::mutex                                                  mutex;
	std::vector<std::shared_ptr<SimulatedVideoOutRegistration>> queues;
	std::atomic_uint32_t&                                       stage;
	std::atomic_uint32_t&                                       destroy_count;
	uint64_t                                                    marker = 0xfedcba9876543210ull;
};

void DetachSimulatedVideoOutEvent(EventQueue::KernelEqueue       queue,
                                  EventQueue::KernelEqueueEvent* event) {
	auto* registration = static_cast<SimulatedVideoOutRegistration*>(event->filter.data);
	Check(registration != nullptr && registration->handle == queue,
	      "simulated registration identity");
	auto state = registration->state;
	Check(state != nullptr, "simulated event owns shared state");
	state->stage.store(1, std::memory_order_release);
	while (state->stage.load(std::memory_order_acquire) != 2) {
		std::this_thread::yield();
	}

	{
		std::lock_guard lock(state->mutex);
		const auto      entry = std::find_if(
		    state->queues.begin(), state->queues.end(),
		    [registration](const auto& candidate) { return candidate.get() == registration; });
		if (entry != state->queues.end()) {
			state->queues.erase(entry);
		}
	}
	event->filter.owner.reset();
	Check(state->marker == 0xfedcba9876543210ull && registration->marker == 0x123456789abcdef0ull,
	      "callback state survives simulated port destruction");
}

void TestCallbackStateOutlivesPort() {
	EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&queue, "shared-port-state") == OK,
	      "create shared port state queue");

	std::atomic_uint32_t stage {0};
	std::atomic_uint32_t destroy_count {0};
	auto port_state = std::make_shared<SimulatedVideoOutEventState>(stage, destroy_count);
	std::weak_ptr<SimulatedVideoOutEventState> weak_state = port_state;
	auto registration = std::make_shared<SimulatedVideoOutRegistration>();
	std::weak_ptr<SimulatedVideoOutRegistration> weak_registration = registration;
	registration->handle                                           = queue;
	registration->state                                            = port_state;
	port_state->queues.push_back(registration);

	EventQueue::KernelEqueueEvent event {};
	event.event.ident              = 8;
	event.event.filter             = EventQueue::KERNEL_EVFILT_VIDEO_OUT;
	event.filter.data              = registration.get();
	event.filter.owner             = registration;
	event.filter.delete_event_func = DetachSimulatedVideoOutEvent;
	Check(EventQueue::KernelAddEvent(queue, event) == OK, "add shared port state event");
	event.filter.owner.reset();

	std::jthread close([&] {
		Check(EventQueue::KernelDeleteEqueue(queue) == OK, "delete shared port state queue");
	});
	while (stage.load(std::memory_order_acquire) != 1) {
		std::this_thread::yield();
	}

	std::vector<std::shared_ptr<SimulatedVideoOutRegistration>> detached;
	{
		std::lock_guard lock(port_state->mutex);
		detached = std::move(port_state->queues);
	}
	registration.reset();
	detached.clear();
	port_state.reset();
	Check(!weak_state.expired(), "callback state outlives simulated port object");
	Check(!weak_registration.expired(), "registration outlives simulated port object");

	stage.store(2, std::memory_order_release);
	close.join();
	Check(weak_registration.expired(), "detached registration is released");
	Check(weak_state.expired(), "detached event state is released");
	Check(destroy_count.load(std::memory_order_relaxed) == 1,
	      "shared event state is destroyed exactly once");
}

struct OwnedCallbackPayload {
	OwnedCallbackPayload(std::atomic_uint32_t& stage, std::atomic_uint32_t& delete_count,
	                     std::atomic_uint32_t& destroy_count)
	    : stage(stage), delete_count(delete_count), destroy_count(destroy_count) {}

	std::atomic_uint32_t& stage;
	std::atomic_uint32_t& delete_count;
	std::atomic_uint32_t& destroy_count;
	uint64_t              marker = 0xc0dec0dec0dec0deull;

	~OwnedCallbackPayload() { destroy_count.fetch_add(1, std::memory_order_relaxed); }
};

void DeleteOwnedEvent(EventQueue::KernelEqueue queue, EventQueue::KernelEqueueEvent* event) {
	auto* payload = static_cast<OwnedCallbackPayload*>(event->filter.data);
	Check(payload != nullptr, "owned callback payload");
	Check(!EventQueue::KernelPinEqueue(queue), "owned callback runs after registry removal");
	payload->delete_count.fetch_add(1, std::memory_order_relaxed);
	event->filter.owner.reset();
	payload->stage.store(1, std::memory_order_release);
	while (payload->stage.load(std::memory_order_acquire) != 2) {
		std::this_thread::yield();
	}
	Check(payload->marker == 0xc0dec0dec0dec0deull, "owned callback payload remains valid");
}

void TestCallbackOwnsPayload() {
	EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&queue, "owned-callback") == OK,
	      "create owned callback queue");

	std::atomic_uint32_t stage {0};
	std::atomic_uint32_t delete_count {0};
	std::atomic_uint32_t destroy_count {0};
	auto registration = std::make_shared<OwnedCallbackPayload>(stage, delete_count, destroy_count);
	std::weak_ptr<OwnedCallbackPayload>                weak_registration = registration;
	std::vector<std::shared_ptr<OwnedCallbackPayload>> port_registrations {registration};
	{
		EventQueue::KernelEqueueEvent event {};
		event.event.ident              = 2;
		event.event.filter             = EventQueue::KERNEL_EVFILT_VIDEO_OUT;
		event.filter.data              = registration.get();
		event.filter.owner             = registration;
		event.filter.delete_event_func = DeleteOwnedEvent;
		Check(EventQueue::KernelAddEvent(queue, event) == OK, "add owned callback event");
	}

	std::jthread close(
	    [&] { Check(EventQueue::KernelDeleteEqueue(queue) == OK, "delete owned callback queue"); });
	while (stage.load(std::memory_order_acquire) != 1) {
		std::this_thread::yield();
	}

	Check(!EventQueue::KernelPinEqueue(queue),
	      "owned callback queue removed while callback blocked");
	port_registrations.clear();
	registration.reset();
	Check(!weak_registration.expired(), "delete callback retains detached payload");

	stage.store(2, std::memory_order_release);
	close.join();
	Check(delete_count.load(std::memory_order_relaxed) == 1, "owned callback runs exactly once");
	Check(weak_registration.expired(), "owned callback payload released with event");
	Check(destroy_count.load(std::memory_order_relaxed) == 1,
	      "owned callback payload destroyed exactly once");
}

void TestPinnedClose() {
	EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&queue, "pinned-close") == OK, "create pinned queue");

	std::atomic_uint32_t          delete_count {0};
	EventQueue::KernelEqueueEvent event {};
	event.event.ident              = 1;
	event.event.filter             = EventQueue::KERNEL_EVFILT_VIDEO_OUT;
	event.filter.data              = &delete_count;
	event.filter.delete_event_func = CountDeletedEvent;
	Check(EventQueue::KernelAddEvent(queue, event) == OK, "add callback event");

	auto owner = EventQueue::KernelPinEqueue(queue);
	Check(owner != nullptr, "pin live queue");
	Check(EventQueue::KernelDeleteEqueue(queue) == OK, "delete pinned queue");
	Check(delete_count.load(std::memory_order_relaxed) == 1, "close invokes callback once");
	Check(!EventQueue::KernelPinEqueue(queue), "deleted queue leaves registry");
	Check(EventQueue::KernelTriggerEvent(queue, 1, EventQueue::KERNEL_EVFILT_VIDEO_OUT, nullptr) ==
	          KERNEL_ERROR_EBADF,
	      "stale trigger rejected");
	Check(EventQueue::KernelDeleteEqueue(queue) == KERNEL_ERROR_EBADF,
	      "second queue delete rejected");

	owner.reset();
	Check(delete_count.load(std::memory_order_relaxed) == 1,
	      "deferred destruction does not repeat callback");
}

void TestStaleHandleNeverAliasesNewQueue() {
	EventQueue::KernelEqueue stale = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&stale, "stale-handle") == OK, "create stale queue");
	Check(EventQueue::KernelDeleteEqueue(stale) == OK, "delete stale queue");

	EventQueue::KernelEqueue replacement = EventQueue::KERNEL_EQUEUE_INVALID;
	Check(EventQueue::KernelCreateEqueue(&replacement, "replacement") == OK,
	      "create replacement queue");
	Check(stale != replacement, "queue handles are never recycled");
	Check(!EventQueue::KernelPinEqueue(stale), "stale handle does not pin");
	Check(EventQueue::KernelAddUserEvent(stale, 11) == KERNEL_ERROR_EBADF,
	      "stale handle cannot mutate replacement");
	Check(EventQueue::KernelAddUserEvent(replacement, 11) == OK,
	      "replacement handle remains valid");
	Check(EventQueue::KernelTriggerUserEvent(stale, 11, nullptr) == KERNEL_ERROR_EBADF,
	      "stale handle cannot trigger replacement");
	Check(EventQueue::KernelTriggerUserEvent(replacement, 11, nullptr) == OK,
	      "replacement event triggers");
	Check(EventQueue::KernelDeleteEqueue(replacement) == OK, "delete replacement queue");
}

void TestConcurrentCloseCallback() {
	for (uint32_t iteration = 0; iteration < 64; iteration++) {
		EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
		Check(EventQueue::KernelCreateEqueue(&queue, "callback-race") == OK,
		      "create callback race queue");

		std::atomic_uint32_t          delete_count {0};
		EventQueue::KernelEqueueEvent callback_event {};
		callback_event.event.ident              = 9;
		callback_event.event.filter             = EventQueue::KERNEL_EVFILT_GRAPHICS;
		callback_event.filter.data              = &delete_count;
		callback_event.filter.delete_event_func = CountDeletedEvent;
		Check(EventQueue::KernelAddEvent(queue, callback_event) == OK, "add callback race event");

		std::atomic_bool start {false};
		std::jthread     trigger([&] {
			while (!start.load(std::memory_order_acquire)) {
				std::this_thread::yield();
			}
			for (uint32_t i = 0; i < 256; i++) {
				CheckConcurrentResult(EventQueue::KernelTriggerEvent(
				                          queue, 9, EventQueue::KERNEL_EVFILT_GRAPHICS, nullptr),
				                      "callback race trigger result");
			}
		});

		start.store(true, std::memory_order_release);
		Check(EventQueue::KernelDeleteEqueue(queue) == OK, "callback race queue delete");
		trigger.join();
		Check(delete_count.load(std::memory_order_relaxed) == 1,
		      "concurrent close invokes callback exactly once");
	}
}

void TestConcurrentDelete() {
	for (uint32_t iteration = 0; iteration < 64; iteration++) {
		EventQueue::KernelEqueue queue = EventQueue::KERNEL_EQUEUE_INVALID;
		Check(EventQueue::KernelCreateEqueue(&queue, "concurrent-delete") == OK,
		      "create concurrent queue");
		EventQueue::KernelEqueueEvent event {};
		event.event.ident  = 7;
		event.event.filter = EventQueue::KERNEL_EVFILT_USER;
		Check(EventQueue::KernelAddEvent(queue, event) == OK, "add concurrent event");

		std::atomic_bool start {false};
		std::jthread     mutate([&] {
			while (!start.load(std::memory_order_acquire)) {
				std::this_thread::yield();
			}
			for (uint32_t i = 0; i < 64; i++) {
				CheckConcurrentResult(EventQueue::KernelAddEvent(queue, event),
				                      "concurrent add result");
				CheckConcurrentResult(EventQueue::KernelTriggerEvent(
				                          queue, 7, EventQueue::KERNEL_EVFILT_USER, nullptr),
				                      "concurrent trigger result");
				CheckConcurrentResult(
				    EventQueue::KernelDeleteEvent(queue, 7, EventQueue::KERNEL_EVFILT_USER),
				    "concurrent event delete result");
			}
		});
		std::jthread     trigger([&] {
			while (!start.load(std::memory_order_acquire)) {
				std::this_thread::yield();
			}
			for (uint32_t i = 0; i < 128; i++) {
				CheckConcurrentResult(EventQueue::KernelTriggerEvent(
				                          queue, 7, EventQueue::KERNEL_EVFILT_USER, nullptr),
				                      "parallel trigger result");
			}
		});

		start.store(true, std::memory_order_release);
		Check(EventQueue::KernelDeleteEqueue(queue) == OK, "concurrent queue delete");
		mutate.join();
		trigger.join();
		Check(!EventQueue::KernelPinEqueue(queue), "concurrent queue removed from registry");
	}
}

} // namespace

int main() {
	TestDuplicateAddPreservesEventState();
	TestCallbackStateOutlivesPort();
	TestCallbackOwnsPayload();
	TestPinnedClose();
	TestStaleHandleNeverAliasesNewQueue();
	TestConcurrentCloseCallback();
	TestConcurrentDelete();
	std::printf("EventQueueLifetimeTests: all cases passed\n");
	return 0;
}