FrankYang0529 commented on code in PR #75538:
URL: https://github.com/apache/airflow/pull/75538#discussion_r4499838352
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Logger.kt:
##########
@@ -220,44 +224,110 @@ 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 adding a
message to messages after
+// configure has passed the messages in that queue to the sender thread. The
lock also keeps a new message
+// from being written before the messages that were queued earlier.
internal object LogSender {
+ /** The number of characters that the queued lines can hold before a caller
has to wait for room. */
+ internal const val PENDING_CHARS = 8 * 1024 * 1024
+
private var writer: ByteWriteChannel? = null
internal val messages: ConcurrentLinkedDeque<LogMessage> =
ConcurrentLinkedDeque()
+ // Only the sender thread writes to the channel, so a caller never waits in
runBlocking. While
+ // runBlocking waits, it also runs other coroutines that are ready on the
calling thread. On the thread
+ // that Server.serve blocks, one of those coroutines runs the task. If a
caller waited in runBlocking on
+ // that thread, the task would run 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.
+ private val pending = LinkedBlockingQueue<() -> Unit>()
+
+ // A queued line holds one permit for each of its characters until the
sender thread has written the line.
+ private val room = Semaphore(PENDING_CHARS)
+
+ init {
+ thread(name = "Airflow log sender", isDaemon = true) {
+ while (true) {
+ try {
+ pending.take().invoke()
+ } catch (e: Throwable) {
+ // If the sender thread ended, flush would wait forever, and so
would every caller that waits for room.
+ }
+ }
+ }
+ }
+
+ @Synchronized
fun configure(channel: ByteWriteChannel) {
writer = channel
if (!channel.isClosedForWrite) {
while (messages.isNotEmpty()) {
- sendTo(channel, messages.removeFirst())
+ // Waiting for room would save no memory here, because the earlier
messages are already in memory.
+ // Waiting would also block the thread that serves the task.
+ queueLine(channel, messages.removeFirst(), waitForRoom = false)
}
}
}
+ @Synchronized
fun send(message: LogMessage) {
val channel = writer
if (channel == null || channel.isClosedForWrite) {
messages.addLast(message)
} else {
- sendTo(channel, message)
+ queueLine(channel, message, waitForRoom = true)
}
}
+ /** Waits until the lines queued so far are written. Keeps waiting if the
calling thread is interrupted. */
+ fun flush() {
+ val written = CompletableFuture<Unit>()
+ pending.add { written.complete(Unit) }
+ written.join()
+ }
+
+ /**
+ * Stops queuing lines for the channel and waits until the queued lines are
written. Lines logged once
+ * queuing stops stay in [messages]. After this call, closing the channel
cannot cut a line that the sender
+ * thread is writing.
+ */
+ fun detachAndFlush() {
+ synchronized(this) { writer = null }
+ flush()
+ }
+
internal fun reset() {
writer = null
messages.clear()
}
- private fun sendTo(
+ private fun queueLine(
writer: ByteWriteChannel,
message: LogMessage,
+ waitForRoom: Boolean,
) {
val map = message.arguments.toMutableMap()
map["event"] = message.event
map["level"] = message.level.name.lowercase()
map["logger"] = message.logger
map["timestamp"] = message.timestamp.toString()
- // TODO: Can this be done asynchronously instead?
- runBlocking { writer.writeString("${map.toJsonElement()}\n") }
+ // Build the line now, so that it shows the arguments as they were when
the caller logged them.
+ val line = "${map.toJsonElement()}\n"
+ // A line longer than the limit waits until the sender thread has written
every other queued line.
+ val chars = if (waitForRoom) minOf(line.length, PENDING_CHARS) else 0
+ // Code that restores a thread's interrupt status often logs next, so keep
waiting if the calling
+ // thread is interrupted. acquireUninterruptibly also keeps the calling
thread's interrupt status.
+ room.acquireUninterruptibly(chars)
Review Comment:
Thanks for the review. `send` now waits for room outside `LogSender`'s lock.
It takes the lock twice, and each time only briefly. The first time, `send`
checks whether `LogSender` still queues lines. If `LogSender` does, `send`
releases the lock and waits for room. The second time, `send` adds the line to
the queue that the sender thread writes from. So no thread holds the lock while
it waits for room.
--
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]
