Skip to content
Merged
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,49 @@ import pekko.stream.stage._
private def shutdownCounters: String =
shutdownCounter.map(x => if (x >= KeepGoingFlag) s"${x & KeepGoingMask}(KeepGoing)" else x.toString).mkString(",")

/**
* INTERNAL API
*
* Reports a stage error: logs it according to the configured level, fails the active stage, and aborts any
* in-flight event chasing by re-enqueuing chased events. Lifted out of [[execute]] to shrink the hot loop's
* bytecode and give the JIT a tighter compile target for inlining `processPush`/`processPull` — the nested
* `def` that previously lived inside the loop was already compiled as a synthetic method on this class (Scala
* 2.13 does not allocate a fresh closure per iteration when the body only references class fields), so the
* gain here is purely in method-body size, not in allocation reduction.
*/
private def reportStageError(e: Throwable): Unit = {
if (activeStage eq null) throw e
else {
val logAt: Logging.LogLevel = activeStage.attributes.get[LogLevels] match {
case Some(levels) => levels.onFailure
case None => defaultErrorReportingLogLevel
}
logAt match {
case Logging.ErrorLevel =>
log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.WarningLevel =>
log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.InfoLevel =>
log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.DebugLevel =>
log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case _ => // Off, nop
}
activeStage.failStage(e)

// Abort chasing
chaseCounter = 0
if (chasedPush ne NoEvent) {
enqueue(chasedPush)
chasedPush = NoEvent
}
if (chasedPull ne NoEvent) {
enqueue(chasedPull)
chasedPull = NoEvent
}
}
}

/**
* Executes pending events until the given limit is met. If there were remaining events, isSuspended will return
* true.
Expand All @@ -382,39 +425,6 @@ import pekko.stream.stage._
eventsRemaining -= 1
chaseCounter = math.min(ChaseLimit, eventsRemaining)

def reportStageError(e: Throwable): Unit = {
if (activeStage eq null) throw e
else {
val logAt: Logging.LogLevel = activeStage.attributes.get[LogLevels] match {
case Some(levels) => levels.onFailure
case None => defaultErrorReportingLogLevel
}
logAt match {
case Logging.ErrorLevel =>
log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.WarningLevel =>
log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.InfoLevel =>
log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case Logging.DebugLevel =>
log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage)
case _ => // Off, nop
}
activeStage.failStage(e)

// Abort chasing
chaseCounter = 0
if (chasedPush ne NoEvent) {
enqueue(chasedPush)
chasedPush = NoEvent
}
if (chasedPull ne NoEvent) {
enqueue(chasedPull)
chasedPull = NoEvent
}
}
}

/*
* This is the "normal" event processing code which dequeues directly from the internal event queue. Since
* most execution paths tend to produce either a Push that will be propagated along a longer chain we take
Expand Down