Merge pull request #83 from natefaubion/peekVar-fix

peekVar fix

Author: natefaubion

package.json

@@ -2,14 +2,14 @@
   "private": true,
   "scripts": {
     "clean": "rimraf output && rimraf .pulp-cache",
-    "build": "jshint src && jscs src && pulp build --censor-lib --strict",
+    "build": "jshint src && jscs src && pulp build -- --censor-lib --strict",
     "test": "pulp test"
   },
   "devDependencies": {
     "jscs": "^3.0.7",
     "jshint": "^2.9.4",
-    "pulp": "^9.0.1",
-    "purescript-psa": "^0.3.9",
+    "pulp": "^10.0.0",
+    "purescript-psa": "^0.4.0",
     "purescript": "^0.10.1",
     "rimraf": "^2.5.4"
   }

src/Control/Monad/Aff.js

@@ -81,17 +81,13 @@ exports._forkAff = function (nonCanceler, aff) {
 exports._forkAll = function (nonCanceler, foldl, affs) {
   var voidF = function () {};
 
-  return function (success, error) {
-    try {
-      var cancelers = foldl(function (acc) {
-        return function (aff) {
-          acc.push(aff(voidF, voidF));
-          return acc;
-        };
-      })([])(affs);
-    } catch (err) {
-      error(err);
-    }
+  return function (success) {
+    var cancelers = foldl(function (acc) {
+      return function (aff) {
+        acc.push(aff(voidF, voidF));
+        return acc;
+      };
+    })([])(affs);
 
     var canceler = function (e) {
       return function (success, error) {
@@ -162,18 +158,9 @@ exports._throwError = function (nonCanceler, e) {
 
 exports._fmap = function (f, aff) {
   return function (success, error) {
-    try {
-      return aff(function (v) {
-        try {
-          var v2 = f(v);
-        } catch (err) {
-          error(err);
-        }
-        success(v2);
-      }, error);
-    } catch (err) {
-      error(err);
-    }
+    return aff(function (v) {
+      success(f(v));
+    }, error);
   };
 };
 
@@ -224,25 +211,54 @@ exports._bind = function (alwaysCanceler, aff, f) {
 
 exports._attempt = function (Left, Right, aff) {
   return function (success) {
-    try {
-      return aff(function (v) {
-        success(Right(v));
-      }, function (e) {
-        success(Left(e));
-      });
-    } catch (err) {
-      success(Left(err));
-    }
+    return aff(function (v) {
+      success(Right(v));
+    }, function (e) {
+      success(Left(e));
+    });
   };
 };
 
 exports._runAff = function (errorT, successT, aff) {
+  // If errorT or successT throw, and an Aff is comprised only of synchronous
+  // effects, then it's possible for makeAff/liftEff to accidentally catch
+  // it, which may end up rerunning the Aff depending on error recovery
+  // behavior. To mitigate this, we observe synchronicity using mutation. If
+  // an Aff is observed to be synchronous, we let the stack reset and run the
+  // handlers outside of the normal callback flow.
   return function () {
-    return aff(function (v) {
-      successT(v)();
+    var status = 0;
+    var result, success;
+
+    var canceler = aff(function (v) {
+      if (status === 2) {
+        successT(v)();
+      } else {
+        status = 1;
+        result = v;
+        success = true;
+      }
     }, function (e) {
-      errorT(e)();
+      if (status === 2) {
+        errorT(e)();
+      } else {
+        status = 1;
+        result = e;
+        success = false;
+      }
     });
+
+    if (status === 1) {
+      if (success) {
+        successT(result)();
+      } else {
+        errorT(result)();
+      }
+    } else {
+      status = 2;
+    }
+
+    return canceler;
   };
 };
 
@@ -284,11 +300,7 @@ exports._tailRecM = function (isLeft, f, a) {
           if (isLeft(v)) {
             go(v.value0);
           } else {
-            try {
-              success(v.value0);
-            } catch (err) {
-              error(err);
-            }
+            success(v.value0);
           }
         }
       };
@@ -307,11 +319,7 @@ exports._tailRecM = function (isLeft, f, a) {
             acc = result.value0;
             continue;
           } else {
-            try {
-              success(result.value0);
-            } catch (err) {
-              error(err);
-            }
+            success(result.value0);
           }
         } else {
           // If the status has not resolved yet, then we have observed an

src/Control/Monad/Aff/Internal.js

@@ -1,17 +1,12 @@
 "use strict";
 
 exports._makeVar = function (nonCanceler) {
-  return function (success, error) {
-    try {
-      success({
-        consumers: [],
-        producers: [],
-        error: undefined
-      });
-    } catch (err) {
-      error(err);
-    }
-
+  return function (success) {
+    success({
+      consumers: [],
+      producers: [],
+      error: undefined
+    });
     return nonCanceler;
   };
 };
@@ -21,11 +16,9 @@ exports._takeVar = function (nonCanceler, avar) {
     if (avar.error !== undefined) {
       error(avar.error);
     } else if (avar.producers.length > 0) {
-      var producer = avar.producers.shift();
-
-      producer(success, error);
+      avar.producers.shift()(success, error);
     } else {
-      avar.consumers.push({ success: success, error: error });
+      avar.consumers.push({ peek: false, success: success, error: error });
     }
 
     return nonCanceler;
@@ -37,8 +30,7 @@ exports._peekVar = function (nonCanceler, avar) {
     if (avar.error !== undefined) {
       error(avar.error);
     } else if (avar.producers.length > 0) {
-      var producer = avar.producers[0];
-      producer(success, error);
+      avar.producers[0](success, error);
     } else {
       avar.consumers.push({ peek: true, success: success, error: error });
     }
@@ -50,28 +42,34 @@ exports._putVar = function (nonCanceler, avar, a) {
   return function (success, error) {
     if (avar.error !== undefined) {
       error(avar.error);
-    } else if (avar.consumers.length === 0) {
-      avar.producers.push(function (success, error) {
-        try {
-          success(a);
-        } catch (err) {
-          error(err);
-        }
-      });
-
-      success({});
     } else {
-
+      var shouldQueue = true;
+      var consumers = [];
       var consumer;
-      do {
+
+      while (true) {
         consumer = avar.consumers.shift();
-        try {
-          consumer.success(a);
-        } catch (err) {
-          error(err);
-          return;
+        if (consumer) {
+          consumers.push(consumer);
+          if (consumer.peek) {
+            continue;
+          } else {
+            shouldQueue = false;
+          }
         }
-      } while (consumer.peek === true);
+        break;
+      }
+
+      if (shouldQueue) {
+        avar.producers.push(function (success) {
+          success(a);
+          return nonCanceler;
+        });
+      }
+
+      for (var i = 0; i < consumers.length; i++) {
+        consumers[i].success(a);
+      }
 
       success({});
     }
@@ -85,22 +83,11 @@ exports._killVar = function (nonCanceler, avar, e) {
     if (avar.error !== undefined) {
       error(avar.error);
     } else {
-      var errors = [];
-
       avar.error = e;
-
-      while (avar.consumers.length > 0) {
-        var consumer = avar.consumers.shift();
-
-        try {
-          consumer.error(e);
-        } catch (err) {
-          errors.push(err);
-        }
+      while (avar.consumers.length) {
+        avar.consumers.shift().error(e);
       }
-
-      if (errors.length > 0) error(errors[0]);
-      else success({});
+      success({});
     }
 
     return nonCanceler;

test/Test/Main.purs

@@ -13,12 +13,24 @@ import Control.Monad.Error.Class (throwError)
 import Control.Monad.Rec.Class (Step(..), tailRecM)
 import Control.Parallel (parallel, sequential)
 import Data.Either (either, fromLeft, fromRight)
+import Data.Maybe (Maybe(..))
 import Data.Unfoldable (replicate)
 import Partial.Unsafe (unsafePartial)
 
 type Test a = forall e. Aff (console :: CONSOLE | e) a
 type TestAVar a = forall e. Aff (console :: CONSOLE, avar :: AVAR | e) a
 
+timeout :: Int → TestAVar Unit → TestAVar Unit
+timeout ms aff = do
+  exn <- makeVar
+  clr1 <- forkAff (later' ms (putVar exn (Just "Timed out")))
+  clr2 <- forkAff (aff *> putVar exn Nothing)
+  res ← takeVar exn
+  log (show res)
+  case res of
+    Nothing -> void (clr1 `cancel` error "Done")
+    Just e -> void (clr2 `cancel` error "Done") *> throwError (error e)
+
 replicateArray :: forall a. Int -> a -> Array a
 replicateArray = replicate
 
@@ -69,20 +81,38 @@ test_putTakeVar = do
 
 test_peekVar :: TestAVar Unit
 test_peekVar = do
-  v <- makeVar
-  forkAff (later $ putVar v 1.0)
-  a1 <- peekVar v
-  a2 <- takeVar v
-  when (a1 /= a2) do
-    throwError (error "Something horrible went wrong - peeked var is not equal to taken var")
-  log ("Success: Peeked value not consumed")
-
-  w <- makeVar
-  putVar w true
-  b <- peekVar w
-  when (not b) do
-    throwError (error "Something horrible went wrong - peeked var is not true")
-  log ("Success: Peeked value read from written var")
+  timeout 1000 do
+    v <- makeVar
+    forkAff (later $ putVar v 1.0)
+    a1 <- peekVar v
+    a2 <- takeVar v
+    when (a1 /= a2) do
+      throwError (error "Something horrible went wrong - peeked var is not equal to taken var")
+    log ("Success: Peeked value not consumed")
+
+  timeout 1000 do
+    w <- makeVar
+    putVar w true
+    b <- peekVar w
+    when (not b) do
+      throwError (error "Something horrible went wrong - peeked var is not true")
+    log ("Success: Peeked value read from written var")
+
+  timeout 1000 do
+    x <- makeVar
+    res <- makeVar' 1
+    forkAff do
+      c <- peekVar x
+      putVar x 1000
+      d <- peekVar x
+      modifyVar (_ + (c + d)) res
+    putVar x 10
+    count <- takeVar res
+    e <- takeVar x
+    f <- takeVar x
+    when (not (count == 21 && e == 10 && f == 1000)) do
+      throwError (error "Something horrible went wrong - peeked consumers/producer ordering")
+    log "Success: peekVar consumer/producer order maintained"
 
 test_killFirstForked :: Test Unit
 test_killFirstForked = do
@@ -191,10 +221,10 @@ loopAndBounce n = do
   log $ "Done: " <> show res
   where
   go 0 = pure (Done 0)
-  go n | mod n 30000 == 0 = do
+  go k | mod k 30000 == 0 = do
     later' 10 (pure unit)
-    pure (Loop (n - 1))
-  go n = pure (Loop (n - 1))
+    pure (Loop (k - 1))
+  go k = pure (Loop (k - 1))
 
 all :: forall eff. Int -> Aff (console :: CONSOLE, avar :: AVAR | eff) Unit
 all n = do