[Concurrency] Decode actor/task flags in signposts, make task_wait an interval, signpost continuations.

Decode all fields from the various flags values, pass each field as a separate argument to the various signposts. We were just passing the raw value of the flags and requiring the signpost client to decode them, which was ugly and required the client to know details they shouldn't need to know.

Strip ptrauth bits from the task resume function when signposting, when ptrauth is supported.

Add signpost events for continuation init/await/resume events.

We also make task_wait into an interval, rather than a single event. The interval ends when the task resumes. As part of this change, we also skip emitting the interval when the wait completed immediately and the task didn't have to suspend.

While we're in there, clean up a few SWIFT_TASK_DEBUG_LOG lines that emitted warnings when built with the logging enabled.

rdar://88658803

Author: mikeash

stdlib/public/Concurrency/Actor.cpp

@@ -817,6 +817,30 @@ class alignas(sizeof(void *) * 2) ActiveActorStatus {
     return offsetof(ActiveActorStatus, DrainLock);
   }
 #endif
+
+  void traceStateChanged(HeapObject *actor) {
+    // Convert our state to a consistent raw value. These values currently match
+    // the enum values, but this explicit conversion provides room for change.
+    uint8_t traceState = 255;
+    switch (getActorState()) {
+    case ActiveActorStatus::Idle:
+      traceState = 0;
+      break;
+    case ActiveActorStatus::Scheduled:
+      traceState = 1;
+      break;
+    case ActiveActorStatus::Running:
+      traceState = 2;
+      break;
+    case ActiveActorStatus::Zombie_ReadyForDeallocation:
+      traceState = 3;
+      break;
+    }
+    concurrency::trace::actor_state_changed(
+        actor, getFirstJob().getRawJob(), getFirstJob().needsPreprocessing(),
+        traceState, isDistributedRemote(), isMaxPriorityEscalated(),
+        static_cast<uint8_t>(getMaxPriority()));
+  }
 };
 
 #if SWIFT_CONCURRENCY_ENABLE_PRIORITY_ESCALATION && SWIFT_POINTER_IS_4_BYTES
@@ -943,7 +967,10 @@ class DefaultActorImpl : public HeapObject {
 
   // Only for static assert use below, not for actual use otherwise
   static constexpr size_t offsetOfActiveActorStatus() {
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Winvalid-offsetof"
     return offsetof(DefaultActorImpl, StatusStorage);
+#pragma clang diagnostic pop
   }
 
 private:
@@ -1124,11 +1151,10 @@ static void traceJobQueue(DefaultActorImpl *actor, Job *first) {
 static SWIFT_ATTRIBUTE_ALWAYS_INLINE void traceActorStateTransition(DefaultActorImpl *actor,
     ActiveActorStatus oldState, ActiveActorStatus newState) {
 
-  SWIFT_TASK_DEBUG_LOG("Actor %p transitioned from %zx to %zx (%s)\n", actor,
-    oldState.getOpaqueFlags(), newState.getOpaqueFlags(), __FUNCTION__);
-  concurrency::trace::actor_state_changed(actor,
-        newState.getFirstJob().getRawJob(), newState.getFirstJob().needsPreprocessing(),
-        newState.getOpaqueFlags());
+  SWIFT_TASK_DEBUG_LOG("Actor %p transitioned from %#x to %#x (%s)\n", actor,
+                       oldState.getOpaqueFlags(), newState.getOpaqueFlags(),
+                       __FUNCTION__);
+  newState.traceStateChanged(actor);
 }
 
 void DefaultActorImpl::destroy() {
@@ -1192,7 +1218,9 @@ void DefaultActorImpl::scheduleActorProcessJob(JobPriority priority, bool useInl
     swift_retain(this);
     job = new ProcessOutOfLineJob(this, priority);
   }
-  SWIFT_TASK_DEBUG_LOG("Scheduling processing job %p for actor %p at priority %#x", job, this, priority);
+  SWIFT_TASK_DEBUG_LOG(
+      "Scheduling processing job %p for actor %p at priority %#zx", job, this,
+      priority);
   swift_task_enqueueGlobal(job);
 }
 
@@ -1259,7 +1287,9 @@ void DefaultActorImpl::enqueue(Job *job, JobPriority priority) {
   // We can do relaxed loads here, we are just using the current head in the
   // atomic state and linking that into the new job we are inserting, we don't
   // need acquires
-  SWIFT_TASK_DEBUG_LOG("Enqueueing job %p onto actor %p at priority %#x", job, this, priority);
+  SWIFT_TASK_DEBUG_LOG("Enqueueing job %p onto actor %p at priority %#zx", job,
+                       this, priority);
+  concurrency::trace::actor_enqueue(this, job);
   auto oldState = _status().load(std::memory_order_relaxed);
   while (true) {
     auto newState = oldState;
@@ -1422,6 +1452,7 @@ Job * DefaultActorImpl::drainOne() {
                             /* failure */ std::memory_order_acquire)) {
       SWIFT_TASK_DEBUG_LOG("Drained first job %p from actor %p", firstJob, this);
       traceActorStateTransition(this, oldState, newState);
+      concurrency::trace::actor_dequeue(this, firstJob);
       return firstJob;
     }
 
@@ -1650,7 +1681,8 @@ static bool canGiveUpThreadForSwitch(ExecutorTrackingInfo *trackingInfo,
 /// do that in runOnAssumedThread.
 static void giveUpThreadForSwitch(ExecutorRef currentExecutor) {
   if (currentExecutor.isGeneric()) {
-    SWIFT_TASK_DEBUG_LOG("Giving up current generic executor %p", currentExecutor);
+    SWIFT_TASK_DEBUG_LOG("Giving up current generic executor %p",
+                         currentExecutor.getIdentity());
     return;
   }
 

stdlib/public/Concurrency/Task.cpp

@@ -107,8 +107,6 @@ FutureFragment::Status AsyncTask::waitFuture(AsyncTask *waitingTask,
   auto queueHead = fragment->waitQueue.load(std::memory_order_acquire);
   bool contextIntialized = false;
   while (true) {
-    concurrency::trace::task_wait(
-        waitingTask, this, static_cast<uintptr_t>(queueHead.getStatus()));
     switch (queueHead.getStatus()) {
     case Status::Error:
     case Status::Success:
@@ -123,6 +121,8 @@ FutureFragment::Status AsyncTask::waitFuture(AsyncTask *waitingTask,
       SWIFT_TASK_DEBUG_LOG("task %p waiting on task %p, going to sleep",
                            waitingTask, this);
       _swift_tsan_release(static_cast<Job *>(waitingTask));
+      concurrency::trace::task_wait(
+          waitingTask, this, static_cast<uintptr_t>(queueHead.getStatus()));
       // Task is not complete. We'll need to add ourselves to the queue.
       break;
     }
@@ -233,6 +233,8 @@ void AsyncTask::completeFuture(AsyncContext *context) {
 
     _swift_tsan_acquire(static_cast<Job *>(waitingTask));
 
+    concurrency::trace::task_resume(waitingTask);
+
     // Enqueue the waiter on the global executor.
     // TODO: allow waiters to fill in a suggested executor
     waitingTask->flagAsAndEnqueueOnExecutor(ExecutorRef::generic());
@@ -470,23 +472,27 @@ const void
         reinterpret_cast<void *>(task_future_wait_resume_adapter);
 
 const void *AsyncTask::getResumeFunctionForLogging() {
+  const void *result = reinterpret_cast<const void *>(ResumeTask);
+
   if (ResumeTask == non_future_adapter) {
     auto asyncContextPrefix = reinterpret_cast<AsyncContextPrefix *>(
         reinterpret_cast<char *>(ResumeContext) - sizeof(AsyncContextPrefix));
-    return reinterpret_cast<const void *>(asyncContextPrefix->asyncEntryPoint);
+    result =
+        reinterpret_cast<const void *>(asyncContextPrefix->asyncEntryPoint);
   } else if (ResumeTask == future_adapter) {
     auto asyncContextPrefix = reinterpret_cast<FutureAsyncContextPrefix *>(
         reinterpret_cast<char *>(ResumeContext) -
         sizeof(FutureAsyncContextPrefix));
-    return reinterpret_cast<const void *>(asyncContextPrefix->asyncEntryPoint);
+    result =
+        reinterpret_cast<const void *>(asyncContextPrefix->asyncEntryPoint);
   } else if (ResumeTask == task_wait_throwing_resume_adapter) {
     auto context = static_cast<TaskFutureWaitAsyncContext *>(ResumeContext);
-    return reinterpret_cast<const void *>(context->ResumeParent);
+    result = reinterpret_cast<const void *>(context->ResumeParent);
   } else if (ResumeTask == task_future_wait_resume_adapter) {
-    return reinterpret_cast<const void *>(ResumeContext->ResumeParent);
+    result = reinterpret_cast<const void *>(ResumeContext->ResumeParent);
   }
 
-  return reinterpret_cast<const void *>(ResumeTask);
+  return __ptrauth_swift_runtime_function_entry_strip(result);
 }
 
 JobPriority swift::swift_task_currentPriority(AsyncTask *task)
@@ -653,7 +659,7 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
      basePriority = JobPriority::Default;
   }
 
-  SWIFT_TASK_DEBUG_LOG("Task's base priority = %#x", basePriority);
+  SWIFT_TASK_DEBUG_LOG("Task's base priority = %#zx", basePriority);
 
   // Figure out the size of the header.
   size_t headerSize = sizeof(AsyncTask);
@@ -788,7 +794,9 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
     futureAsyncContextPrefix->indirectResult = futureFragment->getStoragePtr();
   }
 
-  SWIFT_TASK_DEBUG_LOG("creating task %p with parent %p at base pri %zu", task, parent, basePriority);
+  SWIFT_TASK_DEBUG_LOG("creating task %p ID %" PRIu64
+                       " with parent %p at base pri %zu",
+                       task, task->getTaskId(), parent, basePriority);
 
   // Initialize the task-local allocator.
   initialContext->ResumeParent = reinterpret_cast<TaskContinuationFunction *>(
@@ -1059,6 +1067,8 @@ static AsyncTask *swift_continuation_initImpl(ContinuationAsyncContext *context,
   task->ResumeContext = context;
   task->ResumeTask = context->ResumeParent;
 
+  concurrency::trace::task_continuation_init(task, context);
+
   return task;
 }
 
@@ -1071,6 +1081,8 @@ static void swift_continuation_awaitImpl(ContinuationAsyncContext *context) {
   assert(task->ResumeTask == context->ResumeParent);
 #endif
 
+  concurrency::trace::task_continuation_await(context);
+
   auto &sync = context->AwaitSynchronization;
 
   auto oldStatus = sync.load(std::memory_order_acquire);
@@ -1157,12 +1169,14 @@ static void resumeTaskAfterContinuation(AsyncTask *task,
 SWIFT_CC(swift)
 static void swift_continuation_resumeImpl(AsyncTask *task) {
   auto context = static_cast<ContinuationAsyncContext*>(task->ResumeContext);
+  concurrency::trace::task_continuation_resume(context, false);
   resumeTaskAfterContinuation(task, context);
 }
 
 SWIFT_CC(swift)
 static void swift_continuation_throwingResumeImpl(AsyncTask *task) {
   auto context = static_cast<ContinuationAsyncContext*>(task->ResumeContext);
+  concurrency::trace::task_continuation_resume(context, false);
   resumeTaskAfterContinuation(task, context);
 }
 
@@ -1171,6 +1185,7 @@ SWIFT_CC(swift)
 static void swift_continuation_throwingResumeWithErrorImpl(AsyncTask *task,
                                                 /* +1 */ SwiftError *error) {
   auto context = static_cast<ContinuationAsyncContext*>(task->ResumeContext);
+  concurrency::trace::task_continuation_resume(context, true);
   context->ErrorResult = error;
   resumeTaskAfterContinuation(task, context);
 }

stdlib/public/Concurrency/TaskPrivate.h

@@ -507,7 +507,9 @@ class alignas(2 * sizeof(void*)) ActiveTaskStatus {
   }
 
   void traceStatusChanged(AsyncTask *task) {
-    concurrency::trace::task_status_changed(task, Flags);
+    concurrency::trace::task_status_changed(
+        task, static_cast<uint8_t>(getStoredPriority()), isCancelled(),
+        isStoredPriorityEscalated(), isRunning(), isEnqueued());
   }
 };
 
@@ -771,7 +773,10 @@ inline void AsyncTask::flagAsAndEnqueueOnExecutor(ExecutorRef newExecutor) {
 
   // Set up task for enqueue to next location by setting the Job priority field
   Flags.setPriority(newStatus.getStoredPriority());
-  concurrency::trace::task_flags_changed(this, Flags.getOpaqueValue());
+  concurrency::trace::task_flags_changed(
+      this, static_cast<uint8_t>(Flags.getPriority()), Flags.task_isChildTask(),
+      Flags.task_isFuture(), Flags.task_isGroupChildTask(),
+      Flags.task_isAsyncLetTask());
 
   swift_task_enqueue(this, newExecutor);
 }

stdlib/public/Concurrency/Tracing.h

@@ -22,6 +22,7 @@
 namespace swift {
 class AsyncLet;
 class AsyncTask;
+class ContinuationAsyncContext;
 class ExecutorRef;
 struct HeapObject;
 class Job;
@@ -43,10 +44,13 @@ void actor_enqueue(HeapObject *actor, Job *job);
 
 void actor_dequeue(HeapObject *actor, Job *job);
 
-// The `flags` parameter is the raw values of the actor's
-// DefaultActorImpl::State::Flags.
+// State values are:
+// Idle = 0, Scheduled = 1, Running = 2, Zombie_ReadyForDeallocation = 3,
+// invalid/unknown = 255
 void actor_state_changed(HeapObject *actor, Job *firstJob,
-                         bool needsPreprocessing, uintptr_t flags);
+                         bool needsPreprocessing, uint8_t state,
+                         bool isDistributedRemote, bool isPriorityEscalated,
+                         uint8_t maxPriority);
 
 void actor_note_job_queue(HeapObject *actor, Job *first,
                           Job *(*getNext)(Job *));
@@ -58,17 +62,28 @@ void task_create(AsyncTask *task, AsyncTask *parent, TaskGroup *group,
 
 void task_destroy(AsyncTask *task);
 
-// The `flags` parameter is the raw value of the ActiveTaskStatus::Flags field
-// in the task.
-void task_status_changed(AsyncTask *task, uintptr_t flags);
+void task_status_changed(AsyncTask *task, uint8_t maxPriority, bool isCancelled,
+                         bool isEscalated, bool isRunning, bool isEnqueued);
 
-// The `flags` parameter is the raw value of Job::Flags.
-void task_flags_changed(AsyncTask *task, uint32_t flags);
+void task_flags_changed(AsyncTask *task, uint8_t jobPriority, bool isChildTask,
+                        bool isFuture, bool isGroupChildTask,
+                        bool isAsyncLetTask);
 
 // The `status` parameter is the value of the corresponding
 // FutureFragment::Status.
 void task_wait(AsyncTask *task, AsyncTask *waitingOn, uintptr_t status);
 
+void task_resume(AsyncTask *task);
+
+// The context parameter is the context pointer used to create the continuation.
+// This same pointer will be passed to the corresponding call to
+// task_continuation_await and task_continuation_resume.
+void task_continuation_init(AsyncTask *task, ContinuationAsyncContext *context);
+
+void task_continuation_await(ContinuationAsyncContext *context);
+
+void task_continuation_resume(ContinuationAsyncContext *context, bool error);
+
 void job_enqueue_global(Job *job);
 
 void job_enqueue_global_with_delay(unsigned long long delay, Job *job);