Add FileWriter, and let the Decoder write articles straight to it - #203
Conversation
Owns the descriptor rather than borrowing one, so nothing outside can close it while a write is in flight and a stale descriptor cannot send data into a file that has since claimed that number. Writes take a shared lock and run concurrently; only close takes it exclusively, so it drains rather than pulling the handle away. On Windows this is the point: WriteFile with an OVERLAPPED offset is positional even on a non-overlapped handle, which os.pwrite is not available to provide there, so callers no longer need a lock of their own. Windows also opens through CreateFileW, giving a real HANDLE for both the writes and the sparse ioctl with no msvcrt round trip. Short writes are retried internally. preallocate() matches the existing sparse() behaviour so the two can be swapped without changing what callers see. sparse() itself is unchanged.
Decoder.expect(context, sink) records a request that has gone out. NNTPResponse.context hands the object back, so the pairing lives here rather than in a queue on the Python side kept in step with this one. Two queues can drift, and a drift writes one article's bytes into another article's file at a plausible-looking offset: silent corruption that only par2 would catch. With a sink, the body is decoded through a buffer shared by every response on the connection and written at the offset the yEnc headers declare, instead of building a bytearray per article. response.data is left as None. Bodies larger than the staging buffer are flushed in pieces, and the CRC is folded across them. uu carries no offsets, so a sink is ignored there and data is populated as before. Callers that never use expect() are unaffected. Holding arbitrary Python objects makes cycles through these types reachable - d.expect(d) is enough - so both now support GC, which they did not before. Measured over 60 articles of 700 KB held at once: peak RSS grows 44.4 MB on the bytearray path against 0.2 MB streaming, with the payload on disk instead.
expected reported only requests with nothing received yet, but the pending entry is taken as soon as the first byte of a response arrives. A pipelined connection therefore looked idle while it was still receiving, and SABnzbd stopped reading from the socket: with two requests outstanding only the first ever completed, and the job stalled. It now counts responses finished but not yet collected, the one arriving, and those still untouched. pending reports the same three, oldest first, so the article a connection is fetching is the one being received rather than the next one queued behind it.
Deleting a queue item while it was downloading closed the FileWriter under an article still arriving, and the write error surfaced as ValueError out of Decoder.process(). That killed the receive thread that was running it. The thread was the visible half. The real problem is that raising there abandons the decoder in the middle of a response: the remainder of the article is still in the connection's buffer and would be parsed as the start of the next one. So the failure would cost the whole connection rather than one article, and no amount of catching it in Python helps, because by the time the exception surfaces the stream is already out of step. A sink write failure is fatal to the article and not to the connection. The response is now consumed to its end with the body discarded, and the failure is reported as NNTPResponse.sink_failed once it completes. The article is refetched; anything already written is overwritten at the same offsets, so a retry is idempotent. The test that asserted the old behaviour is replaced by one that decodes a second article on the same connection after a failure, which is what a mid-response abort destroys and what a test for the raise could never have caught. Disk-full took the same path and killed a receive thread the same way.
It was 1 MiB, chosen so a typical ~700 KB article staged whole and was written with one syscall. That reasoning does not survive measurement. Decode CPU is flat from 256 KiB upward across 64 KiB to 4 MiB: the yEnc decode dominates and the saved write syscalls do not register, with 1 MiB measuring no better than 256 KiB. Below 256 KiB they do start to tell, 64 KiB costing about 9% more. So the extra 768 KiB bought nothing, and it is paid once per connection - 200 connections is not unusual, and 200 MB of staging eats the memory saving that streaming exists to deliver. At 256 KiB the same case is 50 MB. 256 KiB is the caller's input buffer size, which is the number that actually matters: one process() call cannot produce more decoded bytes than its input holds, so this absorbs the most any one call can generate and flushes land on call boundaries. The comment claiming the buffer stays resident in L2/L3 is removed rather than reworded. At any realistic connection count it does not, and a buffer this size overflows L1 several times over on every article. The size is now overridable at build time, which is how the above was measured.
Decoder_process reset the ring at the end of every call. The caller writes its next read into [position, size) and that tail is what the buffer protocol exports, so the reset buys nothing until the tail has run down far enough to make the next read small. The memmove is the eye-catching half of that and the irrelevant one. On a yEnc stream the unprocessed remainder is almost always empty, because the decoder carries partial lines in its own state rather than leaving bytes in the buffer, so the branch that actually fired was the free position = consumed = 0 reset. Measured over a realistic article stream, the memmove ran zero times - deferring only the move was a no-op, and the reset wipes the consumed region just as thoroughly. Both are now deferred behind the same threshold. Nothing above the decoder can see this: same decoded bytes, verified by hashing the output of both builds over an article stream at several read sizes, and a live download byte-identical to its source. The point is what it makes possible rather than what it saves. It saves nothing measurable - the eager path was a pointer reset. It leaves the consumed region in front of the read pointer intact between calls, which is where decoded output would live if the decoder ever writes in place instead of into a separate staging buffer. With 48 KiB reads into a 256 KiB ring that region now reaches ~192 KiB before a rewind. The test fails against an always-rewind build, and also pins that the tail never reaches zero: handing the caller an empty buffer would stall the connection.
|
@claude review, I'm a C-noob, so help me determine if it's good code, structurally and functionally. It will be used together with sabnzbd/sabnzbd#3562, take that into account if possible |
|
Claude finished @Safihre's task in 4m 18s —— View job Code review — FileWriter + Decoder sink/pairing
Overall this is solid, careful C/C++ — good structure, and the concurrency story is well thought through and (mostly) consistently applied. One real bug worth fixing, plus a couple of nits. Findings1. Data race on Fix: wrap the handle read in each of these four functions with 2. Minor: duplicate field init — 3. Minor: possibly-redundant reset — Structural notes (why the rest looks good)
Fit with the SABnzbd side (from the description in this PR +
|
|
How will it deal with disk full? I saw it just drops the data, but how would we report in SABnzbd? And can we not detect if SSD using something like kipodd/ssd_checker#1? |
close() mutates it under the exclusive lock with the GIL released, so the accessors were reading it unsynchronised; size could also stat a descriptor that close() had already released.
Fixed in
Fixed
It's not redundant and Claude has commented why to help in the future
That was rather missing, fetched articles into a black hole and failed the job.
System probing will come after this PR; it had already gotten large. Briefly (he says before writing an essay) Also "is it an SSD" is not sufficient, I need to know can it sustain 700KB random writes at download rate which is not just down to the SSD; CPUfor processing, encryption, filesystem, network mount, RAID, etc. A single probe at startup is also not sufficient, you end up just measuring the device's cache unless doing a long test. |
|
Just wondering if there's some way that we don't need the path back to Python anymore. |
tl;dr; maybe and I've had in mind that it's something we might want to do in the future In the current implementation, articles that return to Python with data are the ones queued without a sink: the first article, and any other queued whose filename is unresolved at that time (par2, md5of16k, or not yet seen yenc headers). What we could do, and this abstraction makes much simpler is pass the destination directory in or temporary name, always write to file and rename if we need to once filenames are resoled. Unix allows renaming an open file descriptor, and so does Windows when the file is opened with specific flags - FileWriter already does so. I'm uncertain how that behaves across exotic filesystems but as long as the path degrades into the existing Python path it does not matter much. If we do not take that path, then yes, ArticleCache and eventually Assembler could move to C - but as it stands I don't think it's worth it. The purpose of these changes are to spend less time in the cache or assembler, and on this streamed/direct_decode path the assembler does almost nothing; it fires on the first article and file_done, but usually it has nothing to do except set permissions and some file_done actions (which I may take off the assembler thread) - I measured CPU in the assembler went from 16.3s to 0.1 so there is nothing performance wise to gain and it just makes it more complicated. Keeping the HDD path in C is complicated. You'd still need a cache, simple but you lose a lot of the visibility you get in Python which could lead to bugs like NzbFile removed but articles remain in memory. The core ordered writing of the assembler is only ~80 lines. There is still room to improve the HDD contiguous part, its problem is the gap between download speed and write speed. The cache fills and there is no way back. Which is what this eliminates, writes happen inline with downloading while the GIL is released. A slow disk blocks the receive thread rather than backing up into memory so you cannot download faster than you can write, including while par2, unrar, etc are competing for the same disk. The direct_decode path is 256KB of input plus 256KB of staging per connection, both C-owned and reused for every article passing through, so memory use is very predictable and allocation-free per article. |
|
Good considerations, I agree. |
2 participants
Enables the matching SABnzbd change, which shortens the download path from
Downloader (Py) > Unlocked SSL (C) > Decoder (C) > Article Cache (Py) > Assembler (Py) > Disk write (Py/C)toDownloader (Py) > Unlocked SSL (C) > Decoder (C) > Disk write (C).FileWriter
closed,path,sizeand context-manager supportWriteFilewith anOVERLAPPEDoffset, so callers no longer need a lock to stand in for the missingos.pwriteclose()is exclusivesabctools.sparse(), kept for existing callersDecoder: pairing, and an optional sink
Decoder.expect(context, sink=None)records a sent request;contextcomes back asNNTPResponse.context. Callers no longer keep a queue alongside the decoder's - two queues that drift write one article's bytes into another article's file at a plausible offset, which only par2 would catchsink, the body is written at the offset from the yEnc headers andresponse.dataisNone; bodies over the staging buffer are written in piecesdatais populated as usualDecoder, reused for every article, so streaming costs no per-article allocation. Sized to the caller's input buffer (256 KiB)NNTPResponse.sink_failedreports a failed write and a discarded body. The response still completes and the connection stays usable - abandoning it mid-stream would leave the rest of the article to be parsed as the next one, costing the whole connection instead of one articleNotes
I think I could remove the sparse methods, the same is accomplished using FileWriter.