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);
126 TmThreadsSlotProcessPktFail(
tv, s, NULL);
138 if (s->slot_next != NULL) {
141 TmThreadsSlotProcessPktFail(
tv, s, extra_p);
182 r = TmThreadsSlotProcessPkt(
tv, fw_slot, p);
228 static void *TmThreadsSlotPktAcqLoop(
void *td)
248 SCLogError(
"TmSlot or ThreadVars badly setup: s=%p,"
249 " PktAcqLoop=%p, tmqh_in=%p,"
253 pthread_exit((
void *) -1);
257 for (slot = s; slot != NULL; slot = slot->
slot_next) {
259 void *slot_data = NULL;
282 pthread_exit((
void *) -1);
297 pthread_exit((
void *) -1);
332 TmThreadTimeoutLoop(
tv, s);
339 for (slot = s; slot != NULL; slot = slot->
slot_next) {
356 pthread_exit((
void *) 0);
361 pthread_exit((
void *) -1);
365 static void *TmThreadsSlotVar(
void *td)
386 pthread_exit((
void *) -1);
392 void *slot_data = NULL;
412 pthread_exit((
void *) -1);
427 pthread_exit((
void *) -1);
477 TmThreadsHandleInjectedPackets(
tv);
510 pthread_exit((
void *) 0);
515 pthread_exit((
void *) -1);
519 static void *TmThreadsManagement(
void *td)
538 void *slot_data = NULL;
542 pthread_exit((
void *) -1);
573 pthread_exit((
void *) -1);
579 pthread_exit((
void *) 0);
597 printf(
"Both slot name and function pointer can't be NULL inside "
598 "TmThreadSetSlots\n");
605 if (strcmp(name,
"varslot") == 0) {
607 }
else if (strcmp(name,
"pktacqloop") == 0) {
609 }
else if (strcmp(name,
"management") == 0) {
611 }
else if (strcmp(name,
"command") == 0) {
613 }
else if (strcmp(name,
"custom") == 0) {
618 printf(
"Error: Slot \"%s\" not supported\n", name);
642 memset(slot, 0,
sizeof(
TmSlot));
682 #if !defined __CYGWIN__ && !defined OS_WIN32 && !defined __OpenBSD__ && !defined sun
683 static int SetCPUAffinitySet(cpu_set_t *cs)
685 #if defined OS_FREEBSD
686 int r = cpuset_setaffinity(CPU_LEVEL_WHICH, CPU_WHICH_TID,
689 int r = thread_policy_set(mach_thread_self(), THREAD_AFFINITY_POLICY,
690 (
void*)cs, THREAD_AFFINITY_POLICY_COUNT);
692 pid_t tid = syscall(SYS_gettid);
693 int r = sched_setaffinity(tid,
sizeof(cpu_set_t), cs);
697 printf(
"Warning: sched_setaffinity failed (%" PRId32
"): %s\n", r,
714 static int SetCPUAffinity(uint16_t cpuid)
716 #if defined __OpenBSD__ || defined sun
719 int cpu = (int)cpuid;
721 #if defined OS_WIN32 || defined __CYGWIN__
724 int r = (0 == SetThreadAffinityMask(GetCurrentThread(), cs));
726 printf(
"Warning: sched_setaffinity failed (%" PRId32
"): %s\n", r,
730 SCLogDebug(
"CPU Affinity for thread %lu set to CPU %" PRId32,
737 memset(&cs, 0,
sizeof(cs));
741 return SetCPUAffinitySet(&cs);
773 tv->
name, strerror(errno));
775 SCLogDebug(
"Priority set to %"PRId32
" for thread %s",
785 SCLogDebug(
"Nice value set to %"PRId32
" for thread %s",
846 SCLogPerf(
"Setting affinity for thread \"%s\"to cpu/core "
852 #if !defined __CYGWIN__ && !defined OS_WIN32 && !defined __OpenBSD__ && !defined sun
865 }
else if (CPU_ISSET(cpu, &taf->
hiprio_cpu)) {
870 SCLogPerf(
"Setting prio %d for thread \"%s\" to cpu/core "
874 SetCPUAffinitySet(&taf->
cpu_set);
876 SCLogPerf(
"Setting prio %d for thread \"%s\", "
903 const char *outq_name,
const char *outqh_name,
const char *slots,
904 void * (*fn_p)(
void *),
int mucond)
910 SCLogDebug(
"creating thread \"%s\"...", name);
928 if (inq_name != NULL && strcmp(inq_name,
"packetpool") != 0) {
943 if (inqh_name != NULL) {
960 if (outqh_name != NULL) {
975 if (outq_name != NULL && strcmp(outq_name,
"packetpool") != 0) {
1033 const char *inqh_name,
const char *outq_name,
1034 const char *outqh_name,
const char *slots)
1094 tv =
TmThreadCreate(name, NULL, NULL, NULL, NULL,
"management", NULL, mucond);
1127 tv =
TmThreadCreate(name, NULL, NULL, NULL, NULL,
"command", NULL, mucond);
1164 if (t->
next == NULL) {
1236 if (
tv->
inq != NULL) {
1260 pthread_join(
tv->
t, NULL);
1271 static void TmThreadDrainPacketThreads(
void)
1274 struct timeval start_ts;
1275 struct timeval cur_ts;
1276 gettimeofday(&start_ts, NULL);
1279 gettimeofday(&cur_ts, NULL);
1280 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1282 "to process their packets in time");
1291 if (ThreadStillHasPackets(
tv)) {
1341 struct timeval start_ts;
1342 struct timeval cur_ts;
1343 gettimeofday(&start_ts, NULL);
1346 gettimeofday(&cur_ts, NULL);
1347 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1348 FatalError(
"Engine unable to disable detect "
1349 "thread - \"%s\". Killing engine",
1367 while (slots != NULL) {
1380 if (ThreadStillHasPackets(
tv)) {
1420 if (
tv->
inq != NULL) {
1445 TmThreadDrainPacketThreads();
1449 #ifdef DEBUG_VALIDATION
1450 static void TmThreadDumpThreads(
void);
1453 static void TmThreadDebugValidateNoMorePackets(
void)
1455 #ifdef DEBUG_VALIDATION
1458 if (ThreadStillHasPackets(
tv)) {
1460 TmThreadDumpThreads();
1473 struct timeval start_ts;
1474 struct timeval cur_ts;
1477 TmThreadDrainPacketThreads();
1482 TmThreadDebugValidateNoMorePackets();
1484 gettimeofday(&start_ts, NULL);
1486 gettimeofday(&cur_ts, NULL);
1487 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1488 FatalError(
"Engine unable to disable packet "
1489 "threads. Killing engine");
1500 if (
tv->
inq != NULL) {
1518 #define MIN_WAIT_TIME 100
1519 #define MAX_WAIT_TIME 999999
1532 int r = TmThreadKillThread(
tv);
1546 #undef MIN_WAIT_TIME
1547 #undef MAX_WAIT_TIME
1553 for (i = 0; i <
TVT_MAX; i++) {
1576 TmThreadDeinitMC(
tv);
1605 char *thread_group_name = NULL;
1613 thread_group_name =
SCStrdup(name);
1614 if (
unlikely(thread_group_name == NULL)) {
1626 if ((family < 0) || (family >=
TVT_MAX))
1648 pthread_attr_t attr;
1654 pthread_attr_init(&attr);
1656 pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
1662 FatalError(
"Unable to increase stack size to %" PRIu64
" in thread attributes",
1669 FatalError(
"Unable to create thread with pthread_create() is %" PRId32, rc);
1672 #if DEBUG && HAVE_PTHREAD_GETATTR_NP
1674 if (pthread_getattr_np(
tv->
t, &attr) == 0) {
1677 pthread_attr_getstack(&attr, &stack_addr, &stack_size);
1678 SCLogDebug(
"stack: %p; size %" PRIu64, stack_addr, (uintmax_t)stack_size);
1680 SCLogDebug(
"Unable to retrieve current stack-size for display; return code from "
1681 "pthread_getattr_np() is %" PRId32,
1704 FatalError(
"Fatal error encountered in TmThreadInitMC. "
1709 printf(
"Error initializing the tv->m mutex\n");
1714 FatalError(
"Fatal error encountered in TmThreadInitMC. "
1719 FatalError(
"Error initializing the tv->cond condition "
1792 uint16_t RX_num = 0;
1794 uint16_t FM_num = 0;
1795 uint16_t FR_num = 0;
1796 uint16_t TX_num = 0;
1798 struct timeval start_ts;
1799 struct timeval cur_ts;
1800 gettimeofday(&start_ts, NULL);
1804 for (
int i = 0; i <
TVT_MAX; i++) {
1806 while (
tv != NULL) {
1811 "start: flags %04x",
1821 gettimeofday(&cur_ts, NULL);
1822 if ((cur_ts.tv_sec - start_ts.tv_sec) > 60) {
1824 "start in time: flags %04x",
1837 for (
int i = 0; i <
TVT_MAX; i++) {
1855 uint16_t app_len = 32;
1856 uint16_t buf_len = 256;
1858 char append_str[app_len];
1859 char thread_counts[buf_len];
1861 strlcpy(thread_counts,
"Threads created -> ", strlen(
"Threads created -> ") + 1);
1863 snprintf(append_str, app_len,
"RX: %u ", RX_num);
1864 strlcat(thread_counts, append_str, buf_len);
1867 snprintf(append_str, app_len,
"W: %u ", W_num);
1868 strlcat(thread_counts, append_str, buf_len);
1871 snprintf(append_str, app_len,
"TX: %u ", TX_num);
1872 strlcat(thread_counts, append_str, buf_len);
1875 snprintf(append_str, app_len,
"FM: %u ", FM_num);
1876 strlcat(thread_counts, append_str, buf_len);
1879 snprintf(append_str, app_len,
"FR: %u ", FR_num);
1880 strlcat(thread_counts, append_str, buf_len);
1882 snprintf(append_str, app_len,
" Engine started.");
1883 strlcat(thread_counts, append_str, buf_len);
1895 for (
int i = 0; i <
TVT_MAX; i++) {
1897 while (
tv != NULL) {
1912 for (
int i = 0; i <
TVT_MAX; i++) {
1935 struct timeval start_ts;
1936 struct timeval cur_ts;
1937 gettimeofday(&start_ts, NULL);
1941 for (
int i = 0; i <
TVT_MAX; i++) {
1943 while (
tv != NULL) {
1948 "initialize: flags %04x",
1956 gettimeofday(&cur_ts, NULL);
1957 if ((cur_ts.tv_sec - start_ts.tv_sec) > 120) {
1959 "initialize in time: flags %04x",
2000 for (
int i = 0; i <
TVT_MAX; i++) {
2002 while (
tv != NULL) {
2013 #ifdef DEBUG_VALIDATION
2023 static void TmThreadDumpThreads(
void)
2026 for (
int i = 0; i <
TVT_MAX; i++) {
2028 while (
tv != NULL) {
2030 SCLogNotice(
"tv %p: type %u name %s tmm_flags %02X flags %X stream_pq %p",
2036 SCLogNotice(
"tv %p: ==> stream_pq_local: pq.len %u packet src %s",
2041 SCLogNotice(
"tv %p: ==> decode_pq: decode_pq.len %u packet src %s",
2044 TmThreadDoDumpSlots(
tv);
2071 static Threads thread_store = { NULL, 0, 0 };
2077 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2079 if (t == NULL || t->
in_use == 0)
2082 SCLogNotice(
"Thread %"PRIuMAX
", %s type %d, tv %p in_use %d",
2087 SCLogNotice(
"tv %p type %u name %s tmm_flags %02X flags %X",
2101 if (thread_store.
threads == NULL) {
2124 thread_store.
threads = newmem;
2185 struct timeval systs;
2186 gettimeofday(&systs, NULL);
2195 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2210 struct timeval systs;
2211 gettimeofday(&systs, NULL);
2213 for (
size_t s = 0; s < thread_store.
threads_size; s++) {
2225 struct timeval local = { 0 };
2226 static struct timeval nullts;
2229 struct timeval systs;
2230 gettimeofday(&systs, NULL);
2239 if (!(timercmp(&pkttv, &nullts, ==))) {
2256 SCLogDebug(
"ts->tv_sec %"PRIuMAX, (uintmax_t)
ts->tv_sec);
2264 if (thread_max == 0)
2268 if (thread_max > 1024) {
2269 SCLogWarning(
"limited number of 'worker' threads to 1024. Wanted %d", thread_max);
2272 return (uint16_t)thread_max;
2308 if (
tv->
inq != NULL) {
2311 ThreadBreakLoop(
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)
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)
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).
#define THREAD_SET_PRIORITY
void TmqhOutputPacketpool(ThreadVars *t, Packet *p)
thread_local uint64_t rww_lock_contention
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(* 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.
int threading_set_cpu_affinity
void(* OutHandlerCtxFree)(void *)
struct ThreadVars_ * next
#define SCTIME_TO_TIMEVAL(tv, t)
int TmModuleGetIDForTM(TmModule *tm)
Given a TM Module, returns its id.
void TmThreadKillThreads(void)
#define PACKET_PROFILING_TMM_END(p, id)
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.
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...
#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)
@ PKT_SRC_DETECT_RELOAD_FLUSH
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.
#define StatsSyncCounters(tv)
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.