GC safepoints and @preserve for read and write (#119)

* Add GC safepoints on read and write

* Another safepoint

* Move safepoint from flushbuffer to writedata!

* GC safety

Author: ancapdev

src/buffer.jl

@@ -187,7 +187,7 @@ end
 # Copy data from `data` to `buf`.
 function copydata!(buf::Buffer, data::Ptr{UInt8}, nbytes::Integer)
     makemargin!(buf, nbytes)
-    unsafe_copyto!(marginptr(buf), data, nbytes)
+    GC.@preserve buf unsafe_copyto!(marginptr(buf), data, nbytes)
     supplied!(buf, nbytes)
     return buf
 end
@@ -197,7 +197,7 @@ function copydata!(data::Ptr{UInt8}, buf::Buffer, nbytes::Integer)
     # NOTE: It's caller's responsibility to ensure that the buffer has at least
     # nbytes.
     @assert buffersize(buf) ≥ nbytes
-    unsafe_copyto!(data, bufferptr(buf), nbytes)
+    GC.@preserve buf unsafe_copyto!(data, bufferptr(buf), nbytes)
     consumed!(buf, nbytes)
     return data
 end
@@ -206,14 +206,14 @@ end
 function insertdata!(buf::Buffer, data::Ptr{UInt8}, nbytes::Integer)
     makemargin!(buf, nbytes)
     copyto!(buf.data, buf.bufferpos + nbytes, buf.data, buf.bufferpos, buffersize(buf))
-    unsafe_copyto!(bufferptr(buf), data, nbytes)
+    GC.@preserve buf unsafe_copyto!(bufferptr(buf), data, nbytes)
     supplied!(buf, nbytes)
     return buf
 end
 
 # Find the first occurrence of a specific byte.
 function findbyte(buf::Buffer, byte::UInt8)
-    p = ccall(
+    p = GC.@preserve buf ccall(
         :memchr,
         Ptr{UInt8},
         (Ptr{UInt8}, Cint, Csize_t),

src/stream.jl

@@ -334,7 +334,7 @@ function Base.readuntil(stream::TranscodingStream, delim::UInt8; keep::Bool=fals
             @assert filled == 0
             ret = Vector{UInt8}(undef, sz)
         end
-        copydata!(pointer(ret, filled+1), buffer1, sz)
+        GC.@preserve ret copydata!(pointer(ret, filled+1), buffer1, sz)
         filled += sz
         if found
             if !keep
@@ -360,6 +360,7 @@ function Base.unsafe_read(stream::TranscodingStream, output::Ptr{UInt8}, nbytes:
         m = min(buffersize(buffer), p_end - p)
         copydata!(p, buffer, m)
         p += m
+        GC.safepoint()
     end
     if p < p_end && eof(stream)
         throw(EOFError())
@@ -376,7 +377,7 @@ function Base.readbytes!(stream::TranscodingStream, b::AbstractArray{UInt8}, nb=
             resize!(b, min(length(b) * 2, nb))
             resized = true
         end
-        filled += unsafe_read(stream, pointer(b, filled+1), min(length(b), nb)-filled)
+        filled += GC.@preserve b unsafe_read(stream, pointer(b, filled+1), min(length(b), nb)-filled)
     end
     if resized
         resize!(b, filled)
@@ -392,7 +393,7 @@ end
 function Base.readavailable(stream::TranscodingStream)
     n = bytesavailable(stream)
     data = Vector{UInt8}(undef, n)
-    unsafe_read(stream, pointer(data), n)
+    GC.@preserve data unsafe_read(stream, pointer(data), n)
     return data
 end
 
@@ -405,7 +406,7 @@ The next `read(stream, sizeof(data))` call will read data that are just
 inserted.
 """
 function unread(stream::TranscodingStream, data::ByteData)
-    unsafe_unread(stream, pointer(data), sizeof(data))
+    GC.@preserve data unsafe_unread(stream, pointer(data), sizeof(data))
 end
 
 """
@@ -462,6 +463,7 @@ function Base.unsafe_write(stream::TranscodingStream, input::Ptr{UInt8}, nbytes:
         m = min(marginsize(buffer1), p_end - p)
         copydata!(buffer1, p, m)
         p += m
+        GC.safepoint()
     end
     return Int(p - input)
 end
@@ -636,8 +638,8 @@ end
 function callprocess(stream::TranscodingStream, inbuf::Buffer, outbuf::Buffer)
     state = stream.state
     input = buffermem(inbuf)
-    makemargin!(outbuf, minoutsize(stream.codec, input))
-    Δin, Δout, state.code = process(stream.codec, input, marginmem(outbuf), state.error)
+    GC.@preserve inbuf makemargin!(outbuf, minoutsize(stream.codec, input))
+    Δin, Δout, state.code = GC.@preserve inbuf outbuf process(stream.codec, input, marginmem(outbuf), state.error)
     @debug(
         "called process()",
         code = state.code,
@@ -677,7 +679,7 @@ function readdata!(input::IO, output::Buffer)
         navail = bytesavailable(input)
     end
     n = min(navail, marginsize(output))
-    Base.unsafe_read(input, marginptr(output), n)
+    GC.@preserve output Base.unsafe_read(input, marginptr(output), n)
     supplied!(output, n)
     nread += n
     return nread
@@ -691,10 +693,12 @@ function writedata!(output::IO, input::Buffer)
     end
     nwritten::Int = 0
     while buffersize(input) > 0
-        n = Base.unsafe_write(output, bufferptr(input), buffersize(input))
+        n = GC.@preserve input Base.unsafe_write(output, bufferptr(input), buffersize(input))
         consumed!(input, n)
         nwritten += n
+        GC.safepoint()
     end
+    GC.safepoint()
     return nwritten
 end