move processor-count core to rktio

Also, repair various problems with places in RacketCS, but make
`(processor-count)` still return 1 for now, since there are still
problems.

pkgs/racket-test/tests/racket/place-channel.rkt

@@ -244,7 +244,6 @@
     (place-channel-put p (list flx flv bs in out))
     (place-wait p))
 
-
   (let ()
     (define p1 (place ch
                       (define in (place-channel-get ch))

racket/collects/setup/parallel-build.rkt

@@ -110,7 +110,7 @@
                   (append-error cc "making" #f out err "output"))
                 ;(when last (printer (current-output-port) "made" "~a" (cc-name cc)))
                 #t]
-              [else (eprintf "Failed trying to match:\n~e\n" result-type)]))]
+              [else (eprintf "Failed trying to match:\n~s\n" result-type)]))]
         [(list _ (list 'ADD fn))
          ;; Currently ignoring queued individual files
          #f]

racket/src/cs/Makefile

@@ -111,6 +111,8 @@ linklet-demo: $(BUILDDIR)linklet.so
 LINKLET_SRCS = linklet/read.ss \
                linklet/write.ss \
                linklet/performance.ss \
+               linklet/annotation.ss \
+               linklet/compress.ss \
                linklet/db.ss
 
 $(BUILDDIR)linklet.so: linklet.sls $(LINKLET_SRCS) $(LINKLET_DEPS) $(COMPILE_FILE_DEPS)

racket/src/cs/linklet/annotation.ss

@@ -38,18 +38,26 @@
                            stripped-e))
         e)))
 
-(define sfd-cache (make-weak-hash))
+(define sfd-cache-box (unsafe-make-place-local #f))
 
 (define (source->sfd src)
-  (or (hash-ref sfd-cache src #f)
-      (let ([str (if (path? src)
-                     (path->string src)
-                     src)])
-        ;; We'll use a file-position object in source objects, so
-        ;; the sfd checksum doesn't matter
-        (let ([sfd (source-file-descriptor str 0)])
-          (hash-set! sfd-cache src sfd)
-          sfd))))
+  (let ([sfd-cache (unsafe-place-local-ref sfd-cache-box)])
+    (cond
+     [sfd-cache
+      (or (hash-ref sfd-cache src #f)
+          (let ([str (if (path? src)
+                         (path->string src)
+                         src)])
+            ;; We'll use a file-position object in source objects, so
+            ;; the sfd checksum doesn't matter
+            (let ([sfd (source-file-descriptor str 0)])
+              (hash-set! sfd-cache src sfd)
+              sfd)))]
+     [else
+      ;; There's a race here at the level of Racket threads,
+      ;; but that seems ok for setting up a cache
+      (unsafe-place-local-set! sfd-cache-box (make-weak-hash))
+      (source->sfd src)])))
 
 ;; --------------------------------------------------
 

racket/src/cs/rumble/flvector.ss

@@ -103,12 +103,13 @@
         (create-flvector bstr)))]))
 
 (define/who (shared-flvector . xs)
-  (do-flvector who xs))
+  (register-place-shared (do-flvector who xs)))
 
 (define make-shared-flvector
   (case-lambda
    [(size) (make-shared-flvector size 0.0)]
-   [(size init) (do-make-flvector 'make-shared-flvector size init)]))
+   [(size init)
+    (register-place-shared (do-make-flvector 'make-shared-flvector size init))]))
 
 ;; ----------------------------------------
 

racket/src/cs/rumble/place.ss

@@ -88,6 +88,13 @@
         (esc v)
         (#%exit v))))
 
-(define (place-shared? v)
-  #f)
+(define place-shared (make-weak-eq-hashtable))
 
+(define (place-shared? v)
+  (with-global-lock
+   (hashtable-ref place-shared v #f)))
+
+(define (register-place-shared v)
+  (with-global-lock
+   (hashtable-set! place-shared v #t))
+  v)

racket/src/cs/rumble/vector.ss

@@ -396,5 +396,11 @@
    [else
     (raise-argument-error who "vector?" v)]))
 
-(define shared-fxvector fxvector)
-(define make-shared-fxvector make-fxvector)
+(define (shared-fxvector . args)
+  (register-place-shared (apply fxvector args)))
+
+(define make-shared-fxvector
+  (case-lambda
+   [(size) (make-shared-fxvector size 0)]
+   [(size init)
+    (register-place-shared (make-fxvector size init))]))

racket/src/io/host/processor-count.rkt

@@ -0,0 +1,7 @@
+#lang racket/base
+(require "rktio.rkt")
+
+(provide processor-count)
+
+(define (processor-count)
+  1 #;(rktio_processor_count rktio))

racket/src/io/logger/main.rkt

@@ -1,6 +1,7 @@
 #lang racket/base
 (require "../common/check.rkt"
          "../host/thread.rkt"
+         "../host/place-local.rkt"
          "logger.rkt"
          "level.rkt"
          "wanted.rkt"
@@ -18,11 +19,14 @@
          log-receiver?
          make-log-receiver
          add-stderr-log-receiver!
-         add-stdout-log-receiver!)
+         add-stdout-log-receiver!
+         logger-init!)
 
-(define root-logger
+(define (make-root-logger)
   (create-logger #:topic #f #:parent #f #:propagate-filters 'none))
 
+(define-place-local root-logger (make-root-logger))
+
 (define current-logger
   (make-parameter root-logger
                   (lambda (l)
@@ -30,6 +34,10 @@
                       (raise-argument-error 'current-logger "logger?" l))
                     l)))
 
+(define (logger-init!)
+  (set! root-logger (make-root-logger))
+  (current-logger root-logger))
+
 (define (make-logger [topic #f] [parent #f] . filters)
   (unless (or (not topic) (symbol? topic))
     (raise-argument-error 'make-logger "(or/c symbol? #f)" topic))

racket/src/io/main.rkt

@@ -16,6 +16,7 @@
          "envvar/main.rkt"
          "sha/main.rkt"
          "subprocess/main.rkt"
+         "host/processor-count.rkt"
          "network/main.rkt"
          "foreign/main.rkt"
          "unsafe/main.rkt"
@@ -41,6 +42,7 @@
          (all-from-out "envvar/main.rkt")
          (all-from-out "sha/main.rkt")
          (all-from-out "subprocess/main.rkt")
+         (all-from-out "host/processor-count.rkt")
          (all-from-out "network/main.rkt")
          (all-from-out "foreign/main.rkt")
          (all-from-out "unsafe/main.rkt")
@@ -51,6 +53,7 @@
 (define (io-place-init! in-fd out-fd err-fd cust plumber)
   (sandman-place-init!)
   (rktio-place-init!)
+  (logger-init!)
   (init-current-ports! in-fd out-fd err-fd cust plumber))
 
 (module main racket/base)

racket/src/io/network/tcp-port.rkt

@@ -4,7 +4,8 @@
          "../port/port.rkt"
          "../port/input-port.rkt"
          "../port/output-port.rkt"
-         "../port/fd-port.rkt")
+         "../port/fd-port.rkt"
+         "../port/place-message.rkt")
 
 (provide open-input-output-tcp
          tcp-port?
@@ -12,7 +13,17 @@
 
 (struct tcp-data (abandon-in? abandon-out?)
   #:mutable
-  #:authentic)
+  #:authentic
+  #:property
+  prop:fd-extra-data-place-message
+  (lambda (port)
+    (if (input-port? port)
+        (lambda (fd name)
+          (open-input-fd fd name
+                         #:extra-data (tcp-data #f #t)))
+        (lambda (fd name)
+          (open-output-fd fd name
+                          #:extra-data (tcp-data #t #f))))))
 
 (define (open-input-output-tcp fd name #:close? [close? #t])
   (define refcount (box (if close? 2 3)))

racket/src/io/port/fd-port.rkt

@@ -351,13 +351,16 @@
      (define input? (input-port? port))
      (define fd-dup (dup-port-fd port))
      (define name (core-port-name port))
+     (define opener (or (fd-extra-data->opener (fd-data-extra (core-port-data port))
+                                               port)
+                        (if input?
+                            (lambda (port name) (open-input-fd port name))
+                            (lambda (port name) (open-output-fd port name)))))
      (end-atomic)
      (lambda ()
        (atomically
         (define fd (claim-dup fd-dup))
-        (if input?
-            (open-input-fd fd name)
-            (open-output-fd fd name))))]))
+        (opener fd name)))]))
 
   ;; in atomic mode
 (define (dup-port-fd port)

racket/src/io/port/place-message.rkt

@@ -1,7 +1,13 @@
 #lang racket/base
 
+;; To make certain kinds of ports allowed as a place message, a
+;; `prop:place-message` property has to chain through properties on
+;; the `data` field and, for an fd port, the `extra-data` field
+
 (provide prop:data-place-message
-         data->place-message)
+         data->place-message
+         prop:fd-extra-data-place-message
+         fd-extra-data->opener)
 
 (define-values (prop:data-place-message data-place-message? data-place-message-ref)
   (make-struct-type-property 'data-place-message))
@@ -10,3 +16,11 @@
   (if (data-place-message? data)
       ((data-place-message-ref data) port)
       #f))
+
+(define-values (prop:fd-extra-data-place-message fd-extra-data-place-message? fd-extra-data-place-message-ref)
+  (make-struct-type-property 'fd-extra-data-place-message))
+
+(define (fd-extra-data->opener extra-data port)
+  (if (fd-extra-data-place-message? extra-data)
+      ((fd-extra-data-place-message-ref extra-data) port)
+      #f))

racket/src/racket/src/future.c

@@ -20,6 +20,7 @@
 
 #include "schpriv.h"
 #include "schmach.h"
+#include "schrktio.h"
 
 static Scheme_Object *future_p(int argc, Scheme_Object *argv[])
 {
@@ -423,7 +424,6 @@ static Scheme_Object *reset_future_logs_for_tracking(int argc, Scheme_Object *ar
 static Scheme_Object *mark_future_trace_end(int argc, Scheme_Object *argv[]);
 
 READ_ONLY static int cpucount;
-static void init_cpucount(void);
 
 #ifdef MZ_PRECISE_GC
 # define scheme_future_setjmp(newbuf) scheme_jit_setjmp((newbuf).jb)
@@ -555,8 +555,6 @@ void scheme_init_futures(Scheme_Startup_Env *newenv)
 
 void scheme_init_futures_once()
 {
-  init_cpucount();
-
   REGISTER_SO(bad_multi_result_proc);
   bad_multi_result_proc = scheme_make_prim_w_arity(bad_multi_result, "bad-multi-result", 0, -1);
 }
@@ -582,6 +580,9 @@ void futures_init(void)
   Scheme_Struct_Type *stype;
   int pool_size;
 
+  if (cpucount < 1)
+    cpucount = rktio_processor_count(scheme_rktio);
+  
   fs = (Scheme_Future_State *)malloc(sizeof(Scheme_Future_State));
   memset(fs, 0, sizeof(Scheme_Future_State));
   scheme_future_state = fs;
@@ -2166,38 +2167,6 @@ Scheme_Object *touch(int argc, Scheme_Object *argv[])
   }
 }
 
-#if defined(__linux__) || defined(__QNX__)
-# include <unistd.h>
-#elif defined(OS_X)
-# include <sys/param.h>
-# include <sys/sysctl.h>
-#elif defined(DOS_FILE_SYSTEM)
-# include <windows.h>
-#endif 
-
-static void init_cpucount(void)
-/* Called in runtime thread */
-{
-#if defined(__linux__) || defined(__QNX__)
-  cpucount = sysconf(_SC_NPROCESSORS_ONLN);
-#elif defined(OS_X)
-  size_t size = sizeof(cpucount);
-
-  if (sysctlbyname("hw.ncpu", &cpucount, &size, NULL, 0))
-    cpucount = 2;
-#elif defined(DOS_FILE_SYSTEM)
-  SYSTEM_INFO sysinfo;
-  GetSystemInfo(&sysinfo);
-  cpucount = sysinfo.dwNumberOfProcessors;
-#else
-  /* Conservative guess! */
-  /* A result of 1 is not conservative, because claiming a
-     uniprocessor means that atomic cmpxchg operations are not used
-     for setting pair flags and hash codes. */
-  cpucount = 2;
-#endif
-}
-
 int scheme_is_multithreaded(int now)
 {
   if (!now)
@@ -2211,7 +2180,7 @@ int scheme_is_multithreaded(int now)
 Scheme_Object *processor_count(int argc, Scheme_Object *argv[])
 /* Called in runtime thread */
 {
-  return scheme_make_integer(cpucount);
+  return scheme_make_integer(rktio_processor_count(scheme_rktio));
 }
 
 Scheme_Object *scheme_current_future(int argc, Scheme_Object *argv[])

racket/src/rktio/Makefile.in

@@ -38,6 +38,7 @@ OBJS = rktio_fs.@LTO@ \
        rktio_flock.@LTO@ \
        rktio_shellex.@LTO@ \
        rktio_time.@LTO@ \
+       rktio_cpu.@LTO@ \
        rktio_syslog.@LTO@ \
        rktio_convert.@LTO@ \
        rktio_sha1.@LTO@ \
@@ -109,6 +110,9 @@ rktio_shellex.@LTO@: $(srcdir)/rktio_shellex.c $(RKTIO_HEADERS)
 rktio_time.@LTO@: $(srcdir)/rktio_time.c $(RKTIO_HEADERS)
 	$(CC) $(CFLAGS) -I$(srcdir) -I. -o rktio_time.@LTO@ -c $(srcdir)/rktio_time.c
 
+rktio_cpu.@LTO@: $(srcdir)/rktio_cpu.c $(RKTIO_HEADERS)
+	$(CC) $(CFLAGS) -I$(srcdir) -I. -o rktio_cpu.@LTO@ -c $(srcdir)/rktio_cpu.c
+
 rktio_syslog.@LTO@: $(srcdir)/rktio_syslog.c $(RKTIO_HEADERS)
 	$(CC) $(CFLAGS) -I$(srcdir) -I. -o rktio_syslog.@LTO@ -c $(srcdir)/rktio_syslog.c
 

racket/src/rktio/rktio.def

@@ -167,6 +167,7 @@ rktio_seconds_to_date
 rktio_shell_execute
 rktio_path_to_wide_path
 rktio_wide_path_to_path
+rktio_processor_count
 rktio_syslog
 rktio_convert_properties
 rktio_converter_open

racket/src/rktio/rktio.h

@@ -1034,6 +1034,13 @@ RKTIO_EXTERN_NOERR char *rktio_wide_path_to_path(rktio_t *rktio, const rktio_cha
    functions are useful only on Windows. The `rktio_path_to_wide_path`
    function can fail and report `RKTIO_ERROR_INVALID_PATH`. */
 
+/*************************************************/
+/* Processor count                               */
+
+RKTIO_EXTERN_NOERR int rktio_processor_count(rktio_t *rktio);
+/* Returns the number of processing units, either as CPUs, cores, or
+   hyoperthreads. */
+
 /*************************************************/
 /* Logging                                       */
 

racket/src/rktio/rktio.inc

@@ -167,6 +167,7 @@ Sforeign_symbol("rktio_seconds_to_date", (void *)rktio_seconds_to_date);
 Sforeign_symbol("rktio_shell_execute", (void *)rktio_shell_execute);
 Sforeign_symbol("rktio_path_to_wide_path", (void *)rktio_path_to_wide_path);
 Sforeign_symbol("rktio_wide_path_to_path", (void *)rktio_wide_path_to_path);
+Sforeign_symbol("rktio_processor_count", (void *)rktio_processor_count);
 Sforeign_symbol("rktio_syslog", (void *)rktio_syslog);
 Sforeign_symbol("rktio_convert_properties", (void *)rktio_convert_properties);
 Sforeign_symbol("rktio_converter_open", (void *)rktio_converter_open);

racket/src/rktio/rktio.rktl

@@ -1205,6 +1205,7 @@
  (ref char)
  rktio_wide_path_to_path
  (((ref rktio_t) rktio) ((*ref rktio_char16_t) wp)))
+(define-function () int rktio_processor_count (((ref rktio_t) rktio)))
 (define-function/errno
  #f
  ()

racket/src/rktio/rktio_cpu.c

@@ -0,0 +1,41 @@
+#include "rktio.h"
+#include "rktio_private.h"
+
+#if defined(__linux__) || defined(__QNX__)
+# include <unistd.h>
+#elif defined(OS_X)
+# include <sys/param.h>
+# include <sys/sysctl.h>
+#elif defined(DOS_FILE_SYSTEM)
+# include <windows.h>
+#endif 
+
+void rktio_init_cpu(rktio_t *rktio)
+{
+  int processor_count;
+  
+#if defined(__linux__) || defined(__QNX__)
+  processor_count = sysconf(_SC_NPROCESSORS_ONLN);
+#elif defined(OS_X)
+  size_t size = sizeof(processor_count);
+
+  if (sysctlbyname("hw.ncpu", &processor_count, &size, NULL, 0))
+    processor_count = 2;
+#elif defined(DOS_FILE_SYSTEM)
+  SYSTEM_INFO sysinfo;
+  GetSystemInfo(&sysinfo);
+  processor_count = sysinfo.dwNumberOfProcessors;
+#else
+  /* Conservative guess! */
+  /* A result of 1 is not conservative, because that's claiming a
+     uniprocessor. */
+  processor_count = 2;
+#endif
+
+  rktio->processor_count = processor_count;
+}
+
+int rktio_processor_count(rktio_t *rktio)
+{
+  return rktio->processor_count;
+}

racket/src/rktio/rktio_main.c

@@ -34,6 +34,7 @@ rktio_t *rktio_init(void)
   
   rktio_init_time(rktio);
   rktio_init_wide(rktio);
+  rktio_init_cpu(rktio);
 
   rktio_syslog_init(rktio);
 

racket/src/rktio/rktio_private.h

@@ -114,6 +114,8 @@ struct rktio_t {
 
   int pending_os_signals[RKTIO_NUM_OS_SIGNALS];
 
+  int processor_count;
+
   struct rktio_dll_t *all_dlls;
   struct rktio_hash_t *dlls_by_name;
 #ifdef RKTIO_SYSTEM_UNIX
@@ -334,6 +336,8 @@ void rktio_stop_fs_change(rktio_t *rktio);
 
 void rktio_init_time(rktio_t *rktio);
 
+void rktio_init_cpu(rktio_t *rktio);
+
 #ifdef RKTIO_SYSTEM_WINDOWS
 int rktio_winsock_init(rktio_t *rktio);
 void rktio_winsock_done(rktio_t *rktio);

racket/src/thread/future.rkt

@@ -11,7 +11,6 @@
          "lock.rkt")
 
 (provide futures-enabled?
-         processor-count
          current-future
          future
          future?
@@ -51,9 +50,6 @@
           id
           (get-next-id)))))
 
-(define (processor-count)
-  1)
-
 (define futures-enabled? threaded?)
 
 (struct future* (id cond lock prompt

racket/src/thread/main.rkt

@@ -179,7 +179,6 @@
          unsafe-add-post-custodian-shutdown
 
          futures-enabled?
-         processor-count
          future
          future?
          touch

racket/src/thread/place-message.rkt

@@ -195,7 +195,9 @@
          [else
           (for/hash ([(k v) (in-hash v)])
             (values (loop k) (loop v)))])]
-      [(cpointer? v)
+      [(and (cpointer? v)
+            v ; not #f
+            (not (bytes? v)))
        (ptr-add v 0)]
       [(message-ized? v)
        ((message-ized-unmessage v))]

racket/src/thread/place.rkt

@@ -97,11 +97,11 @@
   ;; Start the new place
   (host:fork-place
    (lambda ()
+     (set-root-custodian! orig-cust)
      (define finish (host:start-place child-pch path sym
                                       child-in-fd child-out-fd child-err-fd
                                       orig-cust orig-plumber))
      (call-in-another-main-thread
-      orig-cust
       (lambda ()
         (set! current-place new-place)
         (current-plumber orig-plumber)

racket/src/thread/schedule.rkt

@@ -33,9 +33,8 @@
   (select-thread!))
 
 ;; Initializes the thread system in a new place:
-(define (call-in-another-main-thread c thunk)
+(define (call-in-another-main-thread thunk)
   (make-another-initial-thread-group)
-  (set-root-custodian! c)
   (call-in-main-thread thunk))
 
 ;; ----------------------------------------

racket/src/worksp/librktio/librktio.vcproj

@@ -170,6 +170,10 @@
 				RelativePath="..\..\rktio\rktio_time.c"
 				>
 			</File>
+			<File
+				RelativePath="..\..\rktio\rktio_cpu.c"
+				>
+			</File>
 			<File
 				RelativePath="..\..\rktio\rktio_syslog.c"
 				>

racket/src/worksp/librktio/librktio.vcxproj

@@ -131,6 +131,7 @@
     <ClCompile Include="..\..\rktio\rktio_flock.c" />
     <ClCompile Include="..\..\rktio\rktio_shellex.c" />
     <ClCompile Include="..\..\rktio\rktio_time.c" />
+    <ClCompile Include="..\..\rktio\rktio_cpu.c" />
     <ClCompile Include="..\..\rktio\rktio_syslog.c" />
     <ClCompile Include="..\..\rktio\rktio_convert.c" />
     <ClCompile Include="..\..\rktio\rktio_sha1.c" />