v0.11.120.00+145

src/core_thread.ml

@@ -56,3 +56,35 @@ let num_threads () =
 let block_forever () =
   Event.sync (Event.receive (Event.new_channel ()))
 ;;
+
+[%%import "config.h"]
+
+[%%ifdef JSC_PTHREAD_NP]
+external setaffinity_self_exn : int Array.t -> unit        = "pthread_np_setaffinity_self"
+external getaffinity_self_exn : unit        -> int Array.t = "pthread_np_getaffinity_self"
+
+let setaffinity_self_exn =
+  let setaffinity_self_exn cpuset =
+    setaffinity_self_exn (Int.Set.to_array cpuset)
+  in
+  Ok setaffinity_self_exn
+;;
+
+let getaffinity_self_exn =
+  let getaffinity_self_exn () =
+    Int.Set.of_array (getaffinity_self_exn ())
+  in
+  Ok getaffinity_self_exn
+;;
+[%%else]
+
+let not_supported name =
+  Error.of_string
+    (sprintf
+       "%s: non-portable pthread extension is not supported on this platform"
+       name)
+;;
+
+let setaffinity_self_exn = Error (not_supported "pthread_setaffinity_np")
+let getaffinity_self_exn = Error (not_supported "pthread_getaffinity_np")
+[%%endif]

src/core_thread.mli

@@ -8,17 +8,15 @@ type t [@@deriving sexp_of]
 (** {6 Thread creation and termination} *)
 
 val create : ('a -> 'b) -> 'a -> t
-(** [Thread.create funct arg] creates a new thread of control,
-    in which the function application [funct arg]
-    is executed concurrently with the other threads of the program.
-    The application of [Thread.create]
-    returns the handle of the newly created thread.
-    The new thread terminates when the application [funct arg]
-    returns, either normally or by raising an uncaught exception.
-    In the latter case, the exception is printed on standard error,
-    but not propagated back to the parent thread. Similarly, the
-    result of the application [funct arg] is discarded and not
-    directly accessible to the parent thread. *)
+(** [Thread.create funct arg] creates a new thread of control, in which the
+    function application [funct arg] is executed concurrently with the other
+    threads of the program. The application of [Thread.create] returns the
+    handle of the newly created thread. The new thread terminates when the
+    application [funct arg] returns, either normally or by raising an uncaught
+    exception. In the latter case, the exception is printed on standard error,
+    but not propagated back to the parent thread. Similarly, the result of the
+    application [funct arg] is discarded and not directly accessible to the
+    parent thread. *)
 
 val self : unit -> t
 (** Return the thread currently executing. *)
@@ -119,3 +117,40 @@ val num_threads : unit -> int option
 
 (** [block_forever ()] will block the calling thread forever. *)
 val block_forever : unit -> 'a
+
+(** {2 Non-portable pthread extensions}
+
+    The following operations may not be supported on all platforms. Before you
+    can use them, you must first check that they do not contain error values.
+    For example, if you wanted to use [setaffinity_self_exn] then you would
+    first do:
+
+    {[
+      let setaffinity_self_exn =
+        match Thread.setaffinity_self_exn with
+        | Ok f -> f
+        | Error err -> (* raise or provide a default implementation. *)
+    ]}
+
+    If your application requires that one of these operations be present then,
+    you could just do this instead:
+
+    {[
+      let setaffinity_self_exn = Or_error.ok_exn Thread.setaffinity_self_exn
+    ]}
+*)
+
+(** Sets the core affinity of the currently-running thread to the set
+    specified.
+
+    This function is implemented using [pthread_setaffinity_np(3)], when
+    available. See the man page for situations when this function may return an
+    error, and therefore raise. *)
+val setaffinity_self_exn : (Int.Set.t -> unit) Or_error.t
+
+(** Gets the core affinity of the currently-running thread.
+
+    This function is implemented using [pthread_getaffinity_np(3)], when
+    available. See the man page for situations when this function may return an
+    error, and therefore raise. *)
+val getaffinity_self_exn : (unit      -> Int.Set.t) Or_error.t

src/jbuild

@@ -20,6 +20,7 @@
     iobuf_stubs
     linux_ext_stubs
     ocaml_utils_stubs
+    pthread_np_stubs
     recvmmsg
     signal_stubs
     time_stamp_counter_stubs

src/pthread_np_stubs.c

@@ -0,0 +1,69 @@
+#define _GNU_SOURCE
+
+#include "config.h"
+#include "ocaml_utils.h"
+
+#ifdef JSC_PTHREAD_NP
+
+#include <string.h>
+#include <pthread.h>
+#include <assert.h>
+
+#if defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__) || defined(__APPLE__)
+#include <pthread_np.h>
+#include <sys/cpuset.h>
+#else
+#include <sched.h>
+#endif
+
+CAMLprim value pthread_np_setaffinity_self(value cpuids)
+{
+  int result;
+  mlsize_t length, i;
+  cpu_set_t cpuset;
+  CPU_ZERO(&cpuset);
+
+  length = Wosize_val(cpuids);
+  for (i = 0; i < length; i++) 
+  {
+    CPU_SET(Int_val(Field(cpuids, i)), &cpuset);
+  }
+
+  result = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
+  if (result < 0) 
+  {
+    uerror("pthread_setaffinity_np", Nothing);
+  }
+  return Val_unit;
+}
+
+CAMLprim value pthread_np_getaffinity_self()
+{
+  CAMLparam0();
+  CAMLlocal1(v_cpus);
+  int result;
+  mlsize_t cpu_count, i;
+  cpu_set_t cpuset;
+
+  CPU_ZERO(&cpuset);
+
+  result = pthread_getaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
+  if (result < 0)
+  {
+    uerror("pthread_getaffinity_np", Nothing);
+  }
+
+  cpu_count = CPU_COUNT(&cpuset);
+  v_cpus = caml_alloc_tuple(cpu_count);
+
+  for (i = 0; i < CPU_SETSIZE; i++)
+  {
+    if (CPU_ISSET(i, &cpuset))
+    {
+      assert(cpu_count >= 1);
+      Store_field(v_cpus, --cpu_count, Val_long(i));
+    }
+  }
+  CAMLreturn(v_cpus);
+}
+#endif /* JSC_PTHREAD_NP */