Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Reduce memory & redundant work for concurrent TimeZones construction #356

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+59 −24
Open

Reduce memory & redundant work for concurrent TimeZones construction #356

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5bffed5
Reduce memory & redundant work for concurrent TimeZones construction
NHDaly Aug 19, 2021
6c0d3ea
Add `@lock` macro on julia 1.0
NHDaly Aug 20, 2021
beacaaa
PR Feedback from review.
NHDaly Jan 2, 2022
96252ab
Merge branch 'master' into nhd/share-cache
NHDaly Jan 2, 2022
4f444eb
Drop compat for `Base.@lock` now that we're on julia 1.6+ as LTS :)
NHDaly Jan 2, 2022
0258929
PR feedback
NHDaly Feb 13, 2022
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
18 changes: 17 additions & 1 deletion src/TimeZones.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,22 @@ using Serialization
using RecipesBase: RecipesBase, @recipe
using Unicode

if VERSION >= v"1.3-"
using Base: @lock
else
macro lock(l, e)
quote
ll = $(esc(l));
$lock(ll);
try
$(esc(e))
finally
$unlock(ll)
end
end
end
end

import Dates: TimeZone, UTC

export TimeZone, @tz_str, istimezone, FixedTimeZone, VariableTimeZone, ZonedDateTime,
Expand Down Expand Up @@ -39,7 +55,7 @@ abstract type Local <: TimeZone end

function __init__()
# Initialize the thread-local TimeZone cache (issue #342)
_reset_tz_cache()
_tz_cache_init()

# Base extension needs to happen everytime the module is loaded (issue #24)
Dates.CONVERSION_SPECIFIERS['z'] = TimeZone
Expand Down
68 changes: 53 additions & 15 deletions src/types/timezone.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,12 @@
# to the cache, while still being thread-safe.
const THREAD_TZ_CACHES = Vector{Dict{String,Tuple{TimeZone,Class}}}()

# Holding a lock during construction of a specific TimeZone prevents multiple Tasks (on the
# same or different threads) from attempting to construct the same TimeZone object, and
# allows them all to share the result.
const tz_cache_mutex = ReentrantLock()
NHDaly marked this conversation as resolved.
Show resolved Hide resolved
const TZ_CACHE_FUTURES = Dict{String,Channel{Tuple{TimeZone,Class}}}() # Guarded by: tz_cache_mutex

# Based upon the thread-safe Global RNG implementation in the Random stdlib:
# https://github.com/JuliaLang/julia/blob/e4fcdf5b04fd9751ce48b0afc700330475b42443/stdlib/Random/src/RNGs.jl#L369-L385
@inline _tz_cache() = _tz_cache(Threads.threadid())
Expand All @@ -19,10 +25,22 @@ const THREAD_TZ_CACHES = Vector{Dict{String,Tuple{TimeZone,Class}}}()
end
@noinline _tz_cache_length_assert() = @assert false "0 < tid <= length(THREAD_TZ_CACHES)"

function _reset_tz_cache()
# ensures that we didn't save a bad object
function _tz_cache_init()
NHDaly marked this conversation as resolved.
Show resolved Hide resolved
resize!(empty!(THREAD_TZ_CACHES), Threads.nthreads())
end
# ensures that we didn't save a bad object
function _reset_tz_cache()
# Since we use thread-local caches, we spawn a task on _each thread_ to clear that
# thread's local cache.
Threads.@threads for i in 1:Threads.nthreads()
@assert Threads.threadid() === i "TimeZones.TZData.compile() must be called from the main, top-level Task."
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How is this guaranteed when calling compile/_reset_tz_cache from the main task?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't remember / understand the reason, but the Threads.@threads macro only works from the top-level task.

The behavior of @threads is that it evenly divides the for-loop across the number of threads, so if you have exactly nthreads() iterations, exactly one iteration will go on each thread.

help?> Threads.@threads
  Threads.@threads [schedule] for ... end

  A macro to parallelize a for loop to run with multiple threads. Splits the iteration space among multiple tasks and runs those tasks on threads according
  to a scheduling policy. A barrier is placed at the end of the loop which waits for all tasks to finish execution.

  The schedule argument can be used to request a particular scheduling policy. The only currently supported value is :static, which creates one task per
  thread and divides the iterations equally among them. Specifying :static is an error if used from inside another @threads loop or from a thread other than
  1.

  The default schedule (used when no schedule argument is present) is subject to change.

  │ Julia 1.5
  │
  │  The schedule argument is available as of Julia 1.5.

It only works from thread 1, for reasons i can't quite remember, but so this basically means you have to start it from the main Task. (I didn't remember that it was a "thread 1" requirement - i thought it was actually a "main Task" requirement.. i can consider changing the assertion message, perhaps? But i think it's easier guidance to say "don't call this concurrently, dude")

empty!(_tz_cache())
end
@lock tz_cache_mutex begin
NHDaly marked this conversation as resolved.
Show resolved Hide resolved
empty!(TZ_CACHE_FUTURES)
end
return nothing
end

"""
TimeZone(str::AbstractString) -> TimeZone
Expand Down Expand Up @@ -68,20 +86,40 @@ function TimeZone(str::AbstractString, mask::Class=Class(:DEFAULT))
# Note: If the class `mask` does not match the time zone we'll still load the
# information into the cache to ensure the result is consistent.
tz, class = get!(_tz_cache(), str) do
tz_path = joinpath(TZData.COMPILED_DIR, split(str, "/")...)

if isfile(tz_path)
open(deserialize, tz_path, "r")
elseif occursin(FIXED_TIME_ZONE_REGEX, str)
FixedTimeZone(str), Class(:FIXED)
elseif !isdir(TZData.COMPILED_DIR) || isempty(readdir(TZData.COMPILED_DIR))
# Note: Julia 1.0 supresses the build logs which can hide issues in time zone
# compliation which result in no tzdata time zones being available.
throw(ArgumentError(
"Unable to find time zone \"$str\". Try running `TimeZones.build()`."
))
# Even though we're using Thread-local caches, we still need to lock during
# construction to prevent multiple tasks redundantly constructing the same object,
# and potential thread safety violations due to Tasks migrating threads.
# NOTE that we only grab the lock if the TZ doesn't exist, so the mutex contention
# is not on the critical path for most constructors. :)
constructing = false
# We lock the mutex, but for only a short, *constant time* duration, to grab the
# future for this TimeZone, or create the future if it doesn't exist.
future = @lock tz_cache_mutex begin
get!(TZ_CACHE_FUTURES, str) do
constructing = true
Channel{Tuple{TimeZone,Class}}(1)
end
end
if constructing
tz_path = joinpath(TZData.COMPILED_DIR, split(str, "/")...)

tz = if isfile(tz_path)
NHDaly marked this conversation as resolved.
Show resolved Hide resolved
open(deserialize, tz_path, "r")
elseif occursin(FIXED_TIME_ZONE_REGEX, str)
FixedTimeZone(str), Class(:FIXED)
elseif !isdir(TZData.COMPILED_DIR) || isempty(readdir(TZData.COMPILED_DIR))
# Note: Julia 1.0 supresses the build logs which can hide issues in time zone
# compliation which result in no tzdata time zones being available.
throw(ArgumentError(
"Unable to find time zone \"$str\". Try running `TimeZones.build()`."
))
else
throw(ArgumentError("Unknown time zone \"$str\""))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exceptions while constructing will cause threads to be blocked upon waiting for a channel that will never be populated

end

put!(future, tz)
else
throw(ArgumentError("Unknown time zone \"$str\""))
fetch(future)
end
end

Expand Down
2 changes: 1 addition & 1 deletion src/tzdata/TZData.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
module TZData

using Printf
using ...TimeZones: DEPS_DIR
using ...TimeZones: DEPS_DIR, _reset_tz_cache

import Pkg

Expand Down
9 changes: 2 additions & 7 deletions src/tzdata/compile.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ using Dates
using Serialization
using Dates: parse_components

using ...TimeZones: _tz_cache
using ...TimeZones: _reset_tz_cache
using ...TimeZones: TimeZones, TimeZone, FixedTimeZone, VariableTimeZone, Transition, Class
using ...TimeZones: rename
using ..TZData: TimeOffset, ZERO, MIN_GMT_OFFSET, MAX_GMT_OFFSET, MIN_SAVE, MAX_SAVE,
Expand Down Expand Up @@ -696,12 +696,7 @@ function compile(tz_source::TZSource, dest_dir::AbstractString; kwargs...)
isdir(dest_dir) || error("Destination directory doesn't exist")
# When we recompile the TimeZones from a new source, we clear all the existing cached
# TimeZone objects, so that newly constructed objects pick up the newly compiled rules.
# Since we use thread-local caches, we spawn a task on _each thread_ to clear that
# thread's local cache.
Threads.@threads for i in 1:Threads.nthreads()
@assert Threads.threadid() === i "TimeZones.TZData.compile() must be called from the main, top-level Task."
empty!(_tz_cache())
end
_reset_tz_cache()

for (tz, class) in results
parts = split(TimeZones.name(tz), '/')
Expand Down