Handle large read/write/sendfile to File; return actual bytes written

base/filesystem.jl

@@ -219,7 +219,7 @@ function sendfile(dst::File, src::File, src_offset::Int64, bytes::Int)
     check_open(dst)
     check_open(src)
     while true
-        result = ccall(:jl_fs_sendfile, Int32, (OS_HANDLE, OS_HANDLE, Int64, Csize_t),
+        result = ccall(:jl_fs_sendfile, Cssize_t, (OS_HANDLE, OS_HANDLE, Int64, Csize_t),
                        src.handle, dst.handle, src_offset, bytes)
         uv_error("sendfile", result)
         nsent = result
@@ -232,10 +232,10 @@ end
 
 function unsafe_write(f::File, buf::Ptr{UInt8}, len::UInt, offset::Int64=Int64(-1))
     check_open(f)
-    err = ccall(:jl_fs_write, Int32, (OS_HANDLE, Ptr{UInt8}, Csize_t, Int64),
+    ret = ccall(:jl_fs_write, Cssize_t, (OS_HANDLE, Ptr{UInt8}, Csize_t, Int64),
                 f.handle, buf, len, offset)
-    uv_error("write", err)
-    return len
+    uv_error("write", ret)
+    return ret
 end
 
 write(f::File, c::UInt8) = write(f, Ref{UInt8}(c))
@@ -265,7 +265,7 @@ end
 function read(f::File, ::Type{UInt8})
     check_open(f)
     p = Ref{UInt8}()
-    ret = ccall(:jl_fs_read, Int32, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
+    ret = ccall(:jl_fs_read, Cssize_t, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
                 f.handle, p, 1)
     uv_error("read", ret)
     @assert ret <= sizeof(p) == 1
@@ -298,7 +298,7 @@ read(f::File, ::Type{T}) where {T<:AbstractChar} = T(read(f, Char)) # fallback
 
 function unsafe_read(f::File, p::Ptr{UInt8}, nel::UInt)
     check_open(f)
-    ret = ccall(:jl_fs_read, Int32, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
+    ret = ccall(:jl_fs_read, Cssize_t, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
                 f.handle, p, nel)
     uv_error("read", ret)
     ret == nel || throw(EOFError())
@@ -314,7 +314,7 @@ function readbytes!(f::File, b::MutableDenseArrayType{UInt8}, nb=length(b))
     if length(b) < nr
         resize!(b, nr)
     end
-    ret = ccall(:jl_fs_read, Int32, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
+    ret = ccall(:jl_fs_read, Cssize_t, (OS_HANDLE, Ptr{Cvoid}, Csize_t),
                 f.handle, b, nr)
     uv_error("read", ret)
     return ret

src/jl_uv.c

@@ -583,16 +583,16 @@ JL_DLLEXPORT int jl_fs_rename(const char *src_path, const char *dst_path)
     return ret;
 }
 
-JL_DLLEXPORT int jl_fs_sendfile(uv_os_fd_t src_fd, uv_os_fd_t dst_fd,
-                                int64_t in_offset, size_t len)
+JL_DLLEXPORT ssize_t jl_fs_sendfile(uv_os_fd_t src_fd, uv_os_fd_t dst_fd,
+                                    int64_t in_offset, size_t len)
 {
     uv_fs_t req;
     JL_SIGATOMIC_BEGIN();
     int ret = uv_fs_sendfile(unused_uv_loop_arg, &req, dst_fd, src_fd,
                              in_offset, len, NULL);
     uv_fs_req_cleanup(&req);
     JL_SIGATOMIC_END();
-    return ret;
+    return req.result ? req.result : ret;
 }
 
 JL_DLLEXPORT int jl_fs_hardlink(char *path, char *new_path)
@@ -635,7 +635,15 @@ JL_DLLEXPORT int jl_fs_access(char *path, int mode)
     return ret;
 }
 
-JL_DLLEXPORT int jl_fs_write(uv_os_fd_t handle, const char *data, size_t len,
+/*
+ * A note about the uv_fs_* functions that return a count of bytes written:
+ * uv_fs_t.result is a ssize_t, but the function returns int.  The result field
+ * needs to be checked so we can distinguish a real error from a count that
+ * overflows an int, but we can't use it directly---some error conditions will
+ * leave it set to 0 but return an error.
+ */
+
+JL_DLLEXPORT ssize_t jl_fs_write(uv_os_fd_t handle, const char *data, size_t len,
                              int64_t offset) JL_NOTSAFEPOINT
 {
     jl_task_t *ct = jl_get_current_task();
@@ -654,18 +662,18 @@ JL_DLLEXPORT int jl_fs_write(uv_os_fd_t handle, const char *data, size_t len,
         jl_io_loop = uv_default_loop();
     int ret = uv_fs_write(unused_uv_loop_arg, &req, handle, buf, 1, offset, NULL);
     uv_fs_req_cleanup(&req);
-    return ret;
+    return req.result ? req.result : ret;
 }
 
-JL_DLLEXPORT int jl_fs_read(uv_os_fd_t handle, char *data, size_t len)
+JL_DLLEXPORT ssize_t jl_fs_read(uv_os_fd_t handle, char *data, size_t len)
 {
     uv_fs_t req;
     uv_buf_t buf[1];
     buf[0].base = data;
     buf[0].len = len;
     int ret = uv_fs_read(unused_uv_loop_arg, &req, handle, buf, 1, -1, NULL);
     uv_fs_req_cleanup(&req);
-    return ret;
+    return req.result ? req.result : ret;
 }
 
 JL_DLLEXPORT int jl_fs_close(uv_os_fd_t handle)

test/file.jl

@@ -2200,3 +2200,38 @@ end
 end
 
 @test Base.infer_return_type(stat, (String,)) == Base.Filesystem.StatStruct
+
+# Partial write() to File should return the actual number of bytes written
+if !Sys.iswindows()
+    @test let p = Pipe()
+        # Create a non-blocking pipe that will fill and return EAGAIN
+        Base.link_pipe!(p; writer_supports_async=true, reader_supports_async=true)
+        try
+            f = Base.Filesystem.File(Base._fd(p.in))
+            n_read, n_write = 0, 0
+            @sync begin
+                @async begin
+                    n_write = write(f, rand(UInt8, 1000000))
+                    close(p.in)
+                end
+                n_read = length(read(p))
+            end
+            n_write == n_read
+        finally
+            close(p)
+        end
+    end
+end
+
+# Very large read() shouldn't return EOFError because the return value was truncated.
+@test_skip let
+    FS = Base.Filesystem
+    f = FS.open("/dev/urandom", FS.JL_O_RDONLY)
+    try
+        v = Vector{UInt8}(undef, 5 * 2^30) # 5 GiB (larger than INT32_MAX)
+        read!(f, v)
+        true
+    finally
+        close(f)
+    end
+end