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     bool isValid = true;
42 };
43 
44 struct SendTaskInfo {
45     bool delayFlag = false;
46     Priority taskPrio = Priority::LOW;
47 };
48 
49 using TaskListByTarget = std::map<std::string, std::list<SendTask>>;
50 
51 class SendTaskScheduler {
52 public:
53     SendTaskScheduler() = default; // Default constructor must be explicitly provided due to DISABLE_COPY_ASSIGN_MOVE
54     ~SendTaskScheduler();
55 
56     DISABLE_COPY_ASSIGN_MOVE(SendTaskScheduler);
57 
58     void Initialize();
59 
60     // This method for consumer
61     void Finalize();
62 
63     // This method for producer, support multiple thread
64     int AddSendTaskIntoSchedule(const SendTask &inTask, Priority inPrio);
65 
66     // This method for consumer, not recommend for multiple thread
67     int ScheduleOutSendTask(SendTask &outTask, uint32_t &totalLength);
68     int ScheduleOutSendTask(SendTask &outTask, SendTaskInfo &outTaskInfo, uint32_t &totalLength);
69 
70     // This method for consumer, call ScheduleOutSendTask at least one time before each calling this
71     int FinalizeLastScheduleTask();
72 
73     // These two mothods influence the task that will be schedule out next time
74     int DelayTaskByTarget(const std::string &inTarget);
75     int NoDelayTaskByTarget(const std::string &inTarget);
76 
77     uint32_t GetTotalTaskCount() const;
78     uint32_t GetNoDelayTaskCount() const;
79 
80     void InvalidSendTask(const std::string &target);
81     void SetSoftBusErrCode(const std::string &target, int softBusErrCode);
82 
83 private:
84     int ScheduleDelayTask(SendTask &outTask, SendTaskInfo &outTaskInfo);
85     int ScheduleNoDelayTask(SendTask &outTask, SendTaskInfo &outTaskInfo);
86 
87     mutable std::mutex overallMutex_;
88     uint32_t curTotalSizeByByte_ = 0;
89     uint32_t curTotalSizeByTask_ = 0;
90     uint32_t delayTaskCount_ = 0;
91 
92     std::vector<Priority> priorityOrder_;
93     std::map<Priority, uint32_t> extraCapacityInByteByPrio_;
94     std::map<std::string, TargetPolicy> policyMap_;
95     std::map<std::string, uint32_t> totalBytesByTarget_;
96 
97     std::map<Priority, uint32_t> taskCountByPrio_;
98     std::map<Priority, uint32_t> taskDelayCountByPrio_;
99     std::map<Priority, std::list<std::string>> taskOrderByPrio_;
100     std::map<Priority, TaskListByTarget> taskGroupByPrio_;
101 
102     bool scheduledFlag_ = false;
103     std::string lastScheduleTarget_;
104     Priority lastSchedulePriority_ = Priority::LOW;
105 
106     std::map<std::string, int> softBusErrCodeMap_;
107 };
108 }
109 
110 #endif