Skip to content

New async implementation #7

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

Merged
merged 2 commits into from
Feb 23, 2016
Merged

New async implementation #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8536342
Merge remote-tracking branch 'crossroadlabs/develop'
ypopovych Feb 23, 2016
25dd1e6
New async run loop implementation
ypopovych Feb 23, 2016
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
4 changes: 2 additions & 2 deletions ExecutionContext/PThreadExecutionContext.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,15 +83,15 @@

sema.wait()

self.rl = RunLoop(runLoop!, autoStop: true)
self.rl = RunLoop(runLoop!)
}

init(runLoop:RunLoop) {
rl = runLoop
}

func async(task:SafeTask) {
rl.addSource(RunLoopSource(task), mode: RunLoop.defaultMode)
rl.addTask(task)
}

func async(after:Double, task:SafeTask) {
Expand Down
143 changes: 116 additions & 27 deletions ExecutionContext/RunLoop.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,54 @@ import CoreFoundation
var cfString: CFString { return unsafeBitCast(self, CFString.self) }
}

private class TaskQueueElement {
private let task : SafeTask
private let source: RunLoopSource
var next: TaskQueueElement? = nil

init(_ task: SafeTask, runLoopSource: RunLoopSource) {
self.task = task
self.source = runLoopSource
}

func run() {
task()
}
}

private class TaskQueue {
private let lock = NSLock()
private var head:TaskQueueElement? = nil
private var tail:TaskQueueElement? = nil

func enqueue(elem: TaskQueueElement) {
defer {
lock.unlock()
}
lock.lock()
if tail == nil {
head = elem
tail = elem
} else {
tail!.next = elem
tail = elem
}
}

func dequeue() -> TaskQueueElement? {
defer {
lock.unlock()
}
lock.lock()
let elem = head
head = head?.next
if head == nil {
tail = nil
}
return elem
}
}

private class RunLoopCallbackInfo {
private var task: SafeTask
private var runLoops: [RunLoop] = []
Expand All @@ -37,10 +85,19 @@ import CoreFoundation
}

private func runLoopCallbackInfoRun(i: UnsafeMutablePointer<Void>) {
let info = Unmanaged<RunLoopCallbackInfo>.fromOpaque(COpaquePointer(i)).takeRetainedValue()
let info = Unmanaged<RunLoopCallbackInfo>.fromOpaque(COpaquePointer(i)).takeUnretainedValue()
info.run()
}

private func runLoopCallbackInfoRetain(i: UnsafePointer<Void>) -> UnsafePointer<Void> {
Unmanaged<RunLoopCallbackInfo>.fromOpaque(COpaquePointer(i)).retain()
return i
}

private func runLoopCallbackInfoRelease(i: UnsafePointer<Void>) {
Unmanaged<RunLoopCallbackInfo>.fromOpaque(COpaquePointer(i)).release()
}

private protocol RunLoopCallback {
var info : RunLoopCallbackInfo { get }
var cfObject: AnyObject { get }
Expand All @@ -56,9 +113,9 @@ import CoreFoundation
if _source == nil {
var context = CFRunLoopSourceContext(
version: 0,
info: UnsafeMutablePointer<Void>(Unmanaged.passRetained(info).toOpaque()),
retain: nil,
release: nil,
info: UnsafeMutablePointer<Void>(Unmanaged.passUnretained(info).toOpaque()),
retain: runLoopCallbackInfoRetain,
release: runLoopCallbackInfoRelease,
copyDescription: nil,
equal: nil,
hash: nil,
Expand All @@ -76,6 +133,19 @@ import CoreFoundation
self.info = RunLoopCallbackInfo(task)
self.priority = priority
}

deinit {
if _source != nil && CFRunLoopSourceIsValid(_source) {
CFRunLoopSourceInvalidate(_source)
_source = nil
}
}

func signal() {
if _source != nil {
CFRunLoopSourceSignal(_source)
}
}
}

private func timerRunCallback(timer: CFRunLoopTimer!, i: UnsafeMutablePointer<Void>) {
Expand All @@ -92,9 +162,9 @@ import CoreFoundation
if _timer == nil {
var context = CFRunLoopTimerContext(
version: 0,
info: UnsafeMutablePointer<Void>(Unmanaged.passRetained(info).toOpaque()),
retain: nil,
release: nil,
info: UnsafeMutablePointer<Void>(Unmanaged.passUnretained(info).toOpaque()),
retain: runLoopCallbackInfoRetain,
release: runLoopCallbackInfoRelease,
copyDescription: nil
)
_timer = CFRunLoopTimerCreate(nil, CFAbsoluteTimeGetCurrent()+delay, -1, 0, 0, timerRunCallback, &context)
Expand All @@ -111,35 +181,42 @@ import CoreFoundation

class RunLoop {
private let cfRunLoop: CFRunLoop!
private let autoStop: Bool

private var taskQueueSource: RunLoopSource
private var taskQueue: TaskQueue

#if !os(Linux)
static let defaultMode:NSString = "kCFRunLoopDefaultMode" as NSString
#else
static let defaultMode:NSString = "kCFRunLoopDefaultMode".bridge()
#endif

init(_ cfRunLoop: CFRunLoop, autoStop: Bool = true) {
init(_ cfRunLoop: CFRunLoop) {
self.cfRunLoop = cfRunLoop
self.autoStop = autoStop

let queue = TaskQueue()

taskQueueSource = RunLoopSource({
var elem = queue.dequeue()
while elem != nil {
elem!.run()
elem = queue.dequeue()
}
})
taskQueue = queue
addSource(taskQueueSource, mode: RunLoop.defaultMode, retainLoop: false)
}

convenience init(_ runLoop: AnyObject, autoStop: Bool = true) {
self.init(unsafeBitCast(runLoop, CFRunLoop.self), autoStop: autoStop)
convenience init(_ runLoop: AnyObject) {
self.init(unsafeBitCast(runLoop, CFRunLoop.self))
}

deinit {
if autoStop && cfRunLoop != nil {
CFRunLoopStop(cfRunLoop)
}
}

static func currentRunLoop(autoStop: Bool = false) -> RunLoop {
return RunLoop(CFRunLoopGetCurrent(), autoStop: autoStop)
static func currentRunLoop() -> RunLoop {
return RunLoop(CFRunLoopGetCurrent())
}

static func mainRunLoop() -> RunLoop {
return RunLoop(CFRunLoopGetMain(), autoStop: false)
return RunLoop(CFRunLoopGetMain())
}

static func currentCFRunLoop() -> AnyObject {
Expand All @@ -156,33 +233,45 @@ import CoreFoundation

static func runInMode(mode: NSString) {
#if !os(Linux)
while CFRunLoopRunInMode(mode.cfString, Double.infinity, false) != .Stopped {}
var result:CFRunLoopRunResult
repeat {
result = CFRunLoopRunInMode(mode.cfString, Double.infinity, false)
} while result != .Finished && result != .Stopped
#else
while CFRunLoopRunInMode(mode.cfString, Double.infinity, false) != Int32(kCFRunLoopRunStopped) {}
var result:Int32
repeat {
result = CFRunLoopRunInMode(mode.cfString, Double.infinity, false)
} while result != Int32(kCFRunLoopRunStopped) && result != Int32(kCFRunLoopRunFinished)
#endif
}

@noreturn static func runForever() {
while true { run() }
}

func addSource(rls: RunLoopSource, mode: NSString) {
func addSource(rls: RunLoopSource, mode: NSString, retainLoop: Bool = true) {
let crls = unsafeBitCast(rls.cfObject, CFRunLoopSource.self)
if CFRunLoopSourceIsValid(crls) {
CFRunLoopAddSource(cfRunLoop, crls, mode.cfString)
rls.info.runLoops.append(self)
CFRunLoopSourceSignal(crls)
if retainLoop { rls.info.runLoops.append(self) }
rls.signal()
CFRunLoopWakeUp(cfRunLoop)
}
}

func addDelay(rld: RunLoopDelay, mode: NSString) {
let crld = unsafeBitCast(rld.cfObject, CFRunLoopTimer.self)
if CFRunLoopTimerIsValid(crld) {
if CFRunLoopTimerIsValid(crld) && (rld.info.runLoops.count == 0 || rld.info.runLoops[0] === self) {
CFRunLoopAddTimer(cfRunLoop, crld, mode.cfString)
rld.info.runLoops.append(self)
CFRunLoopWakeUp(cfRunLoop)
}
}

func addTask(task: SafeTask) {
taskQueue.enqueue(TaskQueueElement(task, runLoopSource: taskQueueSource))
taskQueueSource.signal()
CFRunLoopWakeUp(cfRunLoop)
}
}
//#endif