[bugfix](core) runtime state is destroyed when spill task is run (apa…

…che#39566)

==17355==ERROR: AddressSanitizer: heap-use-after-free on address
0x61c00388c269 at pc 0x55b47f79e5c4 bp 0x7fde7dd98e30 sp 0x7fde7dd98e28
READ of size 1 at 0x61c00388c269 thread T604 (SpillIOThreadPo)
#0 0x55b47f79e5c3 in doris::QueryContext::is_nereids() const
/root/doris/be/src/runtime/query_context.h:229:38
#1 0x55b47f79e5c3 in doris::RuntimeState::is_nereids() const
/root/doris/be/src/runtime/runtime_state.cpp:542:24
apache#2 0x55b47f89afed in doris::AttachTask::AttachTask(doris::RuntimeState*)
/root/doris/be/src/runtime/thread_context.cpp:48:50
apache#3 0x55b4b039e7e5 in doris::pipeline::SpillRunnable::run()
/root/doris/be/src/pipeline/exec/spill_utils.h:43:9
apache#4 0x55b47fdf739d in doris::ThreadPool::dispatch_thread()
/root/doris/be/src/util/threadpool.cpp:543:24
apache#5 0x55b47fdd05de in std::function<void ()>::operator()() const
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_function.h:560:9
apache#6 0x55b47fdd05de in doris::Thread::supervise_thread(void*)
/root/doris/be/src/util/thread.cpp:498:5
apache#7 0x7fe1bcb5e608 in start_thread
/build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
apache#8 0x7fe1bce0b132 in __clone
/build/glibc-SzIz7B/glibc-2.31/misc/../sysdeps/unix/sysv/linux/x86_64/clone.S:95

Co-authored-by: yiguolei <[email protected]>

be/src/pipeline/exec/spill_utils.h

@@ -40,17 +40,18 @@ class SpillRunnable : public Runnable {
     ~SpillRunnable() override = default;
 
     void run() override {
+        // Should lock task context before scope task, because the _state maybe
+        // destroyed when run is called.
+        auto task_context_holder = _task_context_holder.lock();
+        if (!task_context_holder) {
+            return;
+        }
         SCOPED_ATTACH_TASK(_state);
         Defer defer([&] {
             std::function<void()> tmp;
             std::swap(tmp, _func);
         });
 
-        auto task_context_holder = _task_context_holder.lock();
-        if (!task_context_holder) {
-            return;
-        }
-
         auto shared_state_holder = _shared_state_holder.lock();
         if (!shared_state_holder) {
             return;