@@ -0,0 +1,1763 @@
1+package purescript_aff
2+
3+import (
4+ "errors"
5+ "fmt"
6+ "os"
7+ "reflect"
8+ "runtime"
9+ "strings"
10+ "sync"
11+ "time"
12+
13+ . "github.com/purescript-native/go-runtime"
14+)
15+
16+// Whether to log errors or not
17+const TRACE = false
18+
19+func Trace(stuff ...Any) {
20+ if TRACE {
21+ fmt.Println(stuff...)
22+ }
23+}
24+
25+func dontRecover() Any {
26+ return nil
27+}
28+func TraceAnyway(stuff ...Any) {
29+ fmt.Println(stuff...)
30+}
31+
32+var (
33+ Info = Teal
34+ Warn = Yellow
35+ Fata = Red
36+)
37+
38+var (
39+ Black = Color("\033[1;30m%s\033[0m")
40+ Red = Color("\033[1;31m%s\033[0m")
41+ Green = Color("\033[1;32m%s\033[0m")
42+ Yellow = Color("\033[1;33m%s\033[0m")
43+ Purple = Color("\033[1;34m%s\033[0m")
44+ Magenta = Color("\033[1;35m%s\033[0m")
45+ Teal = Color("\033[1;36m%s\033[0m")
46+ White = Color("\033[1;37m%s\033[0m")
47+)
48+
49+func Color(colorString string) func(...interface{}) string {
50+ sprint := func(args ...interface{}) string {
51+ return fmt.Sprintf(colorString,
52+ fmt.Sprint(args...))
53+ }
54+ return sprint
55+}
56+
57+type Fiber = Dict
58+
59+// Pure a
60+type Pure struct {
61+ Pure bool
62+ value Any
63+}
64+
65+// type memo interface {
66+// getMemo() Any
67+// setMemo(v Any)
68+// }
69+
70+// func (p Pure) getMemo() Any {
71+// return p.value
72+// }
73+
74+// func (p Pure) setMemo(m Any) {
75+// p.value = m
76+// }
77+
78+// func (p AbstractAff) getMemo() Any {
79+// return p._3
80+// }
81+
82+// func (p AbstractAff) setMemo(m Any) {
83+// p._3 = m
84+// }
85+
86+func (p Pure) String() string {
87+ return fmt.Sprintf("Pure %s", p.value)
88+}
89+
90+// Throw Error
91+type Throw struct {
92+ Throw bool
93+ err error
94+}
95+
96+func (t Throw) String() string {
97+ return fmt.Sprintf("Throw %s", t.err)
98+}
99+
100+// Catch (Aff a) (Error -> Aff a)
101+type Catch struct {
102+ Catch bool
103+ aff Any
104+ errorToAff func(Any) Any
105+}
106+
107+func (c Catch) String() string {
108+ return fmt.Sprintf("Catch %s", c.aff)
109+}
110+
111+// Sync (Effect a)
112+type Sync struct {
113+ Sync bool
114+ eff func() Any
115+}
116+
117+func (s Sync) String() string {
118+ return fmt.Sprintf("Sync")
119+}
120+
121+type K = Any
122+type Unit = Any
123+type EffectUnit = func() Unit
124+type EitherErrorOrAToEffectUnit = func(Any) EffectUnit
125+type EffectCanceler = func() Canceler
126+type Canceler = func(err Any) Any
127+type AsyncFn = func(EitherErrorOrAToEffectUnit) EffectCanceler
128+
129+// Async ((Either Error a -> Effect Unit) -> Effect (Canceler))
130+type Async struct {
131+ Async bool
132+ asyncFn Any
133+}
134+
135+func (a Async) String() string {
136+ return fmt.Sprintf("Async")
137+}
138+
139+// forall b. Bind (Aff b) (b -> Aff a)
140+type Bind struct {
141+ Bind bool
142+ affOfB Any
143+ bToAff func(Any) Any
144+}
145+
146+func (b Bind) String() string {
147+ return fmt.Sprintf("Bind %d", b.affOfB)
148+}
149+
150+// forall b. Bracket (Aff b) (BracketConditions b) (b -> Aff a)
151+type Bracket struct {
152+ Bracket bool
153+ acquire Any
154+ bracketConditions Dict
155+ k func(Any) Any
156+}
157+
158+func (b Bracket) String() string {
159+ return fmt.Sprintf("Bracket acquire:%d conditions:%d",
160+ b.acquire,
161+ b.bracketConditions)
162+}
163+
164+// forall b. Fork Boolean (Aff b) ?(Fiber b -> a)
165+type Fork struct {
166+ Fork bool
167+ immediate bool
168+ affOfB Any
169+}
170+
171+func (f Fork) String() string {
172+ return fmt.Sprintf("Fork %d", f.affOfB)
173+}
174+
175+// Sequential (ParAff a)
176+type Sequential struct {
177+ Sequential bool
178+ parAff Any
179+}
180+
181+type Return struct {
182+ Return bool
183+}
184+
185+func (r Return) String() string {
186+ return fmt.Sprintf("Return")
187+}
188+
189+type Resume struct {
190+ Resume bool
191+ b Cons
192+}
193+
194+func (r Resume) String() string {
195+ return fmt.Sprintf("Resume %d", r.b.head)
196+}
197+
198+type Release struct {
199+ Release bool
200+ bracketConditions Dict
201+ result Any
202+}
203+
204+func (r Release) String() string {
205+ return fmt.Sprintf("Release %d", r.result)
206+}
207+
208+type Finalizer struct {
209+ Finalizer bool
210+ finalizer Any
211+}
212+
213+func (f Finalizer) String() string {
214+ return fmt.Sprintf("Finalizer")
215+}
216+
217+type Finalized struct {
218+ Finalizer bool
219+ step Any
220+ fail Any
221+}
222+
223+func (f Finalized) String() string {
224+ return fmt.Sprintf("Finalized %d %d", f.step, f.fail)
225+}
226+
227+type AbstractAff struct {
228+ tag string
229+ _1 Any
230+ _2 Any
231+ _3 Any
232+}
233+
234+func (aa AbstractAff) String() string {
235+ return fmt.Sprintf("\nAff:\t%s\n\t_1: %d\n\t_2: %d\n\t_3: %d",
236+ aa.tag, aa._1, aa._2, aa._3)
237+}
238+
239+var Empty AbstractAff = AbstractAff{tag: "empty"}
240+
241+// forall b. Map (b -> a) (ParAff b)
242+// forall b. Apply (ParAff (b -> a)) (ParAff b)
243+// Alt (ParAff a) (ParAff a)
244+type ParAlt struct {
245+ parAffFn Any
246+ parAff Any
247+ memo Any
248+}
249+
250+type OnComplete = Dict
251+
252+type Cons struct {
253+ head Any
254+ tail *Cons
255+}
256+
257+func printCons(cons *Cons) {
258+ var goFn func(depth int, currHead Any, currTail *Cons) = nil
259+ goFn = func(depth int, currHead Any, currTail *Cons) {
260+ if currHead != nil {
261+ Trace(Yellow(currHead))
262+ if currTail != nil {
263+ goFn(depth+1, currTail.head, currTail.tail)
264+ }
265+ }
266+ }
267+ goFn(0, cons.head, cons.tail)
268+}
269+
270+type InterruptCons struct {
271+ head Any
272+ interrupt *Fiber
273+ tail *InterruptCons
274+}
275+
276+const SUSPENDED = 0 // Suspended, pending a join.
277+const CONTINUE = 1 // Interpret the next instruction.
278+const STEP_BIND = 2 // Apply the next bind.
279+const STEP_RESULT = 3 // Handle potential failure from a result.
280+const PENDING = 4 // An async effect is running.
281+const RETURN = 5 // The current stack has returned.
282+const COMPLETED = 6 // The entire fiber has completed.
283+
284+func statusName(st int) string {
285+ switch st {
286+ case SUSPENDED:
287+ return "SUSPENDED"
288+ case CONTINUE:
289+ return "CONTINUE"
290+ case STEP_BIND:
291+ return "STEP_BIND"
292+ case STEP_RESULT:
293+ return "STEP_RESULT"
294+ case PENDING:
295+ return "PENDING"
296+ case RETURN:
297+ return "RETURN"
298+ case COMPLETED:
299+ return "COMPLETED"
300+ }
301+ return "!!!!! UNKNOWN !!!!! "
302+}
303+
304+func nonCanceler(err Any) Any {
305+ return Pure{Pure: true, value: nil}
306+}
307+
308+func runEff(eff func() Any) {
309+ Trace(Teal("func: runEff"))
310+ defer func() {
311+ if err := recover(); err != nil {
312+ Trace(Magenta("Caught error in runEff: ", err))
313+ var msg string
314+ switch errTyped := err.(type) {
315+ case error:
316+ msg = errTyped.Error()
317+ case string:
318+ msg = errTyped
319+ default:
320+ Trace(Red("Unexpected error type in runEff ", reflect.TypeOf(err), err))
321+ os.Exit(0)
322+ }
323+ if msg == "runtime error: invalid memory address or nil pointer dereference" {
324+ Trace(Red("invalid mem", msg))
325+ panic(err)
326+ }
327+ if strings.HasPrefix(msg, "interface conversion:") {
328+ Trace(Red("Wrong cast ", msg))
329+ panic(err)
330+ }
331+ if strings.HasPrefix(msg, "Failed pattern match") {
332+ Trace(Red("FPM ", msg))
333+ panic(err)
334+ }
335+ Trace("\tError> ", err)
336+ go func() {
337+ panic(err)
338+ }()
339+ }
340+ }()
341+ eff()
342+}
343+
344+func runSync(left func(Any) Any, right func(Any) Any, eff func() Any) Any {
345+ Trace(Teal("func: runSync with ", reflect.TypeOf(eff)))
346+ var res Any = left(errors.New("Error wasn't set properly"))
347+ defer func() {
348+ if err := recover(); err != nil {
349+ Trace(Magenta("Caught error in runSync: ", err))
350+ var msg string
351+ switch errTyped := err.(type) {
352+ case error:
353+ msg = errTyped.Error()
354+ case string:
355+ msg = errTyped
356+ default:
357+ Trace(Red("Unexpected error type in runSync ", reflect.TypeOf(err), err))
358+ os.Exit(0)
359+ }
360+ if msg == "runtime error: invalid memory address or nil pointer dereference" {
361+ Trace(Red("invalid mem", msg))
362+ panic(err)
363+ }
364+ if strings.HasPrefix(msg, "interface conversion:") {
365+ Trace(Red("Wrong cast ", msg))
366+ panic(err)
367+ }
368+ Trace("\tError> ", err)
369+ res = left(errors.New(msg))
370+ }
371+ }()
372+ res = right(eff())
373+ return res
374+}
375+
376+func runAsync(left func(Any) Any, eff Any, k K) Any {
377+ Trace(Teal("func: runAsync"))
378+ // catch...
379+ var canceler Any = nonCanceler
380+ defer func() {
381+ if err := recover(); err != nil {
382+ Trace(Magenta("Caught error in runAsync: ", err))
383+ var msg string
384+ switch errTyped := err.(type) {
385+ case error:
386+ msg = errTyped.Error()
387+ case string:
388+ msg = errTyped
389+ case func(Any) Any:
390+ Trace(Red("Fn error type in runAsync ", errTyped(nil)))
391+ panic(err)
392+ default:
393+ Trace(Red("Unexpected error type in runAsync ", reflect.TypeOf(err), err))
394+ os.Exit(0)
395+ }
396+ if msg == "runtime error: invalid memory address or nil pointer dereference" {
397+ Trace(Red("invalid mem", msg))
398+ panic(err)
399+ }
400+ if strings.HasPrefix(msg, "interface conversion:") {
401+ Trace(Red("Wrong cast ", msg))
402+ panic(err)
403+ }
404+ Trace("\tError> ", err)
405+ k.(func(Any) Any)(left(err)).(func() Any)()
406+ }
407+ }()
408+ canceler = eff.(func(Any) Any)(k).(func() Any)()
409+ return canceler
410+}
411+
412+func sequential(util Any, supervisor *Supervisor, par Any) Async {
413+ return Async{asyncFn: func(cb Any) Any {
414+ return func() Any {
415+ return runPar(util, supervisor, par, cb)
416+ }
417+ },
418+ }
419+}
420+
421+type Scheduler struct {
422+ isDraining func() bool
423+ enqueue func(cb func())
424+}
425+
426+var scheduler Scheduler = (func() Scheduler {
427+ const limit = 1024
428+ size := 0
429+ ix := 0
430+ var queue [limit](func())
431+ draining := false
432+
433+ drain := func() {
434+ Trace(Teal("func: drain", size))
435+ draining = true
436+ for size != 0 {
437+ size--
438+ thunk := queue[ix]
439+ ix = (ix + 1) % limit
440+ thunk()
441+ }
442+ draining = false
443+ }
444+
445+ isDraining := func() bool { return draining }
446+ enqueue := func(cb func()) {
447+ Trace(Teal("func: enqueue"))
448+ if size == limit {
449+ tmp := draining
450+ drain()
451+ draining = tmp
452+ }
453+ queue[(ix+size)%limit] = cb
454+ size++
455+
456+ if !draining {
457+ drain()
458+ }
459+ }
460+
461+ return Scheduler{isDraining: isDraining, enqueue: enqueue}
462+})()
463+
464+type Supervisor = struct {
465+ register func(fiber Fiber)
466+ isEmpty func() bool
467+ killAll func(killError error, cb func() Any) Any
468+}
469+
470+func mkSupervisor(util_ Any) Supervisor {
471+ // Boilerplate
472+ var util map[string]Any = util_.(map[string]Any)
473+ var isLeft func(Any) Any = util["isLeft"].(func(Any) Any)
474+ var fromLeft func(Any) Any = util["fromLeft"].(func(Any) Any)
475+
476+ // Start here
477+ fibers := NewRegularIntMap()
478+ var fiberID = 0
479+ var count = 0
480+ return Supervisor{
481+ register: func(fiber Fiber) {
482+ Trace("func: supervisor.register", fiber)
483+ fid := fiberID
484+ fiberID++
485+ fiber["onComplete"].(func(Dict) Any)(
486+ Dict{
487+ "rethrow": true,
488+ "handler": func(result Any) Any {
489+ return func() Any {
490+ Trace(Green("🔛✅ Registered on complete handler running"))
491+ count--
492+ fibers.Delete(fid)
493+ return nil
494+ }
495+ },
496+ }).(func() Any)()
497+ fibers.Store(fid, fiber)
498+ count++
499+ },
500+ isEmpty: func() bool {
501+ Trace("func: supervisor.isEmpty")
502+ return count == 0
503+ },
504+ killAll: func(killError error, cb func() Any) Any {
505+ return func() Any {
506+ Trace("func: supervisor.killAll; count = ", count)
507+ if count == 0 {
508+ Trace("\n\nreturning cb")
509+ return cb()
510+ }
511+ killCount := 0
512+ kills := map[int](func() Any){}
513+
514+ kill := func(fid int, fib Fiber) {
515+ Trace(Red("\tMaking killFn for fiber with id ", fiberID))
516+ kills[fid] = fib["kill"].(func(Any, Any) Any)(killError, func(result Any) Any {
517+ return func() Any {
518+ delete(kills, fid)
519+ killCount--
520+ if isLeft(result).(bool) && fromLeft(result) != nil {
521+ go func() {
522+ panic(fromLeft(result)) // [TODO] Panic better?
523+ }()
524+
525+ }
526+ if killCount == 0 {
527+ Trace(Red("\tCalling cb"))
528+ cb()
529+ }
530+ return nil
531+ }
532+ }).(func() Any)().(func() Any)
533+ }
534+
535+ // [TODO] unsafe iteration
536+ for fiberID, _ := range fibers.internal {
537+ killCount++
538+ fib, nope := fibers.Load(fiberID)
539+ Trace(nope, "Just Chuck Testa")
540+ kill(fiberID, fib.(Dict))
541+ }
542+
543+ fibers = NewRegularIntMap()
544+ fiberID = 0
545+ count = 0
546+
547+ return func(err Any) Any {
548+ return Sync{
549+ eff: func() Any {
550+ for _, killThunk := range kills {
551+ killThunk()
552+ }
553+ return nil
554+ }}
555+ }
556+
557+ }
558+ },
559+ }
560+}
561+
562+func mkFiber(util_ Any, supervisor *Supervisor, aff_ Any) Fiber {
563+ var util map[string]Any = util_.(map[string]Any)
564+ var isLeft func(Any) Any = util["isLeft"].(func(Any) Any)
565+ var left func(Any) Any = util["left"].(func(Any) Any)
566+ var right func(Any) Any = util["right"].(func(Any) Any)
567+ var fromRight func(Any) Any = util["fromRight"].(func(Any) Any)
568+ var fromLeft func(Any) Any = util["fromLeft"].(func(Any) Any)
569+ runTick := 0
570+
571+ stepLock := &sync.Mutex{}
572+ step := aff_ // Successful step
573+ setStep := func(who string, newStep Any) {
574+ stepLock.Lock()
575+ Trace(who, "\n\tfrom:\t", step, "\n\tto:\t", newStep)
576+ step = newStep
577+ stepLock.Unlock()
578+ }
579+ var fail Any // Failure step
580+ var interrupt *Dict = nil // Asynchronous interrupt
581+
582+ // Stack of continuations for the current fiber.
583+ var b Cons = Cons{}
584+
585+ // Stack of attempts and finalizers for error recovery. Every `Cons` is also
586+ // tagged with current `interrupt` state. We use this to track which items
587+ // should be ignored or evaluated as a result of a kill.
588+ var attempts *InterruptCons = &InterruptCons{}
589+
590+ // A special state is needed for Bracket, because it cannot be killed. When
591+ // we enter a bracket acquisition or finalizer, we increment the counter,
592+ // and then decrement once complete.
593+ bracketCount := 0
594+
595+ // Each join gets a new id so they can be revoked.
596+ joinId := 0
597+ var joins map[int]OnComplete = nil
598+ var rethrow = true
599+
600+ status := SUSPENDED
601+
602+ var run func(int) Any
603+ run = func(localRunTick int) Any {
604+
605+ Trace("New round", "localRunTick", localRunTick, "bracketCount", bracketCount)
606+ var tmp *Dict
607+ var result Any
608+ var attempt Any
609+ for {
610+ Trace(Purple("\n=======================\nStatus:\t", statusName(status), "\nStep:\t", step, "\n======================="))
611+ tmp = nil
612+ result = nil
613+ attempt = nil
614+ switch status {
615+ case STEP_BIND:
616+ Trace("STEP_BIND", b.head)
617+ status = CONTINUE // next step
618+ headFn := b.head.(func(Any) Any)
619+ // Makeshift try catch block
620+ func() {
621+ defer func() {
622+ if rawErr := recover(); rawErr != nil {
623+ err, ok := rawErr.(error)
624+ if !ok {
625+ Trace(Red("\tNon error error ", rawErr))
626+ panic(error(errors.New("Error of unexpected type")))
627+ }
628+ if err.Error() == "runtime error: invalid memory address or nil pointer dereference" {
629+ Trace(Red("Invalid cast in STEP_BIND", err))
630+ panic(err)
631+ }
632+ if strings.HasPrefix(err.Error(), "interface conversion:") {
633+ Trace(Red("Wrong cast in STEP_BIND ", err))
634+ panic(err)
635+ }
636+ Trace(Red("\tError oh shit ", err))
637+ // early return on error
638+ status = RETURN
639+ fail = left(err)
640+ // step = nil
641+ setStep("bind error", nil)
642+ }
643+ }()
644+ // step = headFn(step)
645+ setStep("bind success", headFn(step))
646+ Trace("\t> Next step", step)
647+ if b.tail == nil {
648+ b.head = nil
649+ } else {
650+ b.head = b.tail.head
651+ b.tail = b.tail.tail
652+ }
653+ }()
654+ case STEP_RESULT:
655+ Trace("STEP_RESULT", step, reflect.TypeOf(step))
656+ if isLeft(step).(bool) {
657+ // early return on error
658+ status = RETURN
659+ fail = step
660+ // step = nil
661+ setStep("result", nil)
662+ } else if b.head == nil {
663+ // happy case done
664+ status = RETURN
665+ Trace("My work here is done")
666+ } else {
667+ // happy case work left
668+ status = STEP_BIND
669+ // step = fromRight(step)
670+ setStep("result", fromRight(step))
671+ Trace("Next step after happy result", step)
672+ }
673+
674+ case CONTINUE:
675+ Trace("CONTINUE")
676+ switch currentStep := step.(type) {
677+
678+ case Bind:
679+ Trace("\tBind")
680+ if b.head != nil {
681+ b.tail = &Cons{head: b.head, tail: b.tail}
682+ }
683+ b.head = currentStep.bToAff
684+ status = CONTINUE
685+ // step = currentStep.affOfB
686+ setStep("continue", currentStep.affOfB)
687+
688+ case Pure:
689+ Trace("\tPure")
690+ if b.head == nil {
691+
692+ Trace("\t> Head nil")
693+ // we're done
694+ status = RETURN
695+ // step = right(currentStep.value)
696+ setStep("pure, return", right(currentStep.value))
697+ } else {
698+ Trace("\t> Head exists", currentStep.value)
699+ // this happens after a bind
700+ // step = currentStep.value
701+ setStep("pure, bind", currentStep.value)
702+ status = STEP_BIND
703+ }
704+
705+ case Sync:
706+ Trace("\tSync")
707+ // step = runSync(left, right, currentStep.eff)
708+ setStep("sync", runSync(left, right, currentStep.eff))
709+ status = STEP_RESULT
710+
711+ case Async:
712+ Trace("\tAsync")
713+ status = PENDING
714+ var asyncCb K = func(theResult Any) Any {
715+ return func() Unit {
716+ if runTick != localRunTick {
717+ Trace("\t\tRun tick != localRunTick")
718+ return nil
719+ }
720+ runTick++
721+ scheduler.enqueue(func() {
722+ if runTick != localRunTick+1 {
723+ Trace("\t\tRun tick != localRunTick + 1")
724+ return
725+ }
726+ // step = theResult
727+ setStep("Async Callback", theResult)
728+ status = STEP_RESULT
729+ run(runTick)
730+ })
731+ return nil
732+ }
733+ }
734+ // runAsync returns a canceller
735+ // step = runAsync(left, currentStep.asyncFn, asyncCb)
736+ setStep("Async ", runAsync(left, currentStep.asyncFn, asyncCb))
737+ return nil
738+
739+ case Throw:
740+ Trace("\tThrow")
741+ status = RETURN
742+ fail = left(currentStep.err)
743+ // step = nil
744+ setStep("Throw", nil)
745+
746+ case Catch:
747+ Trace("\tCatch", currentStep.aff)
748+ if b.head == nil {
749+ Trace("\t\tHead is nil")
750+ attempts = &InterruptCons{head: step, tail: attempts, interrupt: interrupt}
751+ } else {
752+ Trace("\t\tHead is not nil", b.head)
753+ attempts = &InterruptCons{
754+ interrupt: interrupt,
755+ head: step,
756+ tail: &InterruptCons{
757+ interrupt: interrupt,
758+ head: Resume{b: b},
759+ tail: attempts}}
760+ }
761+ b.head = nil
762+ b.tail = nil
763+ status = CONTINUE
764+ // step = currentStep.aff
765+ setStep("Catch", currentStep.aff)
766+
767+ // Enqueue the Bracket so that we can call the appropriate handlers
768+ // after resource acquisition.
769+ case Bracket:
770+ Trace("\tBracket")
771+ bracketCount++
772+ if b.head == nil {
773+ Trace("\t\tb.head == nil")
774+ attempts = &InterruptCons{head: step, tail: attempts, interrupt: interrupt}
775+ } else {
776+ Trace("\t\tb.head /= nil")
777+ attempts = &InterruptCons{
778+ interrupt: interrupt,
779+ head: step,
780+ tail: &InterruptCons{
781+ interrupt: interrupt,
782+ head: Resume{b: b},
783+ tail: attempts}}
784+ }
785+ b.head = nil
786+ b.tail = nil
787+ status = CONTINUE
788+ // step = currentStep.acquire
789+ setStep("Bracket", currentStep.acquire)
790+
791+ case Fork:
792+ Trace("\tFork")
793+ tmpVar := mkFiber(util, supervisor, currentStep.affOfB)
794+ tmp = &tmpVar
795+ if supervisor != nil {
796+ supervisor.register(*tmp)
797+ }
798+ if currentStep.immediate {
799+ (*tmp)["run"].(func() Any)()
800+ }
801+ // step = right(*tmp)
802+ setStep("Fork", right(*tmp))
803+ status = STEP_RESULT
804+
805+ case Sequential:
806+ Trace("\tSequential")
807+ status = CONTINUE
808+ // step = sequential(util, supervisor, currentStep.parAff)
809+ setStep("Sequential", sequential(util, supervisor, currentStep.parAff))
810+
811+ default:
812+ Trace(Purple("Oh god ", reflect.TypeOf(step)))
813+ panic(step)
814+
815+ }
816+
817+ case RETURN:
818+ Trace("RETURN")
819+ b.head = nil
820+ b.tail = nil
821+ Trace("\tAttempts:", attempts)
822+ if attempts.head == nil {
823+ Trace("\tAttempts are nil", interrupt, fail)
824+ status = COMPLETED
825+ if interrupt != nil {
826+ // step = *interrupt
827+ setStep("return interrupt", *interrupt)
828+ } else if fail != nil {
829+ // step = fail
830+ setStep("return fail", fail)
831+ }
832+ } else {
833+ Trace("\tAttempts aren't nil", attempts.head, attempts.tail)
834+ tmp = attempts.interrupt
835+ attempt = attempts.head
836+ attempts = attempts.tail
837+
838+ switch currentAttempt := attempt.(type) {
839+ case Catch:
840+ Trace("\tReturn Catch")
841+ if interrupt != nil && interrupt != tmp && bracketCount == 0 {
842+ Trace("\t\tGonna RETURN")
843+ status = RETURN
844+ } else if fail != nil {
845+ Trace("\t\tGonna CONTINUE", fromLeft(fail))
846+ status = CONTINUE
847+ // step = currentAttempt.errorToAff(fromLeft(fail))
848+ setStep("return catch", currentAttempt.errorToAff(fromLeft(fail)))
849+ fail = nil
850+ }
851+
852+ // We cannot resume from an unmasked interrupt or exception.
853+ case Resume:
854+ Trace("\tResume")
855+ // As with Catch, we only want to ignore in the case of an
856+ // interrupt since enqueing the item.
857+ if (interrupt != nil && interrupt != tmp && bracketCount == 0) || fail != nil {
858+ status = RETURN
859+ } else {
860+ b.head = currentAttempt.b.head
861+ b.tail = currentAttempt.b.tail
862+ status = STEP_BIND
863+ // step = fromRight(step)
864+ setStep("resume ", fromRight(step))
865+ }
866+
867+ case Bracket:
868+ Trace("\tBracket")
869+ bracketCount--
870+ if fail == nil {
871+ result = fromRight(step)
872+ attempts = &InterruptCons{
873+ head: Release{
874+ bracketConditions: currentAttempt.bracketConditions,
875+ result: result,
876+ },
877+ tail: attempts,
878+ interrupt: tmp,
879+ }
880+
881+ if interrupt == tmp || bracketCount > 0 {
882+ Trace("\t\tContinuing")
883+ status = CONTINUE
884+ // step = currentAttempt.k(result)
885+ setStep("Bracket continue", currentAttempt.k(result))
886+ }
887+ }
888+ case Release:
889+ Trace("\tRelease")
890+ attempts = &InterruptCons{
891+ head: Finalized{step: step, fail: fail},
892+ tail: attempts,
893+ interrupt: interrupt,
894+ }
895+ status = CONTINUE
896+ // It has only been killed if the interrupt status has changed
897+ // since we enqueued the item, and the bracket count is 0. If the
898+ // bracket count is non-zero then we are in a masked state so it's
899+ // impossible to be killed.
900+ if interrupt != nil && &interrupt != &tmp && bracketCount == 0 {
901+ Trace("\t\tkilled branch")
902+ // step = currentAttempt.bracketConditions["killed"].(func(Any) Any)(fromLeft(*interrupt)).(func(Any) Any)(currentAttempt.result)
903+ setStep("Release killed", currentAttempt.bracketConditions["killed"].(func(Any) Any)(fromLeft(*interrupt)).(func(Any) Any)(currentAttempt.result))
904+ } else if fail != nil {
905+ Trace("\t\tfailed branch")
906+ // step = currentAttempt.bracketConditions["failed"].(func(Any) Any)(fromLeft(fail)).(func(Any) Any)(currentAttempt.result)
907+ setStep("Release failed", currentAttempt.bracketConditions["failed"].(func(Any) Any)(fromLeft(fail)).(func(Any) Any)(currentAttempt.result))
908+ } else {
909+ Trace("\t\tcompleted branch")
910+ // step = currentAttempt.bracketConditions["completed"].(func(Any) Any)(fromRight(step)).(func(Any) Any)(currentAttempt.result)
911+ setStep("Release completed", currentAttempt.bracketConditions["completed"].(func(Any) Any)(fromRight(step)).(func(Any) Any)(currentAttempt.result))
912+ }
913+ fail = nil
914+ bracketCount++
915+ case Finalizer:
916+ Trace("\tFinalizer")
917+ bracketCount++
918+ attempts = &InterruptCons{
919+ head: Finalized{step: step, fail: fail},
920+ tail: attempts,
921+ interrupt: interrupt,
922+ }
923+ status = CONTINUE
924+ // step = currentAttempt.finalizer
925+ setStep("Finalizer", currentAttempt.finalizer)
926+ case Finalized:
927+ Trace("\tFinalized")
928+ bracketCount--
929+ status = RETURN
930+ // step = currentAttempt.step
931+ setStep("Finalized", currentAttempt.step)
932+ fail = currentAttempt.fail
933+ default:
934+ Trace(Fata("\tUnknown Attempt type hit", reflect.TypeOf(currentAttempt)))
935+ }
936+ }
937+ case COMPLETED:
938+ Trace("COMPLETED", joins)
939+ for _, join := range joins {
940+ if rethrow {
941+ rethrow = join["rethrow"].(bool)
942+ }
943+ Trace(Green("\tRunning onComplete ✅ handler"))
944+ runEff(join["handler"].(func(Any) Any)(step).(func() Any))
945+ }
946+ joins = nil
947+ // If we have an interrupt and a fail, then the thread threw while
948+ // running finalizers. This should always rethrow in a fresh stack.
949+ if interrupt != nil && fail != nil {
950+ go func() {
951+ // [TODO]: How can we avoid this?
952+ // [TODO]: Do we need this here?
953+ // time.Sleep(50 * time.Millisecond) // [TODO]: How can we avoid this?
954+ go panic(fromLeft(fail))
955+ }()
956+
957+ // If we have an unhandled exception, and no other fiber has joined
958+ // then we need to throw the exception in a fresh stack.
959+ } else if isLeft(step).(bool) && rethrow {
960+ go func() {
961+ // Guard on rethrow because a completely synchronous fiber can
962+ // still have an observer which was added after-the-fact.
963+ time.Sleep(80 * time.Millisecond) // [TODO]: How can we avoid this?
964+ runtime.Gosched()
965+ if rethrow {
966+ panic(fromLeft(fail))
967+ }
968+ }()
969+ }
970+ return nil
971+
972+ case SUSPENDED:
973+ Trace("SUSPENDED")
974+ status = CONTINUE
975+ case PENDING:
976+ Trace("PENDING")
977+ return nil
978+
979+ default:
980+ Trace(Fata("Unknown branch hit", step))
981+ }
982+ }
983+
984+ }
985+
986+ onComplete := func(join OnComplete) Any {
987+ return func() Any {
988+ Trace(Teal("func: onComplete"))
989+ if status == COMPLETED {
990+ Trace("\tstatus == COMPLETED")
991+ rethrow = rethrow && join["rethrow"].(bool)
992+ join["handler"].(func(Any) Any)(step).(func() Any)()
993+ return func() Any { return nil }
994+ }
995+ Trace("\tstatus == ", statusName(status))
996+
997+ var jid = joinId
998+ Trace("\tJoin id", jid)
999+ joinId++
1000+ if joins == nil {
1001+ joins = map[int]OnComplete{}
1002+ }
1003+ joins[jid] = join
1004+
1005+ return func() Any {
1006+ if joins != nil {
1007+ delete(joins, jid)
1008+ }
1009+ return nil
1010+ }
1011+ }
1012+ }
1013+
1014+ kill := func(err Any, cb_ Any) Any {
1015+ cb := cb_.(func(Any) Any)
1016+ return func() Any {
1017+ if status == COMPLETED {
1018+ Trace(Green("Completed, no need to cancel"))
1019+ cb(right(nil)).(func() Any)()
1020+ return func() Any { return nil }
1021+ }
1022+
1023+ var canceler (func() Any) = onComplete(Dict{
1024+ "rethrow": false,
1025+ "handler": func(_ Any) Any {
1026+ return cb(right(nil))
1027+ },
1028+ }).(func() Any)().(func() Any)
1029+
1030+ switch status {
1031+ case SUSPENDED:
1032+ interruptValue := left(err).(Dict)
1033+ interrupt = &interruptValue
1034+ status = COMPLETED
1035+ step = *interrupt
1036+ run(runTick)
1037+ case PENDING:
1038+ if interrupt == nil {
1039+ interruptValue := left(err).(Dict)
1040+ interrupt = &interruptValue
1041+ }
1042+ if bracketCount == 0 {
1043+ if status == PENDING {
1044+ attempts = &InterruptCons{
1045+ head: Finalizer{finalizer: step.(func(Any) Any)(err)},
1046+ tail: attempts,
1047+ interrupt: interrupt}
1048+ }
1049+ status = RETURN
1050+ step = nil
1051+ fail = nil
1052+ runTick++
1053+ run(runTick)
1054+ }
1055+ default:
1056+ if interrupt == nil {
1057+ interruptValue := left(err).(Dict)
1058+ interrupt = &interruptValue
1059+ }
1060+ if bracketCount == 0 {
1061+ status = RETURN
1062+ step = nil
1063+ fail = nil
1064+ }
1065+ }
1066+
1067+ return canceler
1068+ }
1069+ }
1070+
1071+ join := func(cb K) Any {
1072+ return func() Any {
1073+ Trace(Teal("func: join 1"))
1074+ Trace("\tjoin: status: ", statusName(status))
1075+ canceler := onComplete(
1076+ Dict{
1077+ "rethrow": false,
1078+ "handler": cb.(func(Any) Any),
1079+ }).(func() Any)()
1080+ if status == SUSPENDED {
1081+ Trace("join: \tcalling run")
1082+ run(runTick)
1083+ }
1084+ return canceler
1085+ }
1086+ }
1087+
1088+ return Fiber{
1089+ "run": func() Any {
1090+ if status == SUSPENDED {
1091+ if !scheduler.isDraining() {
1092+ scheduler.enqueue(func() {
1093+ run(runTick)
1094+ })
1095+ } else {
1096+ run(runTick)
1097+ }
1098+ }
1099+ return nil // Important to avoid panic
1100+ },
1101+ "kill": kill,
1102+ "join": join,
1103+ "onComplete": onComplete,
1104+ "isSuspended": func() Any { return status == SUSPENDED },
1105+ }
1106+}
1107+
1108+func runPar(util_ Any, supervisor *Supervisor, par Any, cb Any) Any {
1109+
1110+ var util map[string]Any = util_.(map[string]Any)
1111+ var isLeft func(Any) Any = util["isLeft"].(func(Any) Any)
1112+ var left func(Any) Any = util["left"].(func(Any) Any)
1113+ var right func(Any) Any = util["right"].(func(Any) Any)
1114+ var fromRight func(Any) Any = util["fromRight"].(func(Any) Any)
1115+ var fromLeft func(Any) Any = util["fromLeft"].(func(Any) Any)
1116+
1117+ fiberID := 0
1118+ fibers := NewRegularIntMap()
1119+
1120+ killID := 0
1121+ kills := NewRegularIntMap()
1122+
1123+ early := errors.New("[ParAff] Early exit")
1124+
1125+ var interrupt Any = nil
1126+
1127+ var root Any = Empty
1128+
1129+ kill := func(err error, par *AbstractAff, cb K) Any {
1130+ step := par
1131+ var head *AbstractAff
1132+ var tail *Cons
1133+
1134+ count := 0
1135+ kills := NewRegularIntMap()
1136+ var tmp Any
1137+ var kid int
1138+
1139+ loop:
1140+ for {
1141+ tmp = nil
1142+ Trace("Step", reflect.TypeOf(step))
1143+
1144+ switch step.tag {
1145+ case "Forked":
1146+ if step._3 == Empty {
1147+ tmp, _ = fibers.Load(step._1.(int))
1148+ tmpKillFn := tmp.(Fiber)["kill"].(func(Any, Any) Any)
1149+ kills.Store(count, tmpKillFn(err, func(result Any) Any {
1150+ return func() Any {
1151+ count--
1152+ if count == 0 {
1153+ cb.(func(Any) Any)(result).(func() Any)()
1154+ }
1155+ return nil
1156+ }
1157+ }))
1158+ count++
1159+ }
1160+ if head == nil {
1161+ break loop
1162+ }
1163+
1164+ step = head._2.(*AbstractAff)
1165+ if tail == nil {
1166+ head = nil
1167+ } else {
1168+ head = tail.head.(*AbstractAff)
1169+ tail = tail.tail
1170+ }
1171+ case "ParMap":
1172+ step = step._2.(*AbstractAff)
1173+ case "ParApply":
1174+ if head != nil {
1175+ tail = &Cons{head: head, tail: tail}
1176+ }
1177+ head = step
1178+ step = step._1.(*AbstractAff)
1179+ case "ParAlt":
1180+ if head != nil {
1181+ tail = &Cons{head: head, tail: tail}
1182+ }
1183+ head = step
1184+ step = step._1.(*AbstractAff)
1185+ }
1186+ }
1187+ if count == 0 {
1188+ cb.(func(Any) Any)(right(nil)).(func() Any)()
1189+ } else {
1190+ // Run the cancelation effects. We alias `count` because it's mutable.
1191+ tmp = count
1192+ for kid = 0; kid < tmp.(int); kid++ {
1193+ kill, ok := kills.Load(kid)
1194+ if ok {
1195+ kills.Store(kid, kill.(func() Any)())
1196+ }
1197+
1198+ }
1199+ }
1200+ return kills
1201+ }
1202+
1203+ var join func(result Any, cons *Cons)
1204+ join = func(result Any, cons *Cons) {
1205+ Trace("Join")
1206+ var head *AbstractAff
1207+ var tail *Cons
1208+ if cons == nil {
1209+ head = nil
1210+ cons = nil
1211+ } else {
1212+ head = cons.head.(*AbstractAff)
1213+ tail = cons.tail
1214+ }
1215+ Trace(Teal("func: join 2 ", result, head))
1216+ var fail Any
1217+ var step Any
1218+ var lhs Any
1219+ var rhs Any
1220+ var tmp Any
1221+ var kid Any
1222+
1223+ if isLeft(result).(bool) {
1224+ fail = result
1225+ step = nil
1226+ } else {
1227+ step = result
1228+ fail = nil
1229+ }
1230+
1231+ for {
1232+ Trace(Magenta("\n############\njoinPar:\t", head, "\nstep:\t", step, "\n#############"))
1233+ // printCons(cons)
1234+ Trace(head)
1235+ Trace("---------\n")
1236+ lhs = nil
1237+ rhs = nil
1238+ tmp = nil
1239+ kid = nil
1240+
1241+ // We should never continue if the entire tree has been interrupted.
1242+ if interrupt != nil {
1243+ return
1244+ }
1245+
1246+ // We've made it all the way to the root of the tree, which means
1247+ // the tree has fully evaluated.
1248+ if head == nil {
1249+ var cbArg Any = nil
1250+ if fail != nil {
1251+ cbArg = fail
1252+ } else {
1253+ cbArg = step
1254+ }
1255+ cb.(func(Any) Any)(cbArg).(func() Any)()
1256+ return
1257+ }
1258+
1259+ if head._3 != Empty {
1260+ return
1261+ }
1262+
1263+ switch head.tag {
1264+
1265+ case "ParMap":
1266+ Trace("Eating ParMap")
1267+ if fail == nil {
1268+ // Unwrap the Right and apply the function in the map and
1269+ // stick it back in a Right
1270+ head._3 = right(head._1.(func(Any) Any)(fromRight(step)))
1271+ step = head._3
1272+ } else {
1273+ head._3 = fail
1274+ }
1275+ case "ParApply":
1276+ Trace("Eating ParApply")
1277+ lhs = head._1.(*AbstractAff)._3
1278+ rhs = head._2.(*AbstractAff)._3
1279+ // If we have a failure we should kill the other side because we
1280+ // can't possible yield a result anymore.
1281+ if fail != nil {
1282+ head._3 = fail
1283+ tmp = true
1284+ kid = killID
1285+ killID++
1286+
1287+ var someTmp Any
1288+ if isLeft(lhs).(bool) && (fromLeft(fail) == fromLeft(lhs)) {
1289+ someTmp = head._2
1290+ } else {
1291+ someTmp = head._1
1292+ }
1293+ kills.Store(kid.(int), kill(early.(error), someTmp.(*AbstractAff), func(unused Any) Any {
1294+ return func() Any {
1295+ kills.Delete(kid.(int))
1296+ if tmp.(bool) {
1297+ tmp = false
1298+ } else if tail == nil {
1299+ join(fail, nil)
1300+ } else {
1301+ join(fail, tail)
1302+ }
1303+ return nil
1304+ }
1305+ }))
1306+
1307+ if tmp != nil {
1308+ tmp = false
1309+ return
1310+ }
1311+ } else if lhs == Empty || rhs == Empty {
1312+ // We can only proceed if both sides have resolved.
1313+ Trace("One side EMPTY", lhs, rhs)
1314+ return
1315+ } else {
1316+ Trace("Both sides not EMPTY", lhs, rhs)
1317+ step = right(fromRight(lhs).(Fn1)(fromRight(rhs)))
1318+ head._3 = step
1319+ }
1320+ case "ParAlt":
1321+ lhs = head._1.(*AbstractAff)._3
1322+ rhs = head._2.(*AbstractAff)._3
1323+ // We can only proceed if both have resolved or we have a success
1324+ if lhs == Empty && isLeft(rhs).(bool) || rhs == Empty && isLeft(lhs).(bool) {
1325+ return
1326+ }
1327+ // If both sides resolve with an error, we should continue with the
1328+ // first error
1329+ if lhs != Empty && isLeft(lhs).(bool) && rhs != Empty && isLeft(rhs).(bool) {
1330+ var tmpStep Any
1331+ if step == lhs {
1332+ tmpStep = rhs
1333+ } else {
1334+ tmpStep = lhs
1335+ }
1336+ fail = &tmpStep
1337+ step = nil
1338+ head._3 = fail
1339+ } else {
1340+ head._3 = step
1341+ tmp = true
1342+ kid = killID
1343+ killID++
1344+ // Once a side has resolved, we need to cancel the side that is still
1345+ // pending before we can continue.
1346+ var headArg Any
1347+ if &step == &lhs {
1348+ headArg = head._2
1349+ } else {
1350+ headArg = head._1
1351+ }
1352+ kills.Store(kid.(int), kill(early.(error), headArg.(*AbstractAff), func(unused Any) Any {
1353+ return func() Any {
1354+ kills.Delete(kid.(int))
1355+ if tmp.(bool) {
1356+ tmp = false
1357+ } else if tail == nil {
1358+ join(step, nil)
1359+ } else {
1360+ join(step, tail)
1361+ }
1362+ return nil
1363+ }
1364+ }))
1365+
1366+ if tmp.(bool) {
1367+ tmp = false
1368+ return
1369+ }
1370+ }
1371+ }
1372+
1373+ if tail == nil {
1374+ head = nil
1375+ } else {
1376+ head = tail.head.(*AbstractAff)
1377+ tail = tail.tail
1378+ }
1379+
1380+ }
1381+ }
1382+
1383+ resolve := func(fiber *AbstractAff) Any {
1384+ return func(result Any) Any {
1385+ return func() Any {
1386+ Trace("Resolving fiber ", fiber._1.(int))
1387+ fibers.Delete(fiber._1.(int))
1388+ fiber._3 = result
1389+ join(result, fiber._2.(*Cons))
1390+ return nil
1391+ }
1392+ }
1393+ }
1394+
1395+ // Walks the applicative tree, substituting non-applicative nodes with
1396+ // `FORKED` nodes. In this tree, all applicative nodes use the `_3` slot
1397+ // as a mutable slot for memoization. In an unresolved state, the `_3`
1398+ // slot is `EMPTY`. In the cases of `ALT` and `APPLY`, we always walk
1399+ // the left side first, because both operations are left-associative. As
1400+ // we `RETURN` from those branches, we then walk the right side.
1401+
1402+ run := func() {
1403+ Trace(Green("Run"))
1404+ status := CONTINUE
1405+ step := par
1406+ var head *AbstractAff = nil
1407+ var tail *Cons = nil
1408+ var tmp Any = nil
1409+ fid := 0
1410+
1411+ loop:
1412+ for {
1413+ tmp = nil
1414+ Trace(Green("\n############\nrunPar:\t", statusName(status)))
1415+ printCons(&Cons{head: head, tail: tail})
1416+ // Trace("step:\t", step)
1417+ // Trace(Green("#############\n\n"))
1418+ switch status {
1419+ case CONTINUE:
1420+ switch currentStep := step.(type) {
1421+ case AbstractAff:
1422+ switch currentStep.tag {
1423+ case "ParMap":
1424+ Trace(Green("ParMap"))
1425+ if head != nil {
1426+ tail = &Cons{head: head, tail: tail}
1427+ }
1428+ head = &AbstractAff{tag: "ParMap", _1: currentStep._1, _2: Empty, _3: Empty}
1429+ step = currentStep._2
1430+ case "ParApply":
1431+ Trace(Green("ParApply"))
1432+ if head != nil {
1433+ tail = &Cons{head: head, tail: tail}
1434+ }
1435+ head = &AbstractAff{tag: "ParApply", _1: Empty, _2: currentStep._2, _3: Empty}
1436+ step = currentStep._1
1437+ case "ParAlt":
1438+ Trace(Green("ParAlt"))
1439+ if head != nil {
1440+ tail = &Cons{head: head, tail: tail}
1441+ }
1442+ head = &AbstractAff{tag: "ParAlt", _1: Empty, _2: currentStep._2, _3: Empty}
1443+ step = currentStep._1
1444+ default:
1445+ panic(errors.New("Unexpected case " + currentStep.tag))
1446+ }
1447+ default:
1448+ Trace(Green("default, step: ", currentStep))
1449+ // When we hit a leaf value, we suspend the stack in the `FORKED`.
1450+ // When the fiber resolves, it can bubble back up the tree.
1451+ fid = fiberID
1452+ fiberID++
1453+ status = RETURN
1454+ tmp = step
1455+ step = &AbstractAff{tag: "Forked", _1: fid, _2: &Cons{head: head, tail: tail}, _3: Empty}
1456+ tmp = mkFiber(util, supervisor, tmp)
1457+ tmp.(Fiber)["onComplete"].(func(Dict) Any)(OnComplete{
1458+ "rethrow": false,
1459+ "handler": resolve(step.(*AbstractAff)).(func(Any) Any),
1460+ }).(EffFn)()
1461+ fibers.Store(fid, tmp.(Fiber))
1462+ if supervisor != nil {
1463+ supervisor.register(tmp.(Fiber))
1464+ }
1465+ }
1466+ case RETURN:
1467+ // Terminal case. We are back at the root
1468+ if head == nil {
1469+ break loop
1470+ }
1471+ // If we are done with the right side, we need to continue down the
1472+ // left. Otherwise we should continue up the stack.
1473+
1474+ if head._1 == Empty {
1475+ Trace("Head._1 is empty")
1476+ head._1 = step
1477+ status = CONTINUE
1478+ step = head._2
1479+ head._2 = Empty
1480+ } else {
1481+ Trace("Head._1 is not empty")
1482+ head._2 = step
1483+ step = head
1484+ if tail == nil {
1485+ head = nil
1486+ } else {
1487+ head = tail.head.(*AbstractAff)
1488+ tail = tail.tail
1489+ }
1490+ }
1491+ }
1492+ }
1493+ // Keep a reference to the tree root so it can be cancelled.
1494+ root = step
1495+
1496+ for fid = 0; fid < fiberID; fid++ {
1497+ Trace(Magenta("Running fiber ", fid))
1498+ Trace("Id", fid)
1499+ fiber, ok := fibers.Load(fid)
1500+ if !ok {
1501+ Trace(fibers.internal)
1502+ } else {
1503+ runFn := fiber.(Dict)["run"].(func() Any)
1504+ // [TODO] How about some real parallelism with `go` here?
1505+ // Well no luck due to "concurrent map writes"
1506+ go runFn()
1507+ }
1508+ }
1509+
1510+ }
1511+
1512+ // Cancels the entire tree. If there are already subtrees being canceled,
1513+ // we need to first cancel those joins. We will then add fresh joins for
1514+ // all pending branches including those that were in the process of being
1515+ // canceled
1516+ cancel := func(cancelErr error, cb K) Any {
1517+ interrupt = left(cancelErr)
1518+ kills.RWMutex.RLock()
1519+ for innerKillID := range kills.internal {
1520+ innerKillsV, ok := kills.Load(innerKillID)
1521+ innerKills := innerKillsV.(*IntMap)
1522+ innerKills.RWMutex.RLock()
1523+ if ok {
1524+ for innerKillID := range innerKills.internal {
1525+ innerKill, ok := innerKills.Load(innerKillID)
1526+ if ok {
1527+ innerKill.(EffFn)()
1528+ }
1529+ }
1530+ }
1531+ innerKills.RWMutex.RUnlock()
1532+ }
1533+ kills.RWMutex.RUnlock()
1534+ kills = nil
1535+ newKills := kill(cancelErr, root.(*AbstractAff), cb)
1536+
1537+ return func(killError Any) Any {
1538+ return Async{asyncFn: func(killCb K) Any {
1539+ return func() Any {
1540+ for _, killFn := range newKills.(map[Any]Any) {
1541+ killFn.(EffFn)()
1542+ }
1543+ return nonCanceler
1544+ }
1545+ }}
1546+ }
1547+ }
1548+
1549+ run()
1550+
1551+ return func(killError Any) Any {
1552+ return Async{asyncFn: func(k K) Any {
1553+ return func() Any {
1554+ return cancel(killError.(error), k)
1555+ }
1556+ }}
1557+ }
1558+
1559+}
1560+
1561+func init() {
1562+ exports := Foreign("Effect.Aff")
1563+
1564+ // ∷ ∀ a b. Aff a → (a → Aff b) → Aff b
1565+ exports["_pure"] = func(value Any) Any {
1566+ return Pure{Pure: true, value: value}
1567+ }
1568+ // ∷ ∀ a. Error → Aff a
1569+ exports["_throwError"] = func(e_ Any) Any {
1570+ e := e_.(error)
1571+ return Throw{Throw: true, err: e}
1572+ }
1573+ // ∷ ∀ a. Aff a → (Error → Aff a) → Aff a
1574+ exports["_catchError"] = func(aff Any) Any {
1575+ return func(k Any) Any {
1576+ return Catch{aff: aff, errorToAff: k.(func(Any) Any)}
1577+ }
1578+ }
1579+ // ∷ ∀ a b. (a → b) → Aff a → Aff b
1580+ exports["_map"] = func(f_ Any) Any {
1581+ Trace(Yellow("map"))
1582+ f := f_.(func(Any) Any)
1583+ return func(aff Any) Any {
1584+ switch a := aff.(type) {
1585+ case Pure:
1586+ return Pure{Pure: true, value: f(a.value)}
1587+ default:
1588+ return Bind{
1589+ Bind: true,
1590+ affOfB: aff,
1591+ bToAff: func(b Any) Any { return Pure{Pure: true, value: f(b)} }}
1592+ }
1593+ }
1594+ }
1595+
1596+ // ∷ ∀ a b. Aff a → (a → Aff b) → Aff b
1597+ exports["_bind"] = func(aff Any) Any {
1598+ Trace(Yellow("bind"))
1599+ return func(f_ Any) Any {
1600+ f := f_.(func(Any) Any)
1601+ bToAff := func(b Any) Any { return f(b) }
1602+ return Bind{affOfB: aff, bToAff: bToAff}
1603+ }
1604+ }
1605+
1606+ exports["_fork"] = func(immediate Any) Any {
1607+ return func(aff Any) Any {
1608+ return Fork{
1609+ immediate: immediate.(bool),
1610+ affOfB: aff,
1611+ }
1612+ }
1613+ }
1614+
1615+ // ∀ a. Effect a → Aff a
1616+ exports["_liftEffect"] = func(effect_ Any) Any {
1617+ effect := effect_.(func() Any)
1618+ return Sync{Sync: true, eff: effect}
1619+ }
1620+
1621+ exports["_parAffMap"] = func(f Any) Any {
1622+ return func(aff Any) Any {
1623+ return AbstractAff{tag: "ParMap", _1: f, _2: aff}
1624+ }
1625+ }
1626+
1627+ exports["_parAffApply"] = func(aff1 Any) Any {
1628+ return func(aff2 Any) Any {
1629+ return AbstractAff{tag: "ParApply", _1: aff1, _2: aff2}
1630+ }
1631+ }
1632+
1633+ exports["_parAffAlt"] = func(aff1 Any) Any {
1634+ return func(aff2 Any) Any {
1635+ return AbstractAff{tag: "ParAlt", _1: aff1, _2: aff2}
1636+ }
1637+ }
1638+
1639+ // ∷ ∀ a. Fn.Fn2 FFIUtil (Aff a) (Effect (Fiber a))
1640+ exports["_makeFiber"] = func(util Any, aff Any) Any {
1641+ return func() Any {
1642+ return mkFiber(util, nil, aff)
1643+ }
1644+ }
1645+
1646+ // ∷ ∀ a. Fn.Fn2 FFIUtil (Aff a) (Effect (Fiber a))
1647+ exports["_makeSupervisedFiber"] = func(util Any, aff Any) Any {
1648+ return func() Any {
1649+ supervisor := mkSupervisor(util)
1650+ return Dict{
1651+ "fiber": mkFiber(util, &supervisor, aff),
1652+ "supervisor": supervisor,
1653+ }
1654+ }
1655+ }
1656+
1657+ // Fn.Fn3 Error Supervisor (Effect Unit) (Effect Canceler)
1658+ exports["_killAll"] = func(err_ Any, supervisor_ Any, cb_ Any) Any {
1659+ supervisor := supervisor_.(Supervisor)
1660+ cb := cb_.(func() Any)
1661+ err := err_.(error)
1662+ return supervisor.killAll(err, cb)
1663+ }
1664+
1665+ // ∀ a. Fn.Fn2 (Unit → Either a Unit) Number (Aff Unit)
1666+ exports["_delay"] = (func() Any {
1667+
1668+ setDelay := func(duration time.Duration, k func()) *time.Timer {
1669+ return time.AfterFunc(duration, k)
1670+ }
1671+
1672+ clearDelay := func(timer *time.Timer) {
1673+ if timer != nil {
1674+ if !timer.Stop() {
1675+ <-timer.C
1676+ }
1677+ }
1678+ }
1679+
1680+ return func(right_ Any, millis_ Any) Any {
1681+ right := right_.(func(Any) Any)
1682+ var millis int = int(millis_.(float64))
1683+ return Async{
1684+ asyncFn: func(cb K) Any {
1685+ return func() Any {
1686+ duration := time.Duration(millis) * time.Millisecond
1687+ k := Apply(cb, right(nil))
1688+ t1 := time.Now()
1689+ timer := setDelay(duration, func() {
1690+ Trace("Done after: ", time.Now().Sub(t1))
1691+ k.(func() Any)()
1692+ })
1693+ return func(err Any) Any {
1694+ Trace(Magenta("Almost canceling delay"))
1695+ return Sync{eff: func() Any {
1696+ // Trace(Magenta("Canceling delay"))
1697+ clearDelay(timer)
1698+ return right(nil)
1699+ }}
1700+ }
1701+ }
1702+ },
1703+ }
1704+ }
1705+ })()
1706+
1707+ exports["makeAff"] = func(asyncFn Any) Any {
1708+ Trace("I'm making an aff")
1709+ return Async{asyncFn: asyncFn}
1710+ }
1711+
1712+ exports["generalBracket"] = func(acquire Any) Any {
1713+ return func(options Any) Any {
1714+ bracketConditions := options.(Dict)
1715+ return func(k_ Any) Any {
1716+ k := k_.(func(Any) Any)
1717+ return Bracket{
1718+ acquire: acquire,
1719+ bracketConditions: bracketConditions,
1720+ k: k,
1721+ }
1722+ }
1723+ }
1724+ }
1725+
1726+ // ParAff ~> Aff
1727+ exports["_sequential"] = func(aff Any) Any {
1728+ return Sequential{parAff: aff}
1729+ }
1730+
1731+}
1732+
1733+type IntMap struct {
1734+ sync.RWMutex
1735+ internal map[int]Any
1736+}
1737+
1738+func NewRegularIntMap() *IntMap {
1739+ return &IntMap{
1740+ internal: make(map[int]Any),
1741+ }
1742+}
1743+
1744+func (rm *IntMap) Load(key int) (value Any, ok bool) {
1745+ rm.RLock()
1746+ result, ok := rm.internal[key]
1747+ rm.RUnlock()
1748+ return result, ok
1749+}
1750+
1751+func (rm *IntMap) Delete(key int) {
1752+ Trace(Red("Deleting key ", key))
1753+ rm.Lock()
1754+ delete(rm.internal, key)
1755+ rm.Unlock()
1756+}
1757+
1758+func (rm *IntMap) Store(key int, value Any) {
1759+ Trace(Green("Storing key ", key))
1760+ rm.Lock()
1761+ rm.internal[key] = value
1762+ rm.Unlock()
1763+}