watch-outputs remove debug code

hopefully we are now deadlock-free

pkgs/anoia/Makefile

@@ -1,8 +1,8 @@
 servicedir:=$(shell mktemp -d)
 
-default: fs.lua init.lua nl.lua svc.lua net/constants.lua
+default: fs.lua init.lua nl.lua svc.lua process.lua net/constants.lua
 
-CHECK=fs.fnl init.fnl  svc.fnl
+CHECK=fs.fnl init.fnl svc.fnl process.fnl
 
 check:
 	ln -s . anoia
@@ -22,4 +22,4 @@ net/constants.lua: net/constants.c
 
 
 %.lua: %.fnl
-	fennel --compile $< > $@
+	fennel --add-macro-path './assert.fnl'  --compile $< > $@

pkgs/anoia/default.nix

@@ -1,4 +1,5 @@
 {
+  bc, # for tests
   fennel,
   stdenv,
   linotify,
@@ -11,7 +12,7 @@ in stdenv.mkDerivation {
   inherit pname;
   version = "0.1";
   src = ./.;
-  nativeBuildInputs = [ fennel cpio ];
+  nativeBuildInputs = [ fennel cpio bc ];
   buildInputs = with lua.pkgs; [ linotify lualinux ];
   outputs = [ "out" "dev" ];
 

pkgs/anoia/fs.fnl

@@ -1,4 +1,5 @@
 (local ll (require :lualinux))
+(import-macros { :  define-tests : expect : expect= } :anoia.assert)
 
 (local S_IFMT   0xf000)
 (local S_IFSOCK 0xc000)
@@ -65,29 +66,22 @@
     unknown
     (error (.. "can't remove " pathname " of mode \"" unknown "\""))))
 
-(fn popen2 [pname argv envp]
-  (case (ll.pipe2)
-    (cin-s cin-d)
-    (match (ll.pipe2)
-      (cout-s cout-d)
-      (let [(pid err) (ll.fork)]
-        (if (not pid) (error (.. "error: " err))
-            (= pid 0)
-            (do
-              (ll.close cin-d)
-              (ll.close cout-s)
-              (ll.dup2 cin-s 0)
-              (ll.dup2 cout-d 1)
-              (ll.dup2 cout-d 2)
-              (ll.execve pname argv envp)
-              (error "execve failed"))
-            (> pid 0)
-            (do
-              (ll.close cin-s)
-              (ll.close cout-d)))
-        (values pid cin-d cout-s))
-      (nil err) (error (.. "popen pipe out: " err)))
-    (nil err) (error (.. "popen pipe in: " err))))
+;; lualinux doesn't publish access(2), this is not exactly
+;; the same but will suffice until we can add it
+(fn executable? [f]
+  (let [statbuf {}
+        stat (ll.lstat f statbuf 1)]
+    (and stat (> (band (. stat 3) 73) 0)))) ; \0111
+
+(fn find-executable [exe search-path]
+  (accumulate [full-path nil
+               p (string.gmatch search-path "(.-):")]
+    (or full-path (let [f (.. p "/" exe)] (and (executable? f) f)))))
+
+(define-tests
+  (let [p (find-executable "yes" (os.getenv "PATH"))]
+    (expect (string.match p "coreutils.+bin/yes$"))))
+
 
 {
  : mktree
@@ -95,6 +89,6 @@
  : directory?
  : dir
  : file-type
- : popen2
+ : find-executable
  :symlink (fn [from to] (ll.symlink from to))
  }

pkgs/anoia/process.fnl

@@ -0,0 +1,84 @@
+(local ll (require :lualinux))
+(local { : find-executable } (require :anoia.fs))
+(import-macros { : define-tests : expect : expect= } :anoia.assert)
+
+(macro errno-check [x]
+  `(match ,x
+     val# val#
+     (nil errno#) (assert nil (.. "system call failed, errno=" errno#))
+     ))
+
+(fn popen2 [pname argv envp]
+  (case (ll.pipe2)
+    (cin-s cin-d)
+    (match (ll.pipe2)
+      (cout-s cout-d)
+      (let [(pid err) (ll.fork)]
+        (if (not pid) (error (.. "error: " err))
+            (= pid 0)
+            (do
+              (ll.close cin-d)
+              (ll.close cout-s)
+              (ll.dup2 cin-s 0)
+              (ll.dup2 cout-d 1)
+              (ll.dup2 cout-d 2)
+              (ll.execve pname argv envp)
+              (error "execve failed"))
+            (> pid 0)
+            (do
+              (ll.close cin-s)
+              (ll.close cout-d)))
+        (values pid cin-d cout-s))
+      (nil err) (error (.. "popen pipe out: " err)))
+    (nil err) (error (.. "popen pipe in: " err))))
+
+(fn spawn [pname argv envp callback]
+  (let [(pid in out) (popen2 pname argv envp)
+        pollfds [
+                 (bor (lshift in 32) (lshift 4 16))
+                 (bor (lshift out 32) (lshift 1 16))
+                 ]]
+    (while (or (> (. pollfds 1) 0) (> (. pollfds 2) 0))
+      (ll.poll pollfds)
+      (if
+       (> (band (. pollfds 2) 0x11) 0)  ; POLLIN | POLLHUP
+       (if (not (callback :out out)) (tset pollfds 2 (lshift -1 32)))
+
+       (> (band (. pollfds 1) 4) 0)     ; POLLOUT
+       (if (not (callback :in in)) (tset pollfds 1 (lshift -1 32)))
+       ))
+
+    (match (ll.waitpid pid)
+      (0 status) false
+      (pid status) (rshift (band status 0xff00) 8)
+      (nil errno) (error (.. "waitpid: " errno)))))
+
+
+(define-tests
+  (var buf "4 * 6\n")                   ;; spawn bc to multiply two numbers
+  (let [out []
+        p (spawn
+           (assert (find-executable "bc" (os.getenv "PATH")))
+           ["bc"]
+           (ll.environ)
+           (fn [stream fd]
+             (match stream
+               :out (let [b (ll.read fd)]
+                      (table.insert out b)
+                      (> (# b) 0))
+               :in (if (> (# buf) 0)
+                       (let [bytes (ll.write fd buf)]
+                         (set buf (string.sub buf (+ bytes 1) -1))
+                         true)
+                       (do
+                         (ll.close fd)
+                         false))
+               :err (assert nil (ll.read fd)))))]
+    (expect= (table.concat out) "24\n"))
+  )
+
+
+{
+ : popen2
+ : spawn
+ }

pkgs/tangc/tangc.fnl

@@ -1,41 +1,32 @@
 (local json (require :json))
 (local http (require :fetch))
 (local { : base64  : %%} (require :anoia))
-(local { : popen2 } (require :anoia.fs))
+(local { : spawn } (require :anoia.process))
+(local { : find-executable } (require :anoia.fs))
 (local ll (require :lualinux))
 
 (local CLEVIS_DEFAULT_THP_LEN 43)        ;  Length of SHA-256 thumbprint.
 (local thumbprint-algs ["S256" "S1"])
 
-(fn exited [pid]
-  (match (ll.waitpid pid)
-    (0 status) false
-    (pid status) (rshift (band status 0xff00) 8)
-    (nil errno) (error (.. "waitpid: " errno))))
-
-(fn write-all [fd str]
-  (let [written (ll.write fd str)]
-    (if (< written (# str))
-        (write-all fd (string.sub str (+ written 1) -1)))))
-
-(fn read-all [fd]
-  (let [buf (ll.read fd)]
-    (if (> (# buf) 0) (.. buf (read-all fd)) buf)))
-
 (fn jose [params inputstr]
+  (var buf inputstr)
   (let [env (ll.environ)
         argv (doto params (table.insert 1 "jose"))
-        (pid in out) (popen2 (os.getenv "JOSE_BIN") argv env)]
-    ;; be careful if using this code for commands othert than jose: it
-    ;; may deadlock if we write more than 8k and the command doesn't
-    ;; read it.
-    (when inputstr (write-all in inputstr))
-    (ll.close in)
-    (let [output
-          (accumulate [o ""
-                       buf #(match (read-all out) "" nil s s)]
-            (.. o buf))]
-      (values (exited pid) output))))
+        output []
+        exitstatus
+        (spawn
+         (find-executable "jose" (os.getenv "PATH")) argv envp
+         (fn [stream fd]
+           (match stream
+             :out (let [b (ll.read fd)]
+                    (if (> (# b) 0)
+                        (do (table.insert output b) true)
+                        (do (ll.close fd) false)))
+             :in (let [b (string.sub buf (+ 1 (ll.write fd buf)) -1)]
+                   (if (> (# b) 0)
+                       (do (set buf b) true)
+                       (do (ll.close fd) false))))))]
+    (values exitstatus (table.concat output))))
 
 (fn jose! [params inputstr]
   (let [(exitcode out) (jose params inputstr)]

pkgs/watch-outputs/watch-outputs.fnl

@@ -1,6 +1,5 @@
 (local { : %% : system : assoc : split : table= : dig } (require :anoia))
 (local svc (require :anoia.svc))
-(local { : view } (require :fennel))
 (local { : kill } (require :lualinux))
 
 (fn split-paths [paths]
@@ -45,7 +44,6 @@
     (accumulate [tree (service:output ".")
                  v (service:events)]
       (let [new-tree (service:output ".")]
-        (print :was (view tree) :now (view new-tree))
         (when (changed? paths tree new-tree)
           (print "watched path event:"  action controlled-service)
           (do-action action controlled-service))