Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Channel: drain/drainUpTo include pending puts #980

Merged
merged 14 commits into from
Jan 9, 2025
Merged

Channel: drain/drainUpTo include pending puts #980

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
8a41fc5
working
johnhungerford Jan 6, 2025
3e1d80a
Fixed tests
johnhungerford Jan 6, 2025
907b13c
Update kyo-core/shared/src/main/scala/kyo/Channel.scala
johnhungerford Jan 6, 2025
039b8ce
Update kyo-core/shared/src/main/scala/kyo/Channel.scala
johnhungerford Jan 7, 2025
495095c
Update kyo-core/shared/src/main/scala/kyo/Channel.scala
johnhungerford Jan 7, 2025
cc103bd
Explicit null case
johnhungerford Jan 7, 2025
25cc8ec
fixed drain/drainUpTo issue
johnhungerford Jan 7, 2025
f8baa4a
Merge branch 'drain-take-pending-puts' of github.com:johnhungerford/k…
johnhungerford Jan 7, 2025
8503c25
draining only from queue, but flushing until unable to take; removed …
johnhungerford Jan 7, 2025
96b3fe0
Added tailrec to recursive functions
johnhungerford Jan 8, 2025
82cfd72
Merge branch 'main' into drain-take-pending-puts
johnhungerford Jan 8, 2025
7a4e5e4
last fixes
johnhungerford Jan 9, 2025
4f22b8a
Merge branch 'drain-take-pending-puts' of github.com:johnhungerford/k…
johnhungerford Jan 9, 2025
af9e96e
Merge branch 'main' into drain-take-pending-puts
fwbrasil Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 37 additions & 10 deletions kyo-core/shared/src/main/scala/kyo/Channel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,16 +105,18 @@ object Channel:
def takeExactly(n: Int)(using Frame): Chunk[A] < (Abort[Closed] & Async) =
if n <= 0 then Chunk.empty
else
Loop(Chunk.empty[A]): lastChunk =>
val nextN = n - lastChunk.size
Loop(Chunk.empty[A], 0): (lastChunk, lastSize) =>
val nextN = n - lastSize
Channel.drainUpTo(self)(nextN).map: chunk =>
val chunk1 = lastChunk.concat(chunk)
val size1 = chunk1.size
if chunk1.size == n then Loop.done(chunk1)
else
self.take.map: a =>
val chunk2 = chunk1.append(a)
if chunk2.size == n then Loop.done(chunk2)
else Loop.continue(chunk2)
val size2 = size1 + 1
if size2 == n then Loop.done(chunk2)
else Loop.continue(chunk2, size2)
end if

/** Creates a fiber that puts an element into the channel.
Expand Down Expand Up @@ -328,9 +330,23 @@ object Channel:
end poll

def drainUpTo(max: Int)(using AllowUnsafe) =
val result = queue.drainUpTo(max)
if result.exists(_.nonEmpty) then flush()
result
@tailrec
def loop(current: Chunk[A], i: Int): Result[Closed, Chunk[A]] =
if i == 0 then Result.Success(current)
else
val next = queue.drainUpTo(i)
next match
case Result.Success(c) =>
if c.isEmpty then Result.Success(current)
else
flush()
loop(current.concat(c), i - c.length)
case other => other
end match
end if
end loop

loop(Chunk.empty, max)
end drainUpTo

def putFiber(value: A)(using AllowUnsafe): Fiber.Unsafe[Closed, Unit] =
Expand All @@ -357,9 +373,20 @@ object Channel:
end takeFiber

def drain()(using AllowUnsafe) =
val result = queue.drain()
if result.exists(_.nonEmpty) then flush()
result
@tailrec
def loop(current: Chunk[A]): Result[Closed, Chunk[A]] =
val next = queue.drain()
next match
case Result.Success(c) =>
if c.isEmpty then Result.Success(current)
else
flush()
loop(current.concat(c))
case other => other
end match
end loop

loop(Chunk.empty)
end drain

def close()(using frame: Frame, allow: AllowUnsafe) =
Expand Down
35 changes: 32 additions & 3 deletions kyo-core/shared/src/test/scala/kyo/ChannelTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,33 @@ class ChannelTest extends Test:
r <- c.drain
yield assert(r == Seq(1, 2))
}
"should consider pending puts" in run {
import AllowUnsafe.embrace.danger
IO.Unsafe.evalOrThrow {
for
c <- Channel.init[Int](2)
_ <- c.putFiber(1)
_ <- c.putFiber(2)
_ <- c.putFiber(3)
result <- c.drain
finalSize <- c.size
yield assert(result == Chunk(1, 2, 3) && finalSize == 0)
}
}
"should consider pending puts - zero capacity" in pendingUntilFixed {
import AllowUnsafe.embrace.danger
IO.Unsafe.evalOrThrow {
for
c <- Channel.init[Int](0)
_ <- c.putFiber(1)
_ <- c.putFiber(2)
_ <- c.putFiber(3)
result <- c.drain
finalSize <- c.size
yield assert(result == Chunk(1, 2, 3) && finalSize == 0)
}
()
}
}
"drainUpTo" - {
"zero or negative" in run {
Expand Down Expand Up @@ -347,19 +374,19 @@ class ChannelTest extends Test:
s <- c.size
yield assert(r == Seq(1, 2) && s == 2)
}
"should consider pending puts" in pendingUntilFixed {
"should consider pending puts" in run {
import AllowUnsafe.embrace.danger
IO.Unsafe.evalOrThrow {
for
c <- Channel.init[Int](2)
_ <- c.putFiber(1)
_ <- c.putFiber(2)
_ <- c.putFiber(3)
_ <- c.putFiber(4)
result <- c.drainUpTo(3)
finalSize <- c.size
yield assert(result == Chunk(1, 2, 3) && finalSize == 0)
yield assert(result == Chunk(1, 2, 3) && finalSize == 1)
}
()
}
"should consider pending puts - zero capacity" in pendingUntilFixed {
import AllowUnsafe.embrace.danger
Expand All @@ -369,6 +396,7 @@ class ChannelTest extends Test:
_ <- c.putFiber(1)
_ <- c.putFiber(2)
_ <- c.putFiber(3)
_ <- c.putFiber(4)
result <- c.drainUpTo(3)
finalSize <- c.size
yield assert(result == Chunk(1, 2, 3) && finalSize == 0)
Expand Down Expand Up @@ -670,6 +698,7 @@ class ChannelTest extends Test:
}

"putBatch and takeExactly" in run {
val repeats = 1
(for
size <- Choice.get(Seq(0, 1, 2, 10, 100))
channel <- Channel.init[Int](size)
Expand Down
Loading