Hi,
as you were worried about the idle spinning of the thread, i added blocking on the completions, when the request queue is empty:
The code adds some override for the LLThread::wake() method (which needs to be made virtual and added to the lllfsthread.h).
Now a wake() first calls the base implementation and wakes up the LLCondition of the thread and then (if the io_uring is initialized) queues a NOP/dummy completion with key = 1 on the blocking io_uring call (io_uring_wait_cqe() instead of peek) to wake it up. The threadedUpdate() blocks on the completion port, if no more requests are queued and just peeks (as before) when new requests are incoming.
I did not test the Linux part yet, but the Windows part seems to run nicely, as far as i could tell.
 | | |
| Code: #if LL_IOURING // Lets have some fun with overlapped I/O and io_uring ;-) // (c)2021 Kathrine Jansma // Requests map to OVERLAPPED ReadFile / WriteFile calls basically // Responder => The OVERLAPPED Result gets delivered to the responder // We pick up the completions later and have the OVERLAPPED result delivered to // the responder. Completions MAY happen out of order ! //virtual bool LLLFSThread::runCondition() { return !mIdleThread || mPendingResults || !mRequestQueue.empty(); } //virtual void LLLFSThread::threadedUpdate() { if (!sRingInitialized) { return; } # if LL_LINUX struct io_uring_cqe* cqe; # elif LL_WINDOWS DWORD bytes_processed = 0; LPOVERLAPPED result; DWORD_PTR key; # endif // Dequeue finished results bool have_work = true; while (have_work) { # if LL_LINUX // Check for finished cqe's without blocking; it might be more // efficient to use the batch variant. if (!mRequestQueue.empty()) { have_work = io_uring_peek_cqe(&mRing, &cqe) == 0; } else { // block, as we have no further requests queued have_work = io_uring_wait_cqe(&mRing, &cqe) == 0; } if (have_work) { llassert_always(mPendingResults > 0); --mPendingResults; if (!cqe) // Paranoia ? { break; } S32 bytes_handled = cqe->res; LLLFSThread::Request* req = (LLLFSThread::Request*)io_uring_cqe_get_data(cqe); if (!req || req == 1) // Paranoia or magic wakeup cqe { io_uring_cqe_seen(&mRing, cqe); break; } if (bytes_handled >= 0) { req->setBytesRead(bytes_handled); } else // An I/O error occurred: mark as such. { req->setBytesRead(0); } // true = resquest is over and must auto-complete (self-destruct) setRequestResult(req, true); io_uring_cqe_seen(&mRing, cqe); } # elif LL_WINDOWS DWORD timeout = INFINITE; if (!mRequestQueue.empty()) { timeout = 0; } have_work = GetQueuedCompletionStatus(mIOCPPort, &bytes_processed, &key, &result, timeout); if (key == 1) { // our magic wakeup package --mPendingResults; break; } LLLFSThread::Request* req = (LLLFSThread::Request*)key; if (have_work) { llassert_always(mPendingResults > 0); --mPendingResults; if (req) { req->setBytesRead(bytes_processed); setRequestResult(req, true); } } else if (GetLastError() == WAIT_TIMEOUT) { break; } else if (!result) { if (!req || GetLastError() != ERROR_ABANDONED_WAIT_0) { llwarns << "IOCP GetQueuedCompletionStatus failed with code: " << GetLastError() << llendl; } else { llwarns << "IOCP for '" << req->getFilename() << "' closed while waiting for completion status !" << llendl; } break; } else { llassert_always(mPendingResults > 0); --mPendingResults; if (req) { llwarns << "IOCP operation for " << req->getFilename() << " failed with code: " << GetLastError() << llendl; req->setBytesRead(0); // An error occurred: mark as such. // true = resquest is over and must auto-complete // (self-destruct) setRequestResult(req, true); // We may have further IOCP results pending have_work = true; } } # endif } } /* wake the thread and signal to waiting IOCP/io_uring*/ void LLLFSThread::wake() { LLQueuedThread::wake(); if (!sRingInitialized) { return; } // wakeup the completion ports if those are blocked ++mPendingResults; #if LL_WINDOWS PostQueuedCompletionStatus(mIOCPPort, 0, (ULONG_PTR)1, NULL); #elif LL_LINUX struct io_uring_sqe* sqe = io_uring_get_sqe(&(mThread->mRing)); io_uring_sqe_set_data(sqe, (void*)1); io_uring_prep_nop(sqe); io_uring_submit(&(mThread->mRing)); #endif } #endif // LL_IOURING | |
| | |
