FrankYang0529 commented on code in PR #75538:
URL: https://github.com/apache/airflow/pull/75538#discussion_r4490700479
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt:
##########
@@ -220,10 +222,23 @@ internal class Logger(
}
}
+// Several threads log: the task's own thread, the thread that handles a stop
signal, and the thread
+// that reads replies from the supervisor. The lock keeps send from queuing a
message after configure
+// has written out the queue. It also keeps a new message from being written
before the messages that
+// were queued earlier.
internal object LogSender {
private var writer: ByteWriteChannel? = null
internal val messages: ConcurrentLinkedDeque<LogMessage> =
ConcurrentLinkedDeque()
+ // While runBlocking waits, it also runs other coroutines that are ready on
the calling thread. On the
+ // thread that Server.serve blocks, one of them runs the task. So on that
thread, a runBlocking call
+ // here could run the task while the caller holds this object's lock. The
task would then hang if it
+ // waited for another thread that logs, such as the thread that reads
replies from the supervisor.
+ // Writing on writerThread avoids this, because the waiting caller runs no
coroutines.
+ private val writerThread =
+ Executors.newSingleThreadExecutor { Thread(it, "Airflow log sender").apply
{ isDaemon = true } }
Review Comment:
The logger is asynchronous now. A log call puts the line in a queue and
returns, and one thread writes the lines in order. A caller waits only when the
queued lines already hold 8 Mi characters, so a task that logs faster than the
supervisor reads cannot fill the heap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
