[fix](scheduler) Fix invalid access after freed (apache#48168)

WRITE of size 1 at 0x6160007e86f0 thread T1983 (Pipe_normal [wo)
#0 0x55fc8065b975 in std::__atomic_base<bool>::store(bool,
std::memory_order)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/atomic_base.h:457:2
#1 0x55fc8065b975 in std::__atomic_base<bool>::operator=(bool)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/atomic_base.h:349:2
#2 0x55fc8065b975 in std::atomic<bool>::operator=(bool)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/atomic:80:22
#3 0x55fc8065b975 in doris::pipeline::PipelineTask::set_running(bool)
/root/doris/be/src/pipeline/pipeline_task.h:192:47
#4 0x55fc8065b975 in
doris::pipeline::TaskScheduler::_do_work(int)::$_0::operator()() const
/root/doris/be/src/pipeline/task_scheduler.cpp:121:23
#5 0x55fc8065b975 in
doris::Defer<doris::pipeline::TaskScheduler::_do_work(int)::$_0>::~Defer()
/root/doris/be/src/util/defer_op.h:37:16
#6 0x55fc8065b975 in doris::pipeline::TaskScheduler::_do_work(int)
/root/doris/be/src/pipeline/task_scheduler.cpp:162:5
#7 0x55fc4c57cd19 in doris::ThreadPool::dispatch_thread()
/root/doris/be/src/util/threadpool.cpp:608:24
apache#8 0x55fc4c55395e 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#9 0x55fc4c55395e in doris::Thread::supervise_thread(void*)
/root/doris/be/src/util/thread.cpp:498:5
apache#10 0x7f9ee3d25608 in start_thread
/build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
apache#11 0x7f9ee3fd2132 in __clone
/build/glibc-SzIz7B/glibc-2.31/misc/../sysdeps/unix/sysv/linux/x86_64/clone.S:95

0x6160007e86f0 is located 624 bytes inside of 632-byte region
[0x6160007e8480,0x6160007e86f8)
freed by thread T1981 (Pipe_normal [wo) here:
#0 0x55fc47aa680d in operator delete(void*)
(/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be+0x3376e80d)
(BuildId: 865149e62959581e)
#1 0x55fc8059db84 in
std::default_delete<doris::pipeline::PipelineTask>::operator()(doris::pipeline::PipelineTask*)
const
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:85:2
#2 0x55fc8059db84 in std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >::~unique_ptr()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:361:4
#3 0x55fc8059db84 in void
std::destroy_at<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >
>(std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:88:15
#4 0x55fc8059db84 in void
std::_Destroy<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >
>(std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:138:7
#5 0x55fc8059db84 in void
std::_Destroy_aux<false>::__destroy<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask>
>*>(std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*,
std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:152:6
#6 0x55fc8059db84 in void
std::_Destroy<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask>
>*>(std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*,
std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:184:7
#7 0x55fc8059db84 in void
std::_Destroy<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*,
std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >
>(std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*,
std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >*,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > >&)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/alloc_traits.h:746:7
apache#8 0x55fc8059db84 in
std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >::~vector()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_vector.h:680:2
apache#9 0x55fc8052571c in void
std::destroy_at<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >
>(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:88:15
apache#10 0x55fc8052571c in void
std::_Destroy<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >
>(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:138:7
apache#11 0x55fc8052571c in void
std::_Destroy_aux<false>::__destroy<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > >
>*>(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*,
std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:152:6
apache#12 0x55fc8052571c in void
std::_Destroy<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > >
>*>(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*,
std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_construct.h:184:7
apache#13 0x55fc8052571c in void
std::_Destroy<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*,
std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >
>(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*,
std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*,
std::allocator<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > > >&)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/alloc_traits.h:746:7
apache#14 0x55fc8052571c in
std::vector<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >,
std::allocator<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > > >
>::_M_erase_at_end(std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >*)
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_vector.h:1796:6
apache#15 0x55fc8052571c in
std::vector<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > >,
std::allocator<std::vector<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> >,
std::allocator<std::unique_ptr<doris::pipeline::PipelineTask,
std::default_delete<doris::pipeline::PipelineTask> > > > > >::clear()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_vector.h:1499:9
apache#16 0x55fc8052571c in
doris::pipeline::PipelineFragmentContext::~PipelineFragmentContext()
/root/doris/be/src/pipeline/pipeline_fragment_context.cpp:142:12
apache#17 0x55fc47ad30cc in
std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:168:6
apache#18 0x55fc80658d57 in
std::__shared_count<(__gnu_cxx::_Lock_policy)2>::~__shared_count()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:702:11
apache#19 0x55fc80658d57 in std::__shared_ptr<doris::TaskExecutionContext,
(__gnu_cxx::_Lock_policy)2>::~__shared_ptr()
/var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:1149:31
apache#20 0x55fc80658d57 in
doris::pipeline::close_task(doris::pipeline::PipelineTask*,
doris::Status) /root/doris/be/src/pipeline/task_scheduler.cpp:100:1
apache#21 0x55fc8065aa17 in doris::pipeline::TaskScheduler::_do_work(int)
/root/doris/be/src/pipeline/task_scheduler.cpp:160:36
apache#22 0x55fc4c57cd19 in doris::ThreadPool::dispatch_thread()
/root/doris/be/src/util/threadpool.cpp:608:24
apache#23 0x55fc4c55395e 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#24 0x55fc4c55395e in doris::Thread::supervise_thread(void*)
/root/doris/be/src/util/thread.cpp:498:5
apache#25 0x7f9ee3d25608 in start_thread
/build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8

be/src/pipeline/pipeline_fragment_context.cpp

@@ -1753,7 +1753,7 @@ void PipelineFragmentContext::_close_fragment_instance() {
             std::dynamic_pointer_cast<PipelineFragmentContext>(shared_from_this()));
 }
 
-bool PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
+void PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
     // If all tasks of this pipeline has been closed, upstream tasks is never needed, and we just make those runnable here
     DCHECK(_pip_id_to_pipeline.contains(pipeline_id));
     if (_pip_id_to_pipeline[pipeline_id]->close_task()) {
@@ -1767,9 +1767,7 @@ bool PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
     ++_closed_tasks;
     if (_closed_tasks == _total_tasks) {
         _close_fragment_instance();
-        return true;
     }
-    return false;
 }
 
 Status PipelineFragmentContext::send_report(bool done) {

be/src/pipeline/pipeline_fragment_context.h

@@ -100,7 +100,7 @@ class PipelineFragmentContext : public TaskExecutionContext {
 
     [[nodiscard]] int get_fragment_id() const { return _fragment_id; }
 
-    bool decrement_running_task(PipelineId pipeline_id);
+    void decrement_running_task(PipelineId pipeline_id);
 
     Status send_report(bool);
 

be/src/pipeline/task_scheduler.cpp

@@ -88,15 +88,12 @@ bool close_task(PipelineTask* task, Status exec_status) {
                                     print_id(task->query_context()->query_id()),
                                     exec_status.to_string());
     }
-    // decrement_running_task may delete fragment context and will core in some defer
-    // code, because the defer code will access fragment context itself.
-    auto lock_for_context = task->fragment_context()->shared_from_this();
     Status status = task->close(exec_status);
     if (!status.ok()) {
         task->fragment_context()->cancel(status);
     }
     task->finalize();
-    return task->fragment_context()->decrement_running_task(task->pipeline_id());
+    return true;
 }
 
 void TaskScheduler::_do_work(int index) {
@@ -114,10 +111,20 @@ void TaskScheduler::_do_work(int index) {
         }
         task->log_detail_if_need();
         task->set_running(true);
-        bool fragment_is_finished = false;
+        bool eos = false;
+        auto status = Status::OK();
         Defer task_running_defer {[&]() {
             // If fragment is finished, fragment context will be de-constructed with all tasks in it.
-            if (!fragment_is_finished) {
+            if (eos || !status.ok()) {
+                // decrement_running_task may delete fragment context and will core in some defer
+                // code, because the defer code will access fragment context itself.
+                auto lock_for_context = task->fragment_context()->shared_from_this();
+                bool close = close_task(task, status);
+                task->set_running(false);
+                if (close) {
+                    task->fragment_context()->decrement_running_task(task->pipeline_id());
+                }
+            } else {
                 task->set_running(false);
             }
         }};
@@ -127,12 +134,10 @@ void TaskScheduler::_do_work(int index) {
 
         // Close task if canceled
         if (canceled) {
-            fragment_is_finished = close_task(task, fragment_ctx->get_query_ctx()->exec_status());
+            status = fragment_ctx->get_query_ctx()->exec_status();
+            DCHECK(!status.ok());
             continue;
         }
-
-        bool eos = false;
-        auto status = Status::OK();
         task->set_core_id(index);
 
         // Main logics of execution
@@ -155,10 +160,6 @@ void TaskScheduler::_do_work(int index) {
                 } else { status = task->execute(&eos); },
                 status);
         fragment_ctx->trigger_report_if_necessary();
-
-        if (eos || !status.ok()) {
-            fragment_is_finished = close_task(task, status);
-        }
     }
 }