sync.lua

--- The sync library exposes interfaces for various synchronization structures.
--
-- @module system.sync

local expect = require "expect"
local util = require "util"

local sync = {
    mutex = {},
    semaphore = {},
    conditionVariable = {},
    atomic = {},
    barrier = {},
    rwLock = {},
}

--#region Mutex

--- A mutex is an object that controls access to a variable across multiple threads.
-- It ensures only one thread accesses a resource at a time by blocking other
-- threads from locking the mutex until the current thread unlocks it.
-- @type mutex

--- Creates a new mutex.
-- @tparam[opt] boolean recursive Whether to make the mutex recursive
-- @treturn mutex The new mutex object
function sync.mutex.new(recursive)
    expect(1, recursive, "boolean", "nil")
    return setmetatable({recursive = recursive and 0}, {__name = "mutex", __index = sync.mutex})
end

--- Locks the mutex, waiting if it's currently owned by another thread.
function sync.mutex:lock()
    expect(1, self, "mutex")
    return util.syscall.lockmutex(self)
end

--- Unlocks the mutex. This is only valid from the thread that owns the lock.
function sync.mutex:unlock()
    expect(1, self, "mutex")
    return util.syscall.unlockmutex(self)
end

--- Tries to lock the thread, returning false if it could not be locked.
-- @treturn boolean Whether the mutex is now locked
function sync.mutex:tryLock()
    expect(1, self, "mutex")
    return util.syscall.trylockmutex(self)
end

--- Locks the mutex, waiting until it's unlocked or until the specified timeout.
-- @tparam number timeout The number of seconds to wait
-- @treturn boolean Whether the mutex is now locked
function sync.mutex:tryLockFor(timeout)
    expect(1, self, "mutex")
    expect(2, timeout, "number")
    return util.syscall.timelockmutex(self, timeout)
end

--#endregion
--#region Semaphore

--- A semaphore controls access to a limited number of resources. A function may
-- acquire a resource from the semaphore, decrementing its available count. If
-- the count is zero, it waits until another function releases a resource, at
-- which point it will acquire it and return.
-- @type semaphore

--- Creates a new semaphore.
-- @tparam[opt=1] number init The initial count of the semaphore
-- @treturn semaphore The new semaphore object
function sync.semaphore.new(init)
    expect(1, init, "number", "nil")
    if init then expect.range(init, 0) end
    return setmetatable({count = init or 1}, {__name = "semaphore", __index = sync.semaphore})
end

--- Acquires a resource from the semaphore, waiting until there is one available.
function sync.semaphore:acquire()
    expect(1, self, "semaphore")
    return util.syscall.acquiresemaphore(self)
end

--- Acquires a resource from the semaphore, waiting until there is one available or until a timeout.
-- @tparam number timeout The number of seconds to wait
-- @treturn boolean Whether the resource was acquired
function sync.semaphore:tryAcquireFor(timeout)
    expect(1, self, "semaphore")
    expect(2, timeout, "number")
    return util.syscall.timeacquiresemaphore(self, timeout)
end

--- Releases a resource to the semaphore. This can be called from any thread.
function sync.semaphore:release()
    expect(1, self, "semaphore")
    return util.syscall.releasesemaphore(self)
end

--#endregion
--#region Condition variable

--- A condition variable allows threads to wait until another thread notifies
-- them to resume.
-- @type conditionVariable

--- Creates a new condition variable.
-- @treturn conditionVariable The new condition variable.
function sync.conditionVariable.new()
    return setmetatable({
        lock = sync.mutex.new(),
        sem = sync.semaphore.new(0),
        waiting = 0
    }, {__name = "condition variable", __index = sync.conditionVariable})
end

--- Waits for a notification from another thread.
function sync.conditionVariable:wait()
    expect(1, self, "condition variable")
    self.lock:lock()
    self.waiting = self.waiting + 1
    self.lock:unlock()
    self.sem:acquire()
    self.lock:lock()
    self.waiting = self.waiting - 1
    self.lock:unlock()
end

--- Waits for a notification from another thread, or until a timeout occurs.
-- @tparam number timeout The number of seconds to wait
-- @treturn boolean Whether a notification occurred
function sync.conditionVariable:waitFor(timeout)
    expect(1, self, "condition variable")
    expect(2, timeout, "number")
    self.lock:lock()
    self.waiting = self.waiting + 1
    self.lock:unlock()
    local retval = self.sem:tryAcquireFor(timeout)
    self.lock:lock()
    self.waiting = self.waiting - 1
    self.lock:unlock()
    return retval
end

--- Notifies a single (unspecified) thread to continue.
function sync.conditionVariable:notifyOne()
    expect(1, self, "condition variable")
    self.sem:release()
end

--- Notifies all waiting threads to continue.
function sync.conditionVariable:notifyAll()
    expect(1, self, "condition variable")
    self.lock:lock()
    self.sem.count = self.sem.count + self.waiting - 1
    self.sem:release()
    self.lock:unlock()
end

--#endregion
--#region Atomic variables

-- TODO

--#endregion
--#region Barrier

--- A barrier is a lock that waits for a specific number of threads to wait on
-- the object, at which point all threads will be released together.
-- @type barrier

--- Creates a new barrier object.
-- @tparam number count The number of threads to wait for
-- @treturn barrier A new barrier object
function sync.barrier.new(count)
    expect(1, count, "number")
    expect.range(count, 1)
    return setmetatable({
        cvar = sync.conditionVariable.new(),
        lock = sync.mutex.new(),
        left = count,
        count = count,
        cycles = 0
    }, {__name = "barrier", __index = sync.barrier})
end

--- Adds one to the thread wait count, and waits until it meets the limit.
-- @treturn boolean Whether this call directly resulted in the barrier being met
function sync.barrier:wait()
    expect(1, self, "barrier")
    self.lock:lock()
    self.left = self.left - 1
    if self.left == 0 then
        self.left = self.count
        self.cycles = self.cycles + 1
        self.lock:unlock()
        self.cvar:notifyAll()
        return true
    else
        self.lock:unlock()
        self.cvar:wait()
        return false
    end
end

--#endregion
--#region Readers-writer lock

--- A readers-writer lock implements two related locks: a read lock, which can
-- be held by multiple threads, and a write lock, which can only be held by one
-- thread. Multiple threads can hold a read lock, but a write lock blocks both
-- read and write locks.
-- @type rwlock

--- Creates a new RW lock.
-- @treturn rwlock The new RW lock
function sync.rwLock.new()
    return setmetatable({
        count = 0,
        readLock = sync.mutex.new(),
        globalLock = sync.semaphore.new(1)
    }, {__name = "rwlock", __index = sync.rwLock})
end

--- Acquires the lock for reading, waiting for the write lock to be released first.
function sync.rwLock:lockRead()
    expect(1, self, "rwlock")
    self.readLock:lock()
    self.count = self.count + 1
    if self.count == 1 then self.globalLock:acquire() end
    self.readLock:unlock()
end

--- Releases the lock for reading.
function sync.rwLock:unlockRead()
    expect(1, self, "rwlock")
    self.readLock:lock()
    self.count = self.count - 1
    if self.count == 0 then self.globalLock:release() end
    self.readLock:unlock()
end

--- Acquires the lock for writing, waiting for the read and write locks to be released.
function sync.rwLock:lockWrite()
    expect(1, self, "rwlock")
    self.globalLock:acquire()
end

--- Releases the lock for writing.
function sync.rwLock:unlockWrite()
    expect(1, self, "rwlock")
    self.globalLock:release()
end

--#endregion

--- Calls a function, ensuring that the mutex is locked before calling and unlocked
-- after calling, even if the function returns early or throws an error.
-- @tparam mutex mutex The mutex to lock
-- @tparam function fn The function to call
-- @tparam any ... Any parameters to pass
-- @treturn any... The return values from the function
function sync.lockGuard(mutex, fn, ...)
    expect(1, mutex, "mutex")
    expect(2, fn, "function")
    mutex:lock()
    local res = table.pack(pcall(fn, ...))
    mutex:unlock()
    if not res[1] then error(res[2], 0) end
    return table.unpack(res, 2, res.n)
end

--- Creates a new synchronized table. A synchronized table is a table that's
-- protected by a mutex. The table can only be accessed by calling it as a
-- function, which will lock the mutex and calls the callback with the table.
-- @treturn function(callback:function(any):any) The accessor for the variable
function sync.synctab()
    local tab = {}
    local lock = sync.mutex.new()
    return function(fn)
        expect(1, fn, "function")
        return sync.lockGuard(lock, fn, tab)
    end
end

return sync