图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hive|数据仓库 › 代码分析› Doris源码解析[二、异步任务之Schema Change]
分享

Doris源码解析[二、异步任务之Schema Change]

hyj 2021-2-22 07:21:57 发表于 代码分析 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 2799
About云VIP帮助找工作和提升

问题导读

1.在Doris中，哪些任务都是异步执行？
2.异步执行的主要流程包含哪些？
3.FE端用户修改表结构有哪几个状态？


Doris提交Schema Change策略
在Doris中，许多任务都是异步执行的，例如创建tablet，删除tablet，修改表结构，Broker Load等等，这一类的作业的主要流程为:

  • 在FE根据task_type可以分为多种任务，AgentTask为所有任务的父类，AgentTask及其子类封装了包括backendId, signature, taskType等任务的信息，在RPC调用时FE通过调用toAgentTaskRequest方法，将该类转化为TAgentTaskRequest然后将调用RPC方法接口submit_task将请求下发给BE执行。
  • AgentServer在初始化时，会调用类中根据task_type划分的多个std::unique_ptr<TaskWorkerPool>的实例的start方法，该方法会根据task_type为每个实例绑定不同的回调函数，并根据参数，对每种task生成指定数量的线程去执行对应的任务。BE接收任务请求参数TAgentTaskRequest，AgentServer::submit_task方法会解析TAgentTaskRequest，并根据task_type转发给不同类型的TaskWorkPool实例，该实例会调用TaskWorkPool::submit_task方法，将TAgentTaskRequest提交到std::deque<TAgentTaskRequest>队列中等待调度，而TaskWorkPool类中有多种回调函数，当一个任务提交完成后，同时会调用_worker_thread_condition_lock.notify()方法唤醒一个消费者线程去执行任务。
  • 当线程对应回调方法的函数中的条件变量唤醒时（即worker_pool_this->_worker_thread_condition_lock.wait();），该线程会开始执行任务，并在执行完成后在方法TaskWorkerPool::_finish_task中进行RPC调用，向FE汇报任务完成的结果TFinishTaskRequest，最后调用TaskWorkerPool::_remove_task_info方法将该任务从队列中移除，完成整个Schema Change的任务逻辑



FE端逻辑
以下从源码层面进一步剖析整个作业的执行逻辑，首先从FE入手，在用户修改表结构时，任务有4个状态，即pending，waitingTxn, running及finished，在进行类型转换时，Doris首先会在pending阶段，下发给BE创建replicas的任务，具体的代码如下:


  1. protected void runPendingJob() throws AlterCancelException {

  2.     Preconditions.checkState(jobState == JobState.PENDING, jobState);

  3.     LOG.info("begin to send create replica tasks. job: {}", jobId);

  4.     Database db = Catalog.getCurrentCatalog().getDb(dbId);

  5.     if (db == null) {

  6.         throw new AlterCancelException("Databasee " + dbId + " does not exist");

  7.     }

  8. 

  9.     // 1. create replicas

  10.     AgentBatchTask batchTask = new AgentBatchTask();

  11.     // count total replica num

  12.     int totalReplicaNum = 0;

  13.     for (MaterializedIndex shadowIdx : partitionIndexMap.values()) {

  14.         for (Tablet tablet : shadowIdx.getTablets()) {

  15.             totalReplicaNum += tablet.getReplicas().size();

  16.         }

  17.     }

  18.     MarkedCountDownLatch<Long, Long> countDownLatch = new MarkedCountDownLatch<>(totalReplicaNum);

  19.     db.readLock();

  20.     try {

  21.         OlapTable tbl = (OlapTable) db.getTable(tableId);

  22.         if (tbl == null) {

  23.             throw new AlterCancelException("Table " + tableId + " does not exist");

  24.         }

  25. 

  26.         boolean isStable = tbl.isStable(Catalog.getCurrentSystemInfo(),

  27.                 Catalog.getCurrentCatalog().getTabletScheduler(),

  28.                 db.getClusterName());

  29.         if (!isStable) {

  30.             errMsg = "table is unstable";

  31.             LOG.warn("doing schema change job: " + jobId + " while table is not stable.");

  32.             return;

  33.         }

  34. 

  35.         Preconditions.checkState(tbl.getState() == OlapTableState.SCHEMA_CHANGE);

  36.         for (long partitionId : partitionIndexMap.rowKeySet()) {

  37.             Partition partition = tbl.getPartition(partitionId);

  38.             if (partition == null) {

  39.                 continue;

  40.             }

  41.             TStorageMedium storageMedium = tbl.getPartitionInfo().getDataProperty(partitionId).getStorageMedium();

  42.             

  43.             Map<Long, MaterializedIndex> shadowIndexMap = partitionIndexMap.row(partitionId);

  44.             for (Map.Entry<Long, MaterializedIndex> entry : shadowIndexMap.entrySet()) {

  45.                 long shadowIdxId = entry.getKey();

  46.                 MaterializedIndex shadowIdx = entry.getValue();

  47.                 

  48.                 short shadowShortKeyColumnCount = indexShortKeyMap.get(shadowIdxId);

  49.                 List<Column> shadowSchema = indexSchemaMap.get(shadowIdxId);

  50.                 int shadowSchemaHash = indexSchemaVersionAndHashMap.get(shadowIdxId).second;

  51.                 int originSchemaHash = tbl.getSchemaHashByIndexId(indexIdMap.get(shadowIdxId));

  52.                 

  53.                 for (Tablet shadowTablet : shadowIdx.getTablets()) {

  54.                     long shadowTabletId = shadowTablet.getId();

  55.                     List<Replica> shadowReplicas = shadowTablet.getReplicas();

  56.                     for (Replica shadowReplica : shadowReplicas) {

  57.                         long backendId = shadowReplica.getBackendId();

  58.                         countDownLatch.addMark(backendId, shadowTabletId);

  59.                         CreateReplicaTask createReplicaTask = new CreateReplicaTask(

  60.                                 backendId, dbId, tableId, partitionId, shadowIdxId, shadowTabletId,

  61.                                 shadowShortKeyColumnCount, shadowSchemaHash,

  62.                                 Partition.PARTITION_INIT_VERSION, Partition.PARTITION_INIT_VERSION_HASH,

  63.                                 tbl.getKeysType(), TStorageType.COLUMN, storageMedium,

  64.                                 shadowSchema, bfColumns, bfFpp, countDownLatch);

  65.                         createReplicaTask.setBaseTablet(partitionIndexTabletMap.get(partitionId, shadowIdxId).get(shadowTabletId), originSchemaHash);

  66.                         

  67.                         batchTask.addTask(createReplicaTask);

  68.                     } // end for rollupReplicas

  69.                 } // end for rollupTablets

  70.             }

  71.         }

  72.     } finally {

  73.         db.readUnlock();

  74.     }

  75. 

  76.     if (!FeConstants.runningUnitTest) {

  77.         // send all tasks and wait them finished

  78.         AgentTaskQueue.addBatchTask(batchTask);

  79.         AgentTaskExecutor.submit(batchTask);

  80.         // max timeout is 1 min

  81.         long timeout = Math.min(Config.tablet_create_timeout_second * 1000L * totalReplicaNum, 60000);

  82.         boolean ok = false;

  83.         try {

  84.             ok = countDownLatch.await(timeout, TimeUnit.MILLISECONDS);

  85.         } catch (InterruptedException e) {

  86.             LOG.warn("InterruptedException: ", e);

  87.             ok = false;

  88.         }

  89. 

  90.         if (!ok) {

  91.             // create replicas failed. just cancel the job

  92.             // clear tasks and show the failed replicas to user

  93.             AgentTaskQueue.removeBatchTask(batchTask, TTaskType.CREATE);

  94.             String errMsg = null;

  95.             if (!countDownLatch.getStatus().ok()) {

  96.                 errMsg = countDownLatch.getStatus().getErrorMsg();

  97.             } else {

  98.                 List<Entry<Long, Long>> unfinishedMarks = countDownLatch.getLeftMarks();

  99.                 // only show at most 3 results

  100.                 List<Entry<Long, Long>> subList = unfinishedMarks.subList(0, Math.min(unfinishedMarks.size(), 3));

  101.                 errMsg = "Error replicas:" + Joiner.on(", ").join(subList);

  102.             }

  103.             LOG.warn("failed to create replicas for job: {}, {}", jobId, errMsg);

  104.             throw new AlterCancelException("Create replicas failed. Error: " + errMsg);

  105.         }

  106.     }

  107. 

  108.     // create all replicas success.

  109.     // add all shadow indexes to catalog

  110.     db.writeLock();

  111.     try {

  112.         OlapTable tbl = (OlapTable) db.getTable(tableId);

  113.         if (tbl == null) {

  114.             throw new AlterCancelException("Table " + tableId + " does not exist");

  115.         }

  116.         Preconditions.checkState(tbl.getState() == OlapTableState.SCHEMA_CHANGE);

  117.         addShadowIndexToCatalog(tbl);

  118.     } finally {

  119.         db.writeUnlock();

  120.     }

  121. 

  122.     this.watershedTxnId = Catalog.getCurrentGlobalTransactionMgr().getTransactionIDGenerator().getNextTransactionId();

  123.     this.jobState = JobState.WAITING_TXN;

  124. 

  125.     // write edit log

  126.     Catalog.getCurrentCatalog().getEditLog().logAlterJob(this);

  127.     LOG.info("transfer schema change job {} state to {}, watershed txn id: {}", jobId, this.jobState, watershedTxnId);

  128. }
复制代码
这段代码的主要逻辑就是首先在元数据中检查db、tbl、副本数、表的状态等信息，由于一个be需要生成多个tablet，因此使用MarkedCountDownLatch记录任务数量，并生成对应的CreateReplicaTask任务，然后将多个CreateReplicaTask添加到实现了Runnable接口的AgentBatchTask类的私有成员变量private Map<Long, List<AgentTask>> backendIdToTasks中，在AgentBatchTask.run方法中，会将List<AgentTask>转化为List<TAgentTaskRequest>，然后通过thrift接口AgentService.TAgentResult submit_tasks(1:list<AgentService.TAgentTaskRequest> tasks);向BE发送RPC调用，BE完成后，会调用thrift接口MasterService.TMasterResult finishTask(1:MasterService.TFinishTaskRequest request); 向FE汇报BE的任务执行结果。

随后，该方法会调用AgentTaskQueue.addBatchTask(batchTask);及AgentTaskExecutor.submit(batchTask);方法，记录并提交batchTask，AgentTaskExecutor内部是通过Executors.newCachedThreadPool来创造线程并执行实现了Runnable接口的AgentBatchTask。

在生成了多个执行任务线程后，主线程会调用ok = countDownLatch.await(timeout, TimeUnit.MILLISECONDS);方法，阻塞并等待所有执行提交任务的子线程完成。


BE端逻辑
在BE端，我们首先查看AgentServer类，该类的主要私有成员及构造函数如下，在构造函数中，首先有多个std::unique_ptr<TaskWorkerPool>类的实例化对象，每个实例化对象代表一类任务，在AgentServer类初始化时，会根据配置参数，对每个实例化对象调用TaskWorkerPool::start函数，生成指定数量的线程并绑定对应的回调函数。

  1. std::unique_ptr<TaskWorkerPool> _create_tablet_workers;

  2. std::unique_ptr<TaskWorkerPool> _drop_tablet_workers;

  3. std::unique_ptr<TaskWorkerPool> _push_workers;

  4. std::unique_ptr<TaskWorkerPool> _publish_version_workers;

  5. std::unique_ptr<TaskWorkerPool> _clear_transaction_task_workers;

  6. std::unique_ptr<TaskWorkerPool> _delete_workers;

  7. std::unique_ptr<TaskWorkerPool> _alter_tablet_workers;

  8. std::unique_ptr<TaskWorkerPool> _clone_workers;

  9. std::unique_ptr<TaskWorkerPool> _storage_medium_migrate_workers;

  10. std::unique_ptr<TaskWorkerPool> _check_consistency_workers;

  11. 

  12. // These 3 worker-pool do not accept tasks from FE.

  13. // It is self triggered periodically and reports to Fe master

  14. std::unique_ptr<TaskWorkerPool> _report_task_workers;

  15. std::unique_ptr<TaskWorkerPool> _report_disk_state_workers;

  16. std::unique_ptr<TaskWorkerPool> _report_tablet_workers;

  17. 

  18. std::unique_ptr<TaskWorkerPool> _upload_workers;

  19. std::unique_ptr<TaskWorkerPool> _download_workers;

  20. std::unique_ptr<TaskWorkerPool> _make_snapshot_workers;

  21. std::unique_ptr<TaskWorkerPool> _release_snapshot_workers;

  22. std::unique_ptr<TaskWorkerPool> _move_dir_workers;

  23. std::unique_ptr<TaskWorkerPool> _recover_tablet_workers;

  24. std::unique_ptr<TaskWorkerPool> _update_tablet_meta_info_workers;
复制代码

  1. AgentServer::AgentServer(ExecEnv* exec_env, const TMasterInfo& master_info) :

  2.         _exec_env(exec_env),

  3.         _master_info(master_info),

  4.         _topic_subscriber(new TopicSubscriber()) {

  5.     for (auto& path : exec_env->store_paths()) {

  6.         try {

  7.             string dpp_download_path_str = path.path + DPP_PREFIX;

  8.             boost::filesystem::path dpp_download_path(dpp_download_path_str);

  9.             if (boost::filesystem::exists(dpp_download_path)) {

  10.                 boost::filesystem::remove_all(dpp_download_path);

  11.             }

  12.         } catch (...) {

  13.             LOG(WARNING) << "boost exception when remove dpp download path. path=" << path.path;

  14.         }

  15.     }

  16. 

  17.     // It is the same code to create workers of each type, so we use a macro

  18.     // to make code to be more readable.

  19. 

  20. #ifndef BE_TEST

  21. #define CREATE_AND_START_POOL(type, pool_name)         \

  22.     pool_name.reset(new TaskWorkerPool(                \

  23.                 TaskWorkerPool::TaskWorkerType::type,  \

  24.                 _exec_env,                             \

  25.                 master_info));                         \

  26.     pool_name->start();

  27. #else

  28. #define CREATE_AND_START_POOL(type, pool_name)

  29. #endif // BE_TEST

  30. 

  31.     CREATE_AND_START_POOL(CREATE_TABLE, _create_tablet_workers);

  32.     CREATE_AND_START_POOL(DROP_TABLE, _drop_tablet_workers);

  33.     // Both PUSH and REALTIME_PUSH type use _push_workers

  34.     CREATE_AND_START_POOL(PUSH, _push_workers);

  35.     CREATE_AND_START_POOL(PUBLISH_VERSION, _publish_version_workers);

  36.     CREATE_AND_START_POOL(CLEAR_TRANSACTION_TASK, _clear_transaction_task_workers);

  37.     CREATE_AND_START_POOL(DELETE, _delete_workers);

  38.     CREATE_AND_START_POOL(ALTER_TABLE, _alter_tablet_workers);

  39.     CREATE_AND_START_POOL(CLONE, _clone_workers);

  40.     CREATE_AND_START_POOL(STORAGE_MEDIUM_MIGRATE, _storage_medium_migrate_workers);

  41.     CREATE_AND_START_POOL(CHECK_CONSISTENCY, _check_consistency_workers);

  42.     CREATE_AND_START_POOL(REPORT_TASK, _report_task_workers);

  43.     CREATE_AND_START_POOL(REPORT_DISK_STATE, _report_disk_state_workers);

  44.     CREATE_AND_START_POOL(REPORT_OLAP_TABLE, _report_tablet_workers);

  45.     CREATE_AND_START_POOL(UPLOAD, _upload_workers);

  46.     CREATE_AND_START_POOL(DOWNLOAD, _download_workers);

  47.     CREATE_AND_START_POOL(MAKE_SNAPSHOT, _make_snapshot_workers);

  48.     CREATE_AND_START_POOL(RELEASE_SNAPSHOT, _release_snapshot_workers);

  49.     CREATE_AND_START_POOL(MOVE, _move_dir_workers);

  50.     CREATE_AND_START_POOL(RECOVER_TABLET, _recover_tablet_workers);

  51.     CREATE_AND_START_POOL(UPDATE_TABLET_META_INFO, _update_tablet_meta_info_workers);

  52. #undef CREATE_AND_START_POOL

  53. 

  54. #ifndef BE_TEST

  55.     // Add subscriber here and register listeners

  56.     TopicListener* user_resource_listener = new UserResourceListener(exec_env, master_info);

  57.     LOG(INFO) << "Register user resource listener";

  58.     _topic_subscriber->register_listener(doris::TTopicType::type::RESOURCE, user_resource_listener);

  59. #endif

  60. }
复制代码


接下来在AgentServer中最重要的是AgentServer::submit_task方法，该方法是thrift调用接口的具体实现，接收List<TAgentTaskRequest>参数，并根据每个task的task_type将请求转发给TaskWorkPool::submit_task方法
  1. void AgentServer::submit_tasks(TAgentResult& agent_result, const vector<TAgentTaskRequest>& tasks) {

  2.     Status ret_st;

  3. 

  4.     // TODO check master_info here if it is the same with that of heartbeat rpc

  5.     if (_master_info.network_address.hostname == "" || _master_info.network_address.port == 0) {

  6.         Status ret_st = Status::Cancelled("Have not get FE Master heartbeat yet");

  7.         ret_st.to_thrift(&agent_result.status);

  8.         return;

  9.     }

  10. 

  11.     for (auto task : tasks) {

  12.         VLOG_RPC << "submit one task: " << apache::thrift::ThriftDebugString(task).c_str();

  13.         TTaskType::type task_type = task.task_type;

  14.         int64_t signature = task.signature;

  15. 

  16. #define HANDLE_TYPE(t_task_type, work_pool, req_member)                         \

  17.     case t_task_type:                                                           \

  18.         if (task.__isset.req_member) {                                          \

  19.             work_pool->submit_task(task);                                       \

  20.         } else {                                                                \

  21.             ret_st = Status::InvalidArgument(strings::Substitute(               \

  22.                     "task(signature=$0) has wrong request member", signature)); \

  23.         }                                                                       \

  24.         break;

  25. 

  26.         // TODO(lingbin): It still too long, divided these task types into several categories

  27.         switch (task_type) {

  28.         HANDLE_TYPE(TTaskType::CREATE, _create_tablet_workers, create_tablet_req);

  29.         HANDLE_TYPE(TTaskType::DROP, _drop_tablet_workers, drop_tablet_req);

  30.         HANDLE_TYPE(TTaskType::PUBLISH_VERSION, _publish_version_workers, publish_version_req);

  31.         HANDLE_TYPE(TTaskType::CLEAR_TRANSACTION_TASK,

  32.                     _clear_transaction_task_workers,

  33.                     clear_transaction_task_req);

  34.         HANDLE_TYPE(TTaskType::CLONE, _clone_workers, clone_req);

  35.         HANDLE_TYPE(TTaskType::STORAGE_MEDIUM_MIGRATE,

  36.                     _storage_medium_migrate_workers,

  37.                     storage_medium_migrate_req);

  38.         HANDLE_TYPE(TTaskType::CHECK_CONSISTENCY,

  39.                     _check_consistency_workers,

  40.                     check_consistency_req);

  41.         HANDLE_TYPE(TTaskType::UPLOAD, _upload_workers, upload_req);

  42.         HANDLE_TYPE(TTaskType::DOWNLOAD, _download_workers, download_req);

  43.         HANDLE_TYPE(TTaskType::MAKE_SNAPSHOT, _make_snapshot_workers, snapshot_req);

  44.         HANDLE_TYPE(TTaskType::RELEASE_SNAPSHOT, _release_snapshot_workers, release_snapshot_req);

  45.         HANDLE_TYPE(TTaskType::MOVE, _move_dir_workers, move_dir_req);

  46.         HANDLE_TYPE(TTaskType::RECOVER_TABLET, _recover_tablet_workers, recover_tablet_req);

  47.         HANDLE_TYPE(TTaskType::UPDATE_TABLET_META_INFO,

  48.                     _update_tablet_meta_info_workers,

  49.                     update_tablet_meta_info_req);

  50. 

  51.         case TTaskType::REALTIME_PUSH:

  52.         case TTaskType::PUSH:

  53.             if (!task.__isset.push_req) {

  54.                 ret_st = Status::InvalidArgument(strings::Substitute(

  55.                         "task(signature=$0) has wrong request member", signature));

  56.                 break;

  57.             }

  58.             if (task.push_req.push_type == TPushType::LOAD

  59.                     || task.push_req.push_type == TPushType::LOAD_DELETE) {

  60.                 _push_workers->submit_task(task);

  61.             } else if (task.push_req.push_type == TPushType::DELETE) {

  62.                 _delete_workers->submit_task(task);

  63.             } else {

  64.                 ret_st = Status::InvalidArgument(strings::Substitute(

  65.                         "task(signature=$0, type=$1, push_type=$2) has wrong push_type",

  66.                         signature, task_type, task.push_req.push_type));

  67.             }

  68.             break;

  69.         case TTaskType::ALTER:

  70.             if (task.__isset.alter_tablet_req || task.__isset.alter_tablet_req_v2) {

  71.                 _alter_tablet_workers->submit_task(task);

  72.             } else {

  73.                 ret_st = Status::InvalidArgument(strings::Substitute(

  74.                         "task(signature=$0) has wrong request member", signature));

  75.             }

  76.             break;

  77.         default:

  78.             ret_st = Status::InvalidArgument(strings::Substitute(

  79.                     "task(signature=$0, type=$1) has wrong task type", signature, task_type));

  80.             break;

  81.         }

  82. #undef HANDLE_TYPE

  83. 

  84.         if (!ret_st.ok()) {

  85.             LOG(WARNING) << "fail to submit task. reason: " << ret_st.get_error_msg()

  86.                     << ", task: " << task;

  87.             // For now, all tasks in the batch share one status, so if any task

  88.             // was failed to submit, we can only return error to FE(even when some

  89.             // tasks have already been successfully submitted).

  90.             // However, Fe does not check the return status of submit_tasks() currently,

  91.             // and it is not sure that FE will retry when something is wrong, so here we

  92.             // only print an warning log and go on(i.e. do not break current loop),

  93.             // to ensure every task can be submitted once. It is OK for now, because the

  94.             // ret_st can be error only when it encounters an wrong task_type and

  95.             // req-member in TAgentTaskRequest, which is basically impossible.

  96.             // TODO(lingbin): check the logic in FE again later.

  97.         }

  98.     }

  99. 

  100.     ret_st.to_thrift(&agent_result.status);

  101. }
复制代码
然后我们进入TaskWorkPool函数中，可以看到主要有一下的函数指针，每个函数指针会绑定一个回调函数，该回调函数在TaskWorkPool::start方法中被绑定
  1. static void* _create_tablet_worker_thread_callback(void* arg_this);

  2. static void* _drop_tablet_worker_thread_callback(void* arg_this);

  3. static void* _push_worker_thread_callback(void* arg_this);

  4. static void* _publish_version_worker_thread_callback(void* arg_this);

  5. static void* _clear_transaction_task_worker_thread_callback(void* arg_this);

  6. static void* _alter_tablet_worker_thread_callback(void* arg_this);

  7. static void* _clone_worker_thread_callback(void* arg_this);

  8. static void* _storage_medium_migrate_worker_thread_callback(void* arg_this);

  9. static void* _check_consistency_worker_thread_callback(void* arg_this);

  10. static void* _report_task_worker_thread_callback(void* arg_this);

  11. static void* _report_disk_state_worker_thread_callback(void* arg_this);

  12. static void* _report_tablet_worker_thread_callback(void* arg_this);

  13. static void* _upload_worker_thread_callback(void* arg_this);

  14. static void* _download_worker_thread_callback(void* arg_this);

  15. static void* _make_snapshot_thread_callback(void* arg_this);

  16. static void* _release_snapshot_thread_callback(void* arg_this);

  17. static void* _move_dir_thread_callback(void* arg_this);

  18. static void* _recover_tablet_thread_callback(void* arg_this);

  19. static void* _update_tablet_meta_worker_thread_callback(void* arg_this);
复制代码
最后是被AgentServer::submit_task转发调用的TaskWorkPool::submit_task方法，在该方法中，每个task会被依次塞入std::deque<TAgentTaskRequest> _tasks;队列中（先进先出），然后调用_worker_thread_condition_lock.notify();方法唤醒一个task线程执行任务，即一个典型的生产-消费者模型

  1. void TaskWorkerPool::submit_task(const TAgentTaskRequest& task) {

  2.     const TTaskType::type task_type = task.task_type;

  3.     int64_t signature = task.signature;

  4. 

  5.     std::string type_str;

  6.     EnumToString(TTaskType, task_type, type_str);

  7.     LOG(INFO) << "submitting task. type=" << type_str << ", signature=" << signature;

  8. 

  9.     if (_register_task_info(task_type, signature)) {

  10.         // Set the receiving time of task so that we can determine whether it is timed out later

  11.         (const_cast<TAgentTaskRequest&>(task)).__set_recv_time(time(nullptr));

  12.         size_t task_count_in_queue = 0;

  13.         {

  14.             lock_guard<Mutex> worker_thread_lock(_worker_thread_lock);

  15.             _tasks.push_back(task);

  16.             task_count_in_queue = _tasks.size();

  17.             _worker_thread_condition_lock.notify();

  18.         }

  19.         LOG(INFO) << "success to submit task. type=" << type_str << ", signature=" << signature

  20.                 << ", task_count_in_queue=" << task_count_in_queue;

  21.     } else {

  22.         LOG(INFO) << "fail to register task. type=" << type_str << ", signature=" << signature;

  23.     }

  24. }
复制代码
以_alter_tablet_worker_thread_callback任务为例，该任务是一个消费者，当生产者队列为空时会调用worker_pool_this->_worker_thread_condition_lock.wait();进行阻塞，直到被生产唤醒后，执行具体的任务逻辑，然后调用TaskWorkPool::_finish_task向FE汇报任务完成情况，最后通过TaskWorkPool::_remote_task_info将任务从队列std::deque<TAgentTaskRequest> _tasks;中移除，至此便完成了整个创建replica的逻辑。

  1. void* TaskWorkerPool::_alter_tablet_worker_thread_callback(void* arg_this) {

  2.     TaskWorkerPool* worker_pool_this = (TaskWorkerPool*)arg_this;

  3. 

  4. #ifndef BE_TEST

  5.     while (true) {

  6. #endif

  7.         TAgentTaskRequest agent_task_req;

  8.         {

  9.             lock_guard<Mutex> worker_thread_lock(worker_pool_this->_worker_thread_lock);

  10.             while (worker_pool_this->_tasks.empty()) {

  11.                 worker_pool_this->_worker_thread_condition_lock.wait();

  12.             }

  13. 

  14.             agent_task_req = worker_pool_this->_tasks.front();

  15.             worker_pool_this->_tasks.pop_front();

  16.         }

  17.         int64_t signatrue = agent_task_req.signature;

  18.         LOG(INFO) << "get alter table task, signature: " <<  agent_task_req.signature;

  19.         bool is_task_timeout = false;

  20.         if (agent_task_req.__isset.recv_time) {

  21.             int64_t time_elapsed = time(nullptr) - agent_task_req.recv_time;

  22.             if (time_elapsed > config::report_task_interval_seconds * 20) {

  23.                 LOG(INFO) << "task elapsed " << time_elapsed

  24.                           << " seconds since it is inserted to queue, it is timeout";

  25.                 is_task_timeout = true;

  26.             }

  27.         }

  28.         if (!is_task_timeout) {

  29.             TFinishTaskRequest finish_task_request;

  30.             TTaskType::type task_type = agent_task_req.task_type;

  31.             switch (task_type) {

  32.             case TTaskType::ALTER:

  33.                 worker_pool_this->_alter_tablet(worker_pool_this,

  34.                                             agent_task_req,

  35.                                             signatrue,

  36.                                             task_type,

  37.                                             &finish_task_request);

  38.                 break;

  39.             default:

  40.                 // pass

  41.                 break;

  42.             }

  43.             worker_pool_this->_finish_task(finish_task_request);

  44.         }

  45.         worker_pool_this->_remove_task_info(agent_task_req.task_type, agent_task_req.signature);

  46. #ifndef BE_TEST

  47.     }

  48. #endif

  49.     return (void*)0;

  50. }
复制代码




最新经典文章，欢迎关注公众号


原文链接
https://wingsgo.github.io/2020/0 ... ma_change_task.html
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
hyj

超级版主

关注

  • 1689

    主题

  • 2216

    帖子

  • 467

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000