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 DGW_GROUP_ENTITY_H
12 : #define DGW_GROUP_ENTITY_H
13 :
14 : #include "entity.h"
15 : #include "dgw_client.h"
16 :
17 : namespace dgw {
18 :
19 : struct GroupEntityInfo {
20 : int32_t groupId; // group id
21 : bqs::GroupPolicy groupPolicy; // group policy
22 : int64_t timeout; // timeout interval, us
23 : uint64_t lastTransId; // last route label
24 : uint64_t lastTimestamp; // timestamp of last data migration, us
25 : uint32_t peerInstanceNum;
26 : uint32_t localInstanceIndex;
27 : };
28 :
29 : class GroupEntity : public Entity {
30 : public:
31 : explicit GroupEntity(const EntityMaterial& material, const uint32_t resIndex);
32 27 : virtual ~GroupEntity() = default;
33 : GroupEntity(const GroupEntity&) = delete;
34 : GroupEntity(const GroupEntity&&) = delete;
35 : GroupEntity& operator=(const GroupEntity&) = delete;
36 : GroupEntity& operator=(GroupEntity&&) = delete;
37 :
38 : FsmStatus Dequeue() override;
39 : void SelectDstEntities(
40 : const uint64_t key, std::vector<Entity*>& toPushDstEntities, std::vector<Entity*>& reprocessDstEntities,
41 : std::vector<Entity*>& abnormalDstEntities) override;
42 : void ReprocessInTryPush(const Entity& srcEntity, DynamicRequestPtr& dynamicRequest, uint32_t& schedCfgKey) override;
43 : FsmStatus AbProcessInTryPush() override;
44 : FsmStatus PauseSubscribe(const Entity& fullEntity) override;
45 : FsmStatus ResumeSubscribe(const Entity& notFullEntity) override;
46 : FsmStatus ClearQueue() override;
47 : FsmStatus MakeSureOutputCompletion() override;
48 : uint32_t GetMbufDeviceId() const override;
49 : uint32_t GetMbufQueueType() const override;
50 :
51 : private:
52 : void SetGroupInfo(const uint64_t lastTransId, const uint64_t lastTimestamp);
53 : EntityPtr SelectSrcEntity(FsmStatus& status);
54 : bool Match(const Entity& entity, const uint64_t waitTransId, bool exactlyMatch) const;
55 : FsmStatus PeekFromEntityInGroup(Entity& entity, const uint64_t waitTransId) const;
56 : bool CheckTimeout(const uint64_t waitTransId) const;
57 : EntityPtr SelectEntityWithMinTransId(const std::vector<EntityPtr>& entities) const;
58 :
59 : GroupEntityInfo groupInfo_;
60 : uint32_t mbufDeviceId_;
61 : uint32_t mbufQueueType_;
62 : };
63 : } // namespace dgw
64 : #endif
|