async_task_priority.swift

// RUN: %empty-directory(%t)

// RUN: %target-build-swift %s -Xfrontend -disable-availability-checking -parse-as-library -o %t/async_task_priority
// RUN: %target-codesign %t/async_task_priority
// RUN: %target-run %t/async_task_priority

// REQUIRES: VENDOR=apple
// REQUIRES: executable_test
// REQUIRES: concurrency
// REQUIRES: libdispatch

// rdar://76038845
// REQUIRES: concurrency_runtime
// UNSUPPORTED: back_deployment_runtime
// UNSUPPORTED: back_deploy_concurrency

// rdar://101077408 – Temporarily disable on watchOS & iOS simulator
// UNSUPPORTED: DARWIN_SIMULATOR=watchos
// UNSUPPORTED: DARWIN_SIMULATOR=ios
// UNSUPPORTED: DARWIN_SIMULATOR=tvos

// rdar://107390341 - Temporarily disable for arm64e
// UNSUPPORTED: CPU=arm64e

import Darwin
@preconcurrencyimport Dispatch
import StdlibUnittest

func loopUntil(priority:TaskPriority)async{
varcurrentPriority=Task.currentPriority
while(currentPriority != priority){
print("Current priority = \(currentPriority) != \(priority)")
awaitTask.sleep(1_000_000)
 currentPriority =Task.currentPriority
}
}

func print(_ s:String=""){
fputs("\(s)\n", stderr)
}

func expectedBasePri(priority:TaskPriority)->TaskPriority{
letbasePri=Task.basePriority!
print("Testing basePri matching expected pri - \(basePri) == \(priority)")
expectEqual(basePri, priority)
withUnsafeCurrentTask{ unsafeTask in
guardlet unsafeTask else{
fatalError("Expected to be able to get current task, but could not!")
}
 // The UnsafeCurrentTask must return the same value
expectEqual(basePri, unsafeTask.basePriority)
}

return basePri
}

func expectedEscalatedPri(priority:TaskPriority)->TaskPriority{
letcurPri=Task.currentPriority
print("Testing escalated matching expected pri - \(curPri) == \(priority)")
expectEqual(curPri, priority)

return curPri
}

func testNestedTaskPriority(basePri:TaskPriority, curPri:TaskPriority)async{
let _ =expectedBasePri(priority: basePri)
let _ =expectedEscalatedPri(priority: curPri)
}

func childTaskWaitingForEscalation(sem:DispatchSemaphore, basePri:TaskPriority, curPri :TaskPriority)async{
 sem.wait() /* Wait to be escalated */
let _ =awaittestNestedTaskPriority(basePri: basePri, curPri: curPri)
}

actorTest{
privatevarvalue=0
init(){}

func increment()->Int{
letcur= value
 value = value +1
return cur
}

func blockActorThenIncrement(semToSignal:DispatchSemaphore, semToWait :DispatchSemaphore, priExpected:TaskPriority)->Int{
 semToSignal.signal()

 semToWait.wait();

sleep(1)
 // TODO: insert a test to verify that thread priority has actually escalated
 // to match priExpected
returnincrement()
}

}


@mainstructMain{
staticfunc main()async{

lettop_level=detach{ /* To detach from main actor when running work */

lettests=TestSuite("Task Priority manipulations")
if #available(SwiftStdlib 5.1,*){

 tests.test("Basic escalation test when task is running"){
letparentPri=Task.currentPriority

letsem=DispatchSemaphore(value:0)
lettask=Task(priority:.background){
let _ =expectedBasePri(priority:.background)

 // Wait until task is running before asking to be escalated
 sem.signal()
sleep(1)

let _ =expectedEscalatedPri(priority: parentPri)
}

 // Wait till child runs and asks to be escalated
 sem.wait()
await task.value
}

 tests.test("Basic escalation when task is suspended"){
letparentPri=Task.currentPriority

lettask=Task(priority:.background){
awaitloopUntil(priority: parentPri) /* Suspend task until it is escalated */

let _ =expectedEscalatedPri(priority: parentPri)
}
await task.value // Escalate task BG -> DEF
}

 tests.test("Structured concurrency priority propagation"){
lettask=Task(priority:.background){
awaitloopUntil(priority:.default)

letbasePri=expectedBasePri(priority:.background)
letcurPri=expectedEscalatedPri(priority:.default)

 // Structured concurrency via async let, escalated priority of
 // parent should propagate
print("Testing propagation for async let structured concurrency child")
asyncletchild=testNestedTaskPriority(basePri: basePri, curPri: curPri)
await child

letdispatchGroup=DispatchGroup()
 // Structured concurrency via task groups, escalated priority should
 // propagate
awaitwithTaskGroup(of:Void.self, returning:Void.self){ group in
 dispatchGroup.enter()
 group.addTask{
print("Testing propagation for task group regular child")
let _ =awaittestNestedTaskPriority(basePri: basePri, curPri: curPri)
 dispatchGroup.leave()
return
}

 dispatchGroup.enter()
 group.addTask(priority:.utility){
print("Testing propagation for task group child with specified priority")
let _ =awaittestNestedTaskPriority(basePri:.utility, curPri: curPri)
 dispatchGroup.leave()
return
}

 // Wait for child tasks to finish running, don't await since that
 // will escalate them
 dispatchGroup.wait()
}
}

await task.value // Escalate task BG->DEF
}

 tests.test("Unstructured tasks priority propagation"){
lettask=Task(priority :.background){
awaitloopUntil(priority:.default)

letbasePri=expectedBasePri(priority:.background)
let _ =expectedEscalatedPri(priority:.default)

letgroup=DispatchGroup()

 // Create an unstructured task
 group.enter()
let _ =Task{
let _ =awaittestNestedTaskPriority(basePri: basePri, curPri: basePri)
 group.leave()
return
}

 // Wait for unstructured task to finish running, don't await it
 // since that will escalate
 group.wait()
}

await task.value // Escalate task BG->DEF
}

 tests.test("Task escalation propagation to SC children"){
 // Create a task tree and then escalate the parent
letparentPri=Task.currentPriority
letbasePri:TaskPriority=.background

letsem=DispatchSemaphore(value:0)
letsem2=DispatchSemaphore(value :0)

lettask=Task(priority: basePri){

asyncletchild=childTaskWaitingForEscalation(sem: sem2, basePri: basePri, curPri: parentPri)

awaitwithTaskGroup(of:Void.self, returning:Void.self){ group in
 group.addTask{
let _ =awaitchildTaskWaitingForEscalation(sem: sem2, basePri: basePri, curPri: parentPri)
}
 group.addTask(priority:.utility){
let _ =awaitchildTaskWaitingForEscalation(sem: sem2, basePri:.utility, curPri: parentPri)
}

 sem.signal() // Ask for escalation after creating full task tree
sleep(1)

let _ =expectedBasePri(priority: basePri)
let _ =expectedEscalatedPri(priority: parentPri)

 sem2.signal() // Ask async let child to evaluate
 sem2.signal() // Ask task group child 1 to evaluate
 sem2.signal() // Ask task group child 2 to evaluate
}
}

 // Wait until children are created and then ask for escalation of
 // top level
 sem.wait()
await task.value
}

 tests.test("Simple task escalation to a future"){
lettask1Pri:TaskPriority=.background
lettask2Pri:TaskPriority=.utility
letparentPri:TaskPriority=Task.currentPriority
print("Top level task current priority = \(parentPri)")

 // After task2 has suspended waiting for task1, escalating task2
 // should cause task1 to escalate

lettask1=Task(priority: task1Pri){
 // Wait until task2 has blocked on task1 and escalated it
sleep(1)
expectedEscalatedPri(priority: task2Pri)

 // Wait until task2 itself has been escalated
sleep(5)
expectedEscalatedPri(priority: parentPri)
}

lettask2=Task(priority: task2Pri){
await task1.value
}

 // Wait for task2 and task1 to run and for task2 to now block on
 // task1
sleep(3)

await task2.value
}

 tests.test("Simple task escalation to a future 2"){
 // top level task -> unstructured task2 -> child task -> unstructured
 // task1
lettask1Pri:TaskPriority=.background
lettask2Pri:TaskPriority=.utility
letparentPri:TaskPriority=Task.currentPriority
print("Top level task current priority = \(parentPri)")

lettask1=Task(priority: task1Pri){
awaitloopUntil(priority: parentPri)
}

sleep(1) // Wait for task1 to start running

lettask2=Task(priority: task2Pri){
func childTask()async{
await task1.value
}
asyncletchild=childTask()
}

sleep(1) // Wait for task2 to start running
await task2.value
}

 tests.test("Task escalation of a task enqueued on an actor"){
lettask1Pri:TaskPriority=.background
lettask2Pri:TaskPriority=.background
letparentPri:TaskPriority=Task.currentPriority

letsem1=DispatchSemaphore(value:0) // to unblock enqueue of task2
letsem2=DispatchSemaphore(value:0)
lettestActor=Test()

lettask1=Task(priority: task1Pri){
expectedBasePri(priority: task1Pri);
await testActor.blockActorThenIncrement(semToSignal: sem1, semToWait: sem2, priExpected: parentPri);
}

 sem1.wait() // Wait until task1 is on the actor

lettask2=Task(priority: task2Pri){
expectedBasePri(priority: task2Pri);
await testActor.increment()
}

sleep(1)
 sem2.signal() // task2 is probably enqueued on the actor at this point, unblock task1

await task2.value // Escalate task2 which should be queued behind task1 on the actor
}

}
awaitrunAllTestsAsync()
}

await top_level.value
}
}