Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #ifndef COMM_CHANNEL_QUEUE_H
12 : #define COMM_CHANNEL_QUEUE_H
13 :
14 : #include <atomic>
15 : #include <cstdint>
16 : #include "hccl/hccl_types_in.h"
17 : #include "driver/ascend_hal.h"
18 : #include "fsm/state_define.h"
19 :
20 : namespace dgw {
21 : // max queue depth
22 : constexpr uint32_t MAX_QUEUE_DEPTH = 8U * 1024U * 2U + 1U;
23 :
24 : template<typename T>
25 : class CommChannelQueue {
26 : public:
27 : /**
28 : * @brief Construct a new Comm Channel Queue object
29 : * @param depth queue depth
30 : */
31 112 : explicit CommChannelQueue() : depth_(1U), head_(0U), tail_(0U), ring_(nullptr)
32 112 : {}
33 :
34 : /**
35 : * @brief Destroy the Comm Channel Queue object
36 : */
37 112 : ~CommChannelQueue()
38 : {
39 : try {
40 112 : Uninit();
41 0 : } catch(...) {
42 0 : BQS_LOG_ERROR("CommChannelQueue destructor exception.");
43 : }
44 112 : }
45 :
46 : CommChannelQueue(const CommChannelQueue &) = delete;
47 : CommChannelQueue(const CommChannelQueue &&) = delete;
48 : CommChannelQueue &operator = (const CommChannelQueue &) = delete;
49 : CommChannelQueue &operator = (CommChannelQueue &&) = delete;
50 :
51 : /**
52 : * @brief init queue
53 : * @return FSM_SUCCESS: success, other: failed
54 : */
55 51 : FsmStatus Init(const uint32_t depth)
56 : {
57 51 : if ((depth == 0U) || (depth > MAX_QUEUE_DEPTH)) {
58 3 : BQS_LOG_ERROR("Invalid parameter, depth:[%u].", depth);
59 3 : return FsmStatus::FSM_FAILED;
60 : }
61 48 : depth_ = depth;
62 :
63 48 : ring_ = new (std::nothrow) T[depth_];
64 48 : if (ring_ == nullptr) {
65 0 : BQS_LOG_ERROR("Failed to kzalloc memory for queue, depth=[%u].", depth_);
66 0 : return FsmStatus::FSM_FAILED;
67 : }
68 48 : BQS_LOG_DEBUG("Succes to alloc memory[%zu].", sizeof(T) * depth_);
69 :
70 48 : head_ = 0U;
71 48 : tail_ = 0U;
72 48 : BQS_LOG_DEBUG("Success to init comm channel queue, depth:[%u].", depth_);
73 48 : return FsmStatus::FSM_SUCCESS;
74 : }
75 :
76 : /**
77 : * @brief uninit queue
78 : */
79 167 : void Uninit()
80 : {
81 167 : if (ring_ != nullptr) {
82 48 : delete []ring_;
83 48 : ring_ = nullptr;
84 48 : BQS_LOG_DEBUG("Success to free memory[%zu].", sizeof(T) * depth_);
85 : }
86 167 : head_ = 0U;
87 167 : tail_ = 0U;
88 167 : depth_ = 1U;
89 167 : BQS_LOG_DEBUG("Success to uninit comm channel queue.");
90 167 : }
91 :
92 : /**
93 : * @brief push one element to queue
94 : * @param buff buff
95 : * @return current enqueue success count, 0 failed
96 : */
97 59 : int32_t Push(T &buff)
98 : {
99 177 : BQS_LOG_DEBUG("Push queue, head:[%u], tail:[%u], depth:[%u].",
100 : head_.load(), tail_.load(), depth_);
101 59 : if (IsFull()) {
102 6 : return 0;
103 : }
104 53 : ring_[tail_] = std::move(buff);
105 : // ++tail_ cannot be used because of time sequence problem(pop concurrently)
106 53 : tail_ = (tail_ + 1) % depth_;
107 53 : return 1;
108 : }
109 :
110 : /**
111 : * @brief get first element from queue
112 : * @return T* first element
113 : */
114 73 : T *Front()
115 : {
116 219 : BQS_LOG_DEBUG("Get front element from queue, head:[%u], tail:[%u], depth:[%u].",
117 : head_.load(), tail_.load(), depth_);
118 73 : if (IsEmpty()) {
119 8 : return nullptr;
120 : }
121 65 : return &ring_[head_];
122 : }
123 :
124 : /**
125 : * @brief pop first element from queue
126 : * @return current pop success count, 0 failed
127 : */
128 54 : int32_t Pop()
129 : {
130 162 : BQS_LOG_DEBUG("Pop queue, head:[%u], tail:[%u], depth:[%u].",
131 : head_.load(), tail_.load(), depth_);
132 54 : if (IsEmpty()) {
133 1 : return 0;
134 : }
135 : // ++head_ cannot be used because of time sequence problem(push concurrently)
136 53 : head_ = (head_ + 1) % depth_;
137 53 : return 1;
138 : }
139 :
140 : /**
141 : * @brief check queue empty
142 : * @return true or false
143 : */
144 1226 : bool IsEmpty() const
145 : {
146 3678 : BQS_LOG_DEBUG("Check queue empty, head:[%u], tail:[%u], depth:[%u].",
147 : head_.load(), tail_.load(), depth_);
148 1226 : return (head_ == tail_);
149 : }
150 :
151 : /**
152 : * @brief check queue full
153 : * @return true or false
154 : */
155 83 : bool IsFull() const
156 : {
157 249 : BQS_LOG_DEBUG("Check queue full, head:[%u], tail:[%u], depth:[%u].",
158 : head_.load(), tail_.load(), depth_);
159 83 : return (((tail_ + 1) % depth_) == head_);
160 : }
161 :
162 : /**
163 : * @brief get queue elements count
164 : * @return queue elements count
165 : */
166 62 : uint32_t Size() const
167 : {
168 62 : return ((tail_ - head_) + depth_) % depth_;
169 : }
170 :
171 : private:
172 : // max store (depth_ - 1) elements
173 : uint32_t depth_;
174 : // head: point to where the effective memory on the ring begins
175 : // tail: point to the released ring position
176 : std::atomic<uint32_t> head_;
177 : std::atomic<uint32_t> tail_;
178 : T *ring_;
179 : };
180 : } // namespace dgw
181 : #endif
|
