aiden
/
suricata
mirror of https://github.com/OISF/suricata
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

improve the threading api for the ids

Browse Source
remotes/origin/master-1.0.x
Anoop Saldanha 17 years ago committed by Victor Julien
parent d0e70309c0
commit 53c21410b6
10 changed files with 524 additions and 299 deletions
Show Stats Download Patch File Download Diff File

244
src/counters.c
Unescape Escape View File

@ -15,12 +15,11 @@
/** \todo config api */
#define LOGPATH "/var/log/eidps/stats.log"
static PerfThreadContext *perf_tc = NULL;
static PerfOPIfaceContext *perf_op_ctx = NULL;
/**
* Initializes the perf counter api. Things are hard coded currently.
* More work to be done when we implement multiple interfaces
* \brief Initializes the perf counter api. Things are hard coded currently.
* More work to be done when we implement multiple interfaces
*/
void PerfInitCounterApi()
{
@ -30,7 +29,7 @@ void PerfInitCounterApi()
}
/**
* Initializes the output interface context
* \brief Initializes the output interface context
*/
void PerfInitOPCtx()
{
@ -65,92 +64,49 @@ void PerfInitOPCtx()
}
/**
* Spawns the wakeup, and the management thread
* \brief Spawns the wakeup, and the management thread
*/
void PerfSpawnThreads()
{
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
printf("PerfSpawnThreads: spawning counter threads\n");
if ( (perf_tc = malloc(sizeof(PerfThreadContext))) == NULL) {
printf("Error allocating memory\n");
exit(0);
ThreadVars *tv_wakeup = NULL;
ThreadVars *tv_mgmt = NULL;
/* Spawn the stats wakeup thread */
tv_wakeup = TmThreadCreate("PerfWakeupThread", NULL, NULL, NULL, NULL,
"custom", PerfWakeupThread, 1);
if (tv_wakeup == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
}
memset(perf_tc, 0, sizeof(PerfThreadContext));
perf_tc->flags = PT_RUN;
if (pthread_mutex_init(&perf_tc->wakeup_m, NULL) != 0) {
printf("Error initializing the perf_tc->wakeup_m mutex\n");
exit(0);
if (TmThreadSpawn(tv_wakeup, TVT_MGMT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
if (pthread_mutex_init(&perf_tc->mgmt_m, NULL) != 0) {
printf("Error initializing the perf_tc->mgmt_m mutex\n");
exit(0);
/* Spawn the stats mgmt thread */
tv_mgmt = TmThreadCreate("PerfMgmtThread", NULL, NULL, NULL, NULL,
"custom", PerfMgmtThread, 1);
if (tv_mgmt == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
}
if (pthread_cond_init(&perf_tc->tc_cond, NULL) != 0) {
printf("Error initializing the perf_tc->tc_cond condition variable\n");
exit(0);
}
if (pthread_create(&perf_tc->wakeup_t, &attr, PerfWakeupThread, NULL) != 0) {
printf("Error creating PerfWakeupFunc thread\n");
exit(0);
}
if (pthread_create(&perf_tc->mgmt_t, &attr, PerfMgmtThread, NULL) != 0) {
printf("Error creating PerfWakeupFunc thread\n");
exit(0);
if (TmThreadSpawn(tv_mgmt, TVT_MGMT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
return;
}
/**
* Kills the wakeup and the management threads
*/
void PerfDestroyThreads()
{
perf_tc->flags |= PT_KILL;
/* prematurely wakeup, the mgmt and wakeup threads */
pthread_cond_broadcast(&perf_tc->tc_cond);
pthread_join(perf_tc->wakeup_t, NULL);
pthread_join(perf_tc->mgmt_t, NULL);
if (pthread_mutex_destroy(&perf_tc->wakeup_m) != 0) {
printf("Error destroying the mutex perf_tc->wakeup_m\n");
}
if (pthread_mutex_destroy(&perf_tc->mgmt_m) != 0) {
printf("Error destroying the mutex perf_tc->mgmt_m\n");
}
if (pthread_cond_destroy(&perf_tc->tc_cond) != 0) {
printf("Error destroying the condition variable perf_tc->tc_cond\n");
}
if (perf_tc != NULL) free(perf_tc);
return;
}
/**
* The management thread. This thread is responsible for writing the performance
* stats information.
* \brief The management thread. This thread is responsible for writing the
* performance stats information.
*
* @param arg is NULL always
* \param arg is NULL always
*/
void * PerfMgmtThread(void *arg)
{
ThreadVars *tv_local = (ThreadVars *)arg;
u_int8_t run = 1;
struct timespec cond_time;
@ -162,33 +118,37 @@ void * PerfMgmtThread(void *arg)
}
while (run) {
TmThreadTestThreadUnPaused(tv_local);
cond_time.tv_sec = time(NULL) + MGMTT_TTS;
cond_time.tv_nsec = 0;
pthread_mutex_lock(&perf_tc->mgmt_m);
pthread_cond_timedwait(&perf_tc->tc_cond, &perf_tc->mgmt_m,
&cond_time);
pthread_mutex_unlock(&perf_tc->mgmt_m);
pthread_mutex_lock(tv_local->m);
pthread_cond_timedwait(tv_local->cond, tv_local->m, &cond_time);
pthread_mutex_unlock(tv_local->m);
// sleep(MGMTT_TTS);
PerfOutputCounters();
if (perf_tc->flags & PT_KILL)
if (tv_local->flags & THV_KILL) {
tv_local->flags |= THV_CLOSED;
run = 0;
}
}
return NULL;
}
/**
* Wake up thread. This thread wakes up every TTS(time to sleep) seconds and
* sets the flag for every ThreadVars' PerfContext
* \brief Wake up thread. This thread wakes up every TTS(time to sleep) seconds
* and sets the flag for every ThreadVars' PerfContext
*
* @param arg is NULL always
* \param arg is NULL always
*/
void * PerfWakeupThread(void *arg)
{
ThreadVars *tv_local = (ThreadVars *)arg;
u_int8_t run = 1;
ThreadVars *tv = NULL;
PacketQueue *q = NULL;
@ -197,18 +157,18 @@ void * PerfWakeupThread(void *arg)
printf("PerfWakeupThread: spawned\n");
while (run) {
TmThreadTestThreadUnPaused(tv_local);
cond_time.tv_sec = time(NULL) + WUT_TTS;
cond_time.tv_nsec = 0;
pthread_mutex_lock(&perf_tc->wakeup_m);
pthread_cond_timedwait(&perf_tc->tc_cond, &perf_tc->wakeup_m,
&cond_time);
pthread_mutex_unlock(&perf_tc->wakeup_m);
pthread_mutex_lock(tv_local->m);
pthread_cond_timedwait(tv_local->cond, tv_local->m, &cond_time);
pthread_mutex_unlock(tv_local->m);
// sleep(WUT_TTS);
tv = tv_root;
tv = tv_root[TVT_PPT];
while (tv != NULL) {
if (tv->inq == NULL || tv->pctx.head == NULL) {
tv = tv->next;
@ -226,23 +186,26 @@ void * PerfWakeupThread(void *arg)
tv = tv->next;
}
if (perf_tc->flags & PT_KILL)
if (tv_local->flags & THV_KILL) {
tv_local->flags |= THV_CLOSED;
run = 0;
}
}
return NULL;
}
/**
* Registers a counter
* \brief Registers a counter
*
* @param cname holds the counter name
* @param tm_name holds the tm_name
* @param tid holds the tid running this module
* @param type holds the datatype of this counter variable
* @param head holds the PerfCounter
* \param cname Counter name to be registered
* \param tm_name Thread module name
* \param tid Thread id running this module instance
* \param type Datatype of this counter variable
* \param desc Description of this counter
* \param pctx PerfContext for this tm-tv instance
*
* @returns the counter id
* \retval the counter id
*/
u_int32_t PerfRegisterCounter(char *cname, char *tm_name, int type,
char *desc, PerfContext *pctx)
@ -335,11 +298,11 @@ u_int32_t PerfRegisterCounter(char *cname, char *tm_name, int type,
}
/**
* Adds a TM to the clubbed TM table. Multiple instances of the same TM are
* stacked together in a PCTMI container
* \brief Adds a TM to the clubbed TM table. Multiple instances of the same TM
* are stacked together in a PCTMI container
*
* @param tm_name is the name of the tm to be added
* @param pctx holds the PerfContext associated with the TM tm_name
* \param tm_name Name of the tm to be added to the table
* \param pctx PerfContext associated with the TM tm_name
*/
void PerfAddToClubbedTMTable(char *tm_name, PerfContext *pctx)
{
@ -414,13 +377,13 @@ void PerfAddToClubbedTMTable(char *tm_name, PerfContext *pctx)
/**
* Returns a counter array for counters in this id range(s_id - e_id)
* \brief Returns a counter array for counters in this id range(s_id - e_id)
*
* @param s_id is the start id of the counter
* @param e_id is the end id of the counter
* @param pctx is a pointer to the tv's PerfContext
* \param s_id Counter id of the first counter to be added to the array
* \param e_id Counter id of the last counter to be added to the array
* \param pctx Pointer to the tv's PerfContext
*
* @returns a counter-array in this(s_id-e_id) range for this tm instance
* \retval a counter-array in this(s_id-e_id) range for this TM instance
*/
PerfCounterArray * PerfGetCounterArrayRange(u_int32_t s_id, u_int32_t e_id,
PerfContext *pctx)
@ -471,11 +434,11 @@ PerfCounterArray * PerfGetCounterArrayRange(u_int32_t s_id, u_int32_t e_id,
}
/**
* Returns a counter array for all counters registered for this tm instance
* \brief Returns a counter array for all counters registered for this tm instance
*
* @param pctx is a pointer to the tv's PerfContext
* \param pctx Pointer to the tv's PerfContext
*
* @returns a counter-array for all the counters of this tm instance
* \retval a counter-array for all counters of this tm instance
*/
PerfCounterArray * PerfGetAllCountersArray(PerfContext *pctx)
{
@ -484,13 +447,15 @@ PerfCounterArray * PerfGetAllCountersArray(PerfContext *pctx)
/**
* Updates an individual counter
* \brief Updates an individual counter
*
* \param cname Name of the counter to be synced
* \param tm_name Thread module name
* \param id holds Counter id of the counter to be synced
* \param value Pointer to the local counter from the client thread
* \param pctx PerfContext for this tm-tv instance
*
* @param cname holds the counter name
* @param tm_name holds the tm name
* @param id holds the counter id for this tm
* @param value holds a pointer to the local counter from the client thread
* @param pctx holds the PerfContext associated with this instance of the tm
* \retval 1 on success, 0 on failure
*/
int PerfUpdateCounter(char *cname, char *tm_name, u_int32_t id, void *value,
PerfContext *pctx)
@ -535,11 +500,11 @@ int PerfUpdateCounter(char *cname, char *tm_name, u_int32_t id, void *value,
}
/**
* Syncs the counter array with the global counter variables
* \brief Syncs the counter array with the global counter variables
*
* @param pca holds a pointer to the PerfCounterArray
* @param pctx holds a pointer the the tv's PerfContext
* @param reset_lc indicates whether the local counter has to be reset or not
* \param pca Pointer to the PerfCounterArray
* \param pctx Pointer the the tv's PerfContext
* \param reset_lc Indicates whether the local counter has to be reset or not
*/
int PerfUpdateCounterArray(PerfCounterArray *pca, PerfContext *pctx, int reset_lc)
{
@ -582,7 +547,7 @@ int PerfUpdateCounterArray(PerfCounterArray *pca, PerfContext *pctx, int reset_l
}
/**
* The output interface dispatcher for the counter api
* \brief The output interface dispatcher for the counter api
*/
void PerfOutputCounters()
{
@ -605,11 +570,11 @@ void PerfOutputCounters()
}
/**
* The file output interface for the counter api
* \brief The file output interface for the counter api
*/
int PerfOutputCounterFileIface()
{
ThreadVars *tv = tv_root;
ThreadVars *tv = NULL;
PerfClubTMInst *pctmi = NULL;
PerfCounter *pc = NULL;
PerfCounter **pc_heads;
@ -645,25 +610,28 @@ int PerfOutputCounterFileIface()
"------------------\n");
if (perf_op_ctx->club_tm == 0) {
while (tv != NULL) {
pthread_mutex_lock(&tv->pctx.m);
pc = tv->pctx.head;
while (pc != NULL) {
ui64_cvalue = (u_int64_t *)pc->value->cvalue;
fprintf(perf_op_ctx->fp, "%-25s | %-25s | %-llu\n",
pc->name->cname, pc->name->tm_name, *ui64_cvalue);
//printf("%-10d %-10d %-10s %-llu\n", pc->name->tid, pc->id,
// pc->name->cname, *ui64_cvalue);
pc = pc->next;
for (i = 0; i < TVT_MAX; i++) {
tv = tv_root[i];
while (tv != NULL) {
pthread_mutex_lock(&tv->pctx.m);
pc = tv->pctx.head;
while (pc != NULL) {
ui64_cvalue = (u_int64_t *)pc->value->cvalue;
fprintf(perf_op_ctx->fp, "%-25s | %-25s | %-llu\n",
pc->name->cname, pc->name->tm_name, *ui64_cvalue);
//printf("%-10d %-10d %-10s %-llu\n", pc->name->tid, pc->id,
// pc->name->cname, *ui64_cvalue);
pc = pc->next;
}
pthread_mutex_unlock(&tv->pctx.m);
tv = tv->next;
}
pthread_mutex_unlock(&tv->pctx.m);
tv = tv->next;
fflush(perf_op_ctx->fp);
}
fflush(perf_op_ctx->fp);
return 1;
}
@ -717,12 +685,10 @@ int PerfOutputCounterFileIface()
}
/**
* Kills the perf threads and releases other resources.
* \brief Releases perf api resources.
*/
void PerfReleaseResources()
{
PerfDestroyThreads();
PerfReleaseOPCtx();
return;

26
src/counters.h
Unescape Escape View File

@ -1,4 +1,4 @@
/** Copyright (c) Open Information Security Foundation.
/** Copyright (c) 2009 Open Information Security Foundation.
* \author Anoop Saldanha <poonaatsoc@gmail.com>
*/
@ -6,14 +6,11 @@
#define __COUNTERS_H__
/** Time interval for syncing the local counters with the global ones */
/* Time interval for syncing the local counters with the global ones */
#define WUT_TTS 3
/** Time interval at which the mgmt thread o/p the stats */
/* Time interval at which the mgmt thread o/p the stats */
#define MGMTT_TTS 8
#define PT_RUN 0x01
#define PT_KILL 0x02
/* These 2 macros can only be used when all the registered counters for the tm,
* are in the counter array */
#define PerfCounterIncr(id, pca) do { \
@ -54,21 +51,6 @@ enum {
IFACE_SYSLOG,
};
/* Holds the thread context for the counter api */
typedef struct _PerfThreadContext {
pthread_t wakeup_t;
pthread_t mgmt_t;
/* state of the 2 threads, determined by PT_RUN AND PT_KILL */
u_int32_t flags;
/* need these mutexes just for calling pthread_cond_timewait() on tc_cond */
pthread_mutex_t wakeup_m;
pthread_mutex_t mgmt_m;
pthread_cond_t tc_cond;
} PerfThreadContext;
typedef struct _PerfCounterName {
char *cname;
char *tm_name;
@ -156,8 +138,6 @@ void PerfInitOPCtx(void);
void PerfSpawnThreads(void);
void PerfDestroyThreads(void);
void * PerfMgmtThread(void *);
void * PerfWakeupThread(void *);

158
src/eidps.c
Unescape Escape View File

@ -58,12 +58,12 @@
#include "flow-bit.h"
#include "pkt-var.h"
#include "l7-app-detect.h"
#include "util-cidr.h"
#include "util-unittest.h"
#include "util-time.h"
pthread_attr_t attr;
/*
* we put this here, because we only use it here in main.
*/
@ -202,7 +202,7 @@ int RunModeIdsPcap(char *iface) {
TimeModeSetLive();
/* create the threads */
ThreadVars *tv_receivepcap = TmThreadCreate("ReceivePcap","packetpool","packetpool","pickup-queue","simple","1slot_noinout");
ThreadVars *tv_receivepcap = TmThreadCreate("ReceivePcap","packetpool","packetpool","pickup-queue","simple","1slot_noinout", NULL, 0);
if (tv_receivepcap == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -214,12 +214,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_receivepcap,tm_module,(void *)iface);
if (TmThreadSpawn(tv_receivepcap) != 0) {
if (TmThreadSpawn(tv_receivepcap, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot");
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot", NULL, 0);
if (tv_decode1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Decode1\n");
exit(1);
@ -231,12 +231,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_decode1,tm_module,NULL);
if (TmThreadSpawn(tv_decode1) != 0) {
if (TmThreadSpawn(tv_decode1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot");
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot", NULL, 0);
if (tv_stream1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Stream1\n");
exit(1);
@ -248,12 +248,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_stream1,tm_module,NULL);
if (TmThreadSpawn(tv_stream1) != 0) {
if (TmThreadSpawn(tv_stream1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","verdict-queue","simple","1slot");
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","verdict-queue","simple","1slot", NULL, 0);
if (tv_detect1 == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -265,12 +265,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_detect1,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect1) != 0) {
if (TmThreadSpawn(tv_detect1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","verdict-queue","simple","1slot");
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","verdict-queue","simple","1slot", NULL, 0);
if (tv_detect2 == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -282,12 +282,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_detect2,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect2) != 0) {
if (TmThreadSpawn(tv_detect2, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_rreject = TmThreadCreate("RespondReject","verdict-queue","simple","alert-queue1","simple","1slot");
ThreadVars *tv_rreject = TmThreadCreate("RespondReject","verdict-queue","simple","alert-queue1","simple","1slot", NULL, 0);
if (tv_rreject == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -299,12 +299,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_rreject,tm_module,NULL);
if (TmThreadSpawn(tv_rreject) != 0) {
if (TmThreadSpawn(tv_rreject, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot");
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot", NULL, 0);
if (tv_alert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -323,12 +323,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm2SlotSetFunc2(tv_alert,tm_module,NULL);
if (TmThreadSpawn(tv_alert) != 0) {
if (TmThreadSpawn(tv_alert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot");
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot", NULL, 0);
if (tv_unified == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -348,13 +348,12 @@ int RunModeIdsPcap(char *iface) {
}
Tm2SlotSetFunc2(tv_unified,tm_module,NULL);
if (TmThreadSpawn(tv_unified) != 0) {
if (TmThreadSpawn(tv_unified, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot");
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot", NULL, 0);
if (tv_debugalert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -366,7 +365,7 @@ int RunModeIdsPcap(char *iface) {
}
Tm1SlotSetFunc(tv_debugalert,tm_module,NULL);
if (TmThreadSpawn(tv_debugalert) != 0) {
if (TmThreadSpawn(tv_debugalert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
@ -378,7 +377,7 @@ int RunModeIpsNFQ(void) {
TimeModeSetLive();
/* create the threads */
ThreadVars *tv_receivenfq = TmThreadCreate("ReceiveNFQ","packetpool","packetpool","pickup-queue","simple","1slot_noinout");
ThreadVars *tv_receivenfq = TmThreadCreate("ReceiveNFQ","packetpool","packetpool","pickup-queue","simple","1slot_noinout", NULL, 0);
if (tv_receivenfq == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -390,12 +389,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_receivenfq,tm_module,NULL);
if (TmThreadSpawn(tv_receivenfq) != 0) {
if (TmThreadSpawn(tv_receivenfq, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot");
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot", NULL, 0);
if (tv_decode1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Decode1\n");
exit(1);
@ -407,12 +406,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_decode1,tm_module,NULL);
if (TmThreadSpawn(tv_decode1) != 0) {
if (TmThreadSpawn(tv_decode1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot");
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot", NULL, 0);
if (tv_stream1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Stream1\n");
exit(1);
@ -424,12 +423,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_stream1,tm_module,NULL);
if (TmThreadSpawn(tv_stream1) != 0) {
if (TmThreadSpawn(tv_stream1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","verdict-queue","simple","1slot");
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","verdict-queue","simple","1slot", NULL, 0);
if (tv_detect1 == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -441,12 +440,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_detect1,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect1) != 0) {
if (TmThreadSpawn(tv_detect1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","verdict-queue","simple","1slot");
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","verdict-queue","simple","1slot", NULL, 0);
if (tv_detect2 == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -458,12 +457,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_detect2,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect2) != 0) {
if (TmThreadSpawn(tv_detect2, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_verdict = TmThreadCreate("Verdict","verdict-queue","simple","respond-queue","simple","1slot");
ThreadVars *tv_verdict = TmThreadCreate("Verdict","verdict-queue","simple","respond-queue","simple","1slot", NULL, 0);
if (tv_verdict == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -475,12 +474,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_verdict,tm_module,NULL);
if (TmThreadSpawn(tv_verdict) != 0) {
if (TmThreadSpawn(tv_verdict, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_rreject = TmThreadCreate("RespondReject","respond-queue","simple","alert-queue1","simple","1slot");
ThreadVars *tv_rreject = TmThreadCreate("RespondReject","respond-queue","simple","alert-queue1","simple","1slot", NULL, 0);
if (tv_rreject == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -492,12 +491,12 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_rreject,tm_module,NULL);
if (TmThreadSpawn(tv_rreject) != 0) {
if (TmThreadSpawn(tv_rreject, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot");
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot", NULL, 0);
if (tv_alert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -516,12 +515,12 @@ int RunModeIpsNFQ(void) {
}
Tm2SlotSetFunc2(tv_alert,tm_module,NULL);
if (TmThreadSpawn(tv_alert) != 0) {
if (TmThreadSpawn(tv_alert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot");
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot", NULL, 0);
if (tv_unified == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -541,13 +540,12 @@ int RunModeIpsNFQ(void) {
}
Tm2SlotSetFunc2(tv_unified,tm_module,NULL);
if (TmThreadSpawn(tv_unified) != 0) {
if (TmThreadSpawn(tv_unified, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot");
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot", NULL, 0);
if (tv_debugalert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -559,7 +557,7 @@ int RunModeIpsNFQ(void) {
}
Tm1SlotSetFunc(tv_debugalert,tm_module,NULL);
if (TmThreadSpawn(tv_debugalert) != 0) {
if (TmThreadSpawn(tv_debugalert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
@ -572,7 +570,7 @@ int RunModeFilePcap(char *file) {
TimeModeSetOffline();
/* create the threads */
ThreadVars *tv_receivepcap = TmThreadCreate("ReceivePcapFile","packetpool","packetpool","pickup-queue","simple","1slot");
ThreadVars *tv_receivepcap = TmThreadCreate("ReceivePcapFile","packetpool","packetpool","pickup-queue","simple","1slot", NULL, 0);
if (tv_receivepcap == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -584,12 +582,12 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_receivepcap,tm_module,file);
if (TmThreadSpawn(tv_receivepcap) != 0) {
if (TmThreadSpawn(tv_receivepcap, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot");
ThreadVars *tv_decode1 = TmThreadCreate("Decode1","pickup-queue","simple","decode-queue1","simple","1slot", NULL, 0);
if (tv_decode1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Decode1\n");
exit(1);
@ -601,12 +599,12 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_decode1,tm_module,NULL);
if (TmThreadSpawn(tv_decode1) != 0) {
if (TmThreadSpawn(tv_decode1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
//#if 0
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot");
ThreadVars *tv_stream1 = TmThreadCreate("Stream1","decode-queue1","simple","stream-queue1","simple","1slot", NULL, 0);
if (tv_stream1 == NULL) {
printf("ERROR: TmThreadsCreate failed for Stream1\n");
exit(1);
@ -618,12 +616,12 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_stream1,tm_module,NULL);
if (TmThreadSpawn(tv_stream1) != 0) {
if (TmThreadSpawn(tv_stream1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","alert-queue1","simple","1slot");
ThreadVars *tv_detect1 = TmThreadCreate("Detect1","stream-queue1","simple","packetpool","packetpool","1slot", NULL, 0);
//#endif
//ThreadVars *tv_detect1 = TmThreadCreate("Detect1","decode-queue1","simple","alert-queue1","simple","1slot");
if (tv_detect1 == NULL) {
@ -637,12 +635,12 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_detect1,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect1) != 0) {
if (TmThreadSpawn(tv_detect1, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","alert-queue1","simple","1slot");
ThreadVars *tv_detect2 = TmThreadCreate("Detect2","stream-queue1","simple","packetpool","packetpool","1slot", NULL, 0);
if (tv_detect2 == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -654,12 +652,12 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_detect2,tm_module,(void *)g_de_ctx);
if (TmThreadSpawn(tv_detect2) != 0) {
if (TmThreadSpawn(tv_detect2, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot");
ThreadVars *tv_alert = TmThreadCreate("AlertFastlog&Httplog","alert-queue1","simple","alert-queue2","simple","2slot", NULL, 0);
if (tv_alert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -678,12 +676,12 @@ int RunModeFilePcap(char *file) {
}
Tm2SlotSetFunc2(tv_alert,tm_module,NULL);
if (TmThreadSpawn(tv_alert) != 0) {
if (TmThreadSpawn(tv_alert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot");
ThreadVars *tv_unified = TmThreadCreate("AlertUnifiedLog","alert-queue2","simple","alert-queue3","simple","2slot", NULL, 0);
if (tv_unified == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -703,12 +701,12 @@ int RunModeFilePcap(char *file) {
}
Tm2SlotSetFunc2(tv_unified,tm_module,NULL);
if (TmThreadSpawn(tv_unified) != 0) {
if (TmThreadSpawn(tv_unified, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot");
ThreadVars *tv_debugalert = TmThreadCreate("AlertDebuglog","alert-queue3","simple","packetpool","packetpool","1slot", NULL, 0);
if (tv_debugalert == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -720,7 +718,7 @@ int RunModeFilePcap(char *file) {
}
Tm1SlotSetFunc(tv_debugalert,tm_module,NULL);
if (TmThreadSpawn(tv_debugalert) != 0) {
if (TmThreadSpawn(tv_debugalert, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
@ -735,7 +733,7 @@ int RunModeFilePcap2(char *file) {
TimeModeSetOffline();
/* create the threads */
ThreadVars *tv = TmThreadCreate("PcapFile","packetpool","packetpool","packetpool","packetpool","varslot");
ThreadVars *tv = TmThreadCreate("PcapFile","packetpool","packetpool","packetpool","packetpool","varslot", NULL, 0);
if (tv == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
@ -804,7 +802,7 @@ int RunModeFilePcap2(char *file) {
}
TmVarSlotSetFuncAppend(tv,tm_module,NULL);
if (TmThreadSpawn(tv) != 0) {
if (TmThreadSpawn(tv, TVT_PPT, THV_USE | THV_PAUSE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
@ -814,7 +812,6 @@ int RunModeFilePcap2(char *file) {
int main(int argc, char **argv)
{
int rc;
sigset_t set;
sigaddset(&set, SIGINT);
@ -919,10 +916,6 @@ int main(int argc, char **argv)
FlowInitConfig(FLOW_VERBOSE);
/* Initialize and set thread detached attribute */
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
SigLoadSignatures();
struct timeval start_time;
@ -934,34 +927,18 @@ int main(int argc, char **argv)
RunModeFilePcap(argv[1]);
//RunModeFilePcap2(argv[1]);
ThreadVars tv_flowmgr;
memset(&tv_flowmgr, 0, sizeof(ThreadVars));
printf("Creating FlowManagerThread...\n");
tv_flowmgr.name = "FlowManagerThread";
/* Spawn the flow manager thread */
FlowManagerThreadSpawn();
rc = pthread_create(&tv_flowmgr.t, &attr, FlowManagerThread, (void *)&tv_flowmgr);
if (rc) {
printf("ERROR; return code from pthread_create() is %d\n", rc);
exit(1);
}
TmThreadAppend(&tv_flowmgr);
/* Spawn the L7 App Detect thread */
L7AppDetectThreadSpawn();
#include "l7-app-detect.h"
ThreadVars tv_l7appdetect;
memset(&tv_l7appdetect, 0, sizeof(ThreadVars));
printf("Creating L7 Application layer detect thread (WIP)...\n");
tv_l7appdetect.name = "L7AppDetectThread";
rc = pthread_create(&tv_l7appdetect.t, &attr, L7AppDetectThread, (void *)&tv_l7appdetect);
if (rc) {
printf("ERROR; return code from pthread_create() is %d\n", rc);
exit(1);
}
TmThreadAppend(&tv_l7appdetect);
printf("Creating Stats threads...\n");
/* Spawn the perf counter threads */
PerfSpawnThreads();
/* Un-pause all the paused threads */
TmThreadContinueThreads();
while(1) {
if (sigflags) {
printf("signal received\n");
@ -998,8 +975,9 @@ int main(int argc, char **argv)
printf("time elapsed %lus\n", end_time.tv_sec - start_time.tv_sec);
PerfReleaseResources();
TmThreadKillThreads();
PerfReleaseResources();
#if 0
#ifdef DBG_PERF
printf("th_v[0].nfq_t->dbg_maxreadsize %d\n", th_v[0].nfq_t->dbg_maxreadsize);

24
src/flow.c
Unescape Escape View File

@ -4,6 +4,9 @@
#include "debug.h"
#include "decode.h"
#include "threads.h"
#include "tm-modules.h"
#include "threadvars.h"
#include "tm-threads.h"
#include "util-time.h"
@ -415,6 +418,8 @@ void *FlowManagerThread(void *td)
while (1)
{
TmThreadTestThreadUnPaused(th_v);
if (sleeping >= 100 || flow_flags & FLOW_EMERGENCY)
{
u_int32_t timeout_new = flow_config.timeout_new;
@ -474,3 +479,22 @@ void *FlowManagerThread(void *td)
pthread_exit((void *) 0);
}
void FlowManagerThreadSpawn()
{
ThreadVars *tv_flowmgr = NULL;
tv_flowmgr = TmThreadCreate("FlowManagerThread", NULL, NULL, NULL, NULL,
"custom", FlowManagerThread, 0);
if (tv_flowmgr == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
}
if (TmThreadSpawn(tv_flowmgr, TVT_PPT, THV_USE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
printf("Flow Manager thread spawned\n");
return;
}

2
src/flow.h
Unescape Escape View File

@ -93,5 +93,7 @@ void FlowDecrUsecnt(ThreadVars *, Packet *);
void *FlowManagerThread(void *td);
void FlowManagerThreadSpawn(void);
#endif /* __FLOW_H__ */

24
src/l7-app-detect.c
Unescape Escape View File

@ -4,6 +4,9 @@
#include "debug.h"
#include "decode.h"
#include "threads.h"
#include "tm-modules.h"
#include "threadvars.h"
#include "tm-threads.h"
#include "util-print.h"
#include "util-pool.h"
@ -68,6 +71,8 @@ void *L7AppDetectThread(void *td)
/* main loop */
while(run) {
TmThreadTestThreadUnPaused(tv);
/* grab a msg, can return NULL on signals */
StreamMsg *smsg = StreamMsgGetFromQueue(stream_q);
if (smsg != NULL) {
@ -133,3 +138,22 @@ void *L7AppDetectThread(void *td)
pthread_exit((void *) 0);
}
void L7AppDetectThreadSpawn()
{
ThreadVars *tv_l7appdetect = NULL;
tv_l7appdetect = TmThreadCreate("L7AppDetectThread", NULL, NULL, NULL, NULL,
"custom", L7AppDetectThread, 0);
if (tv_l7appdetect == NULL) {
printf("ERROR: TmThreadsCreate failed\n");
exit(1);
}
if (TmThreadSpawn(tv_l7appdetect, TVT_PPT, THV_USE) != 0) {
printf("ERROR: TmThreadSpawn failed\n");
exit(1);
}
printf("L7_App_Detect thread created\n");
return;
}

2
src/l7-app-detect.h
Unescape Escape View File

@ -3,4 +3,6 @@
void *L7AppDetectThread(void *td);
void L7AppDetectThreadSpawn(void);
#endif /* __L7_APP_DETECT_H__ */

8
src/threadvars.h
Unescape Escape View File

@ -9,8 +9,9 @@
/** Thread flags set and read by threads to control the threads */
#define THV_USE 0x01 /** thread is in use */
#define THV_KILL 0x02 /** thread should stop and prepare tp get joined. */
#define THV_CLOSED 0x04 /** thread done, should be joinable */
#define THV_PAUSE 0x02
#define THV_KILL 0x04
#define THV_CLOSED 0x08 /* thread done, should be joinable */
/** \brief Per thread variable structure */
typedef struct ThreadVars_ {
@ -36,6 +37,9 @@ typedef struct ThreadVars_ {
PerfContext pctx;
PerfCounterArray *pca;
pthread_mutex_t *m;
pthread_cond_t *cond;
struct ThreadVars_ *next;
struct ThreadVars_ *prev;
} ThreadVars;

292
src/tm-threads.c
Unescape Escape View File

@ -1,4 +1,6 @@
/* Copyright (c) 2008 Victor Julien <victor@inliniac.net> */
/** Copyright (c) 2009 Open Information Security Foundation.
* \author Victor Julien <victor@inliniac.net>
*/
#include <sys/types.h> /* for gettid(2) */
#define _GNU_SOURCE
@ -11,13 +13,16 @@
#include "tm-queues.h"
#include "tm-queuehandlers.h"
#include "tm-modules.h"
#include "tm-threads.h"
/* prototypes */
static int SetCPUAffinity(int cpu);
/* root of the threadvars list */
ThreadVars *tv_root = NULL;
ThreadVars *tv_root[TVT_MAX] = { NULL };
/* lock to protect tv_root */
pthread_mutex_t tv_root_lock = PTHREAD_MUTEX_INITIALIZER;
typedef struct TmSlot_ {
/* function pointers */
@ -79,6 +84,8 @@ void *TmThreadsSlot1NoIn(void *td) {
memset(&s->s.slot_pq, 0, sizeof(PacketQueue));
while(run) {
TmThreadTestThreadUnPaused(tv);
r = s->s.SlotFunc(tv, p, s->s.slot_data, &s->s.slot_pq);
while (s->s.slot_pq.len > 0) {
Packet *extra = PacketDequeue(&s->s.slot_pq);
@ -134,6 +141,8 @@ void *TmThreadsSlot1NoOut(void *td) {
memset(&s->s.slot_pq, 0, sizeof(PacketQueue));
while(run) {
TmThreadTestThreadUnPaused(tv);
p = tv->tmqh_in(tv);
r = s->s.SlotFunc(tv, p, s->s.slot_data, /* no outqh no pq */NULL);
@ -185,6 +194,8 @@ void *TmThreadsSlot1NoInOut(void *td) {
memset(&s->s.slot_pq, 0, sizeof(PacketQueue));
while(run) {
TmThreadTestThreadUnPaused(tv);
r = s->s.SlotFunc(tv, NULL, s->s.slot_data, /* no outqh, no pq */NULL);
//printf("%s: TmThreadsSlot1NoInNoOut: r %d\n", tv->name, r);
/* XXX handle error */
@ -239,6 +250,8 @@ void *TmThreadsSlot1(void *td) {
memset(&s->s.slot_pq, 0, sizeof(PacketQueue));
while(run) {
TmThreadTestThreadUnPaused(tv);
/* input a packet */
p = tv->tmqh_in(tv);
@ -317,6 +330,8 @@ void *TmThreadsSlot2(void *td) {
}
while(run) {
TmThreadTestThreadUnPaused(tv);
/* input a packet */
p = tv->tmqh_in(tv);
@ -437,6 +452,8 @@ void *TmThreadsSlot3(void *td) {
}
while(run) {
TmThreadTestThreadUnPaused(tv);
/* input a packet */
p = tv->tmqh_in(tv);
@ -618,6 +635,8 @@ void *TmThreadsSlotVar(void *td) {
}
while(run) {
TmThreadTestThreadUnPaused(tv);
/* input a packet */
p = tv->tmqh_in(tv);
//printf("TmThreadsSlotVar: %p\n", p);
@ -661,9 +680,19 @@ void *TmThreadsSlotVar(void *td) {
pthread_exit((void *) 0);
}
int TmThreadSetSlots(ThreadVars *tv, char *name) {
int TmThreadSetSlots(ThreadVars *tv, char *name, void *(*fn_p)(void *)) {
u_int16_t size = 0;
if (name == NULL) {
if (fn_p == NULL) {
printf("Both slot name and function pointer can't be NULL inside "
"TmThreadSetSlots\n");
goto error;
}
else
name = "custom";
}
if (strcmp(name, "1slot") == 0) {
size = sizeof(Tm1Slot);
tv->tm_func = TmThreadsSlot1;
@ -685,6 +714,12 @@ int TmThreadSetSlots(ThreadVars *tv, char *name) {
} else if (strcmp(name, "varslot") == 0) {
size = sizeof(TmVarSlot);
tv->tm_func = TmThreadsSlotVar;
} else if (strcmp(name, "custom") == 0) {
if (fn_p == NULL)
goto error;
tv->tm_func = fn_p;
return 0;
}
tv->tm_slots = malloc(size);
@ -835,7 +870,25 @@ int TmThreadSetCPUAffinity(ThreadVars *tv, int cpu) {
return 0;
}
ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name, char *outq_name, char *outqh_name, char *slots) {
/**
* \brief Creates and returns the TV instance for a new thread.
*
* \param name Name of this TV instance
* \param inq_name Incoming queue name
* \param inqh_name Incoming queue handler name as set by TmqhSetup()
* \param outq_name Outgoing queue name
* \param outqh_name Outgoing queue handler as set by TmqhSetup()
* \param slots String representation for the slot function to be used
* \param fn_p Pointer to function when \"slots\" is of type \"custom\"
* \param mucond Flag to indicate whether to initialize the condition
* and the mutex variables for this newly created TV.
*
* \retval the newly created TV instance, or NULL on error
*/
ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name,
char *outq_name, char *outqh_name, char *slots,
void * (*fn_p)(void *), int mucond)
{
ThreadVars *tv = NULL;
Tmq *tmq = NULL;
Tmqh *tmqh = NULL;
@ -889,19 +942,28 @@ ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name, char *ou
//printf("TmThreadCreate: tv->tmqh_out %p\n", tv->tmqh_out);
}
if (TmThreadSetSlots(tv, slots) != 0) {
if (TmThreadSetSlots(tv, slots, fn_p) != 0) {
goto error;
}
if (mucond != 0)
TmThreadInitMC(tv);
return tv;
error:
printf("ERROR: failed to setup a thread.\n");
return NULL;
}
void TmThreadAppend(ThreadVars *tv) {
if (tv_root == NULL) {
tv_root = tv;
/**
* \brief Appends this TV to tv_root based on its type
*
* \param type holds the type this TV belongs to.
*/
void TmThreadAppend(ThreadVars *tv, int type)
{
if (tv_root[type] == NULL) {
tv_root[type] = tv;
tv->next = NULL;
tv->prev = NULL;
@ -909,7 +971,7 @@ void TmThreadAppend(ThreadVars *tv) {
return;
}
ThreadVars *t = tv_root;
ThreadVars *t = tv_root[type];
while (t) {
if (t->next == NULL) {
@ -926,62 +988,102 @@ void TmThreadAppend(ThreadVars *tv) {
}
void TmThreadKillThreads(void) {
ThreadVars *t = tv_root;
ThreadVars *t = NULL;
int i = 0;
while (t) {
t->flags |= THV_KILL;
//printf("TmThreadKillThreads: told thread %s to stop\n", t->name);
for (i = 0; i < TVT_MAX; i++) {
t = tv_root[i];
/* XXX hack */
StreamMsgSignalQueueHack();
if (t->inq != NULL) {
int i;
while (t) {
t->flags |= THV_KILL;
printf("TmThreadKillThreads: told thread %s to stop\n", t->name);
//printf("TmThreadKillThreads: t->inq->usecnt %u\n", t->inq->usecnt);
/* XXX hack */
StreamMsgSignalQueueHack();
/* make sure our packet pending counter doesn't block */
pthread_cond_signal(&cond_pending);
if (t->inq != NULL) {
int i;
/* signal the queue for the number of users */
for (i = 0; i < t->inq->usecnt; i++)
pthread_cond_signal(&trans_q[t->inq->id].cond_q);
//printf("TmThreadKillThreads: t->inq->usecnt %u\n", t->inq->usecnt);
/* to be sure, signal more */
int cnt = 0;
while (1) {
if (t->flags & THV_CLOSED) {
//printf("signalled the thread %d times\n", cnt);
break;
}
/* make sure our packet pending counter doesn't block */
pthread_cond_signal(&cond_pending);
cnt++;
/* signal the queue for the number of users */
for (i = 0; i < t->inq->usecnt; i++)
pthread_cond_signal(&trans_q[t->inq->id].cond_q);
usleep(100);
/* to be sure, signal more */
int cnt = 0;
while (1) {
if (t->flags & THV_CLOSED) {
printf("signalled the thread %d times\n", cnt);
break;
}
cnt++;
for (i = 0; i < t->inq->usecnt; i++)
pthread_cond_signal(&trans_q[t->inq->id].cond_q);
usleep(100);
}
printf("TmThreadKillThreads: signalled t->inq->id %u\n", t->inq->id);
}
//printf("TmThreadKillThreads: signalled t->inq->id %u\n", t->inq->id);
}
if (t->cond != NULL ) {
int cnt = 0;
while (1) {
if (t->flags & THV_CLOSED) {
printf("signalled the thread %d times\n", cnt);
break;
}
/* join it */
pthread_join(t->t, NULL);
printf("TmThreadKillThreads: thread %s stopped\n", t->name);
cnt++;
t = t->next;
pthread_cond_broadcast(t->cond);
usleep(100);
}
}
/* join it */
pthread_join(t->t, NULL);
printf("TmThreadKillThreads: thread %s stopped\n", t->name);
t = t->next;
}
}
}
int TmThreadSpawn(ThreadVars *tv) {
/**
* \brief Spawns a thread associated with the ThreadVars instance tv
*
* \param type Type this TV belongs to.
* \param flags Flags that should be set for this thread
*
* \retval 0 on success and -1 on failure
*/
int TmThreadSpawn(ThreadVars *tv, int type, int flags)
{
pthread_attr_t attr;
if (type < 0 || type >= TVT_MAX) {
printf("ThreadVars of category %d does not exist\n", type);
return -1;
}
if (tv->tm_func == NULL) {
printf("ERROR: no thread function set\n");
return -1;
}
tv->flags = flags;
/* Initialize and set thread detached attribute */
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
@ -992,8 +1094,116 @@ int TmThreadSpawn(ThreadVars *tv) {
return -1;
}
TmThreadAppend(tv);
TmThreadAppend(tv, type);
return 0;
}
/**
* \brief Initializes the mutex and condition variables for this TV
*
* \param tv Pointer to a TV instance
*/
void TmThreadInitMC(ThreadVars *tv)
{
if ( (tv->m = malloc(sizeof(pthread_mutex_t))) == NULL) {
printf("Error allocating memory\n");
exit(0);
}
if (pthread_mutex_init(tv->m, NULL) != 0) {
printf("Error initializing the tv->m mutex\n");
exit(0);
}
if ( (tv->cond = malloc(sizeof(pthread_cond_t))) == NULL) {
printf("Error allocating memory\n");
exit(0);
}
if (pthread_cond_init(tv->cond, NULL) != 0) {
printf("Error initializing the tv->cond condition variable\n");
exit(0);
}
}
/**
* \brief Tests if the thread represented in the arg has been unpaused or not.
*
* The function would return if the thread tv has been unpaused or if the
* kill flag for the thread has been set.
*
* \param tv Pointer to the TV instance.
*/
void TmThreadTestThreadUnPaused(ThreadVars *tv)
{
while (tv->flags & THV_PAUSE) {
usleep(100);
if (tv->flags & THV_KILL)
break;
}
return;
}
/**
* \brief Unpauses a thread
*
* \param tv Pointer to a TV instance that has to be unpaused
*/
void TmThreadContinue(ThreadVars *tv)
{
tv->flags &= ~THV_PAUSE;
return;
}
/**
* \brief Unpauses all threads present in tv_root
*/
void TmThreadContinueThreads()
{
ThreadVars *tv = NULL;
int i = 0;
for (i = 0; i < TVT_MAX; i++) {
tv = tv_root[i];
while (tv != NULL) {
TmThreadContinue(tv);
tv = tv->next;
}
}
return;
}
/**
* \brief Pauses a thread
*
* \param tv Pointer to a TV instance that has to be paused
*/
void TmThreadPause(ThreadVars *tv)
{
tv->flags |= THV_PAUSE;
return;
}
/**
* \brief Pauses all threads present in tv_root
*/
void TmThreadPauseThreads()
{
ThreadVars *tv = NULL;
int i = 0;
for (i = 0; i < TVT_MAX; i++) {
tv = tv_root[i];
while (tv != NULL) {
TmThreadPause(tv);
tv = tv->next;
}
}
return;
}

43
src/tm-threads.h
Unescape Escape View File

@ -1,20 +1,55 @@
#ifndef __TM_THREADS_H__
#define __TM_THREADS_H__
extern ThreadVars *tv_root;
/* ThreadVars type */
enum {
TVT_PPT,
TVT_MGMT,
TVT_MAX,
};
extern ThreadVars *tv_root[TVT_MAX];
extern pthread_mutex_t tv_root_lock;
void Tm1SlotSetFunc(ThreadVars *, TmModule *, void *);
void Tm2SlotSetFunc1(ThreadVars *, TmModule *, void *);
void Tm2SlotSetFunc2(ThreadVars *, TmModule *, void *);
void Tm3SlotSetFunc1(ThreadVars *, TmModule *, void *);
void Tm3SlotSetFunc2(ThreadVars *, TmModule *, void *);
void Tm3SlotSetFunc3(ThreadVars *, TmModule *, void *);
void TmVarSlotSetFuncAppend(ThreadVars *, TmModule *, void *);
ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name, char *outq_name, char *outqh_name, char *slots);
int TmThreadSpawn(ThreadVars *);
ThreadVars *TmThreadCreate(char *name, char *inq_name, char *inqh_name,
char *outq_name, char *outqh_name, char *slots,
void *(fn_p)(void *), int);
int TmThreadSpawn(ThreadVars *, int, int);
void TmThreadKillThreads(void);
void TmThreadAppend(ThreadVars *);
void TmThreadAppend(ThreadVars *, int);
int TmThreadSetCPUAffinity(ThreadVars *, int);
void TmThreadInitMC(ThreadVars *);
void TmThreadTestThreadUnPaused(ThreadVars *);
void TmThreadContinue(ThreadVars *);
void TmThreadContinueThreads(void);
void TmThreadPause(ThreadVars *);
void TmThreadPauseThreads(void);
#endif /* __TM_THREADS_H__ */

Write Preview
Loading…
Cancel
Save