path: root/src/flow-worker.c

/* Copyright (C) 2016-2020 Open Information Security Foundation
 *
 * You can copy, redistribute or modify this Program under the terms of
 * the GNU General Public License version 2 as published by the Free
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

/**
 * \file
 *
 * \author Victor Julien <victor@inliniac.net>
 *
 * Flow Workers are single thread modules taking care of (almost)
 * everything related to packets with flows:
 *
 * - Lookup/creation
 * - Stream tracking, reassembly
 * - Applayer update
 * - Detection
 *
 * This all while holding the flow lock.
 */

#include "suricata-common.h"
#include "suricata.h"

#include "action-globals.h"
#include "packet.h"
#include "decode.h"
#include "detect.h"
#include "stream-tcp.h"
#include "app-layer.h"
#include "detect-engine.h"
#include "output.h"
#include "app-layer-parser.h"
#include "app-layer-frames.h"

#include "util-profiling.h"
#include "util-validate.h"
#include "util-time.h"
#include "tmqh-packetpool.h"

#include "flow-util.h"
#include "flow-manager.h"
#include "flow-timeout.h"
#include "flow-spare-pool.h"
#include "flow-worker.h"

typedef DetectEngineThreadCtx *DetectEngineThreadCtxPtr;

typedef struct FlowTimeoutCounters {
    uint32_t flows_aside_needs_work;
    uint32_t flows_aside_pkt_inject;
} FlowTimeoutCounters;

typedef struct FlowWorkerThreadData_ {
    DecodeThreadVars *dtv;

    union {
        StreamTcpThread *stream_thread;
        void *stream_thread_ptr;
    };

    SC_ATOMIC_DECLARE(DetectEngineThreadCtxPtr, detect_thread);

    void *output_thread; /* Output thread data. */
    void *output_thread_flow; /* Output thread data. */

    uint16_t local_bypass_pkts;
    uint16_t local_bypass_bytes;
    uint16_t both_bypass_pkts;
    uint16_t both_bypass_bytes;
    /** Queue to put pseudo packets that have been created by the stream (RST response) and by the
     * flush logic following a protocol change. */
    PacketQueueNoLock pq;
    FlowLookupStruct fls;

    struct {
        uint16_t flows_injected;
        uint16_t flows_injected_max;
        uint16_t flows_removed;
        uint16_t flows_aside_needs_work;
        uint16_t flows_aside_pkt_inject;
    } cnt;
    FlowEndCounters fec;

} FlowWorkerThreadData;

static void FlowWorkerFlowTimeout(ThreadVars *tv, Packet *p, FlowWorkerThreadData *fw, void *detect_thread);
Packet *FlowForceReassemblyPseudoPacketGet(int direction, Flow *f, TcpSession *ssn);

/**
 * \internal
 * \brief Forces reassembly for flow if it needs it.
 *
 *        The function requires flow to be locked beforehand.
 *
 * \param f Pointer to the flow.
 *
 * \retval cnt number of packets injected
 */
static int FlowFinish(ThreadVars *tv, Flow *f, FlowWorkerThreadData *fw, void *detect_thread)
{
    Packet *p1 = NULL, *p2 = NULL;
    const int server = f->ffr_tc;
    const int client = f->ffr_ts;

    /* Get the tcp session for the flow */
    TcpSession *ssn = (TcpSession *)f->protoctx;

    /* The packets we use are based on what segments in what direction are
     * unprocessed.
     * p1 if we have client segments for reassembly purpose only.  If we
     * have no server segments p2 can be a toserver packet with dummy
     * seq/ack, and if we have server segments p2 has to carry out reassembly
     * for server segment as well, in which case we will also need a p3 in the
     * toclient which is now dummy since all we need it for is detection */

    /* insert a pseudo packet in the toserver direction */
    if (client == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION) {
        p1 = FlowForceReassemblyPseudoPacketGet(0, f, ssn);
        if (p1 == NULL) {
            return 0;
        }
        PKT_SET_SRC(p1, PKT_SRC_FFR);

        if (server == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION) {
            p2 = FlowForceReassemblyPseudoPacketGet(1, f, ssn);
            if (p2 == NULL) {
                FlowDeReference(&p1->flow);
                TmqhOutputPacketpool(NULL, p1);
                return 0;
            }
            PKT_SET_SRC(p2, PKT_SRC_FFR);
            p2->flowflags |= FLOW_PKT_LAST_PSEUDO;
        } else {
            p1->flowflags |= FLOW_PKT_LAST_PSEUDO;
        }
    } else {
        if (server == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION) {
            p1 = FlowForceReassemblyPseudoPacketGet(1, f, ssn);
            if (p1 == NULL) {
                return 0;
            }
            PKT_SET_SRC(p1, PKT_SRC_FFR);
            p1->flowflags |= FLOW_PKT_LAST_PSEUDO;
        } else {
            /* impossible */
            BUG_ON(1);
        }
    }
    f->flags |= FLOW_TIMEOUT_REASSEMBLY_DONE;

    FlowWorkerFlowTimeout(tv, p1, fw, detect_thread);
    PacketPoolReturnPacket(p1);
    if (p2) {
        FlowWorkerFlowTimeout(tv, p2, fw, detect_thread);
        PacketPoolReturnPacket(p2);
        return 2;
    }
    return 1;
}

extern uint32_t flow_spare_pool_block_size;

/** \param[in] max_work Max flows to process. 0 if unlimited. */
static void CheckWorkQueue(ThreadVars *tv, FlowWorkerThreadData *fw, FlowTimeoutCounters *counters,
        FlowQueuePrivate *fq, const uint32_t max_work)
{
    FlowQueuePrivate ret_queue = { NULL, NULL, 0 };
    uint32_t i = 0;
    Flow *f;
    while ((f = FlowQueuePrivateGetFromTop(fq)) != NULL) {
        FLOWLOCK_WRLOCK(f);
        f->flow_end_flags |= FLOW_END_FLAG_TIMEOUT; //TODO emerg

        if (f->proto == IPPROTO_TCP) {
            if (!(f->flags & (FLOW_TIMEOUT_REASSEMBLY_DONE | FLOW_ACTION_DROP)) &&
                    !FlowIsBypassed(f) && FlowForceReassemblyNeedReassembly(f) == 1 &&
                    f->ffr != 0) {
                /* read detect thread in case we're doing a reload */
                void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);
                int cnt = FlowFinish(tv, f, fw, detect_thread);
                counters->flows_aside_pkt_inject += cnt;
                counters->flows_aside_needs_work++;
            }
        }

        /* no one is referring to this flow, removed from hash
         * so we can unlock it and pass it to the flow recycler */

        if (fw->output_thread_flow != NULL)
            (void)OutputFlowLog(tv, fw->output_thread_flow, f);

        FlowEndCountersUpdate(tv, &fw->fec, f);
        if (f->proto == IPPROTO_TCP && f->protoctx != NULL) {
            StatsDecr(tv, fw->dtv->counter_tcp_active_sessions);
        }
        StatsDecr(tv, fw->dtv->counter_flow_active);

        FlowClearMemory (f, f->protomap);
        FLOWLOCK_UNLOCK(f);

        if (fw->fls.spare_queue.len >= (flow_spare_pool_block_size * 2)) {
            FlowQueuePrivatePrependFlow(&ret_queue, f);
            if (ret_queue.len == flow_spare_pool_block_size) {
                FlowSparePoolReturnFlows(&ret_queue);
            }
        } else {
            FlowQueuePrivatePrependFlow(&fw->fls.spare_queue, f);
        }

        if (max_work != 0 && ++i == max_work)
            break;
    }
    if (ret_queue.len > 0) {
        FlowSparePoolReturnFlows(&ret_queue);
    }

    StatsAddUI64(tv, fw->cnt.flows_removed, (uint64_t)i);
}

/** \brief handle flow for packet
 *
 *  Handle flow creation/lookup
 */
static inline TmEcode FlowUpdate(ThreadVars *tv, FlowWorkerThreadData *fw, Packet *p)
{
    FlowHandlePacketUpdate(p->flow, p, tv, fw->dtv);

    int state = p->flow->flow_state;
    switch (state) {
#ifdef CAPTURE_OFFLOAD
        case FLOW_STATE_CAPTURE_BYPASSED: {
            StatsAddUI64(tv, fw->both_bypass_pkts, 1);
            StatsAddUI64(tv, fw->both_bypass_bytes, GET_PKT_LEN(p));
            Flow *f = p->flow;
            FlowDeReference(&p->flow);
            FLOWLOCK_UNLOCK(f);
            return TM_ECODE_DONE;
        }
#endif
        case FLOW_STATE_LOCAL_BYPASSED: {
            StatsAddUI64(tv, fw->local_bypass_pkts, 1);
            StatsAddUI64(tv, fw->local_bypass_bytes, GET_PKT_LEN(p));
            Flow *f = p->flow;
            FlowDeReference(&p->flow);
            FLOWLOCK_UNLOCK(f);
            return TM_ECODE_DONE;
        }
        default:
            return TM_ECODE_OK;
    }
}

static TmEcode FlowWorkerThreadDeinit(ThreadVars *tv, void *data);

static TmEcode FlowWorkerThreadInit(ThreadVars *tv, const void *initdata, void **data)
{
    FlowWorkerThreadData *fw = SCCalloc(1, sizeof(*fw));
    if (fw == NULL)
        return TM_ECODE_FAILED;

    SC_ATOMIC_INITPTR(fw->detect_thread);
    SC_ATOMIC_SET(fw->detect_thread, NULL);

    fw->local_bypass_pkts = StatsRegisterCounter("flow_bypassed.local_pkts", tv);
    fw->local_bypass_bytes = StatsRegisterCounter("flow_bypassed.local_bytes", tv);
    fw->both_bypass_pkts = StatsRegisterCounter("flow_bypassed.local_capture_pkts", tv);
    fw->both_bypass_bytes = StatsRegisterCounter("flow_bypassed.local_capture_bytes", tv);

    fw->cnt.flows_aside_needs_work = StatsRegisterCounter("flow.wrk.flows_evicted_needs_work", tv);
    fw->cnt.flows_aside_pkt_inject = StatsRegisterCounter("flow.wrk.flows_evicted_pkt_inject", tv);
    fw->cnt.flows_removed = StatsRegisterCounter("flow.wrk.flows_evicted", tv);
    fw->cnt.flows_injected = StatsRegisterCounter("flow.wrk.flows_injected", tv);
    fw->cnt.flows_injected_max = StatsRegisterMaxCounter("flow.wrk.flows_injected_max", tv);

    fw->fls.dtv = fw->dtv = DecodeThreadVarsAlloc(tv);
    if (fw->dtv == NULL) {
        FlowWorkerThreadDeinit(tv, fw);
        return TM_ECODE_FAILED;
    }

    /* setup TCP */
    if (StreamTcpThreadInit(tv, NULL, &fw->stream_thread_ptr) != TM_ECODE_OK) {
        FlowWorkerThreadDeinit(tv, fw);
        return TM_ECODE_FAILED;
    }

    if (DetectEngineEnabled()) {
        /* setup DETECT */
        void *detect_thread = NULL;
        if (DetectEngineThreadCtxInit(tv, NULL, &detect_thread) != TM_ECODE_OK) {
            FlowWorkerThreadDeinit(tv, fw);
            return TM_ECODE_FAILED;
        }
        SC_ATOMIC_SET(fw->detect_thread, detect_thread);
    }

    /* Setup outputs for this thread. */
    if (OutputLoggerThreadInit(tv, initdata, &fw->output_thread) != TM_ECODE_OK) {
        FlowWorkerThreadDeinit(tv, fw);
        return TM_ECODE_FAILED;
    }
    if (OutputFlowLogThreadInit(tv, NULL, &fw->output_thread_flow) != TM_ECODE_OK) {
        SCLogError("initializing flow log API for thread failed");
        FlowWorkerThreadDeinit(tv, fw);
        return TM_ECODE_FAILED;
    }

    DecodeRegisterPerfCounters(fw->dtv, tv);
    AppLayerRegisterThreadCounters(tv);
    FlowEndCountersRegister(tv, &fw->fec);

    /* setup pq for stream end pkts */
    memset(&fw->pq, 0, sizeof(PacketQueueNoLock));
    *data = fw;
    return TM_ECODE_OK;
}

static TmEcode FlowWorkerThreadDeinit(ThreadVars *tv, void *data)
{
    FlowWorkerThreadData *fw = data;

    DecodeThreadVarsFree(tv, fw->dtv);

    /* free TCP */
    StreamTcpThreadDeinit(tv, (void *)fw->stream_thread);

    /* free DETECT */
    void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);
    if (detect_thread != NULL) {
        DetectEngineThreadCtxDeinit(tv, detect_thread);
        SC_ATOMIC_SET(fw->detect_thread, NULL);
    }

    /* Free output. */
    OutputLoggerThreadDeinit(tv, fw->output_thread);
    OutputFlowLogThreadDeinit(tv, fw->output_thread_flow);

    /* free pq */
    BUG_ON(fw->pq.len);

    Flow *f;
    while ((f = FlowQueuePrivateGetFromTop(&fw->fls.spare_queue)) != NULL) {
        FlowFree(f);
    }

    SCFree(fw);
    return TM_ECODE_OK;
}

TmEcode Detect(ThreadVars *tv, Packet *p, void *data);
TmEcode StreamTcp (ThreadVars *, Packet *, void *, PacketQueueNoLock *pq);

static inline void UpdateCounters(ThreadVars *tv,
        FlowWorkerThreadData *fw, const FlowTimeoutCounters *counters)
{
    if (counters->flows_aside_needs_work) {
        StatsAddUI64(tv, fw->cnt.flows_aside_needs_work,
                (uint64_t)counters->flows_aside_needs_work);
    }
    if (counters->flows_aside_pkt_inject) {
        StatsAddUI64(tv, fw->cnt.flows_aside_pkt_inject,
                (uint64_t)counters->flows_aside_pkt_inject);
    }
}

/** \brief update stream engine
 *
 *  We can be called from both the flow timeout path as well as from the
 *  "real" traffic path. If in the timeout path any additional packets we
 *  forge for flushing pipelines should not leave our scope. If the original
 *  packet is real (or related to a real packet) we need to push the packets
 *  on, so IPS logic stays valid.
 */
static inline void FlowWorkerStreamTCPUpdate(ThreadVars *tv, FlowWorkerThreadData *fw, Packet *p,
        void *detect_thread, const bool timeout)
{
    FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_STREAM);
    StreamTcp(tv, p, fw->stream_thread, &fw->pq);
    FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_STREAM);

    // this is the first packet that sets no payload inspection
    bool setting_nopayload =
            p->flow->alparser &&
            AppLayerParserStateIssetFlag(p->flow->alparser, APP_LAYER_PARSER_NO_INSPECTION) &&
            !(p->flags & PKT_NOPAYLOAD_INSPECTION);
    if (FlowChangeProto(p->flow) || setting_nopayload) {
        StreamTcpDetectLogFlush(tv, fw->stream_thread, p->flow, p, &fw->pq);
        if (setting_nopayload) {
            FlowSetNoPayloadInspectionFlag(p->flow);
        }
        AppLayerParserStateSetFlag(p->flow->alparser, APP_LAYER_PARSER_EOF_TS);
        AppLayerParserStateSetFlag(p->flow->alparser, APP_LAYER_PARSER_EOF_TC);
    }

    /* Packets here can safely access p->flow as it's locked */
    SCLogDebug("packet %"PRIu64": extra packets %u", p->pcap_cnt, fw->pq.len);
    Packet *x;
    while ((x = PacketDequeueNoLock(&fw->pq))) {
        SCLogDebug("packet %"PRIu64" extra packet %p", p->pcap_cnt, x);

        if (detect_thread != NULL) {
            FLOWWORKER_PROFILING_START(x, PROFILE_FLOWWORKER_DETECT);
            Detect(tv, x, detect_thread);
            FLOWWORKER_PROFILING_END(x, PROFILE_FLOWWORKER_DETECT);
        }

        OutputLoggerLog(tv, x, fw->output_thread);

        FramesPrune(x->flow, x);
        /*  Release tcp segments. Done here after alerting can use them. */
        FLOWWORKER_PROFILING_START(x, PROFILE_FLOWWORKER_TCPPRUNE);
        StreamTcpPruneSession(
                x->flow, x->flowflags & FLOW_PKT_TOSERVER ? STREAM_TOSERVER : STREAM_TOCLIENT);
        FLOWWORKER_PROFILING_END(x, PROFILE_FLOWWORKER_TCPPRUNE);

        /* no need to keep a flow ref beyond this point */
        FlowDeReference(&x->flow);

        /* no further work to do for this pseudo packet, so we can return
         * it to the pool immediately. */
        if (timeout) {
            PacketPoolReturnPacket(x);
        } else {
            /* to support IPS verdict logic, in the non-timeout case we need to do a bit more */
            TmqhOutputPacketpool(tv, x);
        }
    }
    if (FlowChangeProto(p->flow) && p->flow->flags & FLOW_ACTION_DROP) {
        // in case f->flags & FLOW_ACTION_DROP was set by one of the dequeued packets
        PacketDrop(p, ACTION_DROP, PKT_DROP_REASON_FLOW_DROP);
    }
}

static void FlowWorkerFlowTimeout(ThreadVars *tv, Packet *p, FlowWorkerThreadData *fw,
        void *detect_thread)
{
    DEBUG_VALIDATE_BUG_ON(p->pkt_src != PKT_SRC_FFR);

    SCLogDebug("packet %"PRIu64" is TCP. Direction %s", p->pcap_cnt, PKT_IS_TOSERVER(p) ? "TOSERVER" : "TOCLIENT");
    DEBUG_VALIDATE_BUG_ON(!(p->flow && PKT_IS_TCP(p)));
    DEBUG_ASSERT_FLOW_LOCKED(p->flow);

    /* handle TCP and app layer */
    FlowWorkerStreamTCPUpdate(tv, fw, p, detect_thread, true);

    PacketUpdateEngineEventCounters(tv, fw->dtv, p);

    /* handle Detect */
    SCLogDebug("packet %"PRIu64" calling Detect", p->pcap_cnt);
    if (detect_thread != NULL) {
        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_DETECT);
        Detect(tv, p, detect_thread);
        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_DETECT);
    }

    // Outputs.
    OutputLoggerLog(tv, p, fw->output_thread);

    FramesPrune(p->flow, p);

    /*  Release tcp segments. Done here after alerting can use them. */
    FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_TCPPRUNE);
    StreamTcpPruneSession(p->flow, p->flowflags & FLOW_PKT_TOSERVER ?
            STREAM_TOSERVER : STREAM_TOCLIENT);
    FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_TCPPRUNE);

    /* run tx cleanup last */
    AppLayerParserTransactionsCleanup(p->flow, STREAM_FLAGS_FOR_PACKET(p));

    FlowDeReference(&p->flow);
    /* flow is unlocked later in FlowFinish() */
}

/** \internal
 *  \brief process flows injected into our queue by other threads
 */
static inline void FlowWorkerProcessInjectedFlows(
        ThreadVars *tv, FlowWorkerThreadData *fw, Packet *p)
{
    /* take injected flows and append to our work queue */
    FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_FLOW_INJECTED);
    FlowQueuePrivate injected = { NULL, NULL, 0 };
    if (SC_ATOMIC_GET(tv->flow_queue->non_empty) == true)
        injected = FlowQueueExtractPrivate(tv->flow_queue);
    if (injected.len > 0) {
        StatsAddUI64(tv, fw->cnt.flows_injected, (uint64_t)injected.len);
        if (p->pkt_src == PKT_SRC_WIRE)
            StatsSetUI64(tv, fw->cnt.flows_injected_max, (uint64_t)injected.len);

        /* move to local queue so we can process over the course of multiple packets */
        FlowQueuePrivateAppendPrivate(&fw->fls.work_queue, &injected);
    }
    FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_FLOW_INJECTED);
}

/** \internal
 *  \brief process flows set aside locally during flow lookup
 */
static inline void FlowWorkerProcessLocalFlows(ThreadVars *tv, FlowWorkerThreadData *fw, Packet *p)
{
    uint32_t max_work = 2;
    if (p->pkt_src == PKT_SRC_SHUTDOWN_FLUSH || p->pkt_src == PKT_SRC_CAPTURE_TIMEOUT)
        max_work = 0;

    FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_FLOW_EVICTED);
    if (fw->fls.work_queue.len) {
        FlowTimeoutCounters counters = { 0, 0, };
        CheckWorkQueue(tv, fw, &counters, &fw->fls.work_queue, max_work);
        UpdateCounters(tv, fw, &counters);
    }
    FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_FLOW_EVICTED);
}

/** \internal
 *  \brief apply Packet::app_update_direction to the flow flags
 */
static void PacketAppUpdate2FlowFlags(Packet *p)
{
    switch ((enum StreamUpdateDir)p->app_update_direction) {
        case UPDATE_DIR_NONE: // NONE implies pseudo packet
            break;
        case UPDATE_DIR_PACKET:
            if (PKT_IS_TOSERVER(p)) {
                p->flow->flags |= FLOW_TS_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TS_APP_UPDATED set", p->pcap_cnt);
            } else {
                p->flow->flags |= FLOW_TC_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TC_APP_UPDATED set", p->pcap_cnt);
            }
            break;
        case UPDATE_DIR_BOTH:
            if (PKT_IS_TOSERVER(p)) {
                p->flow->flags |= FLOW_TS_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TS_APP_UPDATED set", p->pcap_cnt);
            } else {
                p->flow->flags |= FLOW_TC_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TC_APP_UPDATED set", p->pcap_cnt);
            }
            /* fall through */
        case UPDATE_DIR_OPPOSING:
            if (PKT_IS_TOSERVER(p)) {
                p->flow->flags |= FLOW_TC_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TC_APP_UPDATED set", p->pcap_cnt);
            } else {
                p->flow->flags |= FLOW_TS_APP_UPDATED;
                SCLogDebug("pcap_cnt %" PRIu64 ", FLOW_TS_APP_UPDATED set", p->pcap_cnt);
            }
            break;
    }
}

static TmEcode FlowWorker(ThreadVars *tv, Packet *p, void *data)
{
    FlowWorkerThreadData *fw = data;
    void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);

    DEBUG_VALIDATE_BUG_ON(p == NULL);
    DEBUG_VALIDATE_BUG_ON(tv->flow_queue == NULL);

    SCLogDebug("packet %"PRIu64, p->pcap_cnt);

    /* update time */
    if (!(PKT_IS_PSEUDOPKT(p))) {
        TimeSetByThread(tv->id, p->ts);
    }

    /* handle Flow */
    if (p->flags & PKT_WANTS_FLOW) {
        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_FLOW);

        FlowHandlePacket(tv, &fw->fls, p);
        if (likely(p->flow != NULL)) {
            DEBUG_ASSERT_FLOW_LOCKED(p->flow);
            if (FlowUpdate(tv, fw, p) == TM_ECODE_DONE) {
                goto housekeeping;
            }
        }
        /* Flow is now LOCKED */

        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_FLOW);

    /* if PKT_WANTS_FLOW is not set, but PKT_HAS_FLOW is, then this is a
     * pseudo packet created by the flow manager. */
    } else if (p->flags & PKT_HAS_FLOW) {
        FLOWLOCK_WRLOCK(p->flow);
        DEBUG_VALIDATE_BUG_ON(p->pkt_src != PKT_SRC_FFR);
    }

    SCLogDebug("packet %"PRIu64" has flow? %s", p->pcap_cnt, p->flow ? "yes" : "no");

    /* handle TCP and app layer */
    if (p->flow) {
        if (PKT_IS_TCP(p)) {
            SCLogDebug("packet %" PRIu64 " is TCP. Direction %s", p->pcap_cnt,
                    PKT_IS_TOSERVER(p) ? "TOSERVER" : "TOCLIENT");
            DEBUG_ASSERT_FLOW_LOCKED(p->flow);

            /* if detect is disabled, we need to apply file flags to the flow
             * here on the first packet. */
            if (detect_thread == NULL &&
                    ((PKT_IS_TOSERVER(p) && (p->flowflags & FLOW_PKT_TOSERVER_FIRST)) ||
                            (PKT_IS_TOCLIENT(p) && (p->flowflags & FLOW_PKT_TOCLIENT_FIRST)))) {
                DisableDetectFlowFileFlags(p->flow);
            }

            FlowWorkerStreamTCPUpdate(tv, fw, p, detect_thread, false);
            PacketAppUpdate2FlowFlags(p);

            /* handle the app layer part of the UDP packet payload */
        } else if (p->proto == IPPROTO_UDP && !PacketCheckAction(p, ACTION_DROP)) {
            FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_APPLAYERUDP);
            AppLayerHandleUdp(tv, fw->stream_thread->ra_ctx->app_tctx, p, p->flow);
            FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_APPLAYERUDP);
            PacketAppUpdate2FlowFlags(p);
        }
    }

    PacketUpdateEngineEventCounters(tv, fw->dtv, p);

    /* handle Detect */
    DEBUG_ASSERT_FLOW_LOCKED(p->flow);
    SCLogDebug("packet %"PRIu64" calling Detect", p->pcap_cnt);
    if (detect_thread != NULL) {
        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_DETECT);
        Detect(tv, p, detect_thread);
        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_DETECT);
    }

    // Outputs.
    OutputLoggerLog(tv, p, fw->output_thread);

    /*  Release tcp segments. Done here after alerting can use them. */
    if (p->flow != NULL) {
        DEBUG_ASSERT_FLOW_LOCKED(p->flow);

        if (FlowIsBypassed(p->flow)) {
            FlowCleanupAppLayer(p->flow);
            if (p->proto == IPPROTO_TCP) {
                StreamTcpSessionCleanup(p->flow->protoctx);
            }
        } else if (p->proto == IPPROTO_TCP && p->flow->protoctx) {
            FramesPrune(p->flow, p);
            FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_TCPPRUNE);
            StreamTcpPruneSession(p->flow, p->flowflags & FLOW_PKT_TOSERVER ?
                    STREAM_TOSERVER : STREAM_TOCLIENT);
            FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_TCPPRUNE);
        } else if (p->proto == IPPROTO_UDP) {
            FramesPrune(p->flow, p);
        }

        if ((PKT_IS_PSEUDOPKT(p)) ||
                (p->flow->flags & (FLOW_TS_APP_UPDATED | FLOW_TC_APP_UPDATED))) {
            if (PKT_IS_TOSERVER(p)) {
                if (PKT_IS_PSEUDOPKT(p) || (p->flow->flags & (FLOW_TS_APP_UPDATED))) {
                    AppLayerParserTransactionsCleanup(p->flow, STREAM_TOSERVER);
                    p->flow->flags &= ~FLOW_TS_APP_UPDATED;
                }
            } else {
                if (PKT_IS_PSEUDOPKT(p) || (p->flow->flags & (FLOW_TC_APP_UPDATED))) {
                    AppLayerParserTransactionsCleanup(p->flow, STREAM_TOCLIENT);
                    p->flow->flags &= ~FLOW_TC_APP_UPDATED;
                }
            }

        } else {
            SCLogDebug("not pseudo, no app update: skip");
        }

        if (p->flow->flags & FLOW_ACTION_DROP) {
            SCLogDebug("flow drop in place: remove app update flags");
            p->flow->flags &= ~(FLOW_TS_APP_UPDATED | FLOW_TC_APP_UPDATED);
        }

        Flow *f = p->flow;
        FlowDeReference(&p->flow);
        FLOWLOCK_UNLOCK(f);
    }

housekeeping:

    /* take injected flows and add them to our local queue */
    FlowWorkerProcessInjectedFlows(tv, fw, p);

    /* process local work queue */
    FlowWorkerProcessLocalFlows(tv, fw, p);

    return TM_ECODE_OK;
}

void FlowWorkerReplaceDetectCtx(void *flow_worker, void *detect_ctx)
{
    FlowWorkerThreadData *fw = flow_worker;

    SC_ATOMIC_SET(fw->detect_thread, detect_ctx);
}

void *FlowWorkerGetDetectCtxPtr(void *flow_worker)
{
    FlowWorkerThreadData *fw = flow_worker;

    return SC_ATOMIC_GET(fw->detect_thread);
}

const char *ProfileFlowWorkerIdToString(enum ProfileFlowWorkerId fwi)
{
    switch (fwi) {
        case PROFILE_FLOWWORKER_FLOW:
            return "flow";
        case PROFILE_FLOWWORKER_STREAM:
            return "stream";
        case PROFILE_FLOWWORKER_APPLAYERUDP:
            return "app-layer";
        case PROFILE_FLOWWORKER_DETECT:
            return "detect";
        case PROFILE_FLOWWORKER_TCPPRUNE:
            return "tcp-prune";
        case PROFILE_FLOWWORKER_FLOW_INJECTED:
            return "flow-inject";
        case PROFILE_FLOWWORKER_FLOW_EVICTED:
            return "flow-evict";
        case PROFILE_FLOWWORKER_SIZE:
            return "size";
    }
    return "error";
}

static void FlowWorkerExitPrintStats(ThreadVars *tv, void *data)
{
    FlowWorkerThreadData *fw = data;
    OutputLoggerExitPrintStats(tv, fw->output_thread);
}

static bool FlowWorkerIsBusy(ThreadVars *tv, void *flow_worker)
{
    FlowWorkerThreadData *fw = flow_worker;
    if (fw->pq.len)
        return true;
    if (fw->fls.work_queue.len)
        return true;

    if (tv->flow_queue) {
        FQLOCK_LOCK(tv->flow_queue);
        bool fq_done = (tv->flow_queue->qlen == 0);
        FQLOCK_UNLOCK(tv->flow_queue);
        if (!fq_done) {
            return true;
        }
    }

    return false;
}

void TmModuleFlowWorkerRegister (void)
{
    tmm_modules[TMM_FLOWWORKER].name = "FlowWorker";
    tmm_modules[TMM_FLOWWORKER].ThreadInit = FlowWorkerThreadInit;
    tmm_modules[TMM_FLOWWORKER].Func = FlowWorker;
    tmm_modules[TMM_FLOWWORKER].ThreadBusy = FlowWorkerIsBusy;
    tmm_modules[TMM_FLOWWORKER].ThreadDeinit = FlowWorkerThreadDeinit;
    tmm_modules[TMM_FLOWWORKER].ThreadExitPrintStats = FlowWorkerExitPrintStats;
    tmm_modules[TMM_FLOWWORKER].cap_flags = 0;
    tmm_modules[TMM_FLOWWORKER].flags = TM_FLAG_STREAM_TM|TM_FLAG_DETECT_TM;
}