class ThreadPool : noncopyable
{
public:
typedef std::function<void()> Task;
explicit ThreadPool(const std::string &nameArg = std::string("ThreadPool"));
  ~ThreadPool();


void set_max_queue_size(int maxSize) { max_queue_size_ = maxSize; }


int start(int num_threads);
void stop();
int run(Task f);


const std::string &name() const { return name_; }


size_t queue_size() const;


private:
    ...
};

allocator_.set_tenant_id(MTL_ID());

// IRunWrapper 用于创建多租户线程时指定租户上下文
// cgroup_ctrl 和IRunWrapper配合使用，实现多租户线程的CPU隔离
void set_run_wrapper(IRunWrapper *run_wrapper, ThreadCGroup cgroup = ThreadCGroup::FRONT_CGROUP)
  {
    run_wrapper_ = run_wrapper;
    cgroup_ = cgroup;
  }

class OxcoThreadPool : public share::ObThreadPool
{
static const int64_t QUEUE_WAIT_TIME = 100 * 1000;


public:
  OxcoThreadPool();
virtual ~OxcoThreadPool();


int init(const int64_t thread_num, const int64_t task_num_limit, const char *name = "unknow");  初始化
void destroy();  析构
int push(void *task);  添加任务
int64_t get_queue_num() const { return queue_.size(); }


private:
void handle(void *task);  处理任务
void handle_drop(void *task) { handle(task); }


protected:
void run1() override;  启动线程池


private:
const char *name_;
bool is_inited_;
  common::ObLightyQueue queue_;
int64_t total_thread_num_;
int64_t active_thread_num_;
};

class ObLightyQueue
{
public:
  ...
int push(void* p);
int pop(void*& p, int64_t timeout = 0);


private:
  ...
void** data_;


};

int OxcoThreadPool::init(const int64_t thread_num, const int64_t task_num_limit, const char *name){  int ret = OB_SUCCESS;  ...  is_inited_ = true;  lib::ThreadPool::set_run_wrapper(MTL_CTX()); 设置租户上下文  ...  return ret;}

template <typename T, typename Compare>
class ObExternalSort
{
public:
typedef ObMemorySortRound<T, Compare> MemorySortRound;
typedef ObExternalSortRound<T, Compare> ExternalSortRound;
  ObExternalSort();
virtual ~ObExternalSort();
int init(const int64_t mem_limit, const int64_t file_buf_size, const int64_t expire_timestamp,
const uint64_t tenant_id, Compare *compare);
int get_next_item(const T *&item);
void clean_up();
int add_fragment_iter(ObFragmentIterator<T> *iter);
int transfer_final_sorted_fragment_iter(ObExternalSort<T, Compare> &merge_sorter);
int get_current_round(ExternalSortRound *&round);
  TO_STRING_KV(K(is_inited_), K(file_buf_size_), K(buf_mem_limit_), K(expire_timestamp_), K(merge_count_per_round_),
      KP(tenant_id_), KP(compare_));


private:
static const int64_t EXTERNAL_SORT_ROUND_CNT = 2;
bool is_inited_;
int64_t file_buf_size_;
int64_t buf_mem_limit_;
int64_t expire_timestamp_;
int64_t merge_count_per_round_;
  Compare *compare_;
  MemorySortRound memory_sort_round_;
  ExternalSortRound sort_rounds_[EXTERNAL_SORT_ROUND_CNT];
  ExternalSortRound *curr_round_;
  ExternalSortRound *next_round_;
bool is_empty_;
uint64_t tenant_id_;
};

ObSSTableIndexBuilder sstable_index_builder;
ObDataStoreDesc data_store_desc;
data_store_desc.sstable_index_builder_ = &sstable_index_builder;
ObMacroBlockWriter macro_block_writer;
ObMacroDataSeq data_seq;
// 设置顺序
data_seq.set_parallel_degree(parallel_idx);
macro_block_writer.open(data_store_desc, data_seq);

void Thread::wait(){if (pth_ != 0) {pthread_join(pth_, nullptr);destroy_stack();pth_ = 0;pid_ = 0;tid_ = 0;runnable_ = nullptr;}}

class ObMacroBlockWriter {
public:
// 根据data_store_desc中的table_id、partition_id等信息，打开一个宏块写入器
int open(ObDataStoreDesc &data_store_desc, const ObMacroDataSeq &start_seq,
      ObIMacroBlockFlushCallback *callback = nullptr);


// 追加一个宏块，主要会应用在这些场景：
// 1. 在合并时，原来的SSTable的某个宏块没有修改，直接复用到当前SSTable中；
// 2. 并行合并后，也可以用到这个接口，将多个没有重合数据的宏块进行追加。
int append_macro_block(const ObMacroBlockDesc &macro_desc);


// 追加一个微块，和append_macro_block不同的是需要考虑是否存在数据重叠：
// 1. 如果数据不重叠，则将micro_block追加到当前宏块中；
// 2. 如果数据重叠，则需要构建micro_block的reader，将数据按row写到当前宏块中。
int append_micro_block(const ObMicroBlock& micro_block);


// 追加一行数据，会调用ObMicroBlockWriter::append_row
// 主要逻辑为：
//   1. 将行数据写入到微块中，并更新 bloomfilter 和 checksum
//   2. 在当前微块写满的情况下：构建当前的微块，并切换微块写入器
int append_row(const ObDatumRow &row);


// 关闭ObMacroBlockWriter，在关闭之前，会尝试将最后两个宏块合并，节省空间，
// 最后将当前最后的宏块flush到磁盘，并等待刷盘成功（wait_io_finish）
int close();
};

union
  {
int64_t macro_data_seq_;
struct
    {
uint64_t data_seq_ : BIT_DATA_SEQ; // 32bit
uint64_t parallel_idx_ : BIT_PARALLEL_IDX; // 11bit
uint64_t block_type_ : BIT_BLOCK_TYPE; // 3bit
uint64_t merge_type_ : BIT_MERGE_TYPE; // 2bit
uint64_t reserved_ : BIT_RESERVED; // 15bit
uint64_t sign_ : BIT_SIGN; // 1bit
    };
  };

data_seq.set_parallel_degree(parallel_writer_id);
macro_block_writer->open(data_store_desc_, data_seq);