unix: add Cstruct_unix.{read,write,writev,send,recv}

unix/dune

@@ -2,4 +2,8 @@
  (name cstruct_unix)
  (wrapped false)
  (public_name cstruct-unix)
+ (foreign_stubs
+  (language c)
+  (names read_stubs write_stubs writev_stubs send_stubs recv_stubs
+         recvfrom_stubs sendto_stubs))
  (libraries cstruct unix))

unix/read_stubs.c

@@ -0,0 +1,70 @@
+#include <caml/mlvalues.h>
+#include <caml/memory.h>
+#include <caml/custom.h>
+#include <caml/callback.h>
+#include <caml/alloc.h>
+#include <caml/unixsupport.h>
+#include <caml/bigarray.h>
+#include <caml/threads.h>
+
+#include <stdio.h>
+#include <errno.h>
+
+CAMLprim value stub_cstruct_read(value val_fd, value val_c)
+{
+  CAMLparam2(val_fd, val_c);
+  CAMLlocal3(val_buf, val_ofs, val_len);
+
+  val_buf = Field(val_c, 0);
+  val_ofs = Field(val_c, 1);
+  val_len = Field(val_c, 2);
+
+  void *buf = (char *)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
+  size_t len = Long_val(val_len);
+  int n = 0;
+
+#ifdef _WIN32
+  int win32err = 0;
+  switch (Descr_kind_val(val_fd))
+  {
+  case KIND_SOCKET:
+    SOCKET s = Socket_val(val_fd);
+
+    caml_release_runtime_system();
+    n = recv(s, buf, len, 0);
+    win32err = WSAGetLastError();
+    caml_acquire_runtime_system();
+
+    if (n == SOCKET_ERROR)
+    {
+      win32_maperr(win32err);
+      unix_error(errno, "read", Nothing);
+    }
+    break;
+  case KIND_HANDLE:
+    HANDLE h = Handle_val(val_fd);
+    DWORD numread;
+    caml_release_runtime_system();
+    int ok = ReadFile(h, buf, len, &numread, NULL);
+    win32err = GetLastError();
+    n = numread;
+    caml_acquire_runtime_system();
+
+    if (!ok)
+    {
+      win32_maperr(win32err);
+      unix_error(errno, "read", Nothing);
+    }
+    break;
+  default:
+    caml_failwith("unknown Descr_kind_val");
+  }
+#else
+  caml_release_runtime_system();
+  n = read(Int_val(val_fd), buf, len);
+  caml_acquire_runtime_system();
+  if (n < 0)
+    unix_error(errno, "read", Nothing);
+#endif
+  CAMLreturn(Val_int(n));
+}

unix/recv_stubs.c

@@ -0,0 +1,63 @@
+#include <caml/mlvalues.h>
+#include <caml/memory.h>
+#include <caml/custom.h>
+#include <caml/callback.h>
+#include <caml/alloc.h>
+#include <caml/unixsupport.h>
+#include <caml/bigarray.h>
+#include <caml/threads.h>
+
+#include <stdio.h>
+
+#ifdef WIN32
+#define WIN32_LEAN_AND_MEAN
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include <NTSecAPI.h>
+#else
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <errno.h>
+#endif
+
+CAMLprim value stub_cstruct_recv(value val_fd, value val_c)
+{
+    CAMLparam2(val_fd, val_c);
+    CAMLlocal3(val_buf, val_ofs, val_len);
+
+    val_buf = Field(val_c, 0);
+    val_ofs = Field(val_c, 1);
+    val_len = Field(val_c, 2);
+
+    void *buf = (void *)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
+    size_t len = (size_t)Long_val(val_len);
+    int n = 0;
+#ifdef WIN32
+    int win32err = 0;
+    if (Descr_kind_val(val_fd) != KIND_SOCKET)
+        unix_error(EINVAL, "stub_cstruct_recv", Nothing);
+
+    SOCKET s = Socket_val(val_fd);
+
+    caml_release_runtime_system();
+    n = recv(s, buf, len, 0);
+    win32err = WSAGetLastError();
+    caml_acquire_runtime_system();
+
+    if (n == SOCKET_ERROR)
+    {
+        win32_maperr(win32err);
+        unix_error(errno, "recv", Nothing);
+    }
+#else
+    int fd = Int_val(val_fd);
+
+    caml_release_runtime_system();
+    n = recv(fd, buf, len, 0);
+    caml_acquire_runtime_system();
+
+    if (n < 0)
+        unix_error(errno, "recv", Nothing);
+#endif
+    CAMLreturn(Val_int(n));
+}

unix/recvfrom_stubs.c

@@ -0,0 +1,50 @@
+#include <caml/mlvalues.h>
+#include <caml/memory.h>
+#include <caml/custom.h>
+#include <caml/callback.h>
+#include <caml/alloc.h>
+#include <caml/unixsupport.h>
+#include <caml/bigarray.h>
+#include <caml/threads.h>
+#include <caml/socketaddr.h>
+
+#include <sys/socket.h>
+
+static int msg_flag_table[] = {
+  MSG_OOB, MSG_DONTROUTE, MSG_PEEK /* XXX */
+};
+
+CAMLprim value stub_cstruct_recvfrom(value val_fd, value val_c, value val_flags)
+{
+  CAMLparam3(val_fd, val_c, val_flags);
+  CAMLlocal5(val_buf, val_ofs, val_len, val_addr, val_res);
+  uint8_t *buf;
+  size_t len;
+  ssize_t n;
+  int cv_flags;
+  union sock_addr_union addr;
+  socklen_param_type addr_len;
+
+  val_buf = Field(val_c, 0);
+  val_ofs = Field(val_c, 1);
+  val_len = Field(val_c, 2);
+  cv_flags = caml_convert_flag_list(val_flags, msg_flag_table);
+
+  buf = (uint8_t *)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
+  len = Long_val(val_len);
+  addr_len = sizeof(addr);
+
+  caml_release_runtime_system();
+  n = recvfrom(Int_val(val_fd), buf, len, cv_flags, &addr.s_gen, &addr_len);
+  caml_acquire_runtime_system();
+
+  if (n == -1)
+    caml_uerror("recvfrom", Nothing);
+
+  val_addr = caml_unix_alloc_sockaddr(&addr, addr_len, -1);
+  val_res = caml_alloc_small(2, 0);
+  Field(val_res, 0) = Val_int(n);
+  Field(val_res, 1) = val_addr;
+
+  CAMLreturn (val_res);
+}

unix/send_stubs.c

@@ -0,0 +1,62 @@
+#include <caml/mlvalues.h>
+#include <caml/memory.h>
+#include <caml/custom.h>
+#include <caml/callback.h>
+#include <caml/alloc.h>
+#include <caml/unixsupport.h>
+#include <caml/bigarray.h>
+#include <caml/threads.h>
+
+#include <stdio.h>
+
+#ifdef WIN32
+#define WIN32_LEAN_AND_MEAN
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include <NTSecAPI.h>
+#else
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <errno.h>
+#endif
+
+CAMLprim value stub_cstruct_send(value val_fd, value val_c)
+{
+    CAMLparam2(val_fd, val_c);
+    CAMLlocal3(val_buf, val_ofs, val_len);
+
+    val_buf = Field(val_c, 0);
+    val_ofs = Field(val_c, 1);
+    val_len = Field(val_c, 2);
+
+    const char *buf = (char *)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
+    size_t len = (size_t)Long_val(val_len);
+    int n = 0;
+
+#ifdef WIN32
+    int win32err = 0;
+    if (Descr_kind_val(val_fd) != KIND_SOCKET)
+        unix_error(EINVAL, "send", Nothing);
+
+    SOCKET s = Socket_val(val_fd);
+    caml_release_runtime_system();
+    n = send(s, buf, len, 0);
+    win32err = WSAGetLastError();
+    caml_acquire_runtime_system();
+
+    if (n == SOCKET_ERROR)
+    {
+        win32_maperr(win32err);
+        unix_error(errno, "send", Nothing);
+    }
+#else
+    int fd = Int_val(val_fd);
+
+    caml_release_runtime_system();
+    n = send(fd, buf, len, 0);
+    caml_acquire_runtime_system();
+    if (n < 0)
+        unix_error(errno, "send", Nothing);
+#endif
+    CAMLreturn(Val_int(n));
+}

unix/sendto_stubs.c

@@ -0,0 +1,45 @@
+#include <caml/mlvalues.h>
+#include <caml/memory.h>
+#include <caml/custom.h>
+#include <caml/callback.h>
+#include <caml/alloc.h>
+#include <caml/unixsupport.h>
+#include <caml/bigarray.h>
+#include <caml/threads.h>
+#include <caml/socketaddr.h>
+
+#include <sys/socket.h>
+
+static int msg_flag_table[] = { /* XXX */
+  MSG_OOB, MSG_DONTROUTE, MSG_PEEK
+};
+
+CAMLprim value stub_cstruct_sendto(value val_fd, value val_c, value val_flags, value val_daddr)
+{
+  CAMLparam4(val_fd, val_c, val_flags, val_daddr);
+  CAMLlocal5(val_buf, val_ofs, val_len, val_addr, val_res);
+  union sock_addr_union addr;
+  socklen_param_type addr_len;
+  uint8_t *buf;
+  size_t len;
+  ssize_t n;
+  int cv_flags;
+
+  val_buf = Field(val_c, 0);
+  val_ofs = Field(val_c, 1);
+  val_len = Field(val_c, 2);
+
+  buf = (uint8_t *)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
+  len = Long_val(val_len);
+  caml_unix_get_sockaddr(val_daddr, &addr, &addr_len);
+  cv_flags = caml_convert_flag_list(val_flags, msg_flag_table);
+
+  caml_release_runtime_system();
+  n = sendto(Int_val(val_fd), buf, len, cv_flags, &addr.s_gen, addr_len);
+  caml_acquire_runtime_system();
+
+  if (n == -1)
+    caml_uerror("sendto", Nothing);
+
+  CAMLreturn (Val_int(n));
+}

unix/unix_cstruct.ml

@@ -17,3 +17,68 @@
 let of_fd fd =
   let buffer = Bigarray.(array1_of_genarray (Unix.map_file fd char c_layout false [|-1|])) in
   Cstruct.of_bigarray buffer
+
+type buffer = (char, Bigarray.int8_unsigned_elt, Bigarray.c_layout) Bigarray.Array1.t
+
+(* Returns 0 if there is no writev *)
+external stub_iov_max: unit -> int = "stub_cstruct_iov_max"
+
+external stub_write: Unix.file_descr -> (buffer * int * int) -> int = "stub_cstruct_write"
+
+external stub_writev: Unix.file_descr -> (buffer * int * int) list -> int = "stub_cstruct_writev"
+
+let iov_max = stub_iov_max ()
+
+(* return the first n fragments, suitable for writev *)
+let rec first n rev_acc = function
+| [] -> List.rev rev_acc
+| _ when n = 0 -> List.rev rev_acc
+| x :: xs -> first (n - 1) (x :: rev_acc) xs
+
+(* shift a list of fragments along by n bytes *)
+let rec shift t x =
+  if x = 0 then t else match t with
+  | [] -> invalid_arg "foo"
+  | y :: ys ->
+    let y' = Cstruct.length y in
+    if y' > x
+    then Cstruct.shift y x :: ys
+    else shift ys (x - y')
+
+let rec write fd buf =
+  if Cstruct.length buf > 0 then begin
+    let n = stub_write fd (buf.Cstruct.buffer, buf.Cstruct.off, buf.Cstruct.len) in
+    write fd @@ Cstruct.shift buf n
+  end
+
+let writev fd bufs =
+  let rec use_writev = function
+    | [] -> ()
+    | remaining ->
+      (* write at most iov_max at a time *)
+      let to_send = first iov_max [] remaining in
+      let n = stub_writev fd (List.map (fun x -> x.Cstruct.buffer, x.Cstruct.off, x.Cstruct.len) to_send) in
+      let rest = shift remaining n in
+      use_writev rest in
+  let use_write_fallback = List.iter (write fd) in
+  (if iov_max = 0 then use_write_fallback else use_writev) bufs
+
+external stub_send: Unix.file_descr -> (buffer * int * int) -> int = "stub_cstruct_send"
+
+external stub_recv: Unix.file_descr -> (buffer * int * int) -> int = "stub_cstruct_recv"
+
+let send fd x = stub_send fd (x.Cstruct.buffer, x.Cstruct.off, x.Cstruct.len)
+
+let recv fd x = stub_recv fd (x.Cstruct.buffer, x.Cstruct.off, x.Cstruct.len)
+
+external stub_read: Unix.file_descr -> (buffer * int * int) -> int = "stub_cstruct_read"
+
+let read fd x = stub_read fd (x.Cstruct.buffer, x.Cstruct.off, x.Cstruct.len)
+
+external stub_recvfrom : Unix.file_descr -> Cstruct.t -> Unix.msg_flag list -> int * Unix.sockaddr = "stub_cstruct_recvfrom"
+
+let recvfrom fd x fl = stub_recvfrom fd x fl
+
+external stub_sendto : Unix.file_descr -> Cstruct.t -> Unix.msg_flag list -> Unix.sockaddr -> int = "stub_cstruct_sendto"
+
+let sendto fd x fl a = stub_sendto fd x fl a