Merge pull request #86 from JuliaData/jps/mem-reserve-opt

Memory reservation optimizations

.github/workflows/test.yml

@@ -13,16 +13,17 @@ jobs:
     strategy:
       matrix:
         julia-version:
-          - '~1.8'
-          - '~1.9'
-          - '~1.10'
+          - '1.8'
+          - '1.9'
+          - '1.10'
+          - '1.11'
           - 'nightly'
       fail-fast: false
     name: Test with Julia ${{ matrix.julia-version }}
     steps:
       - uses: actions/checkout@v2
       - name: Setup julia
-        uses: julia-actions/setup-julia@v1
+        uses: julia-actions/setup-julia@v2
         with:
           version: ${{ matrix.julia-version }}
       - uses: julia-actions/julia-runtest@v1

src/MemPool.jl

@@ -55,6 +55,7 @@ approx_size(f::FileRef) = f.size
 include("io.jl")
 include("lock.jl")
 include("read_write_lock.jl")
+include("clock.jl")
 include("datastore.jl")
 
 """

src/clock.jl

@@ -0,0 +1,59 @@
+##
+# This file is a part of MemPool.jl. License is MIT
+#
+# Based upon https://github.com/google/benchmark, which is licensed under Apache v2:
+# https://github.com/google/benchmark/blob/master/LICENSE
+#
+# In compliance with the Apache v2 license, here are the original copyright notices:
+# Copyright 2015 Google Inc. All rights reserved.
+##
+
+struct TimeSpec
+    tv_sec::UInt64 # time_t
+    tv_nsec::UInt64
+end
+
+maketime(ts) = ts.tv_sec * UInt(1e9) + ts.tv_nsec
+
+# From bits/times.h on a Linux system
+# Check if those are the same on BSD
+if Sys.islinux()
+    const CLOCK_MONOTONIC          = Cint(1)
+    const CLOCK_PROCESS_CPUTIME_ID = Cint(2)
+    const CLOCK_THREAD_CPUTIME_ID  = Cint(3)
+    const CLOCK_MONOTONIC_COARSE   = Cint(6)
+elseif Sys.isfreebsd() # atleast on FreeBSD 11.1
+    const CLOCK_MONOTONIC          = Cint(4)
+    const CLOCK_PROCESS_CPUTIME_ID = Cint(14)
+elseif Sys.isapple() # Version 10.12 required
+    const CLOCK_MONOTONIC          = Cint(6)
+    const CLOCK_PROCESS_CPUTIME_ID = Cint(12)
+end
+
+@static if Sys.isunix()
+@inline function clock_gettime(cid)
+    ts = Ref{TimeSpec}()
+    ccall(:clock_gettime, Cint, (Cint, Ref{TimeSpec}), cid, ts)
+    return ts[]
+end
+
+@inline function realtime()
+    maketime(clock_gettime(CLOCK_MONOTONIC))
+end
+
+@inline function cputime()
+    maketime(clock_gettime(CLOCK_PROCESS_CPUTIME_ID))
+end
+end
+
+@static if Sys.islinux()
+@inline function cputhreadtime()
+    maketime(clock_gettime(CLOCK_THREAD_CPUTIME_ID))
+end
+@inline function fasttime()
+    maketime(clock_gettime(CLOCK_MONOTONIC_COARSE))
+end
+else
+@inline cputhreadtime() = time_ns() # HACK
+@inline fasttime() = time_ns() # HACK
+end

src/datastore.jl

@@ -382,16 +382,24 @@ isondisk(x::DRef) = isondisk(x.id)
 
 const MEM_RESERVED = Ref{UInt}(512 * (1024^2)) # Reserve 512MB of RAM for OS
 const MEM_RESERVE_LOCK = Threads.ReentrantLock()
+const MEM_RESERVE_SWEEPS = Ref{Int}(3)
 
 """
 When called, ensures that at least `MEM_RESERVED[] + size` bytes are available
 to the OS. If there is not enough memory available, then a variety of calls to
 the GC are performed to free up memory until either the reservation limit is
 satisfied, or `max_sweeps` number of cycles have elapsed.
 """
-function ensure_memory_reserved(size::Integer=0; max_sweeps::Integer=5)
+function ensure_memory_reserved(size::Integer=0; max_sweeps::Integer=MEM_RESERVE_SWEEPS[])
     sat_sub(x::T, y::T) where T = x < y ? zero(T) : x-y
 
+    max_sweeps == 0 && return
+
+    # Do a quick (cached) check, to optimize for many calls to this function when memory isn't tight
+    if Int(storage_available(CPURAMResource())) - size >= MEM_RESERVED[]
+        return
+    end
+
     # Check whether the OS is running tight on memory
     sweep_ctr = 0
     while true
@@ -400,7 +408,7 @@ function ensure_memory_reserved(size::Integer=0; max_sweeps::Integer=5)
         end || break
 
         # We need more memory! Let's encourage the GC to clear some memory...
-        sweep_start = time_ns()
+        sweep_start = fasttime()
         mem_used = with(QUERY_MEM_OVERRIDE => true) do
             storage_utilized(CPURAMResource())
         end
@@ -427,7 +435,7 @@ function ensure_memory_reserved(size::Integer=0; max_sweeps::Integer=5)
         end
 
         sweep_ctr += 1
-        if sweep_ctr == max_sweeps
+        if sweep_ctr >= max_sweeps
             @debug "Made too many sweeps, bailing out..."
             break
         end

src/storage.jl

@@ -185,7 +185,7 @@ end
 QueriedMemInfo() = QueriedMemInfo(UInt64(0), UInt64(0))
 const QUERY_MEM_AVAILABLE = Ref(QueriedMemInfo())
 const QUERY_MEM_CAPACITY = Ref(QueriedMemInfo())
-const QUERY_MEM_PERIOD = 10 * 1000^2 # 10ms
+const QUERY_MEM_PERIOD = Ref(10 * 1000^2) # 10ms
 const QUERY_MEM_OVERRIDE = ScopedValue(false)
 function _query_mem_periodically(kind::Symbol)
     if !(kind in (:available, :capacity))
@@ -197,8 +197,8 @@ function _query_mem_periodically(kind::Symbol)
         QUERY_MEM_CAPACITY
     end
     mem_info = mem_bin[]
-    now_ns = time_ns()
-    if QUERY_MEM_OVERRIDE[] || mem_info.last_ns < now_ns - QUERY_MEM_PERIOD
+    now_ns = fasttime()
+    if QUERY_MEM_OVERRIDE[] || mem_info.last_ns < now_ns - QUERY_MEM_PERIOD[]
         if kind == :available
             new_mem_info = QueriedMemInfo(free_memory(), now_ns)
         elseif kind == :capacity