JuliaConcurrent · tkf · Apr 28, 2022 · Apr 27, 2022 · Apr 28, 2022 · Apr 28, 2022
diff --git a/Project.toml b/Project.toml
@@ -6,6 +6,7 @@ version = "0.1.0-DEV"
 [deps]
 ExternalDocstrings = "e189563c-0753-4f5e-ad5c-be4293c83fb4"
 Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
+Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b"
 Try = "bf1d0ff0-c4a9-496b-85f0-2b0d71c4f32a"
 UnsafeAtomics = "013be700-e6cd-48c3-b4a1-df204f14c38f"
 

diff --git a/docs/src/index.md b/docs/src/index.md
@@ -34,7 +34,7 @@ using ConcurrentUtils
 DocumentationOverview.table_md(
     :[
         var"@tasklet",
-        var"@once",
+        Once,
     ],
     namespace = ConcurrentUtils,
     signature = :name,
@@ -43,7 +43,7 @@ DocumentationOverview.table_md(
 
 ```@docs
 @tasklet
-@once
+Once
 ```
 
 ## Read-write Lock

diff --git a/src/ConcurrentUtils.jl b/src/ConcurrentUtils.jl
@@ -2,13 +2,13 @@ baremodule ConcurrentUtils
 
 export
     # Macros
-    @once,
     @tasklet,
     # Constructors
     Backoff,
     Guard,
     NotSetError,
     OccupiedError,
+    Once,
     Promise,
     ReadWriteGuard,
     ReadWriteLock,
@@ -27,7 +27,6 @@ InternalPrelude.@exported_function try_race_fetch
 InternalPrelude.@exported_function try_race_put!
 InternalPrelude.@exported_function try_race_put_with!
 
-macro once end
 macro tasklet end
 
 struct OccupiedError{T} <: InternalPrelude.Exception
@@ -90,9 +89,10 @@ using Random: Xoshiro
 
 import UnsafeAtomics: UnsafeAtomics, acq_rel
 using ExternalDocstrings: @define_docstrings
+using Serialization: AbstractSerializer, Serialization
 using Try: Try, Ok, Err, @?
 
-import ..ConcurrentUtils: @once, @tasklet
+import ..ConcurrentUtils: @tasklet
 using ..ConcurrentUtils:
     AbstractGuard,
     AbstractReadWriteGuard,
@@ -122,6 +122,7 @@ end
 
 include("utils.jl")
 include("promise.jl")
+include("once.jl")
 include("tasklet.jl")
 include("thread_local_storage.jl")
 
@@ -133,6 +134,7 @@ include("backoff.jl")
 end  # module Internal
 
 const Promise = Internal.Promise
+const Once = Internal.Once
 const ThreadLocalStorage = Internal.ThreadLocalStorage
 const ReadWriteLock = Internal.ReadWriteLock
 const Backoff = Internal.Backoff

diff --git a/src/docs/@once.md b/src/docs/@once.md
diff --git a/src/docs/Once.md b/src/docs/Once.md
@@ -0,0 +1,41 @@
+    Once{T}(f = T)
+    Once(f)
+
+A concurrent object for lazily initializing an object of type `T`.
+
+Given `O = Once{T}(f)`, invoking `O[]` evaluates `v = f()` if `f` has not been called via
+`O[]` and return the value `v`.  Otherwise, `O[]` returns the value `v` returned from the
+first invocation of `O[]`.
+
+# Examples
+
+```julia
+julia> using ConcurrentUtils
+
+julia> O = Once{Vector{Int}}(() -> zeros(Int, 3));
+
+julia> v = O[]
+3-element Vector{Int64}:
+ 0
+ 0
+ 0
+
+julia> v === O[]
+true
+```
+
+# Extended help
+
+When used as in `Once{T}(f)`, the function `f` must always return a value of type `T`.  As
+such, `T() isa T` must hold for type `T` used as in `Once{T}()`.
+
+When used as in `Once(f)`, the function `f` must always return a value of concrete
+consistent type.  If `Once` object is used as a global constant in a package, the type of
+the value returned from `f` must not change for different `julia` processes for each stack
+of Julia environment.  Currently, `Once(f)` also directly invokes `f()` to compute the
+result type but this value is thrown away.  This is because `Once(f)` is assumed to be
+called at the top-level of a package for lazily initializing a global state and serializing
+the computed value in the precompile cache is not desired.
+
+Known limitation: If `O[]` is evaluated for a global `O::Once{T}` during precompilation, the
+resulting value is serialized into the precompile cache.
diff --git a/src/once.jl b/src/once.jl
@@ -0,0 +1,33 @@
+struct Once{T,F}
+    factory::F
+    promise::Promise{T}
+end
+
+function Once{T}() where {T}
+    T isa Type || _once_invalid_type_parameter(T)
+    return Once{T,Type{T}}(T, Promise{T}())
+end
+
+function Once{T}(f) where {T}
+    T isa Type || _once_invalid_type_parameter(T)
+    return Once{T,_typeof(f)}(f, Promise{T}())
+end
+
+@noinline _once_invalid_type_parameter(@nospecialize T) =
+    error("`Once{T}`` expcet a type for `T`; got T = $T")
+
+function Once(f)
+    value = f()
+    T = typeof(value)
+    promise = Promise{T}()
+    # Not using the `value` calculated above in `Promise{T}` to avoid serializing into the
+    # precompiled module just in case it is used as in `const O = Once(f)` in a package.
+    return Once{T,_typeof(f)}(f, promise)
+end
+
+Base.getindex(once::Once) = race_put_with!(once.factory, once.promise)
+
+function Base.delete!(once::Once)
+    @atomic :monotonic once.promise.value = NOTSET
+    return once
+end
diff --git a/src/promise.jl b/src/promise.jl
@@ -1,11 +1,13 @@
 struct NotSet end
+const NOTSET = NotSet()
 
 mutable struct Promise{T}
     @atomic value::Union{T,NotSet}
     @const cond::Threads.Condition
+    global _Promise(::Type{T}, value, cond) where {T} = new{T}(value, cond)
 end
 
-Promise{T}() where {T} = Promise{T}(NotSet(), Threads.Condition())
+Promise{T}() where {T} = _Promise(T, NotSet(), Threads.Condition())
 Promise() = Promise{Any}()
 
 function Base.fetch(promise::Promise)
@@ -75,3 +77,8 @@ function Base.put!(promise::Promise{T}, value) where {T}
         throw(OccupiedError{T}(existing))
     end
 end
+
+function Serialization.serialize(s::AbstractSerializer, promise::Promise{T}) where {T}
+    dummy = _Promise(T, NotSet(), promise.cond)
+    invoke(Serialization.serialize, Tuple{AbstractSerializer,Any}, s, dummy)
+end
diff --git a/src/tasklet.jl b/src/tasklet.jl
@@ -7,12 +7,14 @@ end
 
 Tasklet(thunk::OpaqueClosure{Tuple{},T}) where {T} = Tasklet{T}(thunk, Promise{T}())
 
+#=
 struct TypedTasklet{T,Thunk} <: AbstractTasklet{T}
     thunk::Thunk
     promise::Promise{T}
 end
 
 TypedTasklet{T}(thunk::Thunk) where {T,Thunk} = TypedTasklet{T,Thunk}(thunk, Promise{T}())
+=#
 
 macro tasklet(thunk)
     thunk = Expr(:block, __source__, thunk)
@@ -24,17 +26,3 @@ end
 Base.fetch(tasklet::AbstractTasklet) = fetch(tasklet.promise)
 Base.wait(tasklet::AbstractTasklet) = wait(tasklet.promise)
 ConcurrentUtils.try_race_fetch(tasklet::AbstractTasklet) = try_race_fetch(tasklet.promise)
-
-macro once(ex)
-    @gensym ONCETASK thunk
-    ex = Expr(:block, __source__, ex)
-    toplevel = quote
-        const $ONCETASK = let $thunk = () -> $ex
-            $TypedTasklet{$typeof($thunk())}($thunk)
-            # Using explicitly typed `TypedTasklet` since an opaque closure cannot be
-            # serialized.
-        end
-    end
-    Base.eval(__module__, Expr(:toplevel, __source__, toplevel.args...))
-    return esc(:($ONCETASK()))
-end
diff --git a/test/ConcurrentUtilsTests/src/ConcurrentUtilsTests.jl b/test/ConcurrentUtilsTests/src/ConcurrentUtilsTests.jl
@@ -4,6 +4,7 @@ include("utils.jl")
 
 include("test_promise.jl")
 include("test_tasklet.jl")
+include("test_once.jl")
 include("test_thread_local_storage.jl")
 
 # Locks

diff --git a/test/ConcurrentUtilsTests/src/test_once.jl b/test/ConcurrentUtilsTests/src/test_once.jl
@@ -0,0 +1,27 @@
+module TestOnce
+
+using ConcurrentUtils
+using ConcurrentUtils.Internal: NOTSET
+using Test
+
+const O1 = Once{Vector{Int}}()
+const O2 = Once{Vector{Int}}(() -> zeros(Int, 3))
+const O3 = Once(() -> zeros(Int, 3))
+
+function test_once_identity()
+    @test O1[] === O1[]
+    @test O2[] === O2[]
+    @test O3[] === O3[]
+end
+
+const O1_NOUSE = Once{Vector{Int}}()
+const O2_NOUSE = Once{Vector{Int}}(() -> zeros(Int, 3))
+const O3_NOUSE = Once(() -> zeros(Int, 3))
+
+function test_notset()
+    @test O1_NOUSE.promise.value === NOTSET
+    @test O2_NOUSE.promise.value === NOTSET
+    @test O3_NOUSE.promise.value === NOTSET
+end
+
+end  # module
diff --git a/test/ConcurrentUtilsTests/src/test_tasklet.jl b/test/ConcurrentUtilsTests/src/test_tasklet.jl
@@ -13,10 +13,4 @@ function test_serial_tasklet()
     @test try_race_fetch(t) == Ok(t())
 end
 
-get_a_dict() = @once Dict(:a => 1, :b => 2, :c => 3)
-
-function test_get_a_dict()
-    @test get_a_dict() === get_a_dict()
-end
-
 end  # module