Skip to content

Instantly share code, notes, and snippets.

@KaQuMiQ
Created July 31, 2019 13:54
Show Gist options
  • Save KaQuMiQ/dff88eb03b84da46696caec890384fd5 to your computer and use it in GitHub Desktop.
Save KaQuMiQ/dff88eb03b84da46696caec890384fd5 to your computer and use it in GitHub Desktop.
Functura
import Foundation
public protocol Executor {
func execute(_ task: @escaping () -> Void)
}
extension DispatchQueue: Executor {
@inlinable public func execute(_ task: @escaping () -> Void) {
async(execute: task)
}
}
extension OperationQueue: Executor {
@inlinable public func execute(_ task: @escaping () -> Void) {
if OperationQueue.current == self {
task()
} else {
addOperation(task)
}
}
}
// based on https://github.com/miquido/futura
#if os(Linux)
import Glibc
#else
import Darwin.POSIX
#endif
#if os(Linux)
@usableFromInline internal let nSecMsec: __time_t = 1_000_000
@usableFromInline internal let mSecSec: __time_t = 1_000 * nSecMsec
#else
@usableFromInline internal let nSecMsec: __darwin_time_t = 1_000_000
@usableFromInline internal let mSecSec: __darwin_time_t = 1_000 * nSecMsec
#endif
/// pthread_mutex api wrapper
public enum Mutex {
/// Error thrown on mutex timeout
public struct Timeout: Error { @usableFromInline internal init() {} }
/// pthread_mutex_t pointer type
public typealias Pointer = UnsafeMutablePointer<pthread_mutex_t>
/// Creates new instance of pthread_mutex.
/// It is not automatically managed by ARC. You are responsible
/// to deallocate it manually by calling destroy function.
///
/// - Parameter recursive: Tells if created mutex should be recursive or not.
/// - Returns: Pointer to new mutex instance
@inlinable public static func make(recursive: Bool) -> Pointer {
let pointer: UnsafeMutablePointer<pthread_mutex_t> = .allocate(capacity: 1)
let attr: UnsafeMutablePointer<pthread_mutexattr_t> = .allocate(capacity: 1)
guard pthread_mutexattr_init(attr) == 0 else { preconditionFailure() }
pthread_mutexattr_settype(attr, recursive ? PTHREAD_MUTEX_RECURSIVE : PTHREAD_MUTEX_NORMAL)
pthread_mutexattr_setpshared(attr, PTHREAD_PROCESS_PRIVATE)
guard pthread_mutex_init(pointer, attr) == 0 else { preconditionFailure() }
pthread_mutexattr_destroy(attr)
attr.deinitialize(count: 1)
attr.deallocate()
return pointer
}
/// Deallocates instance of pthread_mutex
///
/// - Parameter pointer: Pointer to mutex to be destroyed.
@inlinable public static func destroy(_ pointer: Pointer) {
pthread_mutex_destroy(pointer)
pointer.deinitialize(count: 1)
pointer.deallocate()
}
/// Locks on instance of pthread_mutex or waits until unlocked if locked.
///
/// - Parameter pointer: Pointer to mutex to be locked.
@inlinable public static func lock(_ pointer: Pointer) {
pthread_mutex_lock(pointer)
}
/// Tries to lock on instance of pthread_mutex. Locks if unlocked or passes if locked.
///
/// - Parameter pointer: Pointer to mutex to be locked.
/// - Returns: Result of trying to lock. True if succeeded, false otherwise.
@inlinable public static func tryLock(_ pointer: Pointer) -> Bool {
return pthread_mutex_trylock(pointer) == 0
}
/// Unlocks on instance of pthread_mutex
///
/// - Parameter pointer: Pointer to mutex to be unlocked.
@inlinable public static func unlock(_ pointer: Pointer) {
pthread_mutex_unlock(pointer)
}
}
public typealias FailableFuture<Success, Failure: Error> = Future<Result<Success, Failure>>
// MARK: Initialization
public extension FailableFuture {
convenience init
<Success, Failure: Error>
(succeededWith value: Success,
executor: Executor? = nil)
where Value == Result<Success, Failure> {
self.init(with: .success(value), executor: executor)
}
convenience init
<Success, Failure: Error>
(failedWith reason: Failure,
executor: Executor? = nil)
where Value == Result<Success, Failure> {
self.init(with: .failure(reason), executor: executor)
}
}
// MARK: Handlers
public extension FailableFuture {
@inlinable func success
<Success, Failure>
(_ handler: @escaping (Success) -> Void) -> Future<Failure>
where Failure: Error, Value == Result<Success, Failure> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let failureFuture: Future<Failure>
if case let .success(value)? = value {
handler(value)
failureFuture = .never()
} else if case let .failure(reason)? = value {
failureFuture = .init(with: reason, executor: executor)
} else {
failureFuture = .init(executor: executor)
observers
.append { result in
switch result {
case let .success(value):
handler(value)
case let .failure(reason):
failureFuture.complete(with: reason) }
}
}
return failureFuture
}
@inlinable func failure
<Success, Failure>
(_ handler: @escaping (Failure) -> Void) -> Future<Success>
where Failure: Error, Value == Result<Success, Failure> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let successFuture: Future<Success>
if case let .success(value)? = value {
successFuture = .init(with: value, executor: executor)
} else if case let .failure(reason)? = value {
handler(reason)
successFuture = .never()
} else {
successFuture = .init(executor: executor)
observers
.append { result in
switch result {
case let .success(value):
successFuture.complete(with: value)
case let .failure(reason):
handler(reason)
}
}
}
return successFuture
}
}
// MARK: Transformations
public extension FailableFuture {
@inlinable func mapSuccess
<Success, Failure, Mapped>
(_ transformation: @escaping (Success) -> Result<Mapped, Failure>) -> FailableFuture<Mapped, Failure>
where Failure: Error, Value == Result<Success, Failure> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let mappedFuture: FailableFuture<Mapped, Failure>
if case let .success(value)? = value {
mappedFuture = .init(with: transformation(value), executor: executor)
} else if case let .failure(reason)? = value {
mappedFuture = .init(with: .failure(reason), executor: executor)
} else {
mappedFuture = .init(executor: executor)
observers
.append { result in
switch result {
case let .success(value):
mappedFuture.complete(with: transformation(value))
case let .failure(reason):
mappedFuture.complete(with: .failure(reason)) }
}
}
return mappedFuture
}
@inlinable func mapSuccess
<Success, Failure, Mapped>
(_ transformation: @escaping (Success) -> Mapped) -> FailableFuture<Mapped, Failure>
where Failure: Error, Value == Result<Success, Failure> {
return mapSuccess { (value) -> Result<Mapped, Failure> in
.success(transformation(value))
}
}
@inlinable func mapSuccess
<Success, Failure, Mapped>
(_ transformation: @escaping (Success) -> Failure) -> FailableFuture<Mapped, Failure>
where Failure: Error, Value == Result<Success, Failure> {
return mapSuccess { (value) -> Result<Mapped, Failure> in
.failure(transformation(value))
}
}
@inlinable func mapSuccess
<Success, Failure, Mapped>
(_ transformation: @escaping (Success) throws -> Mapped) -> FailableFuture<Mapped, Error>
where Failure: Error, Value == Result<Success, Failure> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let mappedFuture: FailableFuture<Mapped, Error>
if case let .success(value)? = value {
do {
mappedFuture = try .init(with: .success(transformation(value)), executor: executor)
} catch {
mappedFuture = .init(with: .failure(error), executor: executor)
}
} else if case let .failure(reason)? = value {
mappedFuture = .init(with: .failure(reason), executor: executor)
} else {
mappedFuture = .init(executor: executor)
observers
.append { result in
switch result {
case let .success(value):
do {
try mappedFuture.complete(with: .success(transformation(value)))
} catch {
mappedFuture.complete(with: .failure(error))
}
case let .failure(reason):
mappedFuture.complete(with: .failure(reason)) }
}
}
return mappedFuture
}
@inlinable func mapFailure
<Success, Failure, Mapped>
(_ transformation: @escaping (Failure) -> Result<Success, Mapped>) -> FailableFuture<Success, Mapped>
where Failure: Error, Mapped: Error, Value == Result<Success, Failure> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let mappedFuture: FailableFuture<Success, Mapped>
if case let .success(value)? = value {
mappedFuture = .init(with: .success(value), executor: executor)
} else if case let .failure(reason)? = value {
mappedFuture = .init(with: transformation(reason), executor: executor)
} else {
mappedFuture = .init(executor: executor)
observers
.append { result in
switch result {
case let .success(value):
mappedFuture.complete(with: .success(value))
case let .failure(reason):
mappedFuture.complete(with: transformation(reason)) }
}
}
return mappedFuture
}
@inlinable func mapFailure
<Success, Failure, Mapped>
(_ transformation: @escaping (Failure) -> Success) -> FailableFuture<Success, Mapped>
where Failure: Error, Mapped: Error, Value == Result<Success, Failure> {
return mapFailure { (reason) -> Result<Success, Mapped> in
.success(transformation(reason))
}
}
@inlinable func mapFailure
<Success, Failure, Mapped>
(_ transformation: @escaping (Failure) -> Mapped) -> FailableFuture<Success, Mapped>
where Failure: Error, Mapped: Error, Value == Result<Success, Failure> {
return mapFailure { (reason) -> Result<Success, Mapped> in
.failure(transformation(reason))
}
}
}
// MARK: Base
public final class Future<Value> {
public typealias Handler = (Value) -> Void
public typealias Promise = (future: Future<Value>, complete: (Value) -> Void)
@usableFromInline internal let mtx: Mutex.Pointer = Mutex.make(recursive: true)
@usableFromInline internal let executor: Executor?
@usableFromInline internal private(set) var value: Value? = nil
@usableFromInline internal var observers: Array<Handler> = .init()
@usableFromInline internal init(with value: Value? = nil,
executor: Executor? = nil) {
self.value = value
self.executor = executor
self.observers.reserveCapacity(1)
}
}
// MARK: Initialization
extension Future {
public convenience init(completedWith value: Value,
executor: Executor? = nil) {
self.init(with: value, executor: executor)
}
public static func never() -> Future {
return .init()
}
public static func promise(of: Value.Type = Value.self,
executor: Executor? = nil) -> Promise {
let future: Future<Value> = .init(executor: executor)
return (future, future.complete(with:))
}
}
// MARK: Completion
internal extension Future {
@usableFromInline @inline(__always) func complete(with value: Value) {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
guard case .none = self.value else { return }
self.value = value
if let executor = executor {
observers.forEach { (handler) in
executor.execute { handler(value) }
}
} else {
observers.forEach { $0(value) }
}
observers = .init()
}
}
// MARK: Handlers
public extension Future {
@inlinable func then(_ handler: @escaping Handler) -> Future {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let nextFuture: Future = .init(executor: executor)
switch value {
case let .some(value):
if let executor = executor {
executor.execute {
handler(value)
nextFuture.complete(with: value)
}
} else {
handler(value)
}
case .none:
observers.append { value in
handler(value)
nextFuture.complete(with: value)
}
}
return nextFuture
}
@discardableResult @inlinable func `do`(_ handler: @escaping Handler) -> Future<Void> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let completionFuture: Future<Void> = .init(executor: executor)
switch value {
case let .some(value):
if let executor = executor {
executor.execute {
handler(value)
completionFuture.complete(with: Void())
}
} else {
handler(value)
completionFuture.complete(with: Void())
}
case .none:
observers.append { value in
handler(value)
completionFuture.complete(with: Void())
}
}
return completionFuture
}
}
// MARK: Transformations
public extension Future {
@inlinable func map<Mapped>(_ transformation: @escaping (Value) -> Mapped) -> Future<Mapped> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let mappedFuture: Future<Mapped>
if let value = value {
mappedFuture = .init(with: transformation(value), executor: executor)
} else {
mappedFuture = .init(executor: executor)
observers.append { mappedFuture.complete(with: transformation($0)) }
}
return mappedFuture
}
@inlinable func flatMap<Mapped>(_ transformation: @escaping (Value) -> Future<Mapped>) -> Future<Mapped> {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let mappedFuture: Future<Mapped> = .init(executor: executor)
if let value = value {
transformation(value).do(mappedFuture.complete(with:))
} else {
observers.append { transformation($0).do(mappedFuture.complete(with:)) }
}
return mappedFuture
}
}
// MARK: Executors
public extension Future {
@inlinable func `switch`(to executor: Executor) -> Future {
Mutex.lock(mtx)
defer { Mutex.unlock(mtx) }
let nextFuture: Future
if let value = value {
nextFuture = .init(with: value, executor: executor)
} else {
nextFuture = .init(executor: executor)
observers.append { nextFuture.complete(with: $0) }
}
return nextFuture
}
}
// MARK: zip
@inlinable public func zip<T1, T2>(_ f1: Future<T1>, _ f2: Future<T2>) -> Future<(T1, T2)> {
var result: (T1?, T2?) = (nil, nil)
let resultFuture: Future<(T1, T2)> = .init()
let mtx: Mutex.Pointer = Mutex.make(recursive: true)
f1.do { (val1) in
Mutex.lock(mtx)
if case let (nil, val2?) = result {
resultFuture.complete(with: (val1, val2))
Mutex.destroy(mtx)
} else {
result = (val1, nil)
Mutex.unlock(mtx)
}
}
f2.do { (val2) in
Mutex.lock(mtx)
if case let (val1?, nil) = result {
resultFuture.complete(with: (val1, val2))
Mutex.destroy(mtx)
} else {
result = (nil, val2)
Mutex.unlock(mtx)
}
}
return resultFuture
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment