Logging: Improve threadsafety (#57591)

Closes #57376
Closes #34037

- Adds a lock in `SimpleLogger` and `ConsoleLogger` for use on maxlog
tracking and stream writes to improve threadsafety.
Closely similar to #54497

- Turns the internal `_min_enabled_level` into a `Threads.Atomic`. There
are [some direct
interactions](https://juliahub.com/ui/Search?type=code&q=_min_enabled_level&w=true)
to this internal in the ecosystem, but they should still work
```
julia> Base.CoreLogging._min_enabled_level[] = Logging.Info+1
LogLevel(1)
```

- Brings tests over from #57448

Performance seems highly similar:

```
julia> @time for i in 1:10000
          @info "foo" maxlog=10000000
       end
[ Info: foo
...
  0.481446 seconds (1.33 M allocations: 89.226 MiB, 0.49% gc time)
```

```
  0.477235 seconds (1.31 M allocations: 79.002 MiB, 1.77% gc time)
```

(cherry picked from commit 9af9650)

Author: IanButterworth

base/Base.jl

@@ -330,10 +330,6 @@ using .Libc: getpid, gethostname, time, memcpy, memset, memmove, memcmp
 const libblas_name = "libblastrampoline" * (Sys.iswindows() ? "-5" : "")
 const liblapack_name = libblas_name
 
-# Logging
-include("logging.jl")
-using .CoreLogging
-
 # Concurrency
 include("linked_list.jl")
 include("condition.jl")
@@ -345,6 +341,10 @@ include("task.jl")
 include("threads_overloads.jl")
 include("weakkeydict.jl")
 
+# Logging
+include("logging.jl")
+using .CoreLogging
+
 include("env.jl")
 
 # functions defined in Random

base/logging.jl

@@ -131,6 +131,7 @@ isless(a::LogLevel, b::LogLevel) = isless(a.level, b.level)
 +(level::LogLevel, inc::Integer) = LogLevel(level.level+inc)
 -(level::LogLevel, inc::Integer) = LogLevel(level.level-inc)
 convert(::Type{LogLevel}, level::Integer) = LogLevel(level)
+convert(::Type{Int32}, level::LogLevel) = level.level
 
 const BelowMinLevel = LogLevel(-1000001)
 """
@@ -160,7 +161,8 @@ const Error         = LogLevel(    2000)
 const AboveMaxLevel = LogLevel( 1000001)
 
 # Global log limiting mechanism for super fast but inflexible global log limiting.
-const _min_enabled_level = Ref{LogLevel}(Debug)
+# Atomic ensures that the value is always consistent across threads.
+const _min_enabled_level = Threads.Atomic{Int32}(Debug)
 
 function show(io::IO, level::LogLevel)
     if     level == BelowMinLevel  print(io, "BelowMinLevel")
@@ -383,7 +385,7 @@ function logmsg_code(_module, file, line, level, message, exs...)
             level = $level
             # simplify std_level code emitted, if we know it is one of our global constants
             std_level = $(level isa Symbol ? :level : :(level isa $LogLevel ? level : convert($LogLevel, level)::$LogLevel))
-            if std_level >= $(_min_enabled_level)[]
+            if std_level.level >= $(_min_enabled_level)[]
                 group = $(log_data._group)
                 _module = $(log_data._module)
                 logger = $(current_logger_for_env)(std_level, group, _module)
@@ -526,7 +528,8 @@ end
 
 Disable all log messages at log levels equal to or less than `level`.  This is
 a *global* setting, intended to make debug logging extremely cheap when
-disabled.
+disabled. Note that this cannot be used to enable logging that is currently disabled
+by other mechanisms.
 
 # Examples
 ```julia
@@ -646,17 +649,21 @@ close(closed_stream)
 Simplistic logger for logging all messages with level greater than or equal to
 `min_level` to `stream`. If stream is closed then messages with log level
 greater or equal to `Warn` will be logged to `stderr` and below to `stdout`.
+
+This Logger is thread-safe, with a lock taken around orchestration of message
+limits i.e. `maxlog`, and writes to the stream.
 """
 struct SimpleLogger <: AbstractLogger
     stream::IO
+    lock::ReentrantLock
     min_level::LogLevel
     message_limits::Dict{Any,Int}
 end
-SimpleLogger(stream::IO, level=Info) = SimpleLogger(stream, level, Dict{Any,Int}())
+SimpleLogger(stream::IO, level=Info) = SimpleLogger(stream, ReentrantLock(), level, Dict{Any,Int}())
 SimpleLogger(level=Info) = SimpleLogger(closed_stream, level)
 
 shouldlog(logger::SimpleLogger, level, _module, group, id) =
-    get(logger.message_limits, id, 1) > 0
+    @lock logger.lock get(logger.message_limits, id, 1) > 0
 
 min_enabled_level(logger::SimpleLogger) = logger.min_level
 
@@ -667,15 +674,14 @@ function handle_message(logger::SimpleLogger, level::LogLevel, message, _module,
     @nospecialize
     maxlog = get(kwargs, :maxlog, nothing)
     if maxlog isa Core.BuiltinInts
-        remaining = get!(logger.message_limits, id, Int(maxlog)::Int)
-        logger.message_limits[id] = remaining - 1
-        remaining > 0 || return
+        @lock logger.lock begin
+            remaining = get!(logger.message_limits, id, Int(maxlog)::Int)
+            remaining == 0 && return
+            logger.message_limits[id] = remaining - 1
+        end
     end
     buf = IOBuffer()
     stream::IO = logger.stream
-    if !(isopen(stream)::Bool)
-        stream = stderr
-    end
     iob = IOContext(buf, stream)
     levelstr = level == Warn ? "Warning" : string(level)
     msglines = eachsplit(chomp(convert(String, string(message))::String), '\n')
@@ -689,7 +695,13 @@ function handle_message(logger::SimpleLogger, level::LogLevel, message, _module,
         println(iob, "│   ", key, " = ", val)
     end
     println(iob, "└ @ ", _module, " ", filepath, ":", line)
-    write(stream, take!(buf))
+    b = take!(buf)
+    @lock logger.lock begin
+        if !(isopen(stream)::Bool)
+            stream = stderr
+        end
+        write(stream, b)
+    end
     nothing
 end
 

stdlib/Logging/src/ConsoleLogger.jl

@@ -9,6 +9,9 @@ interactive work with the Julia REPL.
 
 Log levels less than `min_level` are filtered out.
 
+This Logger is thread-safe, with locks for both orchestration of message
+limits i.e. `maxlog`, and writes to the stream.
+
 Message formatting can be controlled by setting keyword arguments:
 
 * `meta_formatter` is a function which takes the log event metadata
@@ -24,6 +27,7 @@ Message formatting can be controlled by setting keyword arguments:
 """
 struct ConsoleLogger <: AbstractLogger
     stream::IO
+    lock::ReentrantLock # do not log within this lock
     min_level::LogLevel
     meta_formatter
     show_limited::Bool
@@ -33,19 +37,19 @@ end
 function ConsoleLogger(stream::IO, min_level=Info;
                        meta_formatter=default_metafmt, show_limited=true,
                        right_justify=0)
-    ConsoleLogger(stream, min_level, meta_formatter,
+    ConsoleLogger(stream, ReentrantLock(), min_level, meta_formatter,
                   show_limited, right_justify, Dict{Any,Int}())
 end
 function ConsoleLogger(min_level=Info;
                        meta_formatter=default_metafmt, show_limited=true,
                        right_justify=0)
-    ConsoleLogger(closed_stream, min_level, meta_formatter,
+    ConsoleLogger(closed_stream, ReentrantLock(), min_level, meta_formatter,
                   show_limited, right_justify, Dict{Any,Int}())
 end
 
 
 shouldlog(logger::ConsoleLogger, level, _module, group, id) =
-    get(logger.message_limits, id, 1) > 0
+    @lock logger.lock get(logger.message_limits, id, 1) > 0
 
 min_enabled_level(logger::ConsoleLogger) = logger.min_level
 
@@ -109,9 +113,11 @@ function handle_message(logger::ConsoleLogger, level::LogLevel, message, _module
     hasmaxlog = haskey(kwargs, :maxlog) ? 1 : 0
     maxlog = get(kwargs, :maxlog, nothing)
     if maxlog isa Core.BuiltinInts
-        remaining = get!(logger.message_limits, id, Int(maxlog)::Int)
-        logger.message_limits[id] = remaining - 1
-        remaining > 0 || return
+        @lock logger.lock begin
+            remaining = get!(logger.message_limits, id, Int(maxlog)::Int)
+            remaining == 0 && return
+            logger.message_limits[id] = remaining - 1
+        end
     end
 
     # Generate a text representation of the message and all key value pairs,
@@ -175,6 +181,7 @@ function handle_message(logger::ConsoleLogger, level::LogLevel, message, _module
         println(iob)
     end
 
-    write(stream, take!(buf))
+    b = take!(buf)
+    @lock logger.lock write(stream, b)
     nothing
 end

stdlib/Logging/test/runtests.jl

@@ -292,4 +292,47 @@ end
     @test occursin("LogLevel(-500): a", String(take!(buf)))
 end
 
+@testset "Logging when multithreaded" begin
+    n = 10000
+    cmd = `$(Base.julia_cmd()) -t4 --color=no $(joinpath(@__DIR__, "threads_exec.jl")) $n`
+    fname = tempname()
+    @testset "Thread safety" begin
+        f = open(fname, "w")
+        @test success(run(pipeline(cmd, stderr=f)))
+        close(f)
+    end
+
+    @testset "No tearing in log printing" begin
+        # Check for print tearing by verifying that each log entry starts and ends correctly
+        f = open(fname, "r")
+        entry_start = r"┌ (Info|Warning|Error): iteration"
+        entry_end = r"└ "
+
+        open_entries = 0
+        total_entries = 0
+        for line in eachline(fname)
+            starts = count(entry_start, line)
+            starts > 1 && error("Interleaved logs: Multiple log entries started on one line")
+            if starts == 1
+                startswith(line, entry_start) || error("Interleaved logs: Log entry started in the middle of a line")
+                open_entries += 1
+                total_entries += 1
+            end
+
+            ends = count(entry_end, line)
+            starts == 1 && ends == 1 && error("Interleaved logs: Log entry started and and another ended on one line")
+            ends > 1 && error("Interleaved logs: Multiple log entries ended on one line")
+            if ends == 1
+                startswith(line, entry_end) || error("Interleaved logs: Log entry ended in the middle of a line")
+                open_entries -= 1
+            end
+            # Ensure no mismatched log entries
+            open_entries >= 0 || error("Interleaved logs")
+        end
+
+        @test open_entries == 0  # Ensure all entries closed properly
+        @test total_entries == n * 3  # Ensure all logs were printed (3 because @debug is hidden)
+    end
+end
+
 end

stdlib/Logging/test/threads_exec.jl

@@ -0,0 +1,13 @@
+using Logging
+
+function test_threads_exec(n)
+    Threads.@threads for i in 1:n
+        @debug "iteration" maxlog=1 _id=Symbol("$(i)_debug") i Threads.threadid()
+        @info "iteration" maxlog=1 _id=Symbol("$(i)_info") i Threads.threadid()
+        @warn "iteration" maxlog=1 _id=Symbol("$(i)_warn") i Threads.threadid()
+        @error "iteration" maxlog=1 _id=Symbol("$(i)_error") i Threads.threadid()
+    end
+end
+
+n = parse(Int, ARGS[1])
+test_threads_exec(n)

stdlib/Test/src/logging.jl

@@ -107,8 +107,8 @@ function Logging.handle_message(logger::TestLogger, level, msg, _module,
         if maxlog isa Core.BuiltinInts
             @lock logger.lock begin
                 remaining = get!(logger.message_limits, id, Int(maxlog)::Int)
+                remaining == 0 && return
                 logger.message_limits[id] = remaining - 1
-                remaining > 0 || return
             end
         end
     end