Skip to content

Commit 881d22d

Browse files
mpilquistmpilquist
authored
Merge pull request #2864 from typelevel/revert-2828-unwind_bind_bind
Revert "Pull - Cut ties with BindBind, no need for BuildR"
2 parents f504097 + fd1d154 commit 881d22d

File tree

1 file changed

+76
-60
lines changed

1 file changed

+76
-60
lines changed

‎core/shared/src/main/scala/fs2/Pull.scala

+76-60
Original file line numberDiff line numberDiff line change
@@ -654,10 +654,9 @@ object Pull extends PullLowPriority {
654654
def cont(r: Terminal[Unit]): Pull[Pure, INothing, Unit] = r
655655
}
656656

657-
private abstract class Bind[+F[_], +O, X, +R](misstep: Pull[F, O, X])
657+
private abstract class Bind[+F[_], +O, X, +R](val step: Pull[F, O, X])
658658
extends Pull[F, O, R]
659659
with ContP[X, F, O, R] {
660-
def step: Pull[F, O, X] = misstep
661660
def cont(r: Terminal[X]): Pull[F, O, R]
662661
def delegate: Bind[F, O, X, R] = this
663662
}
@@ -694,12 +693,12 @@ object Pull extends PullLowPriority {
694693

695694
// This class is not created by combinators in public Pull API, only during compilation
696695
private class BindBind[F[_], O, X, Y](
697-
var innerBind: Bind[F, O, X, Y],
698-
var endBind: Bind[F, O, Y, Unit]
699-
) extends Bind[F, O, X, Unit](null) {
700-
override def step: Pull[F, O, X] = innerBind.step
701-
def cont(xterm: Terminal[X]): Pull[F, O, Unit] =
702-
try bindBindAux(xterm, this)
696+
step: Pull[F, O, X],
697+
val bb: Bind[F, O, X, Y],
698+
val del: Bind[F, O, Y, Unit]
699+
) extends Bind[F, O, X, Unit](step) {
700+
def cont(tx: Terminal[X]): Pull[F, O, Unit] =
701+
try bindBindAux(bb.cont(tx), del)
703702
catch { case NonFatal(e) => Fail(e) }
704703
}
705704

@@ -712,12 +711,7 @@ object Pull extends PullLowPriority {
712711
case ty: Terminal[_] =>
713712
del match {
714713
case cici: BindBind[F, O, r, Y] =>
715-
val innerBind = cici.innerBind
716-
val endBind = cici.endBind
717-
cici.innerBind = null
718-
cici.endBind = null
719-
val nextStep = innerBind.cont(ty)
720-
bindBindAux[F, O, r, Y](nextStep, endBind)
714+
bindBindAux[F, O, r, Y](cici.bb.cont(ty), cici.del)
721715
case _ => del.cont(ty)
722716
}
723717
case x => new DelegateBind(x, del)
@@ -871,7 +865,7 @@ object Pull extends PullLowPriority {
871865
case b: Bind[G, X, y, Unit] =>
872866
b.step match {
873867
case c: Bind[G, X, x, _] =>
874-
viewL(new BindBind[G, X, x, y](c, b.delegate))
868+
viewL(new BindBind[G, X, x, y](c.step, c.delegate, b.delegate))
875869
case e: Action[G, X, y2] =>
876870
contP = b.delegate
877871
e
@@ -904,22 +898,37 @@ object Pull extends PullLowPriority {
904898

905899
}
906900

907-
trait Run[-G[_], -X] {
908-
def done(scope: Scope[F]): F[B]
909-
def out(head: Chunk[X], scope: Scope[F], tail: Pull[G, X, Unit]): F[B]
910-
def interrupted(inter: Interrupted): F[B]
911-
def fail(e: Throwable): F[B]
901+
trait Run[-G[_], -X, +End] {
902+
def done(scope: Scope[F]): End
903+
def out(head: Chunk[X], scope: Scope[F], tail: Pull[G, X, Unit]): End
904+
def interrupted(inter: Interrupted): End
905+
def fail(e: Throwable): End
912906
}
907+
type CallRun[+G[_], +X, End] = Run[G, X, End] => End
908+
909+
object TheBuildR extends Run[Pure, INothing, F[CallRun[Pure, Nothing, F[INothing]]]] {
910+
type TheRun = Run[Pure, INothing, F[INothing]]
911+
def fail(e: Throwable) = F.raiseError(e)
912+
def done(scope: Scope[F]) =
913+
F.pure((cont: TheRun) => cont.done(scope))
914+
def out(head: Chunk[INothing], scope: Scope[F], tail: Pull[Pure, INothing, Unit]) =
915+
F.pure((cont: TheRun) => cont.out(head, scope, tail))
916+
def interrupted(i: Interrupted) =
917+
F.pure((cont: TheRun) => cont.interrupted(i))
918+
}
919+
920+
def buildR[G[_], X, End]: Run[G, X, F[CallRun[G, X, F[End]]]] =
921+
TheBuildR.asInstanceOf[Run[G, X, F[CallRun[G, X, F[End]]]]]
913922

914-
def go[G[_], X](
923+
def go[G[_], X, End](
915924
scope: Scope[F],
916925
extendedTopLevelScope: Option[Scope[F]],
917926
translation: G ~> F,
918-
runner: Run[G, X],
927+
runner: Run[G, X, F[End]],
919928
stream: Pull[G, X, Unit]
920-
): F[B] = {
929+
): F[End] = {
921930

922-
def interruptGuard(scope: Scope[F], view: Cont[INothing, G, X])(next: => F[B]): F[B] =
931+
def interruptGuard(scope: Scope[F], view: Cont[INothing, G, X])(next: => F[End]): F[End] =
923932
scope.isInterrupted.flatMap {
924933
case None => next
925934
case Some(outcome) =>
@@ -931,18 +940,18 @@ object Pull extends PullLowPriority {
931940
go(scope, extendedTopLevelScope, translation, runner, view(result))
932941
}
933942

934-
def goErr(err: Throwable, view: Cont[Nothing, G, X]): F[B] =
943+
def goErr(err: Throwable, view: Cont[Nothing, G, X]): F[End] =
935944
go(scope, extendedTopLevelScope, translation, runner, view(Fail(err)))
936945

937-
class ViewRunner(val view: Cont[Unit, G, X]) extends Run[G, X] {
946+
class ViewRunner(val view: Cont[Unit, G, X]) extends Run[G, X, F[End]] {
938947
private val prevRunner = runner
939948

940-
def done(doneScope: Scope[F]): F[B] =
949+
def done(doneScope: Scope[F]): F[End] =
941950
go(doneScope, extendedTopLevelScope, translation, prevRunner, view(unit))
942951

943-
def out(head: Chunk[X], scope: Scope[F], tail: Pull[G, X, Unit]): F[B] = {
952+
def out(head: Chunk[X], scope: Scope[F], tail: Pull[G, X, Unit]): F[End] = {
944953
@tailrec
945-
def outLoop(acc: Pull[G, X, Unit], pred: Run[G, X]): F[B] =
954+
def outLoop(acc: Pull[G, X, Unit], pred: Run[G, X, F[End]]): F[End] =
946955
// bit of an ugly hack to avoid a stack overflow when these accummulate
947956
pred match {
948957
case vrun: ViewRunner => outLoop(bindView(acc, vrun.view), vrun.prevRunner)
@@ -951,40 +960,40 @@ object Pull extends PullLowPriority {
951960
outLoop(tail, this)
952961
}
953962

954-
def interrupted(inter: Interrupted): F[B] =
963+
def interrupted(inter: Interrupted): F[End] =
955964
go(scope, extendedTopLevelScope, translation, prevRunner, view(inter))
956965

957-
def fail(e: Throwable): F[B] = goErr(e, view)
966+
def fail(e: Throwable): F[End] = goErr(e, view)
958967
}
959968

960-
class TranslateRunner[H[_]](fk: H ~> G, view: Cont[Unit, G, X]) extends Run[H, X] {
961-
def done(doneScope: Scope[F]): F[B] =
969+
class TranslateRunner[H[_]](fk: H ~> G, view: Cont[Unit, G, X]) extends Run[H, X, F[End]] {
970+
def done(doneScope: Scope[F]): F[End] =
962971
go(doneScope, extendedTopLevelScope, translation, runner, view(unit))
963-
def out(head: Chunk[X], scope: Scope[F], tail: Pull[H, X, Unit]): F[B] = {
972+
def out(head: Chunk[X], scope: Scope[F], tail: Pull[H, X, Unit]): F[End] = {
964973
val next = bindView(Translate(tail, fk), view)
965974
runner.out(head, scope, next)
966975
}
967-
def interrupted(inter: Interrupted): F[B] =
976+
def interrupted(inter: Interrupted): F[End] =
968977
go(scope, extendedTopLevelScope, translation, runner, view(inter))
969-
def fail(e: Throwable): F[B] = goErr(e, view)
978+
def fail(e: Throwable): F[End] = goErr(e, view)
970979
}
971980

972-
abstract class StepRunR[Y, S](view: Cont[Option[S], G, X]) extends Run[G, Y] {
973-
def done(scope: Scope[F]): F[B] =
981+
abstract class StepRunR[Y, S](view: Cont[Option[S], G, X]) extends Run[G, Y, F[End]] {
982+
def done(scope: Scope[F]): F[End] =
974983
interruptGuard(scope, view) {
975984
go(scope, extendedTopLevelScope, translation, runner, view(Succeeded(None)))
976985
}
977986

978-
def interrupted(inter: Interrupted): F[B] =
987+
def interrupted(inter: Interrupted): F[End] =
979988
go(scope, extendedTopLevelScope, translation, runner, view(inter))
980989

981-
def fail(e: Throwable): F[B] = goErr(e, view)
990+
def fail(e: Throwable): F[End] = goErr(e, view)
982991
}
983992

984993
class UnconsRunR[Y](view: Cont[Option[(Chunk[Y], Pull[G, Y, Unit])], G, X])
985994
extends StepRunR[Y, (Chunk[Y], Pull[G, Y, Unit])](view) {
986995

987-
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[B] =
996+
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[End] =
988997
// For a Uncons, we continue in same Scope at which we ended compilation of inner stream
989998
interruptGuard(outScope, view) {
990999
val result = Succeeded(Some((head, tail)))
@@ -995,7 +1004,7 @@ object Pull extends PullLowPriority {
9951004
class StepLegRunR[Y](view: Cont[Option[Stream.StepLeg[G, Y]], G, X])
9961005
extends StepRunR[Y, Stream.StepLeg[G, Y]](view) {
9971006

998-
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[B] =
1007+
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[End] =
9991008
// StepLeg: we shift back to the scope at which we were
10001009
// before we started to interpret the Leg's inner stream.
10011010
interruptGuard(scope, view) {
@@ -1004,7 +1013,8 @@ object Pull extends PullLowPriority {
10041013
}
10051014
}
10061015

1007-
class FlatMapR[Y](view: Cont[Unit, G, X], fun: Y => Pull[G, X, Unit]) extends Run[G, Y] {
1016+
class FlatMapR[Y](view: Cont[Unit, G, X], fun: Y => Pull[G, X, Unit])
1017+
extends Run[G, Y, F[End]] {
10081018
private[this] def unconsed(chunk: Chunk[Y], tail: Pull[G, Y, Unit]): Pull[G, X, Unit] =
10091019
if (chunk.size == 1 && tail.isInstanceOf[Succeeded[_]])
10101020
// nb: If tl is Pure, there's no need to propagate flatMap through the tail. Hence, we
@@ -1030,23 +1040,23 @@ object Pull extends PullLowPriority {
10301040
go(0)
10311041
}
10321042

1033-
def done(scope: Scope[F]): F[B] =
1043+
def done(scope: Scope[F]): F[End] =
10341044
interruptGuard(scope, view) {
10351045
go(scope, extendedTopLevelScope, translation, runner, view(unit))
10361046
}
10371047

1038-
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[B] = {
1048+
def out(head: Chunk[Y], outScope: Scope[F], tail: Pull[G, Y, Unit]): F[End] = {
10391049
val next = bindView(unconsed(head, tail), view)
10401050
go(outScope, extendedTopLevelScope, translation, runner, next)
10411051
}
10421052

1043-
def interrupted(inter: Interrupted): F[B] =
1053+
def interrupted(inter: Interrupted): F[End] =
10441054
go(scope, extendedTopLevelScope, translation, runner, view(inter))
10451055

1046-
def fail(e: Throwable): F[B] = goErr(e, view)
1056+
def fail(e: Throwable): F[End] = goErr(e, view)
10471057
}
10481058

1049-
def goEval[V](eval: Eval[G, V], view: Cont[V, G, X]): F[B] =
1059+
def goEval[V](eval: Eval[G, V], view: Cont[V, G, X]): F[End] =
10501060
scope.interruptibleEval(translation(eval.value)).flatMap { eitherOutcome =>
10511061
val result = eitherOutcome match {
10521062
case Right(r) => Succeeded(r)
@@ -1057,7 +1067,7 @@ object Pull extends PullLowPriority {
10571067
go(scope, extendedTopLevelScope, translation, runner, view(result))
10581068
}
10591069

1060-
def goAcquire[R](acquire: Acquire[G, R], view: Cont[R, G, X]): F[B] = {
1070+
def goAcquire[R](acquire: Acquire[G, R], view: Cont[R, G, X]): F[End] = {
10611071
val onScope = scope.acquireResource[R](
10621072
poll =>
10631073
if (acquire.cancelable) poll(translation(acquire.resource))
@@ -1079,7 +1089,7 @@ object Pull extends PullLowPriority {
10791089
def goInterruptWhen(
10801090
haltOnSignal: F[Either[Throwable, Unit]],
10811091
view: Cont[Unit, G, X]
1082-
): F[B] = {
1092+
): F[End] = {
10831093
val onScope = scope.acquireResource(
10841094
_ => scope.interruptWhen(haltOnSignal),
10851095
(f: Fiber[F, Throwable, Unit], _: ExitCase) => f.cancel
@@ -1100,7 +1110,7 @@ object Pull extends PullLowPriority {
11001110
stream: Pull[G, X, Unit],
11011111
useInterruption: Boolean,
11021112
view: Cont[Unit, G, X]
1103-
): F[B] = {
1113+
): F[End] = {
11041114
def endScope(scopeId: Unique.Token, result: Terminal[Unit]): Pull[G, X, Unit] =
11051115
result match {
11061116
case Succeeded(_) => SucceedScope(scopeId)
@@ -1127,7 +1137,7 @@ object Pull extends PullLowPriority {
11271137
interruptGuard(scope, view)(tail)
11281138
}
11291139

1130-
def goCloseScope(close: CloseScope, view: Cont[Unit, G, X]): F[B] = {
1140+
def goCloseScope(close: CloseScope, view: Cont[Unit, G, X]): F[End] = {
11311141
def addError(err: Throwable, res: Terminal[Unit]): Terminal[Unit] = res match {
11321142
case Succeeded(_) => Fail(err)
11331143
case Fail(err0) => Fail(CompositeFailure(err, err0, Nil))
@@ -1187,9 +1197,9 @@ object Pull extends PullLowPriority {
11871197

11881198
(viewL(stream): @unchecked) match { // unchecked b/c scala 3 erroneously reports exhaustiveness warning
11891199
case tst: Translate[h, G, _] @unchecked => // y = Unit
1190-
val translateRunner: Run[h, X] = new TranslateRunner(tst.fk, getCont[Unit, G, X])
1200+
val translateRunner: Run[h, X, F[End]] = new TranslateRunner(tst.fk, getCont[Unit, G, X])
11911201
val composed: h ~> F = translation.compose[h](tst.fk)
1192-
go[h, X](scope, extendedTopLevelScope, composed, translateRunner, tst.stream)
1202+
go[h, X, End](scope, extendedTopLevelScope, composed, translateRunner, tst.stream)
11931203

11941204
case output: Output[_] =>
11951205
val view = getCont[Unit, G, X]
@@ -1204,13 +1214,17 @@ object Pull extends PullLowPriority {
12041214
case u: Uncons[G, y] @unchecked =>
12051215
val v = getCont[Option[(Chunk[y], Pull[G, y, Unit])], G, X]
12061216
// a Uncons is run on the same scope, without shifting.
1207-
F.unit >> go(scope, extendedTopLevelScope, translation, new UnconsRunR(v), u.stream)
1217+
val runr = buildR[G, y, End]
1218+
F.unit >> go(scope, extendedTopLevelScope, translation, runr, u.stream).attempt
1219+
.flatMap(_.fold(goErr(_, v), _.apply(new UnconsRunR(v))))
12081220

12091221
case s: StepLeg[G, y] @unchecked =>
12101222
val v = getCont[Option[Stream.StepLeg[G, y]], G, X]
1223+
val runr = buildR[G, y, End]
12111224
scope
12121225
.shiftScope(s.scope, s.toString)
1213-
.flatMap(go(_, extendedTopLevelScope, translation, new StepLegRunR(v), s.stream))
1226+
.flatMap(go(_, extendedTopLevelScope, translation, runr, s.stream).attempt)
1227+
.flatMap(_.fold(goErr(_, v), _.apply(new StepLegRunR(v))))
12141228

12151229
case _: GetScope[_] =>
12161230
go(scope, extendedTopLevelScope, translation, runner, getCont(Succeeded(scope)))
@@ -1230,7 +1244,7 @@ object Pull extends PullLowPriority {
12301244

12311245
val initFk: F ~> F = cats.arrow.FunctionK.id[F]
12321246

1233-
class OuterRun(initB: B) extends Run[F, O] { self =>
1247+
class OuterRun(initB: B) extends Run[F, O, F[B]] { self =>
12341248
private[this] var accB: B = initB
12351249

12361250
override def done(scope: Scope[F]): F[B] = F.pure(accB)
@@ -1243,19 +1257,21 @@ object Pull extends PullLowPriority {
12431257
override def out(head: Chunk[O], scope: Scope[F], tail: Pull[F, O, Unit]): F[B] =
12441258
try {
12451259
accB = foldChunk(accB, head)
1246-
go(scope, None, initFk, self, tail)
1260+
go[F, O, B](scope, None, initFk, self, tail)
12471261
} catch {
12481262
case NonFatal(e) =>
12491263
viewL(tail) match {
1250-
case _: Action[F, O, _] => go(scope, None, initFk, self, getCont(Fail(e)))
1264+
case _: Action[F, O, _] =>
1265+
val v = contP.asInstanceOf[ContP[Unit, F, O, Unit]]
1266+
go[F, O, B](scope, None, initFk, self, v(Fail(e)))
12511267
case Succeeded(_) => F.raiseError(e)
12521268
case Fail(e2) => F.raiseError(CompositeFailure(e2, e))
12531269
case Interrupted(_, err) => F.raiseError(err.fold(e)(t => CompositeFailure(e, t)))
12541270
}
12551271
}
12561272
}
12571273

1258-
go[F, O](initScope, None, initFk, new OuterRun(init), stream)
1274+
go[F, O, B](initScope, None, initFk, new OuterRun(init), stream)
12591275
}
12601276

12611277
private[fs2] def flatMapOutput[F[_], F2[x] >: F[x], O, O2](

0 commit comments

Comments
 (0)