remove thread context tasktype

be/src/exec/tablet_sink.cpp

@@ -581,15 +581,15 @@ void NodeChannel::try_send_batch(RuntimeState* state) {
         _add_batch_closure->cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
         _add_batch_closure->cntl.http_request().set_content_type("application/json");
         {
-            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
             _brpc_http_stub->tablet_writer_add_batch_by_http(&_add_batch_closure->cntl, NULL,
                                                              &_add_batch_closure->result,
                                                              _add_batch_closure);
         }
     } else {
         _add_batch_closure->cntl.http_request().Clear();
         {
-            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
             _stub->tablet_writer_add_batch(&_add_batch_closure->cntl, &request,
                                            &_add_batch_closure->result, _add_batch_closure);
         }

be/src/exec/tablet_sink.h

@@ -96,7 +96,7 @@ class ReusableClosure final : public google::protobuf::Closure {
     ~ReusableClosure() override {
         // shouldn't delete when Run() is calling or going to be called, wait for current Run() done.
         join();
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         cntl.Reset();
     }
 
@@ -124,7 +124,7 @@ class ReusableClosure final : public google::protobuf::Closure {
 
     // plz follow this order: reset() -> set_in_flight() -> send brpc batch
     void reset() {
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         cntl.Reset();
         cid = cntl.call_id();
     }

be/src/olap/base_compaction.cpp

@@ -68,7 +68,7 @@ Status BaseCompaction::execute_compact_impl() {
         return Status::OLAPInternalError(OLAP_ERR_BE_CLONE_OCCURRED);
     }
 
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::COMPACTION);
+    SCOPED_ATTACH_TASK(_mem_tracker);
 
     // 2. do base compaction, merge rowsets
     int64_t permits = get_compaction_permits();

be/src/olap/cumulative_compaction.cpp

@@ -70,7 +70,7 @@ Status CumulativeCompaction::execute_compact_impl() {
         return Status::OLAPInternalError(OLAP_ERR_CUMULATIVE_CLONE_OCCURRED);
     }
 
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::COMPACTION);
+    SCOPED_ATTACH_TASK(_mem_tracker);
 
     // 3. do cumulative compaction, merge rowsets
     int64_t permits = get_compaction_permits();

be/src/olap/delta_writer.cpp

@@ -164,7 +164,7 @@ Status DeltaWriter::write(Tuple* tuple) {
         return Status::OLAPInternalError(OLAP_ERR_ALREADY_CANCELLED);
     }
 
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::LOAD);
+    SCOPED_ATTACH_TASK(_mem_tracker);
 
     _mem_table->insert(tuple);
 

be/src/olap/memtable.cpp

@@ -54,7 +54,7 @@ MemTable::MemTable(TabletSharedPtr tablet, Schema* schema, const TabletSchema* t
           _rowset_ids(rowset_ids),
           _cur_max_version(cur_max_version) {
     _mem_tracker_hook->enable_reset_zero();
-    SCOPED_ATTACH_TASK(_mem_tracker_hook, ThreadContext::TaskType::LOAD);
+    SCOPED_ATTACH_TASK(_mem_tracker_hook);
     _mem_tracker_manual = std::make_unique<MemTracker>(
             fmt::format("MemTableManual:tabletId={}", std::to_string(tablet_id())));
     _buffer_mem_pool = std::make_unique<MemPool>(_mem_tracker_manual.get());
@@ -176,7 +176,7 @@ int MemTable::RowInBlockComparator::operator()(const RowInBlock* left,
 }
 
 void MemTable::insert(const vectorized::Block* input_block, const std::vector<int>& row_idxs) {
-    SCOPED_ATTACH_TASK(_mem_tracker_hook, ThreadContext::TaskType::LOAD);
+    SCOPED_ATTACH_TASK(_mem_tracker_hook);
     auto target_block = input_block->copy_block(_column_offset);
     if (_is_first_insertion) {
         _is_first_insertion = false;
@@ -392,7 +392,7 @@ void MemTable::_collect_vskiplist_results() {
 }
 
 void MemTable::shrink_memtable_by_agg() {
-    SCOPED_ATTACH_TASK(_mem_tracker_hook, ThreadContext::TaskType::LOAD);
+    SCOPED_ATTACH_TASK(_mem_tracker_hook);
     if (keys_type() == KeysType::DUP_KEYS) {
         return;
     }
@@ -434,7 +434,7 @@ Status MemTable::_generate_delete_bitmap() {
 }
 
 Status MemTable::flush() {
-    SCOPED_ATTACH_TASK(_mem_tracker_hook, ThreadContext::TaskType::LOAD);
+    SCOPED_ATTACH_TASK(_mem_tracker_hook);
     VLOG_CRITICAL << "begin to flush memtable for tablet: " << tablet_id()
                   << ", memsize: " << memory_usage() << ", rows: " << _rows;
     int64_t duration_ns = 0;

be/src/olap/memtable.h

@@ -54,7 +54,7 @@ class MemTable {
     size_t memory_usage() const { return _mem_tracker_manual->consumption(); }
 
     inline void insert(const Tuple* tuple) {
-        SCOPED_ATTACH_TASK(_mem_tracker_hook, ThreadContext::TaskType::LOAD);
+        SCOPED_ATTACH_TASK(_mem_tracker_hook);
         (this->*_insert_fn)(tuple);
     }
     // insert tuple from (row_pos) to (row_pos+num_rows)

be/src/olap/olap_server.cpp

@@ -117,7 +117,7 @@ Status StorageEngine::start_bg_threads() {
             RETURN_IF_ERROR(Thread::create(
                     "StorageEngine", "path_scan_thread",
                     [this, data_dir]() {
-                        SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+                        SCOPED_ATTACH_TASK(_mem_tracker);
                         this->_path_scan_thread_callback(data_dir);
                     },
                     &path_scan_thread));
@@ -127,7 +127,7 @@ Status StorageEngine::start_bg_threads() {
             RETURN_IF_ERROR(Thread::create(
                     "StorageEngine", "path_gc_thread",
                     [this, data_dir]() {
-                        SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+                        SCOPED_ATTACH_TASK(_mem_tracker);
                         this->_path_gc_thread_callback(data_dir);
                     },
                     &path_gc_thread));

be/src/olap/storage_engine.cpp

@@ -166,7 +166,7 @@ void StorageEngine::load_data_dirs(const std::vector<DataDir*>& data_dirs) {
     std::vector<std::thread> threads;
     for (auto data_dir : data_dirs) {
         threads.emplace_back([this, data_dir] {
-            SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+            SCOPED_ATTACH_TASK(_mem_tracker);
             auto res = data_dir->load();
             if (!res.ok()) {
                 LOG(WARNING) << "io error when init load tables. res=" << res
@@ -212,7 +212,7 @@ Status StorageEngine::_init_store_map() {
                                      _tablet_manager.get(), _txn_manager.get());
         tmp_stores.emplace_back(store);
         threads.emplace_back([this, store, &error_msg_lock, &error_msg]() {
-            SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+            SCOPED_ATTACH_TASK(_mem_tracker);
             auto st = store->init();
             if (!st.ok()) {
                 {

be/src/olap/task/engine_alter_tablet_task.cpp

@@ -34,7 +34,7 @@ EngineAlterTabletTask::EngineAlterTabletTask(const TAlterTabletReqV2& request)
 }
 
 Status EngineAlterTabletTask::execute() {
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+    SCOPED_ATTACH_TASK(_mem_tracker);
     DorisMetrics::instance()->create_rollup_requests_total->increment(1);
 
     Status res = SchemaChangeHandler::process_alter_tablet_v2(_alter_tablet_req);

be/src/olap/task/engine_batch_load_task.cpp

@@ -58,7 +58,7 @@ EngineBatchLoadTask::EngineBatchLoadTask(TPushReq& push_req, std::vector<TTablet
 EngineBatchLoadTask::~EngineBatchLoadTask() {}
 
 Status EngineBatchLoadTask::execute() {
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+    SCOPED_ATTACH_TASK(_mem_tracker);
     Status status;
     if (_push_req.push_type == TPushType::LOAD || _push_req.push_type == TPushType::LOAD_V2) {
         RETURN_IF_ERROR(_init());

be/src/olap/task/engine_checksum_task.cpp

@@ -32,7 +32,7 @@ EngineChecksumTask::EngineChecksumTask(TTabletId tablet_id, TSchemaHash schema_h
 }
 
 Status EngineChecksumTask::execute() {
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+    SCOPED_ATTACH_TASK(_mem_tracker);
     return _compute_checksum();
 } // execute
 

be/src/olap/task/engine_clone_task.cpp

@@ -61,7 +61,7 @@ EngineCloneTask::EngineCloneTask(const TCloneReq& clone_req, const TMasterInfo&
 
 Status EngineCloneTask::execute() {
     // register the tablet to avoid it is deleted by gc thread during clone process
-    SCOPED_ATTACH_TASK(_mem_tracker, ThreadContext::TaskType::STORAGE);
+    SCOPED_ATTACH_TASK(_mem_tracker);
     StorageEngine::instance()->tablet_manager()->register_clone_tablet(_clone_req.tablet_id);
     Status st = _do_clone();
     StorageEngine::instance()->tablet_manager()->unregister_clone_tablet(_clone_req.tablet_id);

be/src/runtime/buffer_control_block.cpp

@@ -30,7 +30,7 @@ void GetResultBatchCtx::on_failure(const Status& status) {
     status.to_protobuf(result->mutable_status());
     {
         // call by result sink
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         done->Run();
     }
     delete this;
@@ -45,7 +45,7 @@ void GetResultBatchCtx::on_close(int64_t packet_seq, QueryStatistics* statistics
     result->set_packet_seq(packet_seq);
     result->set_eos(true);
     {
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         done->Run();
     }
     delete this;
@@ -73,7 +73,7 @@ void GetResultBatchCtx::on_data(const std::unique_ptr<TFetchDataResult>& t_resul
     }
     st.to_protobuf(result->mutable_status());
     {
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         done->Run();
     }
     delete this;

be/src/runtime/data_stream_recvr.cc

@@ -186,10 +186,7 @@ Status DataStreamRecvr::SenderQueue::get_batch(RowBatch** next_batch) {
 
     if (!_pending_closures.empty()) {
         auto closure_pair = _pending_closures.front();
-        {
-            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
-            closure_pair.first->Run();
-        }
+        closure_pair.first->Run();
         _pending_closures.pop_front();
 
         closure_pair.second.stop();
@@ -339,11 +336,8 @@ void DataStreamRecvr::SenderQueue::cancel() {
 
     {
         std::lock_guard<std::mutex> l(_lock);
-        {
-            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
-            for (auto closure_pair : _pending_closures) {
-                closure_pair.first->Run();
-            }
+        for (auto closure_pair : _pending_closures) {
+            closure_pair.first->Run();
         }
         _pending_closures.clear();
     }
@@ -357,11 +351,8 @@ void DataStreamRecvr::SenderQueue::close() {
         std::lock_guard<std::mutex> l(_lock);
         _is_cancelled = true;
 
-        {
-            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
-            for (auto closure_pair : _pending_closures) {
-                closure_pair.first->Run();
-            }
+        for (auto closure_pair : _pending_closures) {
+            closure_pair.first->Run();
         }
         _pending_closures.clear();
     }

be/src/runtime/data_stream_sender.cpp

@@ -138,7 +138,7 @@ Status DataStreamSender::Channel::send_batch(PRowBatch* batch, bool eos) {
         _closure->ref();
     } else {
         RETURN_IF_ERROR(_wait_last_brpc());
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
+        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
         _closure->cntl.Reset();
     }
     VLOG_ROW << "Channel::send_batch() instance_id=" << _fragment_instance_id
@@ -160,7 +160,6 @@ Status DataStreamSender::Channel::send_batch(PRowBatch* batch, bool eos) {
     if (_parent->_transfer_large_data_by_brpc && _brpc_request.has_row_batch() &&
         _brpc_request.row_batch().has_tuple_data() &&
         _brpc_request.ByteSizeLong() > MIN_HTTP_BRPC_SIZE) {
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
         Status st = request_embed_attachment_contain_tuple<PTransmitDataParams,
                                                            RefCountClosure<PTransmitDataResult>>(
                 &_brpc_request, _closure);
@@ -174,11 +173,17 @@ Status DataStreamSender::Channel::send_batch(PRowBatch* batch, bool eos) {
                 brpc_url + "/PInternalServiceImpl/transmit_data_by_http";
         _closure->cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
         _closure->cntl.http_request().set_content_type("application/json");
-        _brpc_http_stub->transmit_data_by_http(&_closure->cntl, NULL, &_closure->result, _closure);
+        {
+            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
+            _brpc_http_stub->transmit_data_by_http(&_closure->cntl, NULL, &_closure->result,
+                                                   _closure);
+        }
     } else {
-        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->bthread_mem_tracker());
         _closure->cntl.http_request().Clear();
-        _brpc_stub->transmit_data(&_closure->cntl, &_brpc_request, &_closure->result, _closure);
+        {
+            SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->orphan_mem_tracker());
+            _brpc_stub->transmit_data(&_closure->cntl, &_brpc_request, &_closure->result, _closure);
+        }
     }
 
     if (batch != nullptr) {

be/src/runtime/exec_env.h

@@ -120,20 +120,17 @@ class ExecEnv {
     std::shared_ptr<MemTrackerLimiter> process_mem_tracker() { return _process_mem_tracker; }
     void set_global_mem_tracker(const std::shared_ptr<MemTrackerLimiter>& process_tracker,
                                 const std::shared_ptr<MemTrackerLimiter>& orphan_tracker,
-                                const std::shared_ptr<MemTrackerLimiter>& nursery_mem_tracker,
                                 const std::shared_ptr<MemTrackerLimiter>& bthread_mem_tracker) {
         _process_mem_tracker = process_tracker;
         _orphan_mem_tracker = orphan_tracker;
         _orphan_mem_tracker_raw = orphan_tracker.get();
-        _nursery_mem_tracker = nursery_mem_tracker;
         _bthread_mem_tracker = bthread_mem_tracker;
     }
     std::shared_ptr<MemTracker> allocator_cache_mem_tracker() {
         return _allocator_cache_mem_tracker;
     }
     std::shared_ptr<MemTrackerLimiter> orphan_mem_tracker() { return _orphan_mem_tracker; }
     MemTrackerLimiter* orphan_mem_tracker_raw() { return _orphan_mem_tracker_raw; }
-    std::shared_ptr<MemTrackerLimiter> nursery_mem_tracker() { return _nursery_mem_tracker; }
     std::shared_ptr<MemTrackerLimiter> bthread_mem_tracker() { return _bthread_mem_tracker; }
     std::shared_ptr<MemTrackerLimiter> query_pool_mem_tracker() { return _query_pool_mem_tracker; }
     std::shared_ptr<MemTrackerLimiter> load_pool_mem_tracker() { return _load_pool_mem_tracker; }
@@ -229,8 +226,6 @@ class ExecEnv {
     // and the consumption of the orphan mem tracker is close to 0, but greater than 0.
     std::shared_ptr<MemTrackerLimiter> _orphan_mem_tracker;
     MemTrackerLimiter* _orphan_mem_tracker_raw;
-    // Parent is orphan, Nursery of orphan memory after manually switching thread mem tracker
-    std::shared_ptr<MemTrackerLimiter> _nursery_mem_tracker;
     // Parent is orphan, bthread default mem tracker
     std::shared_ptr<MemTrackerLimiter> _bthread_mem_tracker;
     // The ancestor for all querys tracker.

be/src/runtime/exec_env_init.cpp

@@ -202,7 +202,6 @@ Status ExecEnv::_init_mem_tracker() {
             std::make_shared<MemTrackerLimiter>(global_memory_limit_bytes, "Process");
     _orphan_mem_tracker = std::make_shared<MemTrackerLimiter>(-1, "Orphan", _process_mem_tracker);
     _orphan_mem_tracker_raw = _orphan_mem_tracker.get();
-    _nursery_mem_tracker = std::make_shared<MemTrackerLimiter>(-1, "Nursery", _orphan_mem_tracker);
     _bthread_mem_tracker = std::make_shared<MemTrackerLimiter>(-1, "Bthread", _orphan_mem_tracker);
     thread_context()->_thread_mem_tracker_mgr->init();
     thread_context()->_thread_mem_tracker_mgr->set_check_attach(false);

be/src/runtime/memory/thread_mem_tracker_mgr.cpp

@@ -29,25 +29,25 @@ void ThreadMemTrackerMgr::attach_limiter_tracker(
         const std::shared_ptr<MemTrackerLimiter>& mem_tracker) {
     DCHECK(mem_tracker);
     flush_untracked_mem<false>();
-    _task_id_stack.push_back(task_id);
-    _fragment_instance_id_stack.push_back(fragment_instance_id);
-    _limiter_tracker_stack.push_back(mem_tracker);
+    _task_id = task_id;
+    _fragment_instance_id = fragment_instance_id;
+    _limiter_tracker = mem_tracker;
     _limiter_tracker_raw = mem_tracker.get();
 }
 
-void ThreadMemTrackerMgr::detach_limiter_tracker() {
-    DCHECK(!_limiter_tracker_stack.empty());
+void ThreadMemTrackerMgr::detach_limiter_tracker(
+        const std::shared_ptr<MemTrackerLimiter>& old_mem_tracker) {
     flush_untracked_mem<false>();
-    _task_id_stack.pop_back();
-    _fragment_instance_id_stack.pop_back();
-    _limiter_tracker_stack.pop_back();
-    _limiter_tracker_raw = _limiter_tracker_stack.back().get();
+    _task_id = "";
+    _fragment_instance_id = TUniqueId();
+    _limiter_tracker = old_mem_tracker;
+    _limiter_tracker_raw = old_mem_tracker.get();
 }
 
 void ThreadMemTrackerMgr::exceeded_cancel_task(const std::string& cancel_details) {
-    if (_fragment_instance_id_stack.back() != TUniqueId()) {
+    if (_fragment_instance_id != TUniqueId()) {
         ExecEnv::GetInstance()->fragment_mgr()->cancel(
-                _fragment_instance_id_stack.back(), PPlanFragmentCancelReason::MEMORY_LIMIT_EXCEED,
+                _fragment_instance_id, PPlanFragmentCancelReason::MEMORY_LIMIT_EXCEED,
                 cancel_details);
     }
 }