Thank Zhihao Tao for your hard work. The document spent countless nights and weekends, using his hard work to make it convenient for everyone.
If you have any questions, please send a email to zhihao.tao@outlook.com

1. 概念

流是从特定源发送到特定单播、泛播或多播目的地的数据包序列。流可以由特定传输连接或媒体流中的所有数据包组成。但是，流不一定是1:1映射到传输连接上。

传统上，流是基于源地址和目的地址、源端口和目的端口，及传输协议类型的5元组来区分。

参见RFC-3697

通常我们也指从一端到另一端的一次网络数据传输过程。其包含连接的建立，数据的传输。一条流往往指的是一次完成的数据传输，其包含了该过程中所有的数据包。但是一个连接可以有多个流，即在连接建立之后，可以有多次的数据传输。

这里的流，更多的强调是一次数据传输过程。重在过程及这个过程中的数据。

2. 流的获取

在suricata，流是非常重要的。其在suricata内部组织数据的方式中发挥了很大作用。流与连接有点类似，只是流更一般。所有具有相同五元祖(协议、源IP、目标IP、源端口、目标端口)的数据包都属于同一个流。属于流的包在内部连接到它。

实现中vlan_id及recursion_level同样要考虑

struct {
    uint32_t addrs[2];
    uint16_t ports[2];
    uint16_t proto;
    uint16_t recur;
    uint16_t vlan_id[2];
};

flow流程

流
元祖

元祖

2.1 流的分配

当suricata收到一个特定协议（ipv6, icmp, sctp,tcp,udp）的packet后，会计算一个流的hash值，设置PKT_WANTS_FLOW标志。

void FlowSetupPacket(Packet *p)
{
    p->flags |= PKT_WANTS_FLOW;
    p->flow_hash = FlowGetHash(p);
}

FlowWorker会基于PKT_WANTS_FLOW标志，进行流的查找或分配。

static TmEcode FlowWorker(ThreadVars *tv, Packet *p, void *data, PacketQueue *preq, PacketQueue *unused)
{
    /* handle Flow */
    if (p->flags & PKT_WANTS_FLOW) {
        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_FLOW);

        FlowHandlePacket(tv, fw->dtv, p);
        if (likely(p->flow != NULL)) {
            DEBUG_ASSERT_FLOW_LOCKED(p->flow);
            if (FlowUpdate(tv, fw, p) == TM_ECODE_DONE)

通过对流进行哈希检索。
- 查找包含流指针的哈希桶。

void FlowHandlePacket(ThreadVars *tv, DecodeThreadVars *dtv, Packet *p)
{
...
    Flow *f = FlowGetFlowFromHash(tv, dtv, p, &p->flow);
...
    p->flags |= PKT_HAS_FLOW;
    return;
}

Flow *FlowGetFlowFromHash(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *p, Flow **dest)
{
    Flow *f = NULL;

    /* get our hash bucket and lock it */
    const uint32_t hash = p->flow_hash;
    FlowBucket *fb = &flow_hash[hash % flow_config.hash_size];
    FBLOCK_LOCK(fb);

如果桶内没有任何流，分配一条新的流。

    if (fb->head == NULL) {
        f = FlowGetNew(tv, dtv, p);
        if (f == NULL) {
            FBLOCK_UNLOCK(fb);
            return NULL;
        }

        /* flow is locked */
        fb->head = f;
        fb->tail = f;

        /* got one, now lock, initialize and return */
        FlowInit(f, p);
        f->flow_hash = hash;
        f->fb = fb;
        FlowUpdateState(f, FLOW_STATE_NEW);

        FlowReference(dest, f);

        FBLOCK_UNLOCK(fb);
        return f;
    }

将包与找到的流进行比较，看它是否是我们需要的流。
- 如果不是对应的流，遍历列表，直到找到正确的流。

    /* see if this is the flow we are looking for */
    if (FlowCompare(f, p) == 0) {
        Flow *pf = NULL; /* previous flow */

        while (f) {
...
            if (FlowCompare(f, p) != 0) {
...
    return f;
}

3. 流的队列

3.1 spare队列

spare队列存储着备用的、未使用的、预分配的流。

FlowQueue flow_spare_q;
FlowQueueInit(&flow_spare_q);

3.1.1 入队

flow_spare_q的入队的操作，主要发生在：

初始化时的预分配

void FlowInitConfig(char quiet)
{
...
    /* pre allocate flows */
    for (i = 0; i < flow_config.prealloc; i++) {
...
        Flow *f = FlowAlloc();
        if (f == NULL) {
            SCLogError(SC_ERR_FLOW_INIT, "preallocating flow failed: %s", strerror(errno));
            exit(EXIT_FAILURE);
        }

        FlowEnqueue(&flow_spare_q,f);
    }

流管理检查时的补足

int FlowUpdateSpareFlows(void)
{
...
    if (len < flow_config.prealloc) {
        toalloc = flow_config.prealloc - len;

        uint32_t i;
        for (i = 0; i < toalloc; i++) {
            Flow *f = FlowAlloc();
            if (f == NULL)
                return 0;

            FlowEnqueue(&flow_spare_q,f);
        }

流的回收，即从flow_recycle_q到flow_spare_q

static TmEcode FlowRecycler(ThreadVars *th_v, void *thread_data)
{
...
       while ((f = FlowDequeue(&flow_recycle_q)) != NULL) {
                FLOWLOCK_WRLOCK(f);

                (void)OutputFlowLog(th_v, ftd->output_thread_data, f);

                FlowClearMemory (f, f->protomap);
                FLOWLOCK_UNLOCK(f);
                FlowMoveToSpare(f);
                recycled_cnt++;
            }
        }

3.1.2 出队

flow_spare_q的出队的操作，主要发生在：

流的分配

static Flow *FlowGetNew(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *p)
{
...
    /* get a flow from the spare queue */
    f = FlowDequeue(&flow_spare_q);

Flow *FlowGetFromFlowKey(FlowKey *key, struct timespec *ttime, const uint32_t hash)
{
...
    /* No existing flow so let's get one new */
    f = FlowDequeue(&flow_spare_q);

释放多余的分配的流

int FlowUpdateSpareFlows(void)
{
...
    } else if (len > flow_config.prealloc) {
        tofree = len - flow_config.prealloc;

        uint32_t i;
        for (i = 0; i < tofree; i++) {
            /* FlowDequeue locks the queue */
            Flow *f = FlowDequeue(&flow_spare_q);
            if (f == NULL)
                return 1;

进程退出

void FlowShutdown(void)
{
...
    /* free queues */
    while((f = FlowDequeue(&flow_spare_q))) {
        FlowFree(f);
    }

3.2 recycle队列

recycle队列存储着将传递到清理、日志线程的流。

FlowQueue flow_recycle_q;
FlowQueueInit(&flow_recycle_q);

3.2.1 入队

flow_recycle_q的入队的操作，主要发生在：

流超时进行回收

static uint32_t FlowManagerHashRowTimeout(Flow *f, struct timeval *ts,
...
            /* no one is referring to this flow, use_cnt 0, removed from hash
             * so we can unlock it and pass it to the flow recycler */
            FLOWLOCK_UNLOCK(f);
            FlowEnqueue(&flow_recycle_q, f);

进程退出时，对流hash桶中的流进行回收处理

static uint32_t FlowManagerHashRowCleanup(Flow *f)
{
...
        /* no one is referring to this flow, use_cnt 0, removed from hash
         * so we can unlock it and move it to the recycle queue. */
        FLOWLOCK_UNLOCK(f);

        FlowEnqueue(&flow_recycle_q, f);

3.2.2 出队

flow_recycle_q的出队的操作，主要发生在：

流的回收，即从flow_recycle_q到flow_spare_q

static TmEcode FlowRecycler(ThreadVars *th_v, void *thread_data)
{
...
       while ((f = FlowDequeue(&flow_recycle_q)) != NULL) {
                FLOWLOCK_WRLOCK(f);

                (void)OutputFlowLog(th_v, ftd->output_thread_data, f);

                FlowClearMemory (f, f->protomap);
                FLOWLOCK_UNLOCK(f);
                FlowMoveToSpare(f);
                recycled_cnt++;
            }
        }

进程退出

void FlowShutdown(void)
{
...
    while((f = FlowDequeue(&flow_recycle_q))) {
        FlowFree(f);
    }

Suricata-流的获取