shifathosn
/
AnyPS5
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #613 from nicolasabosch/feat/ult-blocking-queues 

feat(ult): implement blocking queues and shared data pools
This commit is contained in:
Dean 2026-10-05 18:06:19 +13:00 committed by GitHub
parent f7ecd24ddb 043332ba32
commit 3e92cefb4c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 437 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
core/libs/prx/libSceUlt/CMakeLists.txt Normal file
View File

@ -0,0 +1,24 @@
add_library(libSceUlt SHARED EXCLUDE_FROM_ALL Export.cpp)
target_include_directories(libSceUlt PRIVATE ${LIBS_INCLUDE_DIR})
configure_windows_unwind(libSceUlt)
if(MINGW)
foreach(runtimeLinkFlag IN ITEMS -static-libgcc -static-libstdc++ -static)
string(REPLACE "${runtimeLinkFlag}" "" CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS}")
endforeach()
endif()
if(BUILD_TESTING)
add_test_executable(guest_ult_queue_tests ${LIBS_INCLUDE_DIR}/tests/GuestUltQueue.cpp)
target_include_directories(guest_ult_queue_tests PRIVATE ${LIBS_INCLUDE_DIR})
target_link_libraries(guest_ult_queue_tests PRIVATE libSceUlt libc)
configure_windows_unwind(guest_ult_queue_tests)
foreach(testCase IN ITEMS validation fifo blocking shared-pool finalize stress)
add_test(NAME guest_ult_queue_${testCase} COMMAND guest_ult_queue_tests ${testCase})
set_tests_properties(guest_ult_queue_${testCase} PROPERTIES TIMEOUT 20)
if(WIN32)
set_tests_properties(guest_ult_queue_${testCase} PROPERTIES ENVIRONMENT_MODIFICATION
"PATH=path_list_prepend:$<TARGET_FILE_DIR:libSceUlt>")
endif()
endforeach()
endif()

180
core/libs/prx/libSceUlt/Export.cpp
View File

@ -3,9 +3,10 @@
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <deque>
#include <limits>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <unordered_map>
#include <vector>
@ -23,7 +24,7 @@ std::mutex gMutex;
std::unordered_map<void*, std::shared_ptr<UltMutexState>> gMutexes;
std::unordered_map<void*, std::shared_ptr<UltSemaphoreState>> gSemaphores;
std::unordered_map<void*, UltResourcePoolState> gResourcePools;
std::unordered_map<void*, UltQueueDataPoolState> gQueueDataPools;
std::unordered_map<void*, std::shared_ptr<UltQueueDataPoolState>> gQueueDataPools;
std::unordered_map<void*, std::shared_ptr<UltQueueState>> gQueues;
std::unordered_map<void*, UltRuntimeState> gRuntimes;
std::unordered_map<void*, std::shared_ptr<UltUlthreadState>> gUlthreads;
@ -52,6 +53,59 @@ int queueGetState(void* queue, std::shared_ptr<UltQueueState>* out) {
return ULT_OK;
}
int QueueTransfer(void* queue, void* data, const void* source, bool push, bool blocking) {
std::shared_ptr<UltQueueState> state;
if (int result = queueGetState(queue, &state); result != ULT_OK) return result;
if ((push ? source : data) == nullptr) return ULT_ERROR_NULL;
auto& pool = *state->_pool;
std::unique_lock lock(pool._mutex);
if (!state->_alive) return ULT_ERROR_STATE;
const auto ready = [&] { return !state->_alive || (push ? pool._free : state->_head) != UINT32_MAX; };
if (!ready()) {
if (!blocking) return ULT_ERROR_AGAIN;
++state->_waiters;
try {
(push ? pool._spaceAvailable : state->_dataAvailable).wait(lock, ready);
} catch (...) {
--state->_waiters;
throw;
}
--state->_waiters;
if (!state->_alive) return ULT_ERROR_STATE;
}
if (push) {
const auto slot = pool._free;
std::memcpy(pool._data.data() + slot * pool._dataSize, source, state->_dataSize);
pool._free = pool._next[slot];
pool._next[slot] = UINT32_MAX;
if (state->_tail == UINT32_MAX) state->_head = slot;
else pool._next[state->_tail] = slot;
state->_tail = slot;
} else {
const auto slot = state->_head;
std::memcpy(data, pool._data.data() + slot * pool._dataSize, state->_dataSize);
state->_head = pool._next[slot];
if (state->_head == UINT32_MAX) state->_tail = UINT32_MAX;
pool._next[slot] = pool._free;
pool._free = slot;
}
lock.unlock();
if (push) state->_dataAvailable.notify_one();
else pool._spaceAvailable.notify_one();
return ULT_OK;
}
void ReleaseQueue(UltQueueState& state) {
auto& pool = *state._pool;
if (state._tail != UINT32_MAX) {
pool._next[state._tail] = pool._free;
pool._free = state._head;
}
state._head = state._tail = UINT32_MAX;
state._alive = false;
--pool._queues;
}
void* APS5_VABI ulthreadRunner(void* arg) {
auto* state = static_cast<UltUlthreadState*>(arg);
return reinterpret_cast<void*>(static_cast<std::intptr_t>(state->_entry(state->_arg)));
@ -79,8 +133,15 @@ int APS5_VABI sceUltFinalize() {
gSemaphores.clear();
gMutexes.clear();
gResourcePools.clear();
gQueueDataPools.clear();
for (auto& entry : gQueues) {
auto& state = *entry.second;
std::lock_guard stateLock(state._pool->_mutex);
ReleaseQueue(state);
state._dataAvailable.notify_all();
state._pool->_spaceAvailable.notify_all();
}
gQueues.clear();
gQueueDataPools.clear();
gRuntimes.clear();
gUlthreads.clear();
}
@ -162,38 +223,71 @@ int APS5_VABI sceUltWaitingQueueResourcePoolCreate(void* pool, const char* name,
}
std::uint64_t APS5_VABI sceUltQueueDataResourcePoolGetWorkAreaSize(std::uint32_t numData, std::uint64_t dataSize, std::uint32_t numQueueObject) {
const std::uint64_t dataArea = static_cast<std::uint64_t>(numData) * alignUp(dataSize, 8u);
const std::uint64_t queueArea = static_cast<std::uint64_t>(numQueueObject) * 512u;
if (dataSize > UINT64_MAX - 7u) throw std::out_of_range("ULT queue data size overflow");
const auto stride = alignUp(dataSize, 8u);
const auto queueArea = static_cast<std::uint64_t>(numQueueObject) * 512u;
if (stride != 0 && numData > (UINT64_MAX - queueArea) / stride) {
throw std::out_of_range("ULT queue work area size overflow");
}
const std::uint64_t dataArea = static_cast<std::uint64_t>(numData) * stride;
return alignUp(dataArea + queueArea, 8u);
}
int APS5_VABI sceUltQueueDataResourcePoolCreate(void* pool, const char* name, std::uint32_t numData, std::uint64_t dataSize, std::uint32_t numQueueObject, void* waitingQueueResourcePool, void* workArea, const void* optParam, std::uint32_t buildVersion) {
(void)name;
(void)optParam;
(void)buildVersion;
if (pool == nullptr) {
return ULT_ERROR_NULL;
}
if ((reinterpret_cast<std::uintptr_t>(pool) & 7u) != 0) return ULT_ERROR_ALIGNMENT;
if (numData == 0 || numQueueObject == 0 || dataSize == 0 ||
dataSize > std::numeric_limits<std::size_t>::max() / numData) return ULT_ERROR_RANGE;
if (optParam != nullptr) NotImplemented_nid_no_patch("sceUltQueueDataResourcePoolCreate: optParam");
std::lock_guard<std::mutex> lock(gMutex);
if (waitingQueueResourcePool != nullptr && gResourcePools.find(waitingQueueResourcePool) == gResourcePools.end()) {
return ULT_ERROR_INVALID;
}
if (gQueueDataPools.contains(pool)) return ULT_ERROR_STATE;
auto state = std::make_shared<UltQueueDataPoolState>();
state->_numData = numData;
state->_dataSize = dataSize;
state->_numQueueObject = numQueueObject;
state->_waitingPool = waitingQueueResourcePool;
state->_workArea = workArea;
state->_data.resize(numData * dataSize);
state->_next.resize(numData);
for (std::uint32_t slot = 0; slot < numData; ++slot) {
state->_next[slot] = slot + 1u < numData ? slot + 1u : UINT32_MAX;
}
state->_free = 0;
gQueueDataPools.emplace(pool, std::move(state));
std::memset(pool, 0, 512);
gQueueDataPools[pool] = {numData, dataSize, numQueueObject, waitingQueueResourcePool, workArea};
return ULT_OK;
}
int APS5_VABI sceUltQueueDataResourcePoolDestroy(void* pool) {
if (pool == nullptr) return ULT_ERROR_NULL;
std::lock_guard lock(gMutex);
const auto it = gQueueDataPools.find(pool);
if (it == gQueueDataPools.end()) return ULT_ERROR_STATE;
auto state = it->second;
std::lock_guard stateLock(state->_mutex);
if (state->_queues != 0) return ULT_ERROR_BUSY;
gQueueDataPools.erase(it);
return ULT_OK;
}
int APS5_VABI sceUltQueueCreate(void* queue, const char* name, std::uint64_t dataSize, void* waitingQueueResourcePool, void* queueDataResourcePool, const void* optParam, std::uint32_t buildVersion) {
(void)name;
(void)optParam;
(void)buildVersion;
if (queue == nullptr) {
return ULT_ERROR_NULL;
}
if ((reinterpret_cast<std::uintptr_t>(queue) & 7u) != 0) return ULT_ERROR_ALIGNMENT;
if (dataSize == 0) return ULT_ERROR_RANGE;
if (optParam != nullptr) NotImplemented_nid_no_patch("sceUltQueueCreate: optParam");
auto state = std::make_shared<UltQueueState>();
state->_dataSize = dataSize;
state->_waitingPool = waitingQueueResourcePool;
state->_dataPool = queueDataResourcePool;
std::lock_guard<std::mutex> lock(gMutex);
auto dataPoolIt = gQueueDataPools.find(queueDataResourcePool);
if (dataPoolIt == gQueueDataPools.end()) {
@ -202,48 +296,48 @@ int APS5_VABI sceUltQueueCreate(void* queue, const char* name, std::uint64_t dat
if (waitingQueueResourcePool != nullptr && gResourcePools.find(waitingQueueResourcePool) == gResourcePools.end()) {
return ULT_ERROR_INVALID;
}
state->_capacity = dataPoolIt->second._numData;
if (gQueues.contains(queue)) return ULT_ERROR_STATE;
state->_pool = dataPoolIt->second;
auto& pool = *state->_pool;
std::lock_guard stateLock(pool._mutex);
if (dataSize > pool._dataSize || waitingQueueResourcePool != pool._waitingPool) return ULT_ERROR_INVALID;
if (pool._queues == pool._numQueueObject) return ULT_ERROR_AGAIN;
gQueues.emplace(queue, std::move(state));
++pool._queues;
std::memset(queue, 0, 512);
gQueues[queue] = std::move(state);
return ULT_OK;
}
int APS5_VABI sceUltQueuePush(void* queue, const void* data) {
std::shared_ptr<UltQueueState> state;
if (int ret = queueGetState(queue, &state); ret != ULT_OK) {
return ret;
}
if (data == nullptr && state->_dataSize != 0) {
return ULT_ERROR_NULL;
}
std::lock_guard<std::mutex> lock(state->_mutex);
if (state->_capacity != 0 && state->_items.size() >= state->_capacity) {
return ULT_OK;
}
auto& item = state->_items.emplace_back(static_cast<std::size_t>(state->_dataSize));
if (!item.empty()) {
std::memcpy(item.data(), data, item.size());
}
return ULT_OK;
return QueueTransfer(queue, nullptr, data, true, true);
}
int APS5_VABI sceUltQueueTryPush(void* queue, const void* data) {
return QueueTransfer(queue, nullptr, data, true, false);
}
int APS5_VABI sceUltQueuePop(void* queue, void* data) {
return QueueTransfer(queue, data, nullptr, false, true);
}
int APS5_VABI sceUltQueueTryPop(void* queue, void* data) {
std::shared_ptr<UltQueueState> state;
if (int ret = queueGetState(queue, &state); ret != ULT_OK) {
return ret;
}
if (data == nullptr && state->_dataSize != 0) {
return ULT_ERROR_NULL;
}
std::lock_guard<std::mutex> lock(state->_mutex);
if (state->_items.empty()) {
return ULT_ERROR_AGAIN;
}
auto item = std::move(state->_items.front());
state->_items.pop_front();
if (!item.empty()) {
std::memcpy(data, item.data(), item.size());
return QueueTransfer(queue, data, nullptr, false, false);
}
int APS5_VABI sceUltQueueDestroy(void* queue) {
if (queue == nullptr) return ULT_ERROR_NULL;
std::lock_guard lock(gMutex);
const auto it = gQueues.find(queue);
if (it == gQueues.end()) return ULT_ERROR_STATE;
auto state = it->second;
auto& pool = *state->_pool;
{
std::lock_guard stateLock(pool._mutex);
if (state->_waiters != 0) return ULT_ERROR_BUSY;
ReleaseQueue(*state);
gQueues.erase(it);
}
pool._spaceAvailable.notify_all();
return ULT_OK;
}

24
core/libs/prx/libSceUlt/SceUltTypes.hpp
View File

@ -1,6 +1,13 @@
#ifndef CORE_LIBS_PRX_LIBSCEULT_SCEULTTYPES_HPP
#define CORE_LIBS_PRX_LIBSCEULT_SCEULTTYPES_HPP
#include <condition_variable>
#include <cstdint>
#include <memory>
#include <mutex>
#include <vector>
#include "SceTypes.hpp"
constexpr int ULT_OK = 0;
constexpr int ULT_ERROR_NULL = -2139029503;
constexpr int ULT_ERROR_ALIGNMENT = -2139029502;
@ -30,6 +37,12 @@ struct UltResourcePoolState {
};
struct UltQueueDataPoolState {
std::mutex _mutex;
std::condition_variable _spaceAvailable;
std::vector<std::uint8_t> _data;
std::vector<std::uint32_t> _next;
std::uint32_t _free = UINT32_MAX;
std::uint32_t _queues = 0;
std::uint32_t _numData = 0;
std::uint64_t _dataSize = 0;
std::uint32_t _numQueueObject = 0;
@ -38,12 +51,13 @@ struct UltQueueDataPoolState {
};
struct UltQueueState {
std::mutex _mutex;
std::deque<std::vector<std::uint8_t>> _items;
std::shared_ptr<UltQueueDataPoolState> _pool;
std::condition_variable _dataAvailable;
std::uint32_t _head = UINT32_MAX;
std::uint32_t _tail = UINT32_MAX;
std::uint32_t _waiters = 0;
bool _alive = true;
std::uint64_t _dataSize = 0;
std::uint32_t _capacity = 0;
void* _waitingPool = nullptr;
void* _dataPool = nullptr;
};
struct UltRuntimeState {

255
core/libs/tests/GuestUltQueue.cpp Normal file
View File

@ -0,0 +1,255 @@
#include "SceTypes.hpp"
#include <array>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <future>
#include <stdexcept>
#include <thread>
#include <vector>
extern "C" {
int APS5_VABI sceUltInitialize();
int APS5_VABI sceUltFinalize();
std::uint64_t APS5_VABI sceUltQueueDataResourcePoolGetWorkAreaSize(std::uint32_t, std::uint64_t, std::uint32_t);
int APS5_VABI sceUltQueueDataResourcePoolCreate(void*, const char*, std::uint32_t, std::uint64_t, std::uint32_t, void*, void*, const void*, std::uint32_t);
int APS5_VABI sceUltQueueDataResourcePoolDestroy(void*);
int APS5_VABI sceUltQueueCreate(void*, const char*, std::uint64_t, void*, void*, const void*, std::uint32_t);
int APS5_VABI sceUltQueueDestroy(void*);
int APS5_VABI sceUltQueuePush(void*, const void*);
int APS5_VABI sceUltQueueTryPush(void*, const void*);
int APS5_VABI sceUltQueuePop(void*, void*);
int APS5_VABI sceUltQueueTryPop(void*, void*);
}
static constexpr int Null = -2139029503;
static constexpr int Alignment = -2139029502;
static constexpr int Range = -2139029501;
static constexpr int Invalid = -2139029500;
static constexpr int State = -2139029498;
static constexpr int Busy = -2139029497;
static constexpr int Again = -2139029496;
static void Require(bool value, int line) {
if (!value) {
std::fprintf(stderr, "ULT queue check failed at line %d\n", line);
std::abort();
}
}
#define CHECK(value) Require((value), __LINE__)
struct alignas(8) Object { std::array<std::uint8_t, 512> bytes{}; };
struct Fixture {
Object pool;
Object first;
Object second;
void Create(std::uint32_t slots, std::uint32_t queues = 1, std::uint64_t size = sizeof(std::uint64_t)) {
CHECK(sceUltQueueDataResourcePoolCreate(&pool, "pool", slots, size, queues, nullptr, nullptr, nullptr, 0) == 0);
CHECK(sceUltQueueCreate(&first, "first", size, nullptr, &pool, nullptr, 0) == 0);
if (queues > 1) CHECK(sceUltQueueCreate(&second, "second", size, nullptr, &pool, nullptr, 0) == 0);
}
void Destroy(bool secondQueue = false) {
CHECK(sceUltQueueDestroy(&first) == 0);
if (secondQueue) CHECK(sceUltQueueDestroy(&second) == 0);
CHECK(sceUltQueueDataResourcePoolDestroy(&pool) == 0);
}
};
static void Validation() {
Fixture fixture;
Object extra;
std::uint64_t value = 99;
CHECK(sceUltQueueDataResourcePoolCreate(nullptr, nullptr, 1, 8, 1, nullptr, nullptr, nullptr, 0) == Null);
CHECK(sceUltQueueDataResourcePoolCreate(fixture.pool.bytes.data() + 1, nullptr, 1, 8, 1, nullptr, nullptr, nullptr, 0) == Alignment);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 0, 8, 1, nullptr, nullptr, nullptr, 0) == Range);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 1, 0, 1, nullptr, nullptr, nullptr, 0) == Range);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 1, 8, 0, nullptr, nullptr, nullptr, 0) == Range);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 2, UINT64_MAX, 1, nullptr, nullptr, nullptr, 0) == Range);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 1, 8, 1, &extra, nullptr, nullptr, 0) == Invalid);
CHECK(sceUltQueueCreate(&fixture.first, nullptr, 8, nullptr, &fixture.pool, nullptr, 0) == Invalid);
fixture.Create(2);
CHECK(sceUltQueuePush(&fixture.first, &value) == 0);
CHECK(sceUltQueueDataResourcePoolCreate(&fixture.pool, nullptr, 1, 8, 1, nullptr, nullptr, nullptr, 0) == State);
CHECK(sceUltQueueCreate(&fixture.first, nullptr, 8, nullptr, &fixture.pool, nullptr, 0) == State);
std::uint64_t saved = 0;
CHECK(sceUltQueueTryPop(&fixture.first, &saved) == 0 && saved == value);
CHECK(sceUltQueueCreate(&extra, nullptr, 9, nullptr, &fixture.pool, nullptr, 0) == Invalid);
CHECK(sceUltQueueCreate(&extra, nullptr, 8, &extra, &fixture.pool, nullptr, 0) == Invalid);
CHECK(sceUltQueueCreate(&extra, nullptr, 8, nullptr, &fixture.pool, nullptr, 0) == Again);
CHECK(sceUltQueueCreate(nullptr, nullptr, 8, nullptr, &fixture.pool, nullptr, 0) == Null);
CHECK(sceUltQueueCreate(extra.bytes.data() + 1, nullptr, 8, nullptr, &fixture.pool, nullptr, 0) == Alignment);
CHECK(sceUltQueueCreate(&extra, nullptr, 0, nullptr, &fixture.pool, nullptr, 0) == Range);
CHECK(sceUltQueueTryPop(&fixture.first, &value) == Again && value == 99);
CHECK(sceUltQueueTryPush(&fixture.first, nullptr) == Null);
CHECK(sceUltQueuePop(&fixture.first, nullptr) == Null);
CHECK(sceUltQueuePush(nullptr, &value) == Null);
CHECK(sceUltQueuePop(&extra, &value) == State);
CHECK(sceUltQueueDataResourcePoolDestroy(&fixture.pool) == Busy);
CHECK(sceUltQueueDataResourcePoolDestroy(nullptr) == Null);
CHECK(sceUltQueueDestroy(nullptr) == Null);
CHECK(sceUltQueueDataResourcePoolGetWorkAreaSize(3, 9, 2) == 1072);
bool overflow = false;
try { sceUltQueueDataResourcePoolGetWorkAreaSize(1, UINT64_MAX, 1); }
catch (const std::out_of_range&) { overflow = true; }
CHECK(overflow);
bool rejected = false;
try { sceUltQueueCreate(&extra, nullptr, 8, nullptr, &fixture.pool, &extra, 0); }
catch (const std::runtime_error&) { rejected = true; }
CHECK(rejected);
rejected = false;
try { sceUltQueueDataResourcePoolCreate(&extra, nullptr, 1, 8, 1, nullptr, nullptr, &extra, 0); }
catch (const std::runtime_error&) { rejected = true; }
CHECK(rejected);
overflow = false;
try { sceUltQueueDataResourcePoolGetWorkAreaSize(UINT32_MAX, UINT64_MAX / 2, 1); }
catch (const std::out_of_range&) { overflow = true; }
CHECK(overflow);
fixture.Destroy();
CHECK(sceUltQueueTryPush(&fixture.first, &value) == State);
CHECK(sceUltQueueDestroy(&fixture.first) == State);
CHECK(sceUltQueueDataResourcePoolDestroy(&fixture.pool) == State);
fixture.Create(1);
fixture.Destroy();
}
static void Fifo() {
Fixture fixture;
fixture.Create(3, 1, 13);
std::array<std::array<std::uint8_t, 13>, 4> items{};
for (std::size_t index = 0; index < items.size(); ++index) {
for (std::size_t byte = 0; byte < items[index].size(); ++byte) items[index][byte] = index * 31 + byte;
}
for (int round = 0; round < 100; ++round) {
for (int index = 0; index < 3; ++index) CHECK(sceUltQueueTryPush(&fixture.first, items[index].data()) == 0);
CHECK(sceUltQueueTryPush(&fixture.first, items[3].data()) == Again);
std::array<std::uint8_t, 15> output;
for (int index = 0; index < 3; ++index) {
output.fill(0xAA);
CHECK(sceUltQueuePop(&fixture.first, output.data() + 1) == 0);
CHECK(output.front() == 0xAA && output.back() == 0xAA);
CHECK(std::memcmp(output.data() + 1, items[index].data(), 13) == 0);
}
CHECK(sceUltQueueTryPop(&fixture.first, output.data()) == Again);
}
fixture.Destroy();
}
static void WaitUntilBlocked(std::future<int>& result) {
CHECK(result.wait_for(std::chrono::milliseconds(100)) == std::future_status::timeout);
}
static void Blocking() {
Fixture fixture;
fixture.Create(1);
std::uint64_t first = 11, second = 22, output = 0;
CHECK(sceUltQueuePush(&fixture.first, &first) == 0);
auto producer = std::async(std::launch::async, [&] { return sceUltQueuePush(&fixture.first, &second); });
WaitUntilBlocked(producer);
CHECK(sceUltQueueDestroy(&fixture.first) == Busy);
CHECK(sceUltQueuePop(&fixture.first, &output) == 0 && output == first);
CHECK(producer.get() == 0);
CHECK(sceUltQueuePop(&fixture.first, &output) == 0 && output == second);
auto consumer = std::async(std::launch::async, [&] { return sceUltQueuePop(&fixture.first, &output); });
WaitUntilBlocked(consumer);
CHECK(sceUltQueueDestroy(&fixture.first) == Busy);
CHECK(sceUltQueuePush(&fixture.first, &first) == 0);
CHECK(consumer.get() == 0 && output == first);
fixture.Destroy();
}
static void SharedPool() {
Fixture fixture;
fixture.Create(1, 2);
std::uint64_t first = 111, second = 222, output = 0;
auto consumer = std::async(std::launch::async, [&] { return sceUltQueuePop(&fixture.second, &output); });
WaitUntilBlocked(consumer);
CHECK(sceUltQueuePush(&fixture.first, &first) == 0);
WaitUntilBlocked(consumer);
CHECK(sceUltQueueTryPush(&fixture.second, &second) == Again);
CHECK(sceUltQueueTryPop(&fixture.second, &output) == Again);
auto producer = std::async(std::launch::async, [&] { return sceUltQueuePush(&fixture.second, &second); });
WaitUntilBlocked(producer);
CHECK(sceUltQueueDestroy(&fixture.second) == Busy);
CHECK(sceUltQueueDestroy(&fixture.first) == 0);
CHECK(producer.get() == 0);
CHECK(consumer.get() == 0 && output == second);
CHECK(sceUltQueueCreate(&fixture.first, "replacement", 4, nullptr, &fixture.pool, nullptr, 0) == 0);
const std::uint32_t small = 42;
CHECK(sceUltQueuePush(&fixture.first, &small) == 0);
std::uint32_t smallOutput = 0;
CHECK(sceUltQueuePop(&fixture.first, &smallOutput) == 0 && smallOutput == small);
fixture.Destroy(true);
}
static void Finalize() {
Fixture fixture;
fixture.Create(1, 2);
std::uint64_t value = 7, output = 123;
CHECK(sceUltQueuePush(&fixture.first, &value) == 0);
auto producer = std::async(std::launch::async, [&] { return sceUltQueuePush(&fixture.first, &value); });
auto consumer = std::async(std::launch::async, [&] { return sceUltQueuePop(&fixture.second, &output); });
WaitUntilBlocked(producer);
WaitUntilBlocked(consumer);
CHECK(sceUltQueueDestroy(&fixture.first) == Busy);
CHECK(sceUltQueueDestroy(&fixture.second) == Busy);
CHECK(sceUltFinalize() == 0);
CHECK(producer.get() == State);
CHECK(consumer.get() == State && output == 123);
CHECK(sceUltQueueTryPop(&fixture.first, &output) == State);
CHECK(sceUltInitialize() == 0);
fixture.Create(1, 2);
CHECK(sceUltQueueTryPop(&fixture.first, &output) == Again);
CHECK(sceUltQueuePush(&fixture.second, &value) == 0);
CHECK(sceUltQueuePop(&fixture.second, &output) == 0 && output == value);
fixture.Destroy(true);
}
static void Stress() {
Fixture fixture;
fixture.Create(7);
constexpr std::size_t workers = 4;
constexpr std::size_t perWorker = 2000;
std::array<std::atomic<unsigned>, workers * perWorker> seen{};
std::vector<std::thread> producers, consumers;
for (std::size_t worker = 0; worker < workers; ++worker) {
consumers.emplace_back([&] {
for (std::size_t index = 0; index < perWorker; ++index) {
std::uint64_t value = UINT64_MAX;
CHECK(sceUltQueuePop(&fixture.first, &value) == 0);
CHECK(value < seen.size());
++seen[value];
}
});
producers.emplace_back([&, worker] {
for (std::size_t index = 0; index < perWorker; ++index) {
const std::uint64_t value = worker * perWorker + index;
CHECK(sceUltQueuePush(&fixture.first, &value) == 0);
}
});
}
for (auto& producer : producers) producer.join();
for (auto& consumer : consumers) consumer.join();
for (const auto& count : seen) CHECK(count == 1);
std::uint64_t output = 0;
CHECK(sceUltQueueTryPop(&fixture.first, &output) == Again);
fixture.Destroy();
}
int main(int argc, char** argv) {
CHECK(argc == 2);
CHECK(sceUltInitialize() == 0);
if (std::strcmp(argv[1], "validation") == 0) Validation();
else if (std::strcmp(argv[1], "fifo") == 0) Fifo();
else if (std::strcmp(argv[1], "blocking") == 0) Blocking();
else if (std::strcmp(argv[1], "shared-pool") == 0) SharedPool();
else if (std::strcmp(argv[1], "finalize") == 0) Finalize();
else if (std::strcmp(argv[1], "stress") == 0) Stress();
else CHECK(false);
CHECK(sceUltFinalize() == 0);
}

2
docs/dev/TechnicalDebt.md
View File

@ -70,9 +70,11 @@ Throughout the project, every function at every stage either **does exactly what
- [sceSystemServiceReenableMediaPlay](../../core/libs/prx/libSceSystemService/Export.cpp) (libSceSystemService) - unknown signature
- [sceVoiceSetMuteFlag](../../core/libs/prx/libSceVoice/Export.cpp) (libSceVoice) - unknown signature
- [snwprintf_s](../../core/libs/prx/libc/src/RuntimeSupport.cpp) (libc) - unknown signature
- [libSceUlt queues](../../core/libs/prx/libSceUlt/Export.cpp) - the added `sceUltQueuePop`, `sceUltQueueTryPush`, `sceUltQueueDestroy` and `sceUltQueueDataResourcePoolDestroy` signatures are inferred from the existing push/pop and object APIs. Queue destruction is assumed to reject blocked callers with `ULT_ERROR_BUSY` and otherwise discard queued data; pool exhaustion returns `ULT_ERROR_AGAIN`. These rules, alignment checks and shared slot limits have not been confirmed on PS5 hardware or a title. Non-null queue and data-pool option parameters throw because their layouts are unknown.
### Functional
- [libSceUlt queues](../../core/libs/prx/libSceUlt/Export.cpp) use preallocated host storage instead of the supplied guest work area; its existing size formula and the 512-byte object layouts remain unverified. Waiting-queue pool thread limits and waiter priority ordering are not implemented. Finalization wakes blocked queue callers with `ULT_ERROR_STATE`.
- The shader recompiler [ignores `s_setreg_b32`](../../core/shader/recompiler/Translation/src/ScalarInstructions.cpp): writes to MODE (float rounding and denormal controls) and the other hardware registers have no effect, and `s_getreg_b32` reads of the MODE round fields report round to nearest even even after `s_setreg_b32` wrote another mode. The initial round mode from the shader's program registers is not read either.
- [Shader recompilation](../../core/shader/recompiler/Recompiler.cpp) currently occurs right before it was transferred to Vulkan with caching, but should be moved to the [relinker](../../core/relinker/main.cpp) stage. For this purpose, [shader/recompiler](../../core/shader/recompiler) was written completely independently from [libs/prx](../../core/libs/prx).
- [v_fma_f64](../../core/shader/recompiler/SpirvBackend/src/SpirvAlu/SpirvAluEmitterFloat64.cpp) is fused exactly only for normal inputs: with a subnormal input it is a separate multiply and add, which rounds twice.