Avoid thread death via message receive

10 files changed, 129 insertions, 8 deletions

diff --git a/lib/js/urweb.js b/lib/js/urweb.js
index 5cf159ad..534769a5 100644
--- a/lib/js/urweb.js
+++ b/lib/js/urweb.js
@@ -75,6 +75,14 @@ function flatten(cls, tr) {
     return tr;
 }
 
+function flattenLocal(s) {
+  var cls = {v : null};
+  var r = flatten(cls, s);
+  for (cl = cls.v; cl != null; cl = cl.next)
+    freeClosure(cl.data);
+  return r;
+}
+
 
 
 // Dynamic tree management
@@ -259,7 +267,21 @@ function onError(f) {
 function er(s) {
   for (var ls = errorHandlers; ls; ls = ls.next)
     ls.data(s)(null);
-  throw s;
+  throw {uw_error: s};
+}
+
+var failHandlers = null;
+
+function onFail(f) {
+  failHandlers = cons(f, failHandlers);
+}
+
+function doExn(v) {
+  if (v == null || v.uw_error == null) {
+    var s = (v == null ? "null" : v.toString());
+    for (var ls = failHandlers; ls; ls = ls.next)
+      ls.data(s)(null);
+  }
 }
 
 
@@ -299,11 +321,7 @@ function requestUri(xhr, uri) {
 }
 
 function rc(uri, parse, k) {
-  var cls = {v : null};
-  uri = flatten(cls, uri);
-  for (cl = cls.v; cl != null; cl = cl.next)
-    freeClosure(cl.data);
-
+  uri = flattenLocal(uri);
   var xhr = getXHR();
 
   xhr.onreadystatechange = function() {
@@ -410,7 +428,11 @@ function listener() {
           if (listener == null) {
             enqueue(ch.msgs, msg);
           } else {
-            listener(msg);
+            try {
+              listener(msg);
+            } catch (v) {
+              doExn(v);
+            }
           }
         }
 
@@ -451,7 +473,11 @@ function rv(chn, parse, k) {
   if (msg == null) {
     enqueue(ch.listeners, function(msg) { k(parse(msg))(null); });
   } else {
-    k(parse(msg))(null);
+    try {
+      k(parse(msg))(null);
+    } catch (v) {
+      doExn(v);
+    }
   }
 }
 
diff --git a/lib/ur/basis.urs b/lib/ur/basis.urs
index 173324f0..af1cf972 100644
--- a/lib/ur/basis.urs
+++ b/lib/ur/basis.urs
@@ -553,6 +553,7 @@ val td : other ::: {Unit} -> [other ~ [Body, Tr]] =>
 val error : t ::: Type -> xbody -> t
 
 val onError : (xbody -> transaction unit) -> transaction unit
+val onFail : (string -> transaction unit) -> transaction unit
 (* Client-side only *)
 
 val show_xml : ctx ::: {Unit} -> use ::: {Type} -> bind ::: {Type} -> show (xml ctx use bind)
diff --git a/tests/buffer.ur b/tests/buffer.ur
new file mode 100644
index 00000000..27e2b805
--- /dev/null
+++ b/tests/buffer.ur
@@ -0,0 +1,25 @@
+datatype lines = End | Line of string * source lines
+
+type t = { Head : source lines, Tail : source (source lines) }
+
+val create =
+    head <- source End;
+    tail <- source head;
+    return {Head = head, Tail = tail}
+
+fun renderL lines =
+    case lines of
+        End => <xml/>
+      | Line (line, linesS) => <xml>{[line]}<br/><dyn signal={renderS linesS}/></xml>
+
+and renderS linesS =
+    lines <- signal linesS;
+    return (renderL lines)
+
+fun render t = renderS t.Head
+
+fun write t s =
+    oldTail <- get t.Tail;
+    newTail <- source End;
+    set oldTail (Line (s, newTail));
+    set t.Tail newTail
diff --git a/tests/buffer.urs b/tests/buffer.urs
new file mode 100644
index 00000000..58312bbd
--- /dev/null
+++ b/tests/buffer.urs
@@ -0,0 +1,5 @@
+type t
+
+val create : transaction t
+val render : t -> signal xbody
+val write : t -> string -> transaction unit
diff --git a/tests/roundTrip.ur b/tests/roundTrip.ur
new file mode 100644
index 00000000..26a0113e
--- /dev/null
+++ b/tests/roundTrip.ur
@@ -0,0 +1,36 @@
+table channels : { Client : client, Channel : channel (string * int * float) }
+  PRIMARY KEY Client
+
+fun writeBack v =
+    me <- self;
+    r <- oneRow (SELECT channels.Channel FROM channels WHERE channels.Client = {[me]});
+    send r.Channels.Channel v
+
+fun main () =
+    me <- self;
+    ch <- channel;
+    dml (INSERT INTO channels (Client, Channel) VALUES ({[me]}, {[ch]}));
+    
+    buf <- Buffer.create;
+
+    let
+        fun receiverA () =
+            v <- recv ch;
+            Buffer.write buf ("A:(" ^ v.1 ^ ", " ^ show v.2 ^ ", " ^ show v.3 ^ ")");
+            receiverA ()
+
+        fun receiverB () =
+            v <- recv ch;
+            Buffer.write buf ("B:(" ^ v.1 ^ ", " ^ show v.2 ^ ", " ^ show v.3 ^ ")");
+            error <xml>Bail out!</xml>;
+            receiverB ()
+
+        fun sender s n f =
+            sleep 9;
+            writeBack (s, n, f);
+            sender (s ^ "!") (n + 1) (f + 1.23)
+    in
+        return <xml><body onload={spawn (receiverA ()); spawn (receiverB ()); sender "" 0 0.0}>
+          <dyn signal={Buffer.render buf}/>
+        </body></xml>
+    end
diff --git a/tests/roundTrip.urp b/tests/roundTrip.urp
new file mode 100644
index 00000000..e8f27f2e
--- /dev/null
+++ b/tests/roundTrip.urp
@@ -0,0 +1,5 @@
+database dbname=roundtrip
+sql roundTrip.sql
+
+buffer
+roundTrip
diff --git a/tests/roundTrip.urs b/tests/roundTrip.urs
new file mode 100644
index 00000000..6ac44e0b
--- /dev/null
+++ b/tests/roundTrip.urs
@@ -0,0 +1 @@
+val main : unit -> transaction page
diff --git a/tests/threads.ur b/tests/threads.ur
new file mode 100644
index 00000000..447b7a74
--- /dev/null
+++ b/tests/threads.ur
@@ -0,0 +1,18 @@
+fun main () =
+    buf <- Buffer.create;
+    let
+        fun loop1 () =
+            Buffer.write buf "A";
+            sleep 9;
+            loop1 ()
+
+        fun loop2 () =
+            Buffer.write buf "B";
+            sleep 9;
+            error <xml>Darn</xml>
+            loop2 ()
+    in
+        return <xml><body onload={spawn (loop1 ()); loop2 ()}>
+          <dyn signal={Buffer.render buf}/>
+        </body></xml>
+    end
diff --git a/tests/threads.urp b/tests/threads.urp
new file mode 100644
index 00000000..153e09a9
--- /dev/null
+++ b/tests/threads.urp
@@ -0,0 +1,3 @@
+
+buffer
+threads
diff --git a/tests/threads.urs b/tests/threads.urs
new file mode 100644
index 00000000..6ac44e0b
--- /dev/null
+++ b/tests/threads.urs
@@ -0,0 +1 @@
+val main : unit -> transaction page