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 :
33 : /**
34 : * @brief Destroy the Comm Channel Queue object
35 : */
36 112 : ~CommChannelQueue()
37 : {
38 : try {
39 112 : Uninit();
40 0 : } catch (...) {
41 0 : BQS_LOG_ERROR("CommChannelQueue destructor exception.");
42 : }
43 112 : }
44 :
45 : CommChannelQueue(const CommChannelQueue&) = delete;
46 : CommChannelQueue(const CommChannelQueue&&) = delete;
47 : CommChannelQueue& operator=(const CommChannelQueue&) = delete;
48 : CommChannelQueue& operator=(CommChannelQueue&&) = delete;
49 :
50 : /**
51 : * @brief init queue
52 : * @return FSM_SUCCESS: success, other: failed
53 : */
54 51 : FsmStatus Init(const uint32_t depth)
55 : {
56 51 : if ((depth == 0U) || (depth > MAX_QUEUE_DEPTH)) {
57 3 : BQS_LOG_ERROR("Invalid parameter, depth:[%u].", depth);
58 3 : return FsmStatus::FSM_FAILED;
59 : }
60 48 : depth_ = depth;
61 :
62 48 : ring_ = new (std::nothrow) T[depth_];
63 48 : if (ring_ == nullptr) {
64 0 : BQS_LOG_ERROR("Failed to kzalloc memory for queue, depth=[%u].", depth_);
65 0 : return FsmStatus::FSM_FAILED;
66 : }
67 48 : BQS_LOG_DEBUG("Succes to alloc memory[%zu].", sizeof(T) * depth_);
68 :
69 48 : head_ = 0U;
70 48 : tail_ = 0U;
71 48 : BQS_LOG_DEBUG("Success to init comm channel queue, depth:[%u].", depth_);
72 48 : return FsmStatus::FSM_SUCCESS;
73 : }
74 :
75 : /**
76 : * @brief uninit queue
77 : */
78 167 : void Uninit()
79 : {
80 167 : if (ring_ != nullptr) {
81 48 : delete[] ring_;
82 48 : ring_ = nullptr;
83 48 : BQS_LOG_DEBUG("Success to free memory[%zu].", sizeof(T) * depth_);
84 : }
85 167 : head_ = 0U;
86 167 : tail_ = 0U;
87 167 : depth_ = 1U;
88 167 : BQS_LOG_DEBUG("Success to uninit comm channel queue.");
89 167 : }
90 :
91 : /**
92 : * @brief push one element to queue
93 : * @param buff buff
94 : * @return current enqueue success count, 0 failed
95 : */
96 59 : int32_t Push(T& buff)
97 : {
98 177 : BQS_LOG_DEBUG("Push queue, head:[%u], tail:[%u], depth:[%u].", head_.load(), tail_.load(), depth_);
99 59 : if (IsFull()) {
100 6 : return 0;
101 : }
102 53 : ring_[tail_] = std::move(buff);
103 : // ++tail_ cannot be used because of time sequence problem(pop concurrently)
104 53 : tail_ = (tail_ + 1) % depth_;
105 53 : return 1;
106 : }
107 :
108 : /**
109 : * @brief get first element from queue
110 : * @return T* first element
111 : */
112 73 : T* Front()
113 : {
114 219 : BQS_LOG_DEBUG(
115 : "Get front element from queue, head:[%u], tail:[%u], depth:[%u].", head_.load(), tail_.load(), depth_);
116 73 : if (IsEmpty()) {
117 8 : return nullptr;
118 : }
119 65 : return &ring_[head_];
120 : }
121 :
122 : /**
123 : * @brief pop first element from queue
124 : * @return current pop success count, 0 failed
125 : */
126 54 : int32_t Pop()
127 : {
128 162 : BQS_LOG_DEBUG("Pop queue, head:[%u], tail:[%u], depth:[%u].", head_.load(), tail_.load(), depth_);
129 54 : if (IsEmpty()) {
130 1 : return 0;
131 : }
132 : // ++head_ cannot be used because of time sequence problem(push concurrently)
133 53 : head_ = (head_ + 1) % depth_;
134 53 : return 1;
135 : }
136 :
137 : /**
138 : * @brief check queue empty
139 : * @return true or false
140 : */
141 1226 : bool IsEmpty() const
142 : {
143 3678 : BQS_LOG_DEBUG("Check queue empty, head:[%u], tail:[%u], depth:[%u].", head_.load(), tail_.load(), depth_);
144 1226 : return (head_ == tail_);
145 : }
146 :
147 : /**
148 : * @brief check queue full
149 : * @return true or false
150 : */
151 83 : bool IsFull() const
152 : {
153 249 : BQS_LOG_DEBUG("Check queue full, head:[%u], tail:[%u], depth:[%u].", head_.load(), tail_.load(), depth_);
154 83 : return (((tail_ + 1) % depth_) == head_);
155 : }
156 :
157 : /**
158 : * @brief get queue elements count
159 : * @return queue elements count
160 : */
161 62 : uint32_t Size() const { return ((tail_ - head_) + depth_) % depth_; }
162 :
163 : private:
164 : // max store (depth_ - 1) elements
165 : uint32_t depth_;
166 : // head: point to where the effective memory on the ring begins
167 : // tail: point to the released ring position
168 : std::atomic<uint32_t> head_;
169 : std::atomic<uint32_t> tail_;
170 : T* ring_;
171 : };
172 : } // namespace dgw
173 : #endif
|
