Skip to content

Commit 023bffd

Browse files
djspiewakdjspiewak
authored
Merge pull request #3455 from bplommer/flatmodify-doc
Expand docs for Ref flatModify methods, and use them when appropriate
2 parents ec8b9e0 + 34de870 commit 023bffd

9 files changed

Lines changed: 152 additions & 179 deletions

File tree

‎kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala‎

Lines changed: 37 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -48,50 +48,48 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] {
4848
// start running the effect, or subscribe if it already is
4949
def evalOrSubscribe: F[A] =
5050
deferred[Fiber[F, E, A]] flatMap { deferredFiber =>
51-
uncancelable { poll =>
52-
state.modify {
53-
case Unevaluated() =>
54-
// run the effect, and if this fiber is still relevant set its result
55-
val go = {
56-
def tryComplete(result: Memoize[F, E, A]): F[Unit] = state.update {
57-
case Evaluating(fiber, _) if fiber eq deferredFiber =>
58-
// we are the blessed fiber of this memo
59-
result
60-
case other => // our outcome is no longer relevant
61-
other
62-
}
63-
64-
fa
65-
// hack around functor law breakage
66-
.flatMap(F.pure(_))
67-
.handleErrorWith(F.raiseError(_))
68-
// end hack
69-
.guaranteeCase {
70-
case Outcome.Canceled() =>
71-
tryComplete(Finished(Right(productR(canceled)(never))))
72-
case Outcome.Errored(err) =>
73-
tryComplete(Finished(Left(err)))
74-
case Outcome.Succeeded(fa) =>
75-
tryComplete(Finished(Right(fa)))
76-
}
51+
state.flatModifyFull {
52+
case (poll, Unevaluated()) =>
53+
// run the effect, and if this fiber is still relevant set its result
54+
val go = {
55+
def tryComplete(result: Memoize[F, E, A]): F[Unit] = state.update {
56+
case Evaluating(fiber, _) if fiber eq deferredFiber =>
57+
// we are the blessed fiber of this memo
58+
result
59+
case other => // our outcome is no longer relevant
60+
other
7761
}
7862

79-
val eval = go.start.flatMap { fiber =>
80-
deferredFiber.complete(fiber) *>
81-
poll(fiber.join.flatMap(_.embed(productR(canceled)(never))))
82-
.onCancel(unsubscribe(deferredFiber))
83-
}
63+
fa
64+
// hack around functor law breakage
65+
.flatMap(F.pure(_))
66+
.handleErrorWith(F.raiseError(_))
67+
// end hack
68+
.guaranteeCase {
69+
case Outcome.Canceled() =>
70+
tryComplete(Finished(Right(productR(canceled)(never))))
71+
case Outcome.Errored(err) =>
72+
tryComplete(Finished(Left(err)))
73+
case Outcome.Succeeded(fa) =>
74+
tryComplete(Finished(Right(fa)))
75+
}
76+
}
77+
78+
val eval = go.start.flatMap { fiber =>
79+
deferredFiber.complete(fiber) *>
80+
poll(fiber.join.flatMap(_.embed(productR(canceled)(never))))
81+
.onCancel(unsubscribe(deferredFiber))
82+
}
8483

85-
Evaluating(deferredFiber, 1) -> eval
84+
Evaluating(deferredFiber, 1) -> eval
8685

87-
case Evaluating(fiber, subscribers) =>
88-
Evaluating(fiber, subscribers + 1) ->
89-
poll(fiber.get.flatMap(_.join).flatMap(_.embed(productR(canceled)(never))))
90-
.onCancel(unsubscribe(fiber))
86+
case (poll, Evaluating(fiber, subscribers)) =>
87+
Evaluating(fiber, subscribers + 1) ->
88+
poll(fiber.get.flatMap(_.join).flatMap(_.embed(productR(canceled)(never))))
89+
.onCancel(unsubscribe(fiber))
9190

92-
case finished @ Finished(result) =>
93-
finished -> fromEither(result).flatten
94-
}.flatten
91+
case (_, finished @ Finished(result)) =>
92+
finished -> fromEither(result).flatten
9593
}
9694
}
9795

‎kernel/shared/src/main/scala/cats/effect/kernel/MiniSemaphore.scala‎

Lines changed: 7 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -81,16 +81,13 @@ private[kernel] object MiniSemaphore {
8181
}
8282

8383
def release: F[Unit] =
84-
state
85-
.modify {
86-
case State(waiting, permits) =>
87-
if (waiting.nonEmpty)
88-
State(waiting.tail, permits) -> waiting.head.complete(()).void
89-
else
90-
State(waiting, permits + 1) -> ().pure[F]
91-
}
92-
.flatten
93-
.uncancelable
84+
state.flatModify {
85+
case State(waiting, permits) =>
86+
if (waiting.nonEmpty)
87+
State(waiting.tail, permits) -> waiting.head.complete(()).void
88+
else
89+
State(waiting, permits + 1) -> ().pure[F]
90+
}
9491

9592
def withPermit[A](fa: F[A]): F[A] =
9693
F.uncancelable { poll => poll(acquire) >> poll(fa).guarantee(release) }

‎kernel/shared/src/main/scala/cats/effect/kernel/Ref.scala‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,8 +102,14 @@ abstract class Ref[F[_], A] extends RefSource[F, A] with RefSink[F, A] {
102102
/**
103103
* Like [[modify]] but schedules resulting effect right after modification.
104104
*
105-
* Both modification and finalizer are uncancelable, if you need cancellation mechanic in
106-
* finalizer please see [[flatModifyFull]].
105+
* Useful for implementing effectful transition of a state machine, in which an effect is
106+
* performed based on current state and the state must be updated to reflect that this effect
107+
* will be performed.
108+
*
109+
* Both modification and finalizer are within a single uncancelable region, to prevent
110+
* canceled finalizers from leaving the Ref's value permanently out of sync with effects
111+
* actually performed. if you need cancellation mechanic in finalizer please see
112+
* [[flatModifyFull]].
107113
*
108114
* @see
109115
* [[modify]]
@@ -116,9 +122,12 @@ abstract class Ref[F[_], A] extends RefSource[F, A] with RefSink[F, A] {
116122
/**
117123
* Like [[modify]] but schedules resulting effect right after modification.
118124
*
119-
* Unlike [[flatModify]] finalizer cancellation could be masked via supplied `Poll`.
125+
* Unlike [[flatModify]] finalizer cancellation could be unmasked via supplied `Poll`.
120126
* Modification itself is still uncancelable.
121127
*
128+
* When used as part of a state machine, cancelable regions should usually have an `onCancel`
129+
* finalizer to update the state to reflect that the effect will not be performed.
130+
*
122131
* @see
123132
* [[modify]]
124133
* @see

‎std/shared/src/main/scala/cats/effect/std/CountDownLatch.scala‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -69,12 +69,10 @@ object CountDownLatch {
6969
extends CountDownLatch[F] {
7070

7171
override def release: F[Unit] =
72-
F.uncancelable { _ =>
73-
state.modify {
74-
case Awaiting(n, signal) =>
75-
if (n > 1) (Awaiting(n - 1, signal), F.unit) else (Done(), signal.complete(()).void)
76-
case d @ Done() => (d, F.unit)
77-
}.flatten
72+
state.flatModify {
73+
case Awaiting(n, signal) =>
74+
if (n > 1) (Awaiting(n - 1, signal), F.unit) else (Done(), signal.complete(()).void)
75+
case d @ Done() => (d, F.unit)
7876
}
7977

8078
override def await: F[Unit] =

‎std/shared/src/main/scala/cats/effect/std/CyclicBarrier.scala‎

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -66,26 +66,24 @@ object CyclicBarrier {
6666
new CyclicBarrier[F] {
6767
val await: F[Unit] =
6868
F.deferred[Unit].flatMap { gate =>
69-
F.uncancelable { poll =>
70-
state.modify {
71-
case State(awaiting, epoch, unblock) =>
72-
val awaitingNow = awaiting - 1
69+
state.flatModifyFull {
70+
case (poll, State(awaiting, epoch, unblock)) =>
71+
val awaitingNow = awaiting - 1
7372

74-
if (awaitingNow == 0)
75-
State(capacity, epoch + 1, gate) -> unblock.complete(()).void
76-
else {
77-
val newState = State(awaitingNow, epoch, unblock)
78-
// reincrement count if this await gets canceled,
79-
// but only if the barrier hasn't reset in the meantime
80-
val cleanup = state.update { s =>
81-
if (s.epoch == epoch) s.copy(awaiting = s.awaiting + 1)
82-
else s
83-
}
84-
85-
newState -> poll(unblock.get).onCancel(cleanup)
73+
if (awaitingNow == 0)
74+
State(capacity, epoch + 1, gate) -> unblock.complete(()).void
75+
else {
76+
val newState = State(awaitingNow, epoch, unblock)
77+
// reincrement count if this await gets canceled,
78+
// but only if the barrier hasn't reset in the meantime
79+
val cleanup = state.update { s =>
80+
if (s.epoch == epoch) s.copy(awaiting = s.awaiting + 1)
81+
else s
8682
}
8783

88-
}.flatten
84+
newState -> poll(unblock.get).onCancel(cleanup)
85+
}
86+
8987
}
9088
}
9189
}

‎std/shared/src/main/scala/cats/effect/std/Dequeue.scala‎

Lines changed: 22 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -184,20 +184,17 @@ object Dequeue {
184184
}
185185

186186
private def _tryOffer(update: BankersQueue[A] => BankersQueue[A]) =
187-
state
188-
.modify {
189-
case State(queue, size, takers, offerers) if takers.nonEmpty =>
190-
val (taker, rest) = takers.dequeue
191-
State(update(queue), size, rest, offerers) -> taker.complete(()).as(true)
187+
state.flatModify {
188+
case State(queue, size, takers, offerers) if takers.nonEmpty =>
189+
val (taker, rest) = takers.dequeue
190+
State(update(queue), size, rest, offerers) -> taker.complete(()).as(true)
192191

193-
case State(queue, size, takers, offerers) if size < capacity =>
194-
State(update(queue), size + 1, takers, offerers) -> F.pure(true)
192+
case State(queue, size, takers, offerers) if size < capacity =>
193+
State(update(queue), size + 1, takers, offerers) -> F.pure(true)
195194

196-
case s =>
197-
s -> F.pure(false)
198-
}
199-
.flatten
200-
.uncancelable
195+
case s =>
196+
s -> F.pure(false)
197+
}
201198

202199
private def _take(dequeue: BankersQueue[A] => (BankersQueue[A], Option[A])): F[A] =
203200
F uncancelable { poll =>
@@ -254,22 +251,19 @@ object Dequeue {
254251

255252
private def _tryTake(
256253
dequeue: BankersQueue[A] => (BankersQueue[A], Option[A])): F[Option[A]] =
257-
state
258-
.modify {
259-
case State(queue, size, takers, offerers) if queue.nonEmpty && offerers.isEmpty =>
260-
val (rest, ma) = dequeue(queue)
261-
State(rest, size - 1, takers, offerers) -> F.pure(ma)
262-
263-
case State(queue, size, takers, offerers) if queue.nonEmpty =>
264-
val (rest, ma) = dequeue(queue)
265-
val (release, tail) = offerers.dequeue
266-
State(rest, size - 1, takers, tail) -> release.complete(()).as(ma)
267-
268-
case s =>
269-
s -> F.pure(none[A])
270-
}
271-
.flatten
272-
.uncancelable
254+
state.flatModify {
255+
case State(queue, size, takers, offerers) if queue.nonEmpty && offerers.isEmpty =>
256+
val (rest, ma) = dequeue(queue)
257+
State(rest, size - 1, takers, offerers) -> F.pure(ma)
258+
259+
case State(queue, size, takers, offerers) if queue.nonEmpty =>
260+
val (rest, ma) = dequeue(queue)
261+
val (release, tail) = offerers.dequeue
262+
State(rest, size - 1, takers, tail) -> release.complete(()).as(ma)
263+
264+
case s =>
265+
s -> F.pure(none[A])
266+
}
273267

274268
override def size: F[Int] = state.get.map(_.size)
275269
}

‎std/shared/src/main/scala/cats/effect/std/PQueue.scala‎

Lines changed: 21 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -105,19 +105,16 @@ object PQueue {
105105
}
106106

107107
def tryOffer(a: A): F[Boolean] =
108-
ref
109-
.modify {
110-
case State(heap, size, takers, offerers) if takers.nonEmpty =>
111-
val (taker, rest) = takers.dequeue
112-
State(heap.insert(a), size + 1, rest, offerers) -> taker.complete(()).as(true)
108+
ref.flatModify {
109+
case State(heap, size, takers, offerers) if takers.nonEmpty =>
110+
val (taker, rest) = takers.dequeue
111+
State(heap.insert(a), size + 1, rest, offerers) -> taker.complete(()).as(true)
113112

114-
case State(heap, size, takers, offerers) if size < capacity =>
115-
State(heap.insert(a), size + 1, takers, offerers) -> F.pure(true)
113+
case State(heap, size, takers, offerers) if size < capacity =>
114+
State(heap.insert(a), size + 1, takers, offerers) -> F.pure(true)
116115

117-
case s => s -> F.pure(false)
118-
}
119-
.flatten
120-
.uncancelable
116+
case s => s -> F.pure(false)
117+
}
121118

122119
val take: F[A] =
123120
F.uncancelable { poll =>
@@ -159,22 +156,19 @@ object PQueue {
159156
}
160157

161158
val tryTake: F[Option[A]] =
162-
ref
163-
.modify {
164-
case State(heap, size, takers, offerers) if heap.nonEmpty && offerers.isEmpty =>
165-
val (rest, a) = heap.take
166-
State(rest, size - 1, takers, offerers) -> F.pure(a.some)
167-
168-
case State(heap, size, takers, offerers) if heap.nonEmpty =>
169-
val (rest, a) = heap.take
170-
val (release, tail) = offerers.dequeue
171-
State(rest, size - 1, takers, tail) -> release.complete(()).as(a.some)
172-
173-
case s =>
174-
s -> F.pure(none[A])
175-
}
176-
.flatten
177-
.uncancelable
159+
ref.flatModify {
160+
case State(heap, size, takers, offerers) if heap.nonEmpty && offerers.isEmpty =>
161+
val (rest, a) = heap.take
162+
State(rest, size - 1, takers, offerers) -> F.pure(a.some)
163+
164+
case State(heap, size, takers, offerers) if heap.nonEmpty =>
165+
val (rest, a) = heap.take
166+
val (release, tail) = offerers.dequeue
167+
State(rest, size - 1, takers, tail) -> release.complete(()).as(a.some)
168+
169+
case s =>
170+
s -> F.pure(none[A])
171+
}
178172

179173
def size: F[Int] =
180174
ref.get.map(_.size)

0 commit comments

Comments
 (0)