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

Fix hang in KqueueExecutorScheduler #65

Merged
merged 6 commits into from
Sep 18, 2022
Merged

Fix hang in KqueueExecutorScheduler #65

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
61d2427
Fix #64: Remove a KqExec poll hang
LeeTibbert Sep 16, 2022
fa70f81
Add Arman's socket close no-hang test
LeeTibbert Sep 17, 2022
1c4a0fa
sockets & server_sockets travel in pairs; their test must also
LeeTibbert Sep 17, 2022
63838ca
Revert "Fix #64: Remove a KqExec poll hang"
armanbilge Sep 18, 2022
a9f368f
Add a note about the failure mode
armanbilge Sep 18, 2022
391a739
Use the final kqueue change count in condition
armanbilge Sep 18, 2022
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
34 changes: 17 additions & 17 deletions core/src/main/scala/epollcat/unsafe/KqueueExecutorScheduler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,27 +44,27 @@ private[unsafe] final class KqueueExecutorScheduler(
val timeoutIsInfinite = timeout == Duration.Inf
val timeoutIsZero = timeout == Duration.Zero
val noCallbacks = callbacks.isEmpty
val changeCount = changes.size()

// pre-process the changes to filter canceled ones
val changelist = stackalloc[kevent64_s](changes.size().toLong)
var change = changelist
var changeCount = 0
while (!changes.isEmpty()) {
val evAdd = changes.poll()
if (!evAdd.canceled) {
change.ident = evAdd.fd.toULong
change.filter = evAdd.filter
change.flags = (EV_ADD | EV_CLEAR).toUShort
change.udata = EventNotificationCallback.toPtr(evAdd.cb)
change += 1
changeCount += 1
}
}

if ((timeoutIsInfinite || timeoutIsZero) && noCallbacks && changeCount == 0)
false // nothing to do here. refer to scaladoc on PollingExecutorScheduler#poll
else {

val changelist = stackalloc[kevent64_s](changeCount.toLong)
var change = changelist
var finalChangeCount = 0
while (!changes.isEmpty()) {
val evAdd = changes.poll()
if (!evAdd.canceled) {
change.ident = evAdd.fd.toULong
change.filter = evAdd.filter
change.flags = (EV_ADD | EV_CLEAR).toUShort
change.udata = EventNotificationCallback.toPtr(evAdd.cb)
change += 1
finalChangeCount += 1
}
}

val timeoutSpec =
if (timeoutIsInfinite || timeoutIsZero) null
else {
Expand All @@ -78,7 +78,7 @@ private[unsafe] final class KqueueExecutorScheduler(
val eventlist = stackalloc[kevent64_s](maxEvents.toLong)
val flags = (if (timeoutIsZero) KEVENT_FLAG_IMMEDIATE else KEVENT_FLAG_NONE).toUInt
val triggeredEvents =
kevent64(kqfd, changelist, finalChangeCount, eventlist, maxEvents, flags, timeoutSpec)
kevent64(kqfd, changelist, changeCount, eventlist, maxEvents, flags, timeoutSpec)

if (triggeredEvents >= 0) {
var i = 0
Expand Down
10 changes: 10 additions & 0 deletions tests/shared/src/test/scala/epollcat/TcpSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,4 +238,14 @@ class TcpSuite extends EpollcatSuite {
.timeoutTo(100.millis, IO.unit)
}

test("immediately closing a socket does not hang") {
// note: on failure the test passes, but the test runner hangs
IOSocketChannel.open.use_
}

test("immediately closing a server socket does not hang") {
// note: on failure the test passes, but the test runner hangs
IOServerSocketChannel.open.use_
}

}