fix(arrow/ipc): make Writer and FileWriter safe for concurrent use#853
fix(arrow/ipc): make Writer and FileWriter safe for concurrent use#853zeroshade wants to merge 3 commits into
Conversation
The IPC stream Writer and FileWriter share mutable state across Write and Close -- the started/headerStarted flags, the lastWrittenDicts map, the dictionary mapper, and the underlying payload writer -- with no synchronization. Concurrent Write calls could therefore write the schema/header more than once, interleave and corrupt the output stream, or panic with "concurrent map writes" (apache#55). Add a sync.Mutex to each writer, held for the duration of the exported mutating methods (Write, Close); the unexported start/checkStarted helpers run under that lock, so they remain unlocked to avoid self-deadlock. Records are still written in whatever order callers acquire the lock -- the guarantee is freedom from data races and corruption, matching how the standard library documents gob.Encoder and log.Logger. Adds -race regression tests (TestWriterConcurrentWrite, TestFileWriterConcurrentWrite) that report data races without the lock.
Now that Write and Close are serialized by the mutex, a Write that is ordered after Close must not write past the finalized output. Previously a FileWriter.Write losing the race to Close would still append a record batch after the file footer and return nil -- silently corrupting the file -- and the stream Writer would nil-deref its payload writer and surface a confusing recovered-panic error. Write now returns errClosedWriter when the stream writer's payload writer has been closed (pw == nil) or the FileWriter's footer has been written, and the godoc for both documents the rejection. Adds TestWriteAfterClose for both writers (the FileWriter case also asserts no bytes are appended after Close).
A Writer or FileWriter created without WithSchema and closed before the first Write reached start() with a nil schema and panicked in Mapper.ImportSchema (schema.NumFields() on a nil *arrow.Schema). With the new concurrent Write/Close contract this is reachable when Close wins the mutex before the first schema-inferring Write. start() now returns errNoSchema instead of panicking; Write is unaffected because it sets the schema from the record before calling start(). Adds TestCloseWithoutSchema covering both the stream Writer and FileWriter.
lidavidm
left a comment
lidavidm
left a comment
There was a problem hiding this comment.
Why do we want it to be concurrent though? Wouldn't a channel + goroutine make more sense?
|
Channel + goroutine definitely makes more sense, but I wanted to avoid the corruption/panic. Alternatively we could explicitly document that it is not concurrent safe if you think that's better than actually making it concurrent safe. Thoughts? |
|
Do we want to make all the readers/writers thread-safe? I think that's not a variant that should be expected (and is that a variant expected of e.g. an io.Writer? It doesn't seem so.) And furthermore, does Go normally expect things to be thread-safe when it isn't explicitly documented? |
|
Hmm, that's a good point honestly. the io.Writer interface makes no guarantees about concurrency safety and should only be considered safe if the underlying implementation being used is safe. I think I agree with you that it would be better for us to just add to the documentation that the ipc writer is not concurrency safe and that writes need to get serialized. |
Rationale for this change
Closes #55.
The IPC stream
WriterandFileWriterare not safe for concurrent use. Both share mutable state acrossWrite/Closewith no synchronization:started/headerStartedflags (checked-then-set inWrite/Close→ schema/header can be written more than once),lastWrittenDictsmap (written fromwriteDictionaryPayloads→ Gofatal error: concurrent map writes),mapper, the sharedcompressorsslice, and the underlyingPayloadWriter(concurrentWritePayload→ interleaved/corrupted stream),Closesettingpw = nilwhile aWriteis in flight → nil-deref / use-after-close.As reported, calling
Writefrom multiple goroutines on one writer can write the schema multiple times and crash the sink. Running the new tests under-raceagainst the old code reports many data races.What changes are included in this PR?
sync.MutextoWriterandFileWriter, held for the duration of the exported mutating methodsWriteandClose.start/checkStartedhelpers are only ever called fromWrite/Close, so they intentionally stay unlocked to avoid self-deadlock.Write/Close. Calls are serialized; records are written in the order in which callers acquire the lock. The guarantee is freedom from data races and stream corruption — not a particular record ordering — which matches how the standard library documents the mutex-protectedgob.Encoderandlog.Logger.No public signatures change; this is additive (a new unexported field + locking + godoc).
Are these changes tested?
Yes. Two new
-raceregression tests inarrow/ipc/writer_test.go:TestWriterConcurrentWrite— 16 goroutinesWriteto one streamWriter, then the stream is read back and asserted to contain exactly one schema message + 16 record batches.TestFileWriterConcurrentWrite— same forFileWriter, assertingNumRecords() == 16on read-back.Verified that both fail with data-race warnings on the pre-change code and pass with the lock. The full
arrow/ipc/...suite passes under-race(no deadlock/regression), andgo vetis clean.Are there any user-facing changes?
No breaking changes.
Write/Closekeep their signatures; concurrent callers now get defined, race-free behavior instead of corruption/panics, and single-threaded callers are unaffected (an uncontended mutex).Design note
Apache Arrow C++ (
IpcFormatWriter) and Java (ArrowWriter) treat IPC writers as single-threaded, and most Go stdlib stream writers (bufio,csv,json,gzip) leave synchronization to the caller. I went with the mutex (rather than only documenting non-safety) because the reported failure mode is a hardconcurrent map writescrash, the lock is free when uncontended, and arrow-go already documents concurrency guarantees elsewhere (e.g.FileReader.RecordBatchAt/RecordAt). Happy to switch to a docs-only "caller must synchronize" approach if maintainers prefer matching C++/Java exactly.