From aa162d2ceb5e7d050adf255af4177c136fc763fb Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 12:50:44 -0500 Subject: [PATCH 01/13] CHG: Fibers enter WAITING state after an async result is received. CHG: Killing a Fiber during the WAITING state defers the kill action. --- src/Effect/Aff.js | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index ff8bca2..f2d80e0 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -222,6 +222,7 @@ var Aff = function () { var PENDING = 4; // An async effect is running. var RETURN = 5; // The current stack has returned. var COMPLETED = 6; // The entire fiber has completed. + var WAITING = 7; // Async result received, waiting for scheduling. function Fiber(util, supervisor, aff) { // Monotonically increasing tick, increased on each asynchronous turn. @@ -326,6 +327,7 @@ var Aff = function () { return; } runTick++; + status = WAITING; Scheduler.enqueue(function () { // It's possible to interrupt the fiber between enqueuing and // resuming, so we need to check that the runTick is still @@ -526,6 +528,7 @@ var Aff = function () { status = CONTINUE; break; case PENDING: return; + case WAITING: return; } } } @@ -585,6 +588,8 @@ var Aff = function () { run(++runTick); } break; + case WAITING: + Scheduler.enqueue(function () { kill(error, cb)(); }); default: if (interrupt === null) { interrupt = util.left(error); From 37a9af9dfb8754c37123e5f51fc56d290e7eae1f Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 13:12:47 -0500 Subject: [PATCH 02/13] FIX --- src/Effect/Aff.js | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index f2d80e0..f92ffec 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -590,6 +590,7 @@ var Aff = function () { break; case WAITING: Scheduler.enqueue(function () { kill(error, cb)(); }); + break; default: if (interrupt === null) { interrupt = util.left(error); From 1c41ceebf943177409f1b3f3543727bccde9c545 Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 15:27:15 -0500 Subject: [PATCH 03/13] FIX --- src/Effect/Aff.js | 30 +++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index f92ffec..71bceee 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -327,18 +327,25 @@ var Aff = function () { return; } runTick++; - status = WAITING; - Scheduler.enqueue(function () { - // It's possible to interrupt the fiber between enqueuing and - // resuming, so we need to check that the runTick is still - // valid. - if (runTick !== localRunTick + 1) { - return; - } + // Nothing left to run, so we complete immediately. + if (bhead === null && attempts === null) { status = STEP_RESULT; - step = result; - run(runTick); - }); + step = result; + } + else { + status = WAITING; + Scheduler.enqueue(function () { + // It's possible to interrupt the fiber between enqueuing + // and resuming, so we need to check that the runTick is + // still valid. + if (runTick !== localRunTick + 1) { + return; + } + status = STEP_RESULT; + step = result; + run(runTick); + }); + } }; }); return; @@ -589,6 +596,7 @@ var Aff = function () { } break; case WAITING: + // TODO: this is not quite the right thing to enqueue Scheduler.enqueue(function () { kill(error, cb)(); }); break; default: From f2d414b081a8c9798c42b75e2a8d14edc3126e2a Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 16:12:01 -0500 Subject: [PATCH 04/13] FIX --- src/Effect/Aff.js | 40 ++++++++++------------------------------ 1 file changed, 10 insertions(+), 30 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index 71bceee..04fbc71 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -222,7 +222,6 @@ var Aff = function () { var PENDING = 4; // An async effect is running. var RETURN = 5; // The current stack has returned. var COMPLETED = 6; // The entire fiber has completed. - var WAITING = 7; // Async result received, waiting for scheduling. function Fiber(util, supervisor, aff) { // Monotonically increasing tick, increased on each asynchronous turn. @@ -321,32 +320,18 @@ var Aff = function () { case ASYNC: status = PENDING; - step = runAsync(util.left, step._1, function (result) { - return function () { - if (runTick !== localRunTick) { - return; - } - runTick++; - // Nothing left to run, so we complete immediately. - if (bhead === null && attempts === null) { + Scheduler.enqueue(function () { + if (runTick !== localRunTick) { + return; + } + runTick++; + step = runAsync(util.left, step._1, function (result) { + return function () { status = STEP_RESULT; step = result; - } - else { - status = WAITING; - Scheduler.enqueue(function () { - // It's possible to interrupt the fiber between enqueuing - // and resuming, so we need to check that the runTick is - // still valid. - if (runTick !== localRunTick + 1) { - return; - } - status = STEP_RESULT; - step = result; - run(runTick); - }); - } - }; + run(runTick); + }; + }); }); return; @@ -535,7 +520,6 @@ var Aff = function () { status = CONTINUE; break; case PENDING: return; - case WAITING: return; } } } @@ -595,10 +579,6 @@ var Aff = function () { run(++runTick); } break; - case WAITING: - // TODO: this is not quite the right thing to enqueue - Scheduler.enqueue(function () { kill(error, cb)(); }); - break; default: if (interrupt === null) { interrupt = util.left(error); From 24d256839250f24a5beaa0ca0134d52cb4e2e583 Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 16:13:59 -0500 Subject: [PATCH 05/13] FIX --- src/Effect/Aff.js | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index 04fbc71..abc2aaf 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -320,6 +320,7 @@ var Aff = function () { case ASYNC: status = PENDING; + step = nonCanceler; Scheduler.enqueue(function () { if (runTick !== localRunTick) { return; From 206f3b9049b30281a516430699e2523af991a882 Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Tue, 5 Feb 2019 18:53:49 -0500 Subject: [PATCH 06/13] FIX --- src/Effect/Aff.js | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index abc2aaf..46152cb 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -320,19 +320,32 @@ var Aff = function () { case ASYNC: status = PENDING; + var asyncAction = step._1; step = nonCanceler; Scheduler.enqueue(function () { if (runTick !== localRunTick) { return; } - runTick++; - step = runAsync(util.left, step._1, function (result) { + ++runTick; + var resolved = false; + var canceler = runAsync(util.left, asyncAction, function (result) { return function () { + if (runTick !== localRunTick + 1) { + return; + } + ++runTick; + resolved = true; status = STEP_RESULT; step = result; run(runTick); }; }); + // Only update the canceler if the asynchronous action has not + // resolved synchronously. If it has, then the next status and + // step have already been set. + if (!resolved) { + step = canceler; + } }); return; From bccb409809d6ea8ae4d4d8d6b06e6ba1a2280e7a Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Thu, 7 Feb 2019 12:02:14 -0500 Subject: [PATCH 07/13] FIX: Fixed memory leak on V8 CHG: Applied Nate's code cleanup suggestions. --- src/Effect/Aff.js | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index 46152cb..71e342a 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -320,24 +320,25 @@ var Aff = function () { case ASYNC: status = PENDING; - var asyncAction = step._1; + tmp = step._1; step = nonCanceler; Scheduler.enqueue(function () { if (runTick !== localRunTick) { return; } - ++runTick; var resolved = false; - var canceler = runAsync(util.left, asyncAction, function (result) { + var canceler = runAsync(util.left, tmp, function (result) { return function () { - if (runTick !== localRunTick + 1) { + if (runTick !== localRunTick) { return; } ++runTick; resolved = true; status = STEP_RESULT; step = result; - run(runTick); + // Free us from this callstack. Otherwise a memory leak may + // happen on V8; not sure why. + setTimeout(function () { run(runTick); }, 0); }; }); // Only update the canceler if the asynchronous action has not From 127fd217b2c19ffe5d16c253e87a9e394bab6a5b Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Thu, 7 Feb 2019 13:22:33 -0500 Subject: [PATCH 08/13] CHG: trying something other than setTimeout to fix V8 memory leak --- src/Effect/Aff.js | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index 71e342a..c078789 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -326,6 +326,7 @@ var Aff = function () { if (runTick !== localRunTick) { return; } + var issync = true; var resolved = false; var canceler = runAsync(util.left, tmp, function (result) { return function () { @@ -336,17 +337,24 @@ var Aff = function () { resolved = true; status = STEP_RESULT; step = result; - // Free us from this callstack. Otherwise a memory leak may - // happen on V8; not sure why. - setTimeout(function () { run(runTick); }, 0); + // Do not recurse on run if we are synchronous with runAsync. + if (!issync) { + run(runTick); + } }; }); + issync = false; // Only update the canceler if the asynchronous action has not // resolved synchronously. If it has, then the next status and // step have already been set. if (!resolved) { step = canceler; } + // If runAsync already resolved then the next step needs to be + // run. + else { + run(runTick); + } }); return; From e03b941281abd734065b25f1e96b9ec2e29b735a Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Thu, 7 Feb 2019 20:36:35 -0500 Subject: [PATCH 09/13] FIX: Memory leak in V8 caused by new Error object --- src/Effect/Aff.js | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index c078789..d723f38 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -56,6 +56,12 @@ var Aff = function () { var FIBER = "Fiber"; // Actual fiber reference var THUNK = "Thunk"; // Primed effect, ready to invoke + // Error used for early cancelation on Alt branches. + // This is initialized here (rather than in the Fiber constructor) because + // otherwise, in V8, this Error object indefinitely hangs on to memory that + // otherwise would be garbage collected. + var early = new Error("[ParAff] Early exit"); + function Aff(tag, _1, _2, _3) { this.tag = tag; this._1 = _1; @@ -660,9 +666,6 @@ var Aff = function () { var killId = 0; var kills = {}; - // Error used for early cancelation on Alt branches. - var early = new Error("[ParAff] Early exit"); - // Error used to kill the entire tree. var interrupt = null; From 38fa57a96c8f84f73437e152c43dc668f1605cc4 Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Fri, 8 Feb 2019 16:46:07 -0500 Subject: [PATCH 10/13] CHG: Simplified case ASYNC state. --- src/Effect/Aff.js | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index d723f38..4298c49 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -332,29 +332,29 @@ var Aff = function () { if (runTick !== localRunTick) { return; } - var issync = true; - var resolved = false; + var skipRun = true; var canceler = runAsync(util.left, tmp, function (result) { return function () { if (runTick !== localRunTick) { return; } ++runTick; - resolved = true; status = STEP_RESULT; step = result; // Do not recurse on run if we are synchronous with runAsync. - if (!issync) { + if (skipRun) { + skipRun = false; + } else { run(runTick); } }; }); - issync = false; // Only update the canceler if the asynchronous action has not // resolved synchronously. If it has, then the next status and // step have already been set. - if (!resolved) { + if (skipRun) { step = canceler; + skipRun = false; } // If runAsync already resolved then the next step needs to be // run. From 3601775eb29116ac2cdf00216b871d2b36d386dc Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Mon, 13 May 2019 16:46:09 -0400 Subject: [PATCH 11/13] FIX: moved two more Error instantiations outside of Aff. --- src/Effect/Aff.purs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/src/Effect/Aff.purs b/src/Effect/Aff.purs index 6dbef67..41f03f1 100644 --- a/src/Effect/Aff.purs +++ b/src/Effect/Aff.purs @@ -51,8 +51,8 @@ import Data.Time.Duration (Milliseconds(..)) import Data.Time.Duration (Milliseconds(..)) as Exports import Effect (Effect) import Effect.Class (class MonadEffect, liftEffect) -import Effect.Exception (Error, error) -import Effect.Exception (Error, error, message) as Exports +import Effect.Exception (Error) +import Effect.Exception (Error, message) as Exports import Effect.Unsafe (unsafePerformEffect) import Partial.Unsafe (unsafeCrashWith) import Unsafe.Coerce (unsafeCoerce) @@ -86,8 +86,11 @@ instance monoidAff ∷ Monoid a ⇒ Monoid (Aff a) where instance altAff ∷ Alt Aff where alt a1 a2 = catchError a1 (const a2) +alwaysFailsError ∷ Error +alwaysFailsError = stacklessError "Always fails" + instance plusAff ∷ Plus Aff where - empty = throwError (error "Always fails") + empty = throwError alwaysFailsError -- | This instance is provided for compatibility. `Aff` is always stack-safe -- | within a given fiber. This instance will just result in unnecessary @@ -306,6 +309,9 @@ type Supervised a = , supervisor ∷ Supervisor } +parentOutlivedError ∷ Error +parentOutlivedError = stacklessError "[Aff] Child fiber outlived parent" + -- | Creates a new supervision context for some `Aff`, guaranteeing fiber -- | cleanup when the parent completes. Any pending fibers forked within -- | the context will be killed and have their cancelers run. @@ -313,14 +319,11 @@ supervise ∷ ∀ a. Aff a → Aff a supervise aff = generalBracket (liftEffect acquire) { killed: \err sup → parSequence_ [ killFiber err sup.fiber, killAll err sup ] - , failed: const (killAll killError) - , completed: const (killAll killError) + , failed: const (killAll parentOutlivedError) + , completed: const (killAll parentOutlivedError) } (joinFiber <<< _.fiber) where - killError ∷ Error - killError = - error "[Aff] Child fiber outlived parent" killAll ∷ Error → Supervised a → Aff Unit killAll err sup = makeAff \k → From a99e89649ada2752d8dec212307c2d4da172225c Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Mon, 13 May 2019 16:56:26 -0400 Subject: [PATCH 12/13] FIX: added error back to exports, fixed compile error --- src/Effect/Aff.purs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Effect/Aff.purs b/src/Effect/Aff.purs index 41f03f1..37c7ae0 100644 --- a/src/Effect/Aff.purs +++ b/src/Effect/Aff.purs @@ -51,8 +51,8 @@ import Data.Time.Duration (Milliseconds(..)) import Data.Time.Duration (Milliseconds(..)) as Exports import Effect (Effect) import Effect.Class (class MonadEffect, liftEffect) -import Effect.Exception (Error) -import Effect.Exception (Error, message) as Exports +import Effect.Exception (Error, error) +import Effect.Exception (Error, error, message) as Exports import Effect.Unsafe (unsafePerformEffect) import Partial.Unsafe (unsafeCrashWith) import Unsafe.Coerce (unsafeCoerce) @@ -87,7 +87,7 @@ instance altAff ∷ Alt Aff where alt a1 a2 = catchError a1 (const a2) alwaysFailsError ∷ Error -alwaysFailsError = stacklessError "Always fails" +alwaysFailsError = error "Always fails" instance plusAff ∷ Plus Aff where empty = throwError alwaysFailsError @@ -310,7 +310,7 @@ type Supervised a = } parentOutlivedError ∷ Error -parentOutlivedError = stacklessError "[Aff] Child fiber outlived parent" +parentOutlivedError = error "[Aff] Child fiber outlived parent" -- | Creates a new supervision context for some `Aff`, guaranteeing fiber -- | cleanup when the parent completes. Any pending fibers forked within From 0d7c02c4cc0f521153bbf39f7346d450bfe722ba Mon Sep 17 00:00:00 2001 From: Eric Brisco Date: Wed, 11 Sep 2019 12:16:58 -0400 Subject: [PATCH 13/13] FIX: Supervisor leaks memory by not unregistering Fibers in COMPLETE state. --- src/Effect/Aff.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Effect/Aff.js b/src/Effect/Aff.js index ff8bca2..17e2330 100644 --- a/src/Effect/Aff.js +++ b/src/Effect/Aff.js @@ -160,7 +160,7 @@ var Aff = function () { delete fibers[fid]; }; } - }); + })(); fibers[fid] = fiber; count++; },