Go to the documentation of this file.
48 #ifdef PROFILE_LOCKING
68 #include <sys/param.h>
69 #include <sys/resource.h>
70 #include <sys/cpuset.h>
72 #define cpu_set_t cpuset_t
76 static int SetCPUAffinity(uint16_t cpu);
115 while (decode_pq->
top != NULL) {
121 if (TmThreadsSlotProcessPkt(
tv, slot, extra_p) !=
TM_ECODE_OK) {
142 TmThreadsSlotProcessPktFail(
tv, NULL);
186 r = TmThreadsSlotProcessPkt(
tv, fw_slot, p);
230 static void *TmThreadsSlotPktAcqLoop(
void *td)
250 SCLogError(
"TmSlot or ThreadVars badly setup: s=%p,"
251 " PktAcqLoop=%p, tmqh_in=%p,"
255 pthread_exit((
void *) -1);
259 for (slot = s; slot != NULL; slot = slot->
slot_next) {
261 void *slot_data = NULL;
284 pthread_exit((
void *) -1);
299 pthread_exit((
void *) -1);
334 TmThreadTimeoutLoop(
tv, s);
341 for (slot = s; slot != NULL; slot = slot->
slot_next) {
358 pthread_exit((
void *) 0);
363 pthread_exit((
void *) -1);
367 static void *TmThreadsSlotVar(
void *td)
389 pthread_exit((
void *) -1);
395 void *slot_data = NULL;
415 pthread_exit((
void *) -1);
430 pthread_exit((
void *) -1);
480 TmThreadsHandleInjectedPackets(
tv);
513 pthread_exit((
void *) 0);
518 pthread_exit((
void *) -1);
522 static void *TmThreadsManagement(
void *td)
541 void *slot_data = NULL;
545 pthread_exit((
void *) -1);
576 pthread_exit((
void *) -1);
582 pthread_exit((
void *) 0);
600 printf(
"Both slot name and function pointer can't be NULL inside "
601 "TmThreadSetSlots\n");
608 if (strcmp(name,
"varslot") == 0) {
610 }
else if (strcmp(name,
"pktacqloop") == 0) {
612 }
else if (strcmp(name,
"management") == 0) {
614 }
else if (strcmp(name,
"command") == 0) {
616 }
else if (strcmp(name,
"custom") == 0) {
621 printf(
"Error: Slot \"%s\" not supported\n", name);
684 #if !defined __CYGWIN__ && !defined OS_WIN32 && !defined __OpenBSD__ && !defined sun
685 static int SetCPUAffinitySet(cpu_set_t *cs)
687 #if defined OS_FREEBSD
688 int r = cpuset_setaffinity(CPU_LEVEL_WHICH, CPU_WHICH_TID,
691 int r = thread_policy_set(mach_thread_self(), THREAD_AFFINITY_POLICY,
692 (
void*)cs, THREAD_AFFINITY_POLICY_COUNT);
694 pid_t tid = syscall(SYS_gettid);
695 int r = sched_setaffinity(tid,
sizeof(cpu_set_t), cs);
699 printf(
"Warning: sched_setaffinity failed (%" PRId32
"): %s\n", r,
716 static int SetCPUAffinity(uint16_t cpuid)
718 #if defined __OpenBSD__ || defined sun
721 int cpu = (int)cpuid;
723 #if defined OS_WIN32 || defined __CYGWIN__
726 int r = (0 == SetThreadAffinityMask(GetCurrentThread(), cs));
728 printf(
"Warning: sched_setaffinity failed (%" PRId32
"): %s\n", r,
732 SCLogDebug(
"CPU Affinity for thread %lu set to CPU %" PRId32,
739 memset(&cs, 0,
sizeof(cs));
743 return SetCPUAffinitySet(&cs);
775 tv->
name, strerror(errno));
777 SCLogDebug(
"Priority set to %"PRId32
" for thread %s",
787 SCLogDebug(
"Nice value set to %"PRId32
" for thread %s",
848 SCLogPerf(
"Setting affinity for thread \"%s\"to cpu/core "
854 #if !defined __CYGWIN__ && !defined OS_WIN32 && !defined __OpenBSD__ && !defined sun
867 }
else if (CPU_ISSET(cpu, &taf->
hiprio_cpu)) {
872 SCLogPerf(
"Setting prio %d for thread \"%s\" to cpu/core "
876 SetCPUAffinitySet(&taf->
cpu_set);
878 SCLogPerf(
"Setting prio %d for thread \"%s\", "
905 const char *outq_name,
const char *outqh_name,
const char *slots,
906 void * (*fn_p)(
void *),
int mucond)
912 SCLogDebug(
"creating thread \"%s\"...", name);
929 if (inq_name != NULL && strcmp(inq_name,
"packetpool") != 0) {
944 if (inqh_name != NULL) {
961 if (outqh_name != NULL) {
976 if (outq_name != NULL && strcmp(outq_name,
"packetpool") != 0) {
1034 const char *inqh_name,
const char *outq_name,
1035 const char *outqh_name,
const char *slots)
1095 tv =
TmThreadCreate(name, NULL, NULL, NULL, NULL,
"management", NULL, mucond);
1128 tv =
TmThreadCreate(name, NULL, NULL, NULL, NULL,
"command", NULL, mucond);
1165 if (t->
next == NULL) {
1235 if (
tv->
inq != NULL) {
1263 pthread_join(
tv->
t, NULL);
1286 static void TmThreadDrainPacketThreads(
void)
1289 struct timeval start_ts;
1290 struct timeval cur_ts;
1291 gettimeofday(&start_ts, NULL);
1294 gettimeofday(&cur_ts, NULL);
1295 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1297 "to process their packets in time");
1306 if (ThreadStillHasPackets(
tv)) {
1316 if (ThreadBusy(
tv)) {
1350 struct timeval start_ts;
1351 struct timeval cur_ts;
1352 gettimeofday(&start_ts, NULL);
1355 gettimeofday(&cur_ts, NULL);
1356 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1357 FatalError(
"Engine unable to disable detect "
1358 "thread - \"%s\". Killing engine",
1376 while (slots != NULL) {
1389 if (ThreadStillHasPackets(
tv)) {
1399 if (ThreadBusy(
tv)) {
1424 if (
tv->
inq != NULL) {
1451 TmThreadDrainPacketThreads();
1454 #ifdef DEBUG_VALIDATION
1455 static void TmThreadDumpThreads(
void);
1458 static void TmThreadDebugValidateNoMorePackets(
void)
1460 #ifdef DEBUG_VALIDATION
1463 if (ThreadStillHasPackets(
tv)) {
1465 TmThreadDumpThreads();
1478 struct timeval start_ts;
1479 struct timeval cur_ts;
1482 TmThreadDrainPacketThreads();
1487 TmThreadDebugValidateNoMorePackets();
1489 gettimeofday(&start_ts, NULL);
1491 gettimeofday(&cur_ts, NULL);
1492 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1493 FatalError(
"Engine unable to disable packet "
1494 "threads. Killing engine");
1505 if (
tv->
inq != NULL) {
1524 #define MIN_WAIT_TIME 100
1525 #define MAX_WAIT_TIME 999999
1538 int r = TmThreadKillThread(
tv);
1552 #undef MIN_WAIT_TIME
1553 #undef MAX_WAIT_TIME
1559 for (i = 0; i <
TVT_MAX; i++) {
1580 TmThreadDeinitMC(
tv);
1609 char *thread_group_name = NULL;
1617 thread_group_name =
SCStrdup(name);
1618 if (
unlikely(thread_group_name == NULL)) {
1630 if ((family < 0) || (family >=
TVT_MAX))
1652 pthread_attr_t attr;
1658 pthread_attr_init(&attr);
1660 pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
1666 FatalError(
"Unable to increase stack size to %" PRIu64
" in thread attributes",
1673 FatalError(
"Unable to create thread with pthread_create(): retval %d: %s", rc,
1677 #if DEBUG && HAVE_PTHREAD_GETATTR_NP
1679 if (pthread_getattr_np(
tv->
t, &attr) == 0) {
1682 pthread_attr_getstack(&attr, &stack_addr, &stack_size);
1683 SCLogDebug(
"stack: %p; size %" PRIu64, stack_addr, (uintmax_t)stack_size);
1685 SCLogDebug(
"Unable to retrieve current stack-size for display; return code from "
1686 "pthread_getattr_np() is %" PRId32,
1709 FatalError(
"Fatal error encountered in TmThreadInitMC. "
1714 printf(
"Error initializing the tv->m mutex\n");
1719 FatalError(
"Fatal error encountered in TmThreadInitMC. "
1724 FatalError(
"Error initializing the tv->cond condition "
1782 static TmEcode WaitOnThreadsRunningByType(
const int t)
1784 struct timeval start_ts;
1785 struct timeval cur_ts;
1786 uint32_t thread_cnt = 0;
1797 uint32_t time_budget = 60 + thread_cnt;
1799 gettimeofday(&start_ts, NULL);
1803 while (
tv != NULL) {
1808 "start: flags %04x",
1818 gettimeofday(&cur_ts, NULL);
1819 if (((uint32_t)cur_ts.tv_sec - (uint32_t)start_ts.tv_sec) > time_budget) {
1821 "start in time: flags %04x. Total threads: %u. Time budget %us",
1846 uint16_t RX_num = 0;
1848 uint16_t FM_num = 0;
1849 uint16_t FR_num = 0;
1850 uint16_t TX_num = 0;
1852 for (
int i = 0; i <
TVT_MAX; i++) {
1858 for (
int i = 0; i <
TVT_MAX; i++) {
1876 uint16_t app_len = 32;
1877 uint16_t buf_len = 256;
1879 char append_str[app_len];
1880 char thread_counts[buf_len];
1882 strlcpy(thread_counts,
"Threads created -> ", strlen(
"Threads created -> ") + 1);
1884 snprintf(append_str, app_len,
"RX: %u ", RX_num);
1885 strlcat(thread_counts, append_str, buf_len);
1888 snprintf(append_str, app_len,
"W: %u ", W_num);
1889 strlcat(thread_counts, append_str, buf_len);
1892 snprintf(append_str, app_len,
"TX: %u ", TX_num);
1893 strlcat(thread_counts, append_str, buf_len);
1896 snprintf(append_str, app_len,
"FM: %u ", FM_num);
1897 strlcat(thread_counts, append_str, buf_len);
1900 snprintf(append_str, app_len,
"FR: %u ", FR_num);
1901 strlcat(thread_counts, append_str, buf_len);
1903 snprintf(append_str, app_len,
" Engine started.");
1904 strlcat(thread_counts, append_str, buf_len);
1916 for (
int i = 0; i <
TVT_MAX; i++) {
1918 while (
tv != NULL) {
1932 for (
int i = 0; i <
TVT_MAX; i++) {
1954 struct timeval start_ts;
1955 struct timeval cur_ts;
1956 gettimeofday(&start_ts, NULL);
1960 for (
int i = 0; i <
TVT_MAX; i++) {
1962 while (
tv != NULL) {
1967 "initialize: flags %04x",
1975 gettimeofday(&cur_ts, NULL);
1976 if ((cur_ts.tv_sec - start_ts.tv_sec) > 120) {
1978 "initialize in time: flags %04x",
2019 for (
int i = 0; i <
TVT_MAX; i++) {
2021 while (
tv != NULL) {
2032 #ifdef DEBUG_VALIDATION
2042 static void TmThreadDumpThreads(
void)
2045 for (
int i = 0; i <
TVT_MAX; i++) {
2047 while (
tv != NULL) {
2049 SCLogNotice(
"tv %p: type %u name %s tmm_flags %02X flags %X stream_pq %p",
2055 SCLogNotice(
"tv %p: ==> stream_pq_local: pq.len %u packet src %s",
2060 SCLogNotice(
"tv %p: ==> decode_pq: decode_pq.len %u packet src %s",
2063 TmThreadDoDumpSlots(
tv);
2090 static Threads thread_store = { NULL, 0, 0 };
2096 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2098 if (t == NULL || t->
in_use == 0)
2101 SCLogNotice(
"Thread %"PRIuMAX
", %s type %d, tv %p in_use %d",
2106 SCLogNotice(
"tv %p type %u name %s tmm_flags %02X flags %X",
2120 if (thread_store.
threads == NULL) {
2143 thread_store.
threads = newmem;
2204 struct timeval systs;
2205 gettimeofday(&systs, NULL);
2214 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2231 struct timeval systs;
2232 gettimeofday(&systs, NULL);
2234 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2248 struct timeval local = { 0 };
2249 static struct timeval nullts;
2252 struct timeval systs;
2253 gettimeofday(&systs, NULL);
2265 if (!(timercmp(&pkttv, &nullts, ==))) {
2282 SCLogDebug(
"ts->tv_sec %"PRIuMAX, (uintmax_t)
ts->tv_sec);
2290 if (thread_max == 0)
2294 if (thread_max > 1024) {
2295 SCLogWarning(
"limited number of 'worker' threads to 1024. Wanted %d", thread_max);
2298 return (uint16_t)thread_max;
2317 if (
tv->
inq != NULL) {
2322 TmThreadsCaptureBreakLoop(
tv);
const char * thread_name_workers
TmEcode TmThreadSetCPUAffinity(ThreadVars *tv, uint16_t cpu)
Set the thread options (cpu affinity).
struct FlowQueue_ * flow_queue
bool TmThreadsTimeSubsysIsReady(void)
void TmThreadInitMC(ThreadVars *tv)
Initializes the mutex and condition variables for this TV.
thread_local uint64_t spin_lock_cnt
TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p, TmSlot *slot)
Separate run function so we can call it recursively.
TmEcode TmThreadSpawn(ThreadVars *tv)
Spawns a thread associated with the ThreadVars instance tv.
ThreadVars * TmThreadCreateMgmtThreadByName(const char *name, const char *module, int mucond)
Creates and returns the TV instance for a Management thread(MGMT). This function supports only custom...
int TmqhNameToID(const char *name)
bool threading_set_cpu_affinity
TmEcode TmThreadSetupOptions(ThreadVars *tv)
Set the thread options (cpu affinitythread). Priority should be already set by pthread_create.
const char * thread_name_flow_mgr
#define SC_ATOMIC_INIT(name)
wrapper for initializing an atomic variable.
void TmThreadContinueThreads(void)
Unpauses all threads present in tv_root.
ThreadVars * TmThreadCreatePacketHandler(const char *name, const char *inq_name, const char *inqh_name, const char *outq_name, const char *outqh_name, const char *slots)
Creates and returns a TV instance for a Packet Processing Thread. This function doesn't support custo...
#define SC_ATOMIC_SET(name, val)
Set the value for the atomic variable.
#define SCCtrlMutexDestroy
void TmThreadSetGroupName(ThreadVars *tv, const char *name)
void CaptureStatsSetup(ThreadVars *tv)
thread_local uint64_t rwr_lock_cnt
void TmThreadsSetFlag(ThreadVars *tv, uint32_t flag)
Set a thread flag.
void TmThreadWaitForFlag(ThreadVars *tv, uint32_t flags)
Waits till the specified flag(s) is(are) set. We don't bother if the kill flag has been set or not on...
void PacketEnqueue(PacketQueue *q, Packet *p)
simple fifo queue for packets with mutex and cond Calling the mutex or triggering the cond is respons...
@ PKT_SRC_CAPTURE_TIMEOUT
#define SCSetThreadName(n)
const char * thread_name_flow_rec
void(* OutHandler)(ThreadVars *, Packet *)
uint32_t TmThreadCountThreadsByTmmFlags(uint8_t flags)
returns a count of all the threads that match the flag
const char * thread_name_autofp
int StatsSetupPrivate(ThreadVars *tv)
struct PacketQueue_ * stream_pq_local
ThreadVars * tv_root[TVT_MAX]
void TmThreadDisableReceiveThreads(void)
Disable all threads having the specified TMs.
#define SCCtrlCondDestroy
#define SCMUTEX_INITIALIZER
TmEcode TmThreadSetThreadPriority(ThreadVars *tv, int prio)
Set the thread options (thread priority).
bool(* ThreadBusy)(ThreadVars *tv, void *thread_data)
#define THREAD_SET_PRIORITY
void TmqhOutputPacketpool(ThreadVars *t, Packet *p)
thread_local uint64_t rww_lock_contention
simple fifo queue for packets
void FlowEnqueue(FlowQueue *q, Flow *f)
add a flow to a queue
void TmThreadDisablePacketThreads(void)
Disable all packet threads.
TmEcode(* PktAcqLoop)(ThreadVars *, void *, void *)
void PacketPoolInit(void)
Packet *(* InHandler)(ThreadVars *)
void TmThreadsInjectFlowById(Flow *f, const int id)
inject a flow into a threads flow queue
thread_local uint64_t rwr_lock_contention
void TmThreadsGetMinimalTimestamp(struct timeval *ts)
size_t strlcpy(char *dst, const char *src, size_t siz)
TmEcode TmThreadsProcessDecodePseudoPackets(ThreadVars *tv, PacketQueueNoLock *decode_pq, TmSlot *slot)
TmEcode(* ThreadDeinit)(ThreadVars *, void *)
TmModule * TmModuleGetByName(const char *name)
get a tm module ptr by name
thread_local uint64_t spin_lock_wait_ticks
#define PKT_SET_SRC(p, src_val)
void TmThreadsUnsetFlag(ThreadVars *tv, uint32_t flag)
Unset a thread flag.
void(* InShutdownHandler)(ThreadVars *)
void TmThreadSetPrio(ThreadVars *tv)
Adjusting nice value for threads.
void TmThreadContinue(ThreadVars *tv)
Unpauses a thread.
uint16_t AffinityGetNextCPU(ThreadsAffinityType *taf)
Return next cpu to use for a given thread family.
TmEcode(* PktAcqLoop)(ThreadVars *, void *, void *)
TmEcode TmThreadWaitOnThreadInit(void)
Used to check if all threads have finished their initialization. On finding an un-initialized thread,...
TmEcode(* PktAcqBreakLoop)(ThreadVars *, void *)
size_t strlcat(char *, const char *src, size_t siz)
struct TmSlot_ * tm_slots
Packet * PacketDequeueNoLock(PacketQueueNoLock *qnl)
TmEcode(* Management)(ThreadVars *, void *)
#define SCMutexUnlock(mut)
uint64_t threading_set_stack_size
StatsPublicThreadContext perf_public_ctx
#define PKT_PSEUDO_STREAM_END
void TmThreadsSetThreadTimestamp(const int id, const SCTime_t ts)
thread_local uint64_t rww_lock_cnt
Per thread variable structure.
thread_local uint64_t mutex_lock_contention
#define SCTIME_FROM_TIMEVAL(tv)
Tmq * TmqGetQueueByName(const char *name)
void TmThreadTestThreadUnPaused(ThreadVars *tv)
Tests if the thread represented in the arg has been unpaused or not.
TmEcode(* Management)(ThreadVars *, void *)
thread_local uint64_t spin_lock_contention
TmEcode(* Func)(ThreadVars *, Packet *, void *)
void TmThreadClearThreadsFamily(int family)
ThreadVars * TmThreadCreate(const char *name, const char *inq_name, const char *inqh_name, const char *outq_name, const char *outqh_name, const char *slots, void *(*fn_p)(void *), int mucond)
Creates and returns the TV instance for a new thread.
FlowQueue * FlowQueueNew(void)
const char * PktSrcToString(enum PktSrcEnum pkt_src)
void TmThreadsUnregisterThread(const int id)
int TmThreadsRegisterThread(ThreadVars *tv, const int type)
#define SCLogWarning(...)
Macro used to log WARNING messages.
void(* OutHandlerCtxFree)(void *)
struct ThreadVars_ * next
#define SCTIME_TO_TIMEVAL(tv, t)
int TmModuleGetIDForTM(TmModule *tm)
Given a TM Module, returns its id.
#define SCCtrlMutexLock(mut)
void TmThreadKillThreads(void)
#define PACKET_PROFILING_TMM_END(p, id)
#define TM_FLAG_DECODE_TM
TmModule * TmModuleGetById(int id)
Returns a TM Module by its id.
uint8_t thread_setup_flags
thread_local uint64_t rww_lock_wait_ticks
ThreadVars * TmThreadCreateMgmtThread(const char *name, void *(fn_p)(void *), int mucond)
Creates and returns the TV instance for a Management thread(MGMT). This function supports only custom...
thread_local uint64_t rwr_lock_wait_ticks
#define SCMutexInit(mut, mutattrs)
#define TM_FLAG_RECEIVE_TM
#define SCGetThreadIdLong(...)
#define SCRealloc(ptr, sz)
void TmThreadsInitThreadsTimestamp(const SCTime_t ts)
struct PacketQueue_ * stream_pq
#define SCTIME_CMP_LT(a, b)
void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, const void *data)
Appends a new entry to the slots.
#define SCCtrlMutexUnlock(mut)
void TmThreadKillThreadsFamily(int family)
void TmThreadAppend(ThreadVars *tv, int type)
Appends this TV to tv_root based on its type.
TmEcode TmThreadSetCPU(ThreadVars *tv, uint8_t type)
ThreadsAffinityType thread_affinity[MAX_CPU_SET]
void(* tmqh_out)(struct ThreadVars_ *, struct Packet_ *)
struct TmSlot_ * tm_flowworker
int TmThreadGetNbThreads(uint8_t type)
TmEcode(* SlotThreadInit)(ThreadVars *, const void *, void **)
TmEcode(* ThreadInit)(ThreadVars *, const void *, void **)
const void * slot_initdata
uint16_t TmThreadsGetWorkerThreadMax(void)
ThreadVars * TmThreadCreateCmdThreadByName(const char *name, const char *module, int mucond)
Creates and returns the TV instance for a Command thread (CMD). This function supports only custom sl...
void StatsSyncCounters(ThreadVars *tv)
#define THREAD_SET_AFFTYPE
void(* ThreadExitPrintStats)(ThreadVars *, void *)
Packet * PacketDequeue(PacketQueue *q)
Packet * PacketGetFromAlloc(void)
Get a malloced packet.
thread_local uint64_t mutex_lock_wait_ticks
#define PACKET_PROFILING_TMM_START(p, id)
struct Packet_ *(* tmqh_in)(struct ThreadVars_ *)
void *(* OutHandlerCtxSetup)(const char *)
#define SCLogError(...)
Macro used to log ERROR messages.
Tmq * TmqCreateQueue(const char *name)
TmEcode(* SlotThreadDeinit)(ThreadVars *, void *)
const char * thread_name_verdict
void(* SlotThreadExitPrintStats)(ThreadVars *, void *)
Tmqh * TmqhGetQueueHandlerByID(const int id)
#define SC_ATOMIC_INITPTR(name)
PacketQueueNoLock decode_pq
#define SCCtrlMutexInit(mut, mutattr)
void EngineDone(void)
Used to indicate that the current task is done.
#define THREAD_SET_AFFINITY
TmEcode TmThreadWaitOnThreadRunning(void)
Waits for all threads to be in a running state.
void TmThreadsListThreads(void)
struct TmSlot_ * slot_next
void TmThreadCheckThreadState(void)
Used to check the thread for certain conditions of failure.
#define SC_ATOMIC_GET(name)
Get the value from the atomic variable.
uint16_t UtilCpuGetNumProcessorsOnline(void)
Get the number of cpus online in the system.
void PacketPoolDestroy(void)
thread_local uint64_t mutex_lock_cnt
int TmThreadsCheckFlag(ThreadVars *tv, uint32_t flag)
Check if a thread flag is set.
#define SCLogNotice(...)
Macro used to log NOTICE messages.
Tmqh * TmqhGetQueueHandlerByName(const char *name)
void StatsThreadCleanup(ThreadVars *tv)
void *(* tm_func)(void *)
Packet * PacketGetFromQueueOrAlloc(void)
Get a packet. We try to get a packet from the packetpool first, but if that is empty we alloc a packe...
#define DEBUG_VALIDATE_BUG_ON(exp)
float threading_detect_ratio
#define SC_ATOMIC_AND(name, val)
Bitwise AND a value to our atomic variable.
volatile uint8_t suricata_ctl_flags
#define SC_ATOMIC_OR(name, val)
Bitwise OR a value to our atomic variable.