一、成员变量
- 对齐
  - C++ 11 的 alignas 和 alignof
  - UE 的 Align 模板函数
- volatile
二、构造和析构
- TNode
三、Enqueue
四、Dequeue
五、其他辅助函数

这是一篇关于UE中TQueue队列的文章

一、成员变量

using TNodeVolatilePtr = std::conditional_t<Mode == EQueueMode::SingleThreaded, TNode*, TNode* volatile>;
/** Holds a pointer to the head of the list. */
MS_ALIGN(16) TNodeVolatilePtr Head GCC_ALIGN(16);

/** Holds a pointer to the tail of the list. */
TNode* Tail;

先看成员变量，有两个指针Head和Tail，其中Head在多线程时加了volatile关键字，同时用宏告诉编译器必须16字节对齐

因为TQueue不知道外面代码会用在哪里，成员变量依赖于外面的内存空间，不声明16字节，外面就有可能把内存对到一个诡异的位置上。比如用一个uint8数组来装TQueue，先用掉一个字节，从第二字节开始装TQueue，我们知道指针在64位机器是8字节，那么这个Head就有7字节在前面，一个字节在后面，那么Head指针肯定不是16的倍数，在做一些事情时，如果不接受这样的指针就很可能报错，即使不报错也不那么效率。

对齐

还有更为详细的：UE内存管理系统（内存对齐方式）（一）

struct 和 class 的内存大小，有两个决定因素：

每个成员变量占据一定的字节数
成员变量之间的padding，内存对齐

一个简单的结构体示例：

struct Example2 {
    char a;      // 1 byte
    double b;    // 8 bytes
    int c;       // 4 bytes
};

它的sizeof 是 24 bytes, alignof 是 8 bytes。

它的内存布局可以看出，它是用最大的成员（double b）来对齐的：

偏移量: 0    1    2    3    4    5    6    7    8    9    10   11   12   13   14   15   16   17   18   19   20   21   22   23
       +----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+
内容:   | a  |pad |pad |pad |pad |pad |pad |pad | b  | b  | b  | b  | b  | b  | b  | b  | c  | c  | c  | c  |pad |pad |pad |pad |
       +----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+
说明:   |<-- char -->|<------ padding ----->|<---------- double ---------->|<---- int ---->|<---- padding ---->|
       |<-------------------------- 24 bytes total --------------------------->|

这造成了非常多无用的padding，导致内存浪费。可以调整成员变量的位置来减少浪费：

struct Example2Optimized {
    double b;    // 8 bytes
    int c;       // 4 bytes
    char a;      // 1 byte
    // 只需要 3 字节的填充
};

优化后的内存布局：

偏移量: 0    1    2    3    4    5    6    7    8    9    10   11   12   13   14   15
       +----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+
内容:   | b  | b  | b  | b  | b  | b  | b  | b  | c  | c  | c  | c  | a  |pad |pad |pad |
       +----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+
       |<---------- double ---------->|<---- int ---->|<-char->|<-padding->|
       |<---------------------- 16 bytes total ---------------------->|

它的sizeof 是 16 bytes, alignof 是 8 bytes。大大减少了内存浪费。

尽量将大尺寸类型放在前面可以减少填充字节。

C++ 11 的 alignas 和 alignof

上面的示例可以看出，alignof 通常返回成员变量中最大的成员的大小。 alignas 是一个 C++11引入的关键字，可以用来显式指定 class 或 struct 的对齐方式。

示例：

#include <iostream>
#include <cstddef>

alignas(16) class AlignedExample {
    char c;  // 1字节
    int i;   // 4字节
};

int main() {
    std::cout << "Alignment of AlignedExample: " << alignof(AlignedExample) << std::endl; // 输出 16
    return 0;
}

上述代码中，alignas(16) 强制 AlignedExample 的对齐为 16 字节。

还可以：

alignas(8) uint8 Data[14];  // Data数组起始地址将按8字节对齐
struct alignas(16) VectorRegister4Double { ... }  // 结构体按16字节对齐 
class alignas(16) FAABBVectorized { ... }  // 类按16字节对齐

用于指定变量的内存对齐时：

// 假设我们在内存中连续声明以下变量
uint8 normalVar;          // 假设被分配到地址 1001
alignas(8) uint8 alignedData[14];  // 会被分配到第一个能被8整除的地址，比如 1008
uint8 anotherVar;        // 假设被分配到地址 1022

图解说明：

内存地址:  1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 ...  1021 1022
                                            |
                                            +----- alignedData的起始地址
                                                  (确保是8的倍数)

看下每个变量的位置：
1001: normalVar 
1008: alignedData[0]    <-- 注意这里是8的倍数
1009: alignedData[1]
1010: alignedData[2]
...
1021: alignedData[13]
1022: anotherVar

UE 的 Align 模板函数

简单地说，这个函数用于将一个值向上对齐到Alignment的倍数。比如:

Align(5, 4) = 8
Align(8, 4) = 8
Align(9, 4) = 12

这段代码确保ComponentDataPtr按照缓存行大小对齐:

uint8 Alignment = FMath::Max<uint8>(PLATFORM_CACHE_LINE_SIZE, TypeInfo.Alignment);
ComponentDataPtr = Align(ComponentDataPtr, Alignment);

假设 ComponentDataPtr 当前指向地址 1005

Alignment 是 8

那么 Align(1005, 8) 会返回 1008

// 内存示意图：
// 地址:    1000 1001 1002 1003 1004 1005 1006 1007 1008 1009
//                                    ^                 ^
//                                    |                 |
//                               当前位置          对齐后位置

其他对齐的宏：

// PLATFORM_CACHE_LINE_SIZE 通常是 64 或 128 字节（取决于CPU架构）

struct MS_ALIGN(PLATFORM_CACHE_LINE_SIZE) FAlignedDecomposedValue  // MSVC编译器使用
{
    FDecomposedValue Value;
} GCC_ALIGN(PLATFORM_CACHE_LINE_SIZE);    // GCC编译器使用

volatile

C++中volatile关键字，是为了告诉编译器，这个变量会经常修改，让编译器不要生成带优化的汇编代码，而是生成每次访问都是从内存读取和写入的汇编代码。因为编译器在优化时不会考虑一段作用域内，不考虑多线程之间，如果发现这个值在一个作用域内的代码中从来没改过，或者改过之后再也没有使用过，就很可能把这个变量存成一个常量，赋值后就再也不改了。这样就会导致即使另一个线程改了这个值，但在这个线程就完全感知不到。加了volatile，两个线程在同时访问这块内存时，比如原始值是1，第一个线程写了2，编译器不去优化了，就会生成把2写入内存的汇编指令，第二个线程在读取的时候生成的是读取内存的汇编指令，这样就能感知到这个值为2。那么可以看到这样就保证了这个变量基本是线程安全的。

但是这个关键字并不保证是原子操作。

二、构造和析构

/** Default constructor. */
[[nodiscard]] TQueue()
{
    Head = Tail = new TNode();
}

/** Destructor. */
~TQueue()
{
    while (Tail != nullptr)
    {
        TNode* Node = Tail;
        Tail = Tail->NextNode;

        delete Node;
    }
}

这个没什么好说的，不过构造的时候默认new了一个TNode

还有一个nodiscard：[[nodiscard]] 属性：这个属性可以用于函数或者返回类型。它的作用是告诉编译器：调用这个函数时，它的返回值不应被忽略。如果程序员调用了这样的函数但没有使用它的返回值，编译器会发出警告。这对于那些返回重要状态或错误码的函数特别有用。

TNode

public:
	using FElementType = T;
private:
using TNodeVolatilePtr = std::conditional_t<Mode == EQueueMode::SingleThreaded, TNode*, TNode* volatile>;
struct TNode
{
    /** Holds a pointer to the next node in the list. */
    TNodeVolatilePtr NextNode = nullptr;

    /** Holds the node's item. */
    FElementType Item;

    /** Default constructor. */
    [[nodiscard]] TNode() = default;

    /** Creates and initializes a new node. */
    [[nodiscard]] explicit TNode(const FElementType& InItem)
        : Item(InItem)
    {
    }

    /** Creates and initializes a new node. */
    [[nodiscard]] explicit TNode(FElementType&& InItem)
        : Item(MoveTemp(InItem))
    {
    }
};

构造函数中new的TNode是TQueue里的子结构体，他就是一个普通的链表，一个Item和一个NextNode，注意NextNode也加了volatile。这里没有加align(16)的原因是，TNode外部不能使用，唯一内存来源就是TQueue内部new，new操作来源于UE4的内存池FMemory（文末有截图），内存池已经保证了是对齐的[UE4/5 内存管理：Binned2内存分配器Malloc与Free过程]，所以不用强行加align(16)。

三、Enqueue

/**
 * Adds an item to the head of the queue.
 *
 * @param Item The item to add.
 * @return true if the item was added, false otherwise.
 * @note To be called only from producer thread(s).
 * @see Dequeue, Pop
 */
bool Enqueue(const FElementType& Item)
{
    TNode* NewNode = new TNode(Item);

    if (NewNode == nullptr)
    {
       return false;
    }

    TNode* OldHead;

    if constexpr (Mode == EQueueMode::Mpsc)
    {
       OldHead = (TNode*)FPlatformAtomics::InterlockedExchangePtr((void**)&Head, NewNode);
       TSAN_BEFORE(&OldHead->NextNode);
       FPlatformAtomics::InterlockedExchangePtr((void**)&OldHead->NextNode, NewNode);
    }
    else
    {
       OldHead = Head;
       Head = NewNode;
       TSAN_BEFORE(&OldHead->NextNode);
       FPlatformMisc::MemoryBarrier();
       OldHead->NextNode = NewNode;
    }

    return true;
}

上面说了，队列是链表形式的，通过代码也可以看到，队列链表的尾部是队列的Head，链表的头是队列的Tail

进队列有两个重载，还有一个是右值引用bool Enqueue(FElementType&& Item)，不过两者代码几乎一样，除了第一行右值引用版本需要TNode* NewNode = new TNode(MoveTemp(Item));

其中TSAN_BEFORE(&OldHead->NextNode);是用来查数据竞争的。具体参考：ThreadSanitizer——跟data race说再见 - 知乎 (zhihu.com)

/**
 * Enumerates concurrent queue modes.
 */
enum class EQueueMode
{
    /** Multiple-producers, single-consumer queue. */
    Mpsc,

    /** Single-producer, single-consumer queue. */
    Spsc
};

队列支持两种模式，单生产者-单消费者，多生产者-单消费者，进队列相当于是处理生产者，所以有一个枚举的判断区分开，在多生产者时可能同时操作Head成员，而前面又说到volatile不能保证原子性，所以使用InterlockedExchangePtr保证原子性。这个函数在每个平台都不一样，可以根据平台去查找对应函数，总之就是确保内存操作按顺序执行，也就是确保原子性，函数作用是把Exchange的值给Dest，返回老的Dest，这是一次做完的。

OldHead = (TNode*)FPlatformAtomics::InterlockedExchangePtr((void**)&Head, NewNode);
TSAN_BEFORE(&OldHead->NextNode);
FPlatformAtomics::InterlockedExchangePtr((void**)&OldHead->NextNode, NewNode);

所以在多生产者中就是先一次性把Head指向NewNode，OldHead变为原来的Head，然后一次性把OldHead的NextNode指向了NewNode。相当于是通过两个原子操作。OldHead是局部变量不会变。但是由于这块内存处于队列中可能会有其他线程访问，所以也需要确保原子性。

OldHead = Head;
Head = NewNode;
OldHead->NextNode = NewNode;

对单生产者来说即使不是线程安全的，但是因为只有一个生产者，不会出现多线程同时修改Head这样的操作，所以也是没有问题的。不过注意调用了FPlatformMisc::MemoryBarrier();函数，大概作用是确保此函数前后的代码顺序正常，保证了OldHead->NextNode一定是在Head先指向NewNode后才会指向NewNode。

内存屏障的详细信息可以参考：撕裂写 torn write 对齐地址写多核内存屏障

四、Dequeue

/**
 * Removes and returns the item from the tail of the queue.
 *
 * @param OutValue Will hold the returned value.
 * @return true if a value was returned, false if the queue was empty.
 * @note To be called only from consumer thread.
 * @see Empty, Enqueue, IsEmpty, Peek, Pop
 */
bool Dequeue(FElementType& OutItem)
{
    TNode* Popped = Tail->NextNode;

    if (Popped == nullptr)
    {
       return false;
    }
    
    TSAN_AFTER(&Tail->NextNode);
    OutItem = MoveTemp(Popped->Item);

    TNode* OldTail = Tail;
    Tail = Popped;
    Tail->Item = FElementType();
    delete OldTail;

    return true;
}

这里逻辑就简单很多了，因为出队列本身只有单消费者，不存在多个线程同时访问的情况，所以这里只要处理好和Enqueue的关系，保证操作是安全的就好。具体流程是让Tail指向自己的Next，并把旧数据用默认构造的对象覆盖掉，再删掉更老的无效节点，就完成任务了。可以看到这里的代码连MemoryBarrier都没有加。至于为什么不加，是因为Head只会往Next方向去移动，不会往回移动指向前一个结点，Tail永远不会超过Head，即使追上Head又因为Tail始终指向的无效节点，真正的数据是Tail->NextNode，这样就相当于根本不存在多线程访问Tail的情况，即使编译器做了乱序优化，也能保证操作结果的正确。

五、其他辅助函数

都没有复杂的逻辑都是对Tail操作也不需要原子性，甚至没有内存屏障，这些函数有：Empty, IsEmpty, Peek, Pop

让我们携手共创奇迹！

目录