1 /* 2 * Copyright (c) 2021 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef SEND_TASK_SCHEDULER_H 17 #define SEND_TASK_SCHEDULER_H 18 19 #include <cstdint> 20 #include <list> 21 #include <map> 22 #include <mutex> 23 #include <string> 24 #include <vector> 25 #include "communicator_type_define.h" 26 #include "macro_utils.h" 27 28 namespace DistributedDB { 29 enum class TargetPolicy { 30 NO_DELAY = 0, 31 DELAY = 1, 32 }; 33 34 class SerialBuffer; // Forward Declaration 35 36 struct SendTask { 37 SerialBuffer *buffer = nullptr; 38 std::string dstTarget; 39 OnSendEnd onEnd; 40 uint32_t frameId = 0u; 41 }; 42 43 struct SendTaskInfo { 44 bool delayFlag = false; 45 Priority taskPrio = Priority::LOW; 46 }; 47 48 using TaskListByTarget = std::map<std::string, std::list<SendTask>>; 49 50 class SendTaskScheduler { 51 public: 52 SendTaskScheduler() = default; // Default constructor must be explicitly provided due to DISABLE_COPY_ASSIGN_MOVE 53 ~SendTaskScheduler(); 54 55 DISABLE_COPY_ASSIGN_MOVE(SendTaskScheduler); 56 57 void Initialize(); 58 59 // This method for consumer 60 void Finalize(); 61 62 // This method for producer, support multiple thread 63 int AddSendTaskIntoSchedule(const SendTask &inTask, Priority inPrio); 64 65 // This method for consumer, not recommend for multiple thread 66 int ScheduleOutSendTask(SendTask &outTask, uint32_t &totalLength); 67 int ScheduleOutSendTask(SendTask &outTask, SendTaskInfo &outTaskInfo, uint32_t &totalLength); 68 69 // This method for consumer, call ScheduleOutSendTask at least one time before each calling this 70 int FinalizeLastScheduleTask(); 71 72 // These two mothods influence the task that will be schedule out next time 73 int DelayTaskByTarget(const std::string &inTarget); 74 int NoDelayTaskByTarget(const std::string &inTarget); 75 76 uint32_t GetTotalTaskCount() const; 77 uint32_t GetNoDelayTaskCount() const; 78 79 private: 80 int ScheduleDelayTask(SendTask &outTask, SendTaskInfo &outTaskInfo); 81 int ScheduleNoDelayTask(SendTask &outTask, SendTaskInfo &outTaskInfo); 82 83 mutable std::mutex overallMutex_; 84 uint32_t curTotalSizeByByte_ = 0; 85 uint32_t curTotalSizeByTask_ = 0; 86 uint32_t delayTaskCount_ = 0; 87 88 std::vector<Priority> priorityOrder_; 89 std::map<Priority, uint32_t> extraCapacityInByteByPrio_; 90 std::map<std::string, TargetPolicy> policyMap_; 91 std::map<std::string, uint32_t> totalBytesByTarget_; 92 93 std::map<Priority, uint32_t> taskCountByPrio_; 94 std::map<Priority, uint32_t> taskDelayCountByPrio_; 95 std::map<Priority, std::list<std::string>> taskOrderByPrio_; 96 std::map<Priority, TaskListByTarget> taskGroupByPrio_; 97 98 bool scheduledFlag_ = false; 99 std::string lastScheduleTarget_; 100 Priority lastSchedulePriority_ = Priority::LOW; 101 }; 102 } 103 104 #endif