Add Lwt_unix.pread and pwrite (#768)

Resolves #767.

Author: chambart

src/unix/dune

@@ -45,11 +45,15 @@
   windows_get_page_size
   unix_mincore
   unix_read
+  unix_pread
   windows_read
+  windows_pread
   unix_bytes_read
   windows_bytes_read
   unix_write
+  unix_pwrite
   windows_write
+  windows_pwrite
   unix_bytes_write
   windows_bytes_write
   unix_readv_writev_utils
@@ -84,11 +88,15 @@
   unix_wait_mincore_job
   unix_open_job
   unix_read_job
+  unix_pread_job
   windows_read_job
+  windows_pread_job
   unix_bytes_read_job
   windows_bytes_read_job
   unix_write_job
   windows_write_job
+  unix_pwrite_job
+  windows_pwrite_job
   unix_bytes_write_job
   windows_bytes_write_job
   unix_stat_job_utils

src/unix/lwt_unix.cppo.ml

@@ -631,6 +631,12 @@ let wait_read ch =
 
 external stub_read : Unix.file_descr -> Bytes.t -> int -> int -> int = "lwt_unix_read"
 external read_job : Unix.file_descr -> Bytes.t -> int -> int -> int job = "lwt_unix_read_job"
+external stub_pread :
+  Unix.file_descr -> Bytes.t -> file_offset:int -> int -> int -> int =
+    "lwt_unix_pread"
+external pread_job :
+  Unix.file_descr -> Bytes.t -> file_offset:int -> int -> int -> int job =
+    "lwt_unix_pread_job"
 
 let read ch buf pos len =
   if pos < 0 || len < 0 || pos > Bytes.length buf - len then
@@ -643,6 +649,17 @@ let read ch buf pos len =
     | false ->
       wrap_syscall Read ch (fun () -> stub_read ch.fd buf pos len)
 
+let pread ch buf ~file_offset pos len =
+  if pos < 0 || len < 0 || pos > Bytes.length buf - len then
+    invalid_arg "Lwt_unix.pread"
+  else
+    Lazy.force ch.blocking >>= function
+    | true ->
+      wait_read ch >>= fun () ->
+      run_job (pread_job ch.fd buf ~file_offset pos len)
+    | false ->
+      wrap_syscall Read ch (fun () -> stub_pread ch.fd buf ~file_offset pos len)
+
 external stub_read_bigarray :
   Unix.file_descr -> bigarray -> int -> int -> int = "lwt_unix_bytes_read"
 external read_bigarray_job :
@@ -672,6 +689,12 @@ let wait_write ch =
 
 external stub_write : Unix.file_descr -> Bytes.t -> int -> int -> int = "lwt_unix_write"
 external write_job : Unix.file_descr -> Bytes.t -> int -> int -> int job = "lwt_unix_write_job"
+external stub_pwrite :
+  Unix.file_descr -> Bytes.t -> file_offset:int -> int -> int -> int =
+    "lwt_unix_pwrite"
+external pwrite_job :
+  Unix.file_descr -> Bytes.t -> file_offset:int -> int -> int -> int job =
+    "lwt_unix_pwrite_job"
 
 let write ch buf pos len =
   if pos < 0 || len < 0 || pos > Bytes.length buf - len then
@@ -684,10 +707,25 @@ let write ch buf pos len =
     | false ->
       wrap_syscall Write ch (fun () -> stub_write ch.fd buf pos len)
 
+let pwrite ch buf ~file_offset pos len =
+  if pos < 0 || len < 0 || pos > Bytes.length buf - len then
+    invalid_arg "Lwt_unix.pwrite"
+  else
+    Lazy.force ch.blocking >>= function
+    | true ->
+      wait_write ch >>= fun () ->
+      run_job (pwrite_job ch.fd buf ~file_offset pos len)
+    | false ->
+      wrap_syscall Write ch (fun () -> stub_pwrite ch.fd buf ~file_offset pos len)
+
 let write_string ch buf pos len =
   let buf = Bytes.unsafe_of_string buf in
   write ch buf pos len
 
+let pwrite_string ch buf ~file_offset pos len =
+  let buf = Bytes.unsafe_of_string buf in
+  pwrite ch buf ~file_offset pos len
+
 external stub_write_bigarray :
   Unix.file_descr -> bigarray -> int -> int -> int = "lwt_unix_bytes_write"
 external write_bigarray_job :

src/unix/lwt_unix.cppo.mli

@@ -298,6 +298,17 @@ val read : file_descr -> bytes -> int -> int -> int Lwt.t
     except [Unix.Unix_error Unix.EAGAIN], [Unix.Unix_error Unix.EWOULDBLOCK] or
     [Unix.Unix_error Unix.EINTR]. *)
 
+val pread : file_descr -> bytes -> file_offset:int -> int -> int -> int Lwt.t
+(** [pread fd buf ~file_offset ofs len] on file descriptors allowing seek,
+    reads up to [len] bytes from [fd] at offset [file_offset] from the
+    beginning of the file, and writes them to [buf], starting at offset [ofs].
+
+    On Unix systems, the file descriptor position is unaffected. On Windows
+    it is changed to be just after the last read position.
+
+    The thread can fail with any exception that can be raised by [read] or
+    [lseek]. *)
+
 val write : file_descr -> bytes -> int -> int -> int Lwt.t
 (** [write fd buf ofs len] writes up to [len] bytes to [fd] from [buf], starting
     at buffer offset [ofs]. The function immediately evaluates to an Lwt thread,
@@ -315,9 +326,25 @@ val write : file_descr -> bytes -> int -> int -> int Lwt.t
     [Unix.single_write], except [Unix.Unix_error Unix.EAGAIN],
     [Unix.Unix_error Unix.EWOULDBLOCK] or [Unix.Unix_error Unix.EINTR]. *)
 
+val pwrite : file_descr -> bytes -> file_offset:int -> int -> int -> int Lwt.t
+(** [pwrite fd buf ~file_offset ofs len] on file descriptors allowing seek,
+    writes up to [len] bytes to [fd] from [buf], starting at buffer offset
+    [ofs]. The data is written at offset [file_offset] from the beginning
+    of [fd].
+
+    On Unix systems, the file descriptor position is unaffected. On Windows
+    it is changed to be just after the last written position.
+
+    The thread can fail with any exception that can be raised by [write] or
+    [lseek]. *)
+
 val write_string : file_descr -> string -> int -> int -> int Lwt.t
   (** See {!write}. *)
 
+val pwrite_string :
+  file_descr -> string -> file_offset:int -> int -> int -> int Lwt.t
+  (** See {!pwrite}. *)
+
 (** Sequences of buffer slices for {!writev}. *)
 module IO_vectors :
 sig

src/unix/unix_c/unix_pread.c

@@ -0,0 +1,22 @@
+/* This file is part of Lwt, released under the MIT license. See LICENSE.md for
+   details, or visit https://github.com/ocsigen/lwt/blob/master/LICENSE.md. */
+
+
+
+#include "lwt_config.h"
+
+#if !defined(LWT_ON_WINDOWS)
+
+#include <caml/mlvalues.h>
+#include <caml/unixsupport.h>
+
+CAMLprim value lwt_unix_pread(value val_fd, value val_buf, value val_file_ofs,
+                              value val_ofs, value val_len)
+{
+    long ret;
+    ret = pread(Int_val(val_fd), &Byte(String_val(val_buf), Long_val(val_ofs)),
+                Long_val(val_len), Long_val(val_file_ofs));
+    if (ret == -1) uerror("pread", Nothing);
+    return Val_long(ret);
+}
+#endif

src/unix/unix_c/unix_pread_job.c

@@ -0,0 +1,80 @@
+/* This file is part of Lwt, released under the MIT license. See LICENSE.md for
+   details, or visit https://github.com/ocsigen/lwt/blob/master/LICENSE.md. */
+
+
+
+#include "lwt_config.h"
+
+#if !defined(LWT_ON_WINDOWS)
+
+#include <caml/alloc.h>
+#include <caml/memory.h>
+#include <caml/mlvalues.h>
+#include <caml/unixsupport.h>
+#include <caml/version.h>
+#include <errno.h>
+#include <string.h>
+
+#include "lwt_unix.h"
+
+#if OCAML_VERSION < 40600
+#define Bytes_val(x) String_val(x)
+#endif
+
+struct job_pread {
+    struct lwt_unix_job job;
+    /* The file descriptor. */
+    int fd;
+    /* The amount of data to read. */
+    long length;
+    /* The offset in the file */
+    off_t file_offset;
+    /* The OCaml string. */
+    value string;
+    /* The offset in the string. */
+    long offset;
+    /* The result of the pread syscall. */
+    long result;
+    /* The value of errno. */
+    int error_code;
+    /* The temporary buffer. */
+    char buffer[];
+};
+
+static void worker_pread(struct job_pread *job)
+{
+    job->result = pread(job->fd, job->buffer, job->length, job->file_offset);
+    job->error_code = errno;
+}
+
+static value result_pread(struct job_pread *job)
+{
+    long result = job->result;
+    if (result < 0) {
+        int error_code = job->error_code;
+        caml_remove_generational_global_root(&(job->string));
+        lwt_unix_free_job(&job->job);
+        unix_error(error_code, "pread", Nothing);
+    } else {
+        memcpy(Bytes_val(job->string) + job->offset, job->buffer, result);
+        caml_remove_generational_global_root(&(job->string));
+        lwt_unix_free_job(&job->job);
+        return Val_long(result);
+    }
+}
+
+CAMLprim value lwt_unix_pread_job(value val_fd, value val_buffer,
+                                  value val_file_offset, value val_offset,
+                                  value val_length)
+{
+    long length = Long_val(val_length);
+    LWT_UNIX_INIT_JOB(job, pread, length);
+    job->fd = Int_val(val_fd);
+    job->length = length;
+    job->file_offset = Long_val(val_file_offset);
+    job->string = val_buffer;
+    job->offset = Long_val(val_offset);
+    caml_register_generational_global_root(&(job->string));
+    return lwt_unix_alloc_job(&(job->job));
+}
+#endif

src/unix/unix_c/unix_pwrite.c

@@ -0,0 +1,23 @@
+/* This file is part of Lwt, released under the MIT license. See LICENSE.md for
+   details, or visit https://github.com/ocsigen/lwt/blob/master/LICENSE.md. */
+
+
+
+#include "lwt_config.h"
+
+#if !defined(LWT_ON_WINDOWS)
+
+#include <caml/mlvalues.h>
+#include <caml/unixsupport.h>
+#include <unistd.h>
+
+CAMLprim value lwt_unix_pwrite(value val_fd, value val_buf, value val_file_ofs,
+                               value val_ofs, value val_len)
+{
+    long ret;
+    ret = pwrite(Int_val(val_fd), &Byte(String_val(val_buf), Long_val(val_ofs)),
+                 Long_val(val_len), Long_val(val_file_ofs));
+    if (ret == -1) uerror("pwrite", Nothing);
+    return Val_long(ret);
+}
+#endif

src/unix/unix_c/unix_pwrite_job.c

@@ -0,0 +1,53 @@
+/* This file is part of Lwt, released under the MIT license. See LICENSE.md for
+   details, or visit https://github.com/ocsigen/lwt/blob/master/LICENSE.md. */
+
+
+
+#include "lwt_config.h"
+
+#if !defined(LWT_ON_WINDOWS)
+
+#include <caml/mlvalues.h>
+#include <caml/unixsupport.h>
+#include <errno.h>
+#include <string.h>
+
+#include "lwt_unix.h"
+
+struct job_pwrite {
+    struct lwt_unix_job job;
+    int fd;
+    long length;
+    off_t file_offset;
+    long result;
+    int error_code;
+    char buffer[];
+};
+
+static void worker_pwrite(struct job_pwrite *job)
+{
+    job->result = pwrite(job->fd, job->buffer, job->length, job->file_offset);
+    job->error_code = errno;
+}
+
+static value result_pwrite(struct job_pwrite *job)
+{
+    long result = job->result;
+    LWT_UNIX_CHECK_JOB(job, result < 0, "pwrite");
+    lwt_unix_free_job(&job->job);
+    return Val_long(result);
+}
+
+CAMLprim value lwt_unix_pwrite_job(value val_fd, value val_string,
+                                   value val_file_offset, value val_offset,
+                                   value val_length)
+{
+    long length = Long_val(val_length);
+    LWT_UNIX_INIT_JOB(job, pwrite, length);
+    job->fd = Int_val(val_fd);
+    job->length = length;
+    job->file_offset = Long_val(val_file_offset);
+    memcpy(job->buffer, String_val(val_string) + Long_val(val_offset), length);
+    return lwt_unix_alloc_job(&(job->job));
+}
+#endif

src/unix/windows_c/windows_pread.c

@@ -0,0 +1,50 @@
+/* This file is part of Lwt, released under the MIT license. See LICENSE.md for
+   details, or visit https://github.com/ocsigen/lwt/blob/master/LICENSE.md. */
+
+
+
+#include "lwt_config.h"
+
+#if defined(LWT_ON_WINDOWS)
+
+#include <caml/fail.h>
+#include <caml/memory.h>
+#include <caml/mlvalues.h>
+#include <caml/unixsupport.h>
+
+CAMLprim value lwt_unix_pread(value fd, value buf, value vfile_offset,
+                              value vofs, value vlen)
+{
+    intnat ofs, len, file_offset, written;
+    DWORD numbytes, numwritten;
+    DWORD err = 0;
+
+    Begin_root(buf);
+    ofs = Long_val(vofs);
+    len = Long_val(vlen);
+    file_offset = Long_val(vfile_offset);
+    written = 0;
+    if (len > 0) {
+        numbytes = len;
+        if (Descr_kind_val(fd) == KIND_SOCKET) {
+            caml_invalid_argument("Lwt_unix.pread");
+        } else {
+            HANDLE h = Handle_val(fd);
+            OVERLAPPED overlapped;
+            memset( &overlapped, 0, sizeof(overlapped));
+            overlapped.OffsetHigh = (DWORD)(file_offset >> 32);
+            overlapped.Offset = (DWORD)(file_offset & 0xFFFFFFFFLL);
+            if (!ReadFile(h, &Byte(buf, ofs), numbytes, &numwritten,
+                          &overlapped))
+                err = GetLastError();
+        }
+        if (err) {
+            win32_maperr(err);
+            uerror("pread", Nothing);
+        }
+        written = numwritten;
+    }
+    End_roots();
+    return Val_long(written);
+}
+#endif