rktio: repairs to kqueue-based fs-change events

pkgs/racket-doc/scribblings/inside/ports.scrbl

@@ -792,8 +792,10 @@ Whether @var{close} is zero or not, closing the resulting ports
  So, passing zero for @var{close} and also using the file descriptor
  with other ports or with @cpp{scheme_fd_to_semaphore} will not work right.
 
-@history["6.9.0.6" @elem{Changed ports to always unregister with @cpp{scheme_fd_to_semaphore},
+@history[#:changed "6.9.0.6" @elem{Changed ports to always unregister
-                         since it's not safe to skip that step.}]}
+                                   with @cpp{scheme_fd_to_semaphore},
+                                   since it's not safe to skip that
+                                   step.}]}
 
 
 @function[(Scheme_Object* scheme_fd_to_semaphore

racket/src/racket/include/schthread.h

@@ -195,6 +195,7 @@ typedef struct Thread_Local_Variables {
   struct Scheme_Object *scheme_orig_stderr_port_;
   struct Scheme_Object *scheme_orig_stdin_port_;
   struct rktio_ltps_t *scheme_semaphore_fd_set_;
+  struct Scheme_Object *fs_change_props_;
   struct Scheme_Custodian *new_port_cust_;
   char *read_string_byte_buffer_;
   struct ITimer_Data *itimerdata_;
@@ -589,6 +590,7 @@ XFORM_GC_VARIABLE_STACK_THROUGH_THREAD_LOCAL;
 #define scheme_orig_stderr_port XOA (scheme_get_thread_local_variables()->scheme_orig_stderr_port_)
 #define scheme_orig_stdin_port XOA (scheme_get_thread_local_variables()->scheme_orig_stdin_port_)
 #define scheme_semaphore_fd_set XOA (scheme_get_thread_local_variables()->scheme_semaphore_fd_set_)
+#define fs_change_props XOA (scheme_get_thread_local_variables()->fs_change_props_)
 #define new_port_cust XOA (scheme_get_thread_local_variables()->new_port_cust_)
 #define read_string_byte_buffer XOA (scheme_get_thread_local_variables()->read_string_byte_buffer_)
 #define itimerdata XOA (scheme_get_thread_local_variables()->itimerdata_)

racket/src/racket/src/env.c

@@ -534,6 +534,7 @@ static Scheme_Env *place_instance_init(void *stack_base, int initial_main_os_thr
   scheme_init_error_escape_proc(NULL);
   scheme_init_print_buffers_places();
   scheme_init_thread_places();
+  scheme_init_fd_semaphores();
   scheme_init_string_places();
   scheme_init_logger();
   scheme_init_eval_places();
@@ -541,7 +542,6 @@ static Scheme_Env *place_instance_init(void *stack_base, int initial_main_os_thr
   scheme_init_regexp_places();
   scheme_init_sema_places();
   scheme_init_gmp_places();
-  scheme_init_fd_semaphores();
 #ifndef DONT_USE_FOREIGN
   scheme_init_foreign_places();
 #endif
@@ -647,10 +647,6 @@ void scheme_place_instance_destroy(int force)
   else
     scheme_run_atexit_closers_on_all(force_more_closed_after);
 
-#ifdef WINDOWS_PROCESSES
-  scheme_release_process_job_object();
-#endif
-
   scheme_release_fd_semaphores();
   
   scheme_release_file_descriptor();

racket/src/racket/src/port.c

@@ -4300,6 +4300,11 @@ Scheme_Object *scheme_file_unlock(int argc, Scheme_Object **argv)
 /*                        filesystem change events                        */
 /*========================================================================*/
 
+static void filesystem_change_evt_fnl(void *fc, void *data)
+{
+  scheme_filesystem_change_evt_cancel((Scheme_Object *)fc, NULL);
+}
+
 Scheme_Object *scheme_filesystem_change_evt(Scheme_Object *path, int flags, int signal_errs)
 {
   char *filename;
@@ -4309,7 +4314,7 @@ Scheme_Object *scheme_filesystem_change_evt(Scheme_Object *path, int flags, int
 					   "filesystem-change-evt",
 					   NULL,
 					   SCHEME_GUARD_FILE_EXISTS);
-  rfc = rktio_fs_change(scheme_rktio, filename);
+  rfc = rktio_fs_change(scheme_rktio, filename, scheme_semaphore_fd_set);
 
   if (!rfc) {
     if (scheme_last_error_is_racket(RKTIO_ERROR_UNSUPPORTED)) {
@@ -4339,6 +4344,8 @@ Scheme_Object *scheme_filesystem_change_evt(Scheme_Object *path, int flags, int
     mref = scheme_add_managed(NULL, (Scheme_Object *)fc, scheme_filesystem_change_evt_cancel, NULL, 1);
     fc->mref = mref;
 
+    scheme_add_finalizer(fc, filesystem_change_evt_fnl, NULL);
+
     return (Scheme_Object *)fc;
   }
 }
@@ -4351,6 +4358,11 @@ void scheme_filesystem_change_evt_cancel(Scheme_Object *evt, void *ignored_data)
     rktio_fs_change_forget(scheme_rktio, fc->rfc);
     fc->rfc = NULL;
   }
+
+  if (fc->mref) {
+    scheme_remove_managed(fc->mref, (Scheme_Object *)fc);
+    fc->mref = NULL;
+  }
 }
 
 static int filesystem_change_evt_ready(Scheme_Object *evt, Scheme_Schedule_Info *sinfo)
@@ -4379,6 +4391,9 @@ void scheme_fs_change_properties(int *_supported, int *_scalable, int *_low_late
   int props;
 
   props = rktio_fs_change_properties(scheme_rktio);
+  if ((props & RKTIO_FS_CHANGE_NEED_LTPS) && !scheme_semaphore_fd_set)
+    props = 0;
+  
   *_supported = ((props & RKTIO_FS_CHANGE_SUPPORTED) ? 1 : 0);
   *_scalable = ((props & RKTIO_FS_CHANGE_SCALABLE) ? 1 : 0);
   *_low_latency = ((props & RKTIO_FS_CHANGE_LOW_LATENCY) ? 1 : 0);

racket/src/racket/src/string.c

@@ -389,7 +389,7 @@ SHARED_OK static char *embedding_banner;
 SHARED_OK static Scheme_Object *vers_str;
 SHARED_OK static Scheme_Object *banner_str;
 
-SHARED_OK static Scheme_Object *fs_change_props;
+THREAD_LOCAL_DECL(static Scheme_Object *fs_change_props);
 
 READ_ONLY static Scheme_Object *complete_symbol, *continues_symbol, *aborts_symbol, *error_symbol;
 
@@ -472,31 +472,6 @@ scheme_init_string (Scheme_Env *env)
   REGISTER_SO(vers_str);
   REGISTER_SO(banner_str);
 
-  REGISTER_SO(fs_change_props);
-  {
-    int supported, scalable, low_latency, file_level;
-    Scheme_Object *s;
-    scheme_fs_change_properties(&supported, &scalable, &low_latency, &file_level);
-    fs_change_props = scheme_make_vector(4, scheme_false);
-    if (supported) {
-      s = scheme_intern_symbol("supported");
-      SCHEME_VEC_ELS(fs_change_props)[0] = s;
-    }
-    if (scalable) {
-      s = scheme_intern_symbol("scalable");
-      SCHEME_VEC_ELS(fs_change_props)[1] = s;
-    }
-    if (low_latency) {
-      s = scheme_intern_symbol("low-latency");
-      SCHEME_VEC_ELS(fs_change_props)[2] = s;
-    }
-    if (file_level) {
-      s = scheme_intern_symbol("file-level");
-      SCHEME_VEC_ELS(fs_change_props)[3] = s;
-    }
-    SCHEME_SET_IMMUTABLE(fs_change_props);
-  }
-
   vers_str = scheme_make_utf8_string(scheme_version());
   SCHEME_SET_CHAR_STRING_IMMUTABLE(vers_str);
   banner_str = scheme_make_utf8_string(scheme_banner());
@@ -1014,6 +989,31 @@ scheme_init_string (Scheme_Env *env)
 void scheme_init_string_places(void) {
   REGISTER_SO(current_locale_name_ptr);
   current_locale_name_ptr = (void *)xes_char_string;
+
+  REGISTER_SO(fs_change_props);
+  {
+    int supported, scalable, low_latency, file_level;
+    Scheme_Object *s;
+    scheme_fs_change_properties(&supported, &scalable, &low_latency, &file_level);
+    fs_change_props = scheme_make_vector(4, scheme_false);
+    if (supported) {
+      s = scheme_intern_symbol("supported");
+      SCHEME_VEC_ELS(fs_change_props)[0] = s;
+    }
+    if (scalable) {
+      s = scheme_intern_symbol("scalable");
+      SCHEME_VEC_ELS(fs_change_props)[1] = s;
+    }
+    if (low_latency) {
+      s = scheme_intern_symbol("low-latency");
+      SCHEME_VEC_ELS(fs_change_props)[2] = s;
+    }
+    if (file_level) {
+      s = scheme_intern_symbol("file-level");
+      SCHEME_VEC_ELS(fs_change_props)[3] = s;
+    }
+    SCHEME_SET_IMMUTABLE(fs_change_props);
+  }
 }
 
 /**********************************************************************/

racket/src/rktio/demo.c

@@ -909,12 +909,18 @@ int main(int argc, char **argv)
     char *path = "test1";
     rktio_fs_change_t *fc;
     rktio_poll_set_t *ps;
+    rktio_ltps_t *lt;
     double start;
 
     if (verbose)
       printf("fs change\n");
 
-    fc = rktio_fs_change(rktio, path);
+    if (rktio_fs_change_properties(rktio) & RKTIO_FS_CHANGE_NEED_LTPS)
+      lt = rktio_ltps_open(rktio);
+    else
+      lt = NULL;
+
+    fc = rktio_fs_change(rktio, path, lt);
     check_valid(fc);
 
     check_valid(!rktio_poll_fs_change_ready(rktio, fc));
@@ -945,6 +951,9 @@ int main(int argc, char **argv)
     rktio_poll_set_forget(rktio, ps);
     
     rktio_fs_change_forget(rktio, fc);
+
+    if (lt)
+      rktio_ltps_close(rktio, lt);
   }
 
   if (verbose)

racket/src/rktio/rktio.h

@@ -502,10 +502,20 @@ RKTIO_EXTERN int rktio_fs_change_properties(rktio_t *rktio);
 #define RKTIO_FS_CHANGE_SCALABLE    (1 << 1)
 #define RKTIO_FS_CHANGE_LOW_LATENCY (1 << 2)
 #define RKTIO_FS_CHANGE_FILE_LEVEL  (1 << 3)
+#define RKTIO_FS_CHANGE_NEED_LTPS   (1 << 4)
 
 typedef struct rktio_fs_change_t rktio_fs_change_t;
+struct rktio_ltps_t; /* forward reference */
 
-RKTIO_EXTERN rktio_fs_change_t *rktio_fs_change(rktio_t *rktio, const char *path);
+RKTIO_EXTERN rktio_fs_change_t *rktio_fs_change(rktio_t *rktio, const char *path,
+                                                struct rktio_ltps_t *ltps);
+/* Creates a filesystem-change tracker that reports changes in `path`
+   after creation of the tracker. The properties repotred by
+   `rktio_fs_change_properties` report various aspects of how the
+   tracket behaves. In particular, the `ltps` argument can be NULL
+   unless the `RKTIO_FS_CHANGE_NEED_LTPS` property is reported; if
+   `lt` is provided, then the tracker must be canceled or discovered
+   ready before `ltps` is closed. */
 
 RKTIO_EXTERN void rktio_fs_change_forget(rktio_t *rktio, rktio_fs_change_t *fc);
 
@@ -598,8 +608,8 @@ enum {
   RKTIO_LTPS_REMOVE_VNODE
 };
 
-RKTIO_EXTERN void rktio_ltps_handle_set_data(rktio_t *rktio, rktio_ltps_handle_t *s, void *data);
+RKTIO_EXTERN void rktio_ltps_handle_set_data(rktio_t *rktio, rktio_ltps_handle_t *h, void *data);
-RKTIO_EXTERN void *rktio_ltps_handle_get_data(rktio_t *rktio, rktio_ltps_handle_t *s);
+RKTIO_EXTERN void *rktio_ltps_handle_get_data(rktio_t *rktio, rktio_ltps_handle_t *h);
 
 void rktio_ltps_remove_all(rktio_t *rktio, rktio_ltps_t *lt);
 /* Removes all additions, signaling all handles. */
@@ -610,6 +620,19 @@ RKTIO_EXTERN rktio_ok_t rktio_ltps_poll(rktio_t *rktio, rktio_ltps_t *lt);
 RKTIO_EXTERN rktio_ltps_handle_t *rktio_ltps_get_signaled_handle(rktio_t *rktio, rktio_ltps_t *lt);
 /* Free the returned handle when you're done with it. */
 
+RKTIO_EXTERN void rktio_ltps_handle_set_auto(rktio_t *rktio, rktio_ltps_handle_t *lth, int auto_mode);
+/* An alternative to receiving the handle via `rktio_ltps_get_signaled_handle`;
+   have signaling automatically either zero the handle content (so the
+   client can detect signaling) or free the handle (bcause the client
+   is no longer watching it). If `auto_mode` is `RKTIO_LTPS_HANDLE_NONE`,
+   automatic handling is disabled for the handle. */
+/* `auto_mode` values: */
+enum {
+  RKTIO_LTPS_HANDLE_NONE,
+  RKTIO_LTPS_HANDLE_ZERO,
+  RKTIO_LTPS_HANDLE_FREE
+};
+
 RKTIO_EXTERN void rktio_sleep(rktio_t *rktio, float nsecs, rktio_poll_set_t *fds, rktio_ltps_t *lt);
 /* Waits up to `nsecs` seconds (or forever if `nsecs` is 0) or until
    something registered with `fds` or `lt` is ready. */

racket/src/rktio/rktio_fs_change.c

@@ -52,7 +52,9 @@ int rktio_fs_change_properties(rktio_t *rktio)
 #ifdef NO_FILESYSTEM_CHANGE_EVTS
 #else
   flags |= RKTIO_FS_CHANGE_SUPPORTED;
-# if !defined(HAVE_KQUEUE_SYSCALL)
+# if defined(HAVE_KQUEUE_SYSCALL)
+  flags |= RKTIO_FS_CHANGE_NEED_LTPS;
+# else
   flags |= RKTIO_FS_CHANGE_SCALABLE;
 # endif
 # if !defined(HAVE_INOTIFY_SYSCALL)
@@ -66,12 +68,11 @@ int rktio_fs_change_properties(rktio_t *rktio)
   return flags;
 }
 
-rktio_fs_change_t *rktio_fs_change(rktio_t *rktio, const char *path)
+rktio_fs_change_t *rktio_fs_change(rktio_t *rktio, const char *path, rktio_ltps_t *lt)
 {
   int ok = 0;
 #ifndef NO_FILESYSTEM_CHANGE_EVTS
 # if defined(HAVE_KQUEUE_SYSCALL)
-  rktio_ltps_t *lt;
   rktio_ltps_handle_t *lth;
 # endif
 #endif
@@ -86,24 +87,30 @@ rktio_fs_change_t *rktio_fs_change(rktio_t *rktio, const char *path)
 #elif defined(FILESYSTEM_NEVER_CHANGES)
   ok = 1;
 #elif defined(HAVE_KQUEUE_SYSCALL)
-  do {
+  if (!lt) {
-    fd = open(path, RKTIO_BINARY, 0666);
+    set_racket_error(RKTIO_ERROR_UNSUPPORTED);
-  } while ((fd == -1) && (errno == EINTR));
+    ok = 0;
-  if (fd == -1)
+  } else {
-    get_posix_error();
+    do {
-  else {
+      fd = open(path, RKTIO_BINARY, 0666);
-    rktio_fd_t *rfd;
+    } while ((fd == -1) && (errno == EINTR));
-    rfd = rktio_system_fd(rktio, fd, 0);
+    if (fd == -1)
-    lt = rktio_ltps_open(rktio);
+      get_posix_error();
-    if (lt)
+    else {
+      rktio_fd_t *rfd;
+      rfd = rktio_system_fd(rktio, fd, 0);
       lth = rktio_ltps_add(rktio, lt, rfd, RKTIO_LTPS_CREATE_VNODE);
-    if (!lt || !lth) {
+      if (!lth) {
-      if (lt)
+        rktio_reliably_close(fd);
-        rktio_ltps_close(rktio, lt);
+      } else {
-      rktio_reliably_close(fd);
+        /* Put any pointer in the handle, and set it to auto-handle mode
-    } else
+           to clear the pointer if it gets signalled. */
-      ok = 1;
+        rktio_ltps_handle_set_data(rktio, lth, lth);
-    rktio_forget(rktio, rfd);
+        rktio_ltps_handle_set_auto(rktio, lth, RKTIO_LTPS_HANDLE_ZERO);
+        ok = 1;
+      }
+      rktio_forget(rktio, rfd);
+    }
   }
 #elif defined(HAVE_INOTIFY_SYSCALL)
   do_inotify_init(rktio);
@@ -162,8 +169,18 @@ static void fs_change_release(rktio_t *rktio, rktio_fs_change_t *fc)
 # elif defined(HAVE_INOTIFY_SYSCALL)
   do_inotify_remove(rktio, fc->fd);
 # elif defined(HAVE_KQUEUE_SYSCALL)
-  rktio_ltps_close(rktio, fc->lt); /* frees lth */
+  if (rktio_ltps_handle_get_data(rktio, fc->lth)) {
-  rktio_reliably_close(fc->fd);
+    /* Not zeroed, so never signaled. Change the auto behavior
+       to free the handle, and deregsiter the file descriptor. */
+    rktio_fd_t *rfd;
+    rktio_ltps_handle_set_auto(rktio, fc->lth, RKTIO_LTPS_HANDLE_FREE);
+    rfd = rktio_system_fd(rktio, fc->fd, 0);
+    (void)rktio_ltps_add(rktio, fc->lt, rfd, RKTIO_LTPS_REMOVE_VNODE);
+    rktio_close(rktio, rfd);
+  } else {
+    /* Was signaled, so we need to free it. */
+    free(fc->lth);
+  }
 #endif
 
   fc->done = 1;
@@ -200,9 +217,11 @@ int rktio_poll_fs_change_ready(rktio_t *rktio, rktio_fs_change_t *fc)
   return 0;
 #elif defined(HAVE_KQUEUE_SYSCALL)
   if (!fc->done) {
-    if (rktio_ltps_poll(rktio, fc->lt))
+    (void)rktio_ltps_poll(rktio, fc->lt);
-      if (rktio_ltps_get_signaled_handle(rktio, fc->lt) == fc->lth)
+    if (!rktio_ltps_handle_get_data(rktio, fc->lth)) {
-        fs_change_release(rktio, fc);
+      /* NULL value means that it was signaled; can free, etc. */
+      fs_change_release(rktio, fc);
+    }
   }
 
   return (fc->done ? RKTIO_POLL_READY : 0);

racket/src/rktio/rktio_ltps.c

@@ -43,6 +43,7 @@ struct rktio_ltps_t {
 };
 
 struct rktio_ltps_handle_t {
+  int auto_mode;
   void *data; /* arbitrary data from client */
   struct rktio_ltps_handle_t *next; /* in signaled chain */
 };
@@ -75,6 +76,7 @@ rktio_ltps_handle_t *make_ltps_handle()
 {
   rktio_ltps_handle_t *s;
   s = malloc(sizeof(rktio_ltps_handle_t));
+  s->auto_mode = RKTIO_LTPS_HANDLE_NONE;
   s->data = NULL;
   s->next = NULL;
   return s;
@@ -82,8 +84,18 @@ rktio_ltps_handle_t *make_ltps_handle()
 
 void ltps_signal_handle(rktio_ltps_t *lt, rktio_ltps_handle_t *s)
 {
-  s->next = lt->signaled;
+  switch (s->auto_mode) {
-  lt->signaled = s;
+  case RKTIO_LTPS_HANDLE_NONE:
+    s->next = lt->signaled;
+    lt->signaled = s;
+    break;
+  case RKTIO_LTPS_HANDLE_ZERO:
+    s->data = NULL;
+    break;
+  case RKTIO_LTPS_HANDLE_FREE:
+    free(s);
+    break;
+  }
 }
 
 void rktio_ltps_handle_set_data(rktio_t *rktio, rktio_ltps_handle_t *s, void *data)
@@ -96,6 +108,11 @@ void *rktio_ltps_handle_get_data(rktio_t *rktio, rktio_ltps_handle_t *s)
   return s->data;
 }
 
+void rktio_ltps_handle_set_auto(rktio_t *rktio, rktio_ltps_handle_t *s, int auto_mode)
+{
+  s->auto_mode = auto_mode;
+}
+
 /*========================================================================*/
 
 rktio_ltps_t *rktio_ltps_open(rktio_t *rktio)
@@ -144,12 +161,8 @@ int rktio_ltps_close(rktio_t *rktio, rktio_ltps_t *lt)
 #endif
 
 #if defined(HAVE_KQUEUE_SYSCALL) || defined(HAVE_EPOLL_SYSCALL)
-  if (lt->fd >= 0) {
+  if (lt->fd >= 0)
-    intptr_t rc;
+    rktio_reliably_close(lt->fd);
-    do {
-      rc = close(lt->fd);
-    } while ((rc == -1) && (errno == EINTR));
-  }
   free(lt);
 #else
   rktio_poll_set_forget(rktio, lt->fd_set);