Suricata 3.2.1 源码阅读笔记：数据包队列

小咪咪 2022-06-14 06:24 327阅读 0赞

## 简介 ##

Suricata中使用队列来缓存数据包，包括缓存线程模块内部新产生数据包的线程内队列，以及线程之间用来传递数据包的线程间队列。

用于表示数据包队列的结构体为PacketQueue，其定义如下（省略了调试相关字段）：

typedef struct PacketQueue_ {
        Packet *top;
        Packet *bot;
        uint32_t len;
    #ifdef DBG_PERF
        uint32_t dbg_maxlen;
    #endif /* DBG_PERF */
        SCMutex mutex_q;
        SCCondT cond_q;
    } PacketQueue;

## 线程内队列 ##

线程内队列与线程内的每个slot相绑定，其又可细分为pre队列和post队列，定义如下：

typedef struct TmSlot_ {
        ...
        /* queue filled by the SlotFunc with packets that will
     * be processed futher _before_ the current packet.
     * The locks in the queue are NOT used */
        PacketQueue slot_pre_pq;
        
        /* queue filled by the SlotFunc with packets that will
     * be processed futher _after_ the current packet. The
     * locks in the queue are NOT used */
        PacketQueue slot_post_pq;
    }

关于pre队列和post队列之间的用法区别，注释中已经说得很清楚了。具体使用方式可参考前面的关于数据包源的笔记。

##     ##

## 线程间队列 ##

Suricata中使用了一个全局数组作为所有的线程间队列的存储，定义为：

PacketQueue trans\_q\[256\];

这里为什么使用全局静态数组，而不是更直观更省内存的在运行时按需动态分配呢？主要是出于性能考虑。按需动态分配，就要求将队列组织为链表（当然，动态数组也行，但实现相对复杂，且完全没有必要），而为了支持队列负载均衡，需要将数据包在各个数据包之间进行分配，像hash这种分配方式是需要能够按照队列索引进行随机访问的。并且，由于所需队列个数在编译时就已经确定，因此使用全局静态数据是最合适的了。

然而，由于PacketQueue结构只是单纯用来存储，Suricata中另外使用了一个Tmq结构体实现对队列的管理。

typedef struct Tmq_ { char *name; /* 队列名字 */ uint16_t id; /* 对应的队列存储在trans_q中的索引 */ uint16_t reader_cnt; /* 读这个队列中数据的线程数 */ uint16_t writer_cnt; /* 向这个队列写数据的线程数 */ uint8_t q_type; /* 队列类型， 0为数据包队列，1为数据队列 */ } Tmq;

同样地，与trans\_q对应，Suricata也使用了一个全局数组作为Tmq的存储，与之相关的定义包括：

#define TMQ_MAX_QUEUES 256 static uint16_t tmq_id = 0; static Tmq tmqs[TMQ_MAX_QUEUES];

相应地，创建队列的函数TmqCreateQueue流程为：

1.  获取tmqs\[tmq\_id\]的地址，作为返回的Tmq指针。
2.  设置该Tmq的name、id、type，其中id设为tmq\_id，然后递增tmq\_id。

该函数在TmThreadCreate中被直接或间接调用。

*  对于线程的inq，首先查inq\_name对应的队列是否已存在，若不存在则直接调用TmqCreateQueue创建。
 *  对于线程的outq，则首先要看outqh的OutHandlerCtxSetup是否设置，若设置了则会调用该函数去间接创建队列，否则跟创建inq类似。间接创建目前只有一种情况：使用"flow"作为线程的outqh->TmqhOutputFlowSetupCtx->StoreQueueId解析队列名字符串->TmqCreateQueue。

下面的小节都是关于线程间队列的。

## 队列数量 ##

Suricata中究竟会创造多少个线程间队列呢？这取决于系统的运行模式。

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""><span style="">运行模式&nbsp;&nbsp;</span></td> 
   <td style=""><span style="">线程数</span></td> 
   <td style=""><span style="">队列数</span></td> 
   <td style=""><span style="">说明</span></td> 
  </tr> 
  <tr style=""> 
   <td style="">single</td> 
   <td style="">1</td> 
   <td style="">0</td> 
   <td style="">单线程模式，不需要队列</td> 
  </tr> 
  <tr style=""> 
   <td style="">workers</td> 
   <td style="">IDS: 取决于接口配置&nbsp;<br style=""> IPS: 取决于nqueue</td> 
   <td style="">0</td> 
   <td style="">多线程工人模式，但每个线程做所有事（从抓包到输出日志），因此也不需要队列</td> 
  </tr> 
  <tr style=""> 
   <td style="">auto</td> 
   <td style="">IDS：多个检测线程，其它各类线程均一个&nbsp;<br style=""> IPS：一个解码/重组线程，其它各类线程都是多个</td> 
   <td style="">各类线程间均1个</td> 
   <td style="">多线程自动模式，同一个流的数据包可能被多个检测线程处理。</td> 
  </tr> 
  <tr style=""> 
   <td style="">autofp</td> 
   <td style="">IDS：多个包获取线程，多个其他线程&nbsp;<br style=""> IPS：与IDS下类似，但裁决/响应为单独线程</td> 
   <td style="">多个pickup队列，其余各1个</td> 
   <td style="">多线程自动流绑定负载均衡模式（auto flow pinned load balancing），每个流的包只会被同一个检测线程处理（flow pinned）。其中pickup对立个数与其reader线程数量一致，因为每个线程只能有1个inq。</td> 
  </tr> 
 </tbody> 
</table>

## 队列Handler ##

现在队列有了，线程也创建好了，那么线程与数据包队列之间是如何传递数据包的呢？

例如，前面的笔记中已经记录了包含ReceivePcap模块和DecodePcap模块的线程的执行流程，在这两个模块处理完数据包后，接下来这些数据包怎么送到pickup队列去呢？

这时候，队列Handler就起作用了。每个线程会在其ThreadVars中绑定两个Handler：

*  tmqh\_in：在初始化时绑定为inqh的InHander，用于从上一级队列中获取数据包。
 *  tmqh\_out：在初始化时绑定为outqh的OutHander，用于将处理后的数据包送往下一级队列。

系统初始化时会调用TmqhSetup注册所有队列handler，类似包括：

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""><span style="">类型</span></td> 
   <td style=""><span style="">说明</span></td> 
  </tr> 
  <tr style=""> 
   <td style="">simple</td> 
   <td style="">简单地从inq获取数据包，线程处理完后将包送往唯一的outq。</td> 
  </tr> 
  <tr style=""> 
   <td style="">packetpool*</td> 
   <td style="">从数据包池中获取数据包，将数据包放回数据包池。见下面注释。</td> 
  </tr> 
  <tr style=""> 
   <td style="">flow</td> 
   <td style="">用于autofp模式的handler，实现流的绑定和负载均衡。</td> 
  </tr> 
 </tbody> 
</table>

> 注：目前Suricata中关于packet pool的部分比较乱。例如，虽然包获取线程的inq和inqh都设为"packetpool"，但首先来说，对于包获取线程inq是没意义的，因此这里这么设置给程序中带来很多特殊处理（TmThreadCreate），我认为设为NULL其实更好理解；其次，inqh这么设可能本意是想统一包获取过程，即不区分抓包或者从队列入包，都统一从tmqh\_in来得到包，然而实际上Suricata中却做了特殊区分，对于包获取线程其slot类型为"pktacqloop"，相应的执行函数TmThreadsSlotPktAcqLoop并没有调用tmqh\_in，而是调用其第一个slot中模块的PktAcqLoop。另外，TmqhOutputPacketpool函数作为"packetpool"这个handler的OutHandler，本应该只能由线程的tmqh\_out来调用，但实际上程序中却到处都有调用（主要目的是跳过后续的线程处理，直接返回到数据包池中）。总之，这块是我目前感觉到Suricata中最ugly的。。。

由于autofp是目前几乎全部运行模式类型下的默认模式（参见RunModeDispatch），下面将重点介绍"flow"队列handler的实现。

## "flow" ##

首先，TmqhFlowRegister完成"flow"的注册，包括InHandler、OutHandler、OutHandlerCtxSetup/Free等。

InHandler：由于每个线程只有一个inq，因此"flow"的输入函数TmqhInputFlow与"simple"是一致的，直接从inq取包就可以了。

OutHandler："flow"有多个OutHandle，分别对应不同的负载均衡策略，目前支持的策略如下：

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style=""><span style="">策略</span></td> 
   <td style=""><span style="">对应函数</span></td> 
   <td style=""><span style="">说明</span></td> 
  </tr> 
  <tr style=""> 
   <td style="">round-robin</td> 
   <td style="">TmqhOutputFlowRoundRobin</td> 
   <td style="">轮转调度：按照轮转方式依次选择下一个队列</td> 
  </tr> 
  <tr style=""> 
   <td style="">active-packets</td> 
   <td style="">TmqhOutputFlowHash[两种模式]</td> 
   <td style="">最短匹配 ：选择当前最短（数据包最少）的队列</td> 
  </tr> 
  <tr style=""> 
   <td style="">hash</td> 
   <td style="">TmqhOutputFlowHash</td> 
   <td style=""> <p>随机哈希：按照流指针的Hash值选择队列</p> </td> 
  </tr> 
 </tbody> 
</table>

<table style=""> 
 <tbody style=""> 
  <tr style=""> 
   <td style="">ippair</td> 
   <td style="">TmqhOutputFlowIPPair</td> 
   <td style="">流管理： 二元组相同归属一个队列</td> 
  </tr> 
 </tbody> 
</table>

Suricata支持用户通过"autofp-scheduler"参数对以上策略进行选择，默认为active-packets。

OutHandlerCtxSetup/Free：这个是"flow"模块目前独有的。使用"flow"作为outqh的线程都会有多个输出队列（如pickup1, pickup2, …），因此直接调用TmqCreateQueue创建ThreadVars的outq不太方便，因此Suricata就把这个任务下放到outqh的OutHandlerCtxSetup去做了。

"flow"对应的OutHandlerCtxSetup为TmqhOutputFlowSetupCtx，函数原型为：

/**
     * \brief setup the queue handlers ctx
     *
     * Parses a comma separated string "queuename1,queuename2,etc"
     * and sets the ctx up to devide flows over these queue's.
     *
     * \param queue_str comma separated string with output queue names
     *
     * \retval ctx queues handlers ctx or NULL in error
     */
    void *TmqhOutputFlowSetupCtx(char *queue_str)

函数的功能、参数及返回值在上面的注释已经说得比较清楚了。具体实现流程为：

1.  创建一个TmqhFlowCtx，该结构体将被存储在ThreadVars的outctx中，并在线程调用tmqh\_out时被"flow"的OutHandler所利用。
2.  循环解析queue\_str，将每个qname传给StoreQueueId，以创建相应名字队列并存储到ctx中。

下面介绍各个OutHandler的实现。首先，需要知道Flow中有一个字段是专门用于autofp的：

typedef struct Flow_ { ... /** flow queue id, used with autofp */ SC_ATOMIC_DECLARE(int, autofp_tmqh_flow_qid); }

这个字段用来记录这个流所被绑定到的输出队列的ID，在FLOW\_INITIALIZE中被设为-1，表示还未被绑定。而这个绑定关系，就是由下面这些OutHandler所确定的：

TmqhOutputFlowRoundRobin

1.  首先需要确定一个该包将被送往的qid。
2.  若数据包有对应的flow，则首先获取该flow的autofp\_tmqh\_flow\_qid，若为-1说明还没绑定，就将qid和autofp\_tmqh\_flow\_qid都设为round robin的下一个值（ctx->round\_robin\_idx + 1 % ctx->size），并将。相应队列的total\_flows计数器+1（用于统计负载分配情况）。
3.  若没有对应flow，则简单的使用（ctx->last + 1） % ctx->size，即将所有这种包单独进行round robin。其他OutHandler也都是这么处理的。注意，这种包实际上很少，比如ICMPv4的报错包，或者内存不够导致无法分配flow。
4.  确定好qid了，就把包enqueue到对应PacketQueue就行了，然后cond\_signal这个队列的cond，以唤醒等待这个队列的线程。

下面两个Handler整体流程类似，只是第2步为flow绑定qid的方式不同。

TmqhOutputFlowActivePackets：qid设置为所有输出队列中len最小的队列。

TmqhOutputFlowHash：qid设置为 ((p->flow)>>7 % ctx->size)。个人感觉这样利用指针的hash分布会不均匀，毕竟连续分配的指针的高位很可能是相同的。

最后需要注意一点的是，若两个（组）线程通过数据包队列对接，则前面线程的outqh应该和后面线程的inqh应该保持一致。