raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

chore: Add PekkoManagedBlocker to reduce memory.

Browse source
This commit is contained in:
He-Pin 2024-01-23 16:29:05 +08:00 • committed by He-Pin(kerr)
parent ae210b17d2
commit 4d8aa26404
1 changed files with 14 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

24
actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
View file

@ -173,19 +173,23 @@ object MonitorableThreadFactory {
val doNothing: Thread.UncaughtExceptionHandler = val doNothing: Thread.UncaughtExceptionHandler =
new Thread.UncaughtExceptionHandler() { def uncaughtException(thread: Thread, cause: Throwable) = () } new Thread.UncaughtExceptionHandler() { def uncaughtException(thread: Thread, cause: Throwable) = () }
private class PekkoManagedBlocker[T](thunk: => T)
extends AtomicReference[Option[T]](None) with ForkJoinPool.ManagedBlocker {
final override def block(): Boolean = {
set(Some(thunk))
true
}
final override def isReleasable: Boolean = get().isDefined // Exception intended if None
}
private[pekko] class PekkoForkJoinWorkerThread(_pool: ForkJoinPool) private[pekko] class PekkoForkJoinWorkerThread(_pool: ForkJoinPool)
extends ForkJoinWorkerThread(_pool) extends ForkJoinWorkerThread(_pool)
with BlockContext { with BlockContext {
override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = { final override def blockOn[T](thunk: => T)(implicit permission: CanAwait): T = {
val result = new AtomicReference[Option[T]](None) val blocker = new PekkoManagedBlocker(thunk)
ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker { ForkJoinPool.managedBlock(blocker)
def block(): Boolean = { blocker.get.get // Exception intended if None
result.set(Some(thunk))
true
}
def isReleasable = result.get.isDefined
})
result.get.get // Exception intended if None
} }
} }
} }