From beaa909eb84ea801202eb5e621315e64f37570f2 Mon Sep 17 00:00:00 2001 From: Eric Leblond Date: Mon, 3 Oct 2011 23:56:36 +0200 Subject: [PATCH] Add "workers" runmode. Previous commits have considerabily empowered the "single" mode which could contain multiple threads. This behaviour was not a target for this runmode and the following patch remedies to the situation by introducing the "workers" mode where each thread do all the tasks from acquisition to logging. This runmode is currently implemented for af-packet and pf-ring. --- src/runmode-af-packet.c | 41 ++++++++++++++++ src/runmode-af-packet.h | 1 + src/runmode-pfring.c | 43 ++++++++++++++++- src/runmode-pfring.h | 1 + src/util-runmodes.c | 104 ++++++++++++++++++++++++++++++++-------- src/util-runmodes.h | 8 ++++ 6 files changed, 178 insertions(+), 20 deletions(-) diff --git a/src/runmode-af-packet.c b/src/runmode-af-packet.c index 2a49acff97..3d9a71768b 100644 --- a/src/runmode-af-packet.c +++ b/src/runmode-af-packet.c @@ -69,6 +69,10 @@ void RunModeIdsAFPRegister(void) RunModeRegisterNewRunMode(RUNMODE_AFP_DEV, "single", "Single threaded af-packet mode", RunModeIdsAFPSingle); + RunModeRegisterNewRunMode(RUNMODE_AFP_DEV, "workers", + "Workers af-packet mode, each thread does all" + " tasks from acquisition to logging", + RunModeIdsAFPWorkers); default_mode_autofp = "autofp"; RunModeRegisterNewRunMode(RUNMODE_AFP_DEV, "autofp", "Multi socket AF_PACKET mode. Packets from " @@ -325,3 +329,40 @@ int RunModeIdsAFPSingle(DetectEngineCtx *de_ctx) #endif /* HAVE_AF_PACKET */ SCReturnInt(0); } + +/** + * \brief Workers version of the AF_PACKET processing. + * + * Start N threads with each thread doing all the work. + * + */ +int RunModeIdsAFPWorkers(DetectEngineCtx *de_ctx) +{ +#ifdef HAVE_AF_PACKET + int ret; + char *live_dev = NULL; +#endif + SCEnter(); +#ifdef HAVE_AF_PACKET + + RunModeInitialize(); + TimeModeSetLive(); + + ConfGet("af-packet.live-interface", &live_dev); + + ret = RunModeSetLiveCaptureWorkers(de_ctx, + ParseAFPConfig, + AFPConfigGeThreadsCount, + "ReceiveAFP", + "DecodeAFP", "AFPacket", + live_dev); + if (ret != 0) { + printf("ERROR: Unable to start runmode\n"); + exit(EXIT_FAILURE); + } + + SCLogInfo("RunModeIdsAFPSingle initialised"); + +#endif /* HAVE_AF_PACKET */ + SCReturnInt(0); +} diff --git a/src/runmode-af-packet.h b/src/runmode-af-packet.h index c1afea3577..b6637861ac 100644 --- a/src/runmode-af-packet.h +++ b/src/runmode-af-packet.h @@ -26,6 +26,7 @@ int RunModeIdsAFPAuto(DetectEngineCtx *); int RunModeIdsAFPSingle(DetectEngineCtx *); int RunModeIdsAFPAutoFp(DetectEngineCtx *); +int RunModeIdsAFPWorkers(DetectEngineCtx *); void RunModeIdsAFPRegister(void); const char *RunModeAFPGetDefaultMode(void); diff --git a/src/runmode-pfring.c b/src/runmode-pfring.c index 27e9e3efaf..9a2e91e257 100644 --- a/src/runmode-pfring.c +++ b/src/runmode-pfring.c @@ -70,8 +70,12 @@ void RunModeIdsPfringRegister(void) "detect thread", RunModeIdsPfringAutoFp); RunModeRegisterNewRunMode(RUNMODE_PFRING, "single", - "Single threaded pfringt mode", + "Single threaded pfring mode", RunModeIdsPfringSingle); + RunModeRegisterNewRunMode(RUNMODE_PFRING, "workers", + "Workers pfring mode, each thread does all" + " tasks from acquisition to logging", + RunModeIdsPfringWorkers); return; } @@ -454,3 +458,40 @@ int RunModeIdsPfringSingle(DetectEngineCtx *de_ctx) return 0; } + +int RunModeIdsPfringWorkers(DetectEngineCtx *de_ctx) +{ + SCEnter(); + +/* We include only if pfring is enabled */ +#ifdef HAVE_PFRING + int ret; + char *live_dev = NULL; + ConfigIfaceParserFunc tparser; + + RunModeInitialize(); + + TimeModeSetLive(); + + ret = GetDevAndParser(&live_dev, &tparser); + if (ret != 0) { + printf("ERROR: Unabme to get parser and interface params\n"); + exit(EXIT_FAILURE); + } + + ret = RunModeSetLiveCaptureWorkers(de_ctx, + tparser, + PfringConfigGeThreadsCount, + "ReceivePfring", + "DecodePfring", "RxPFR", + live_dev); + if (ret != 0) { + printf("ERROR: Unable to start runmode\n"); + exit(EXIT_FAILURE); + } + + SCLogInfo("RunModeIdsPfringSingle initialised"); +#endif /* HAVE_PFRING */ + + return 0; +} diff --git a/src/runmode-pfring.h b/src/runmode-pfring.h index aeb999b86a..9ba7d5c93f 100644 --- a/src/runmode-pfring.h +++ b/src/runmode-pfring.h @@ -28,6 +28,7 @@ int RunModeIdsPfringAuto(DetectEngineCtx *); int RunModeIdsPfringAutoFp(DetectEngineCtx *de_ctx); int RunModeIdsPfringSingle(DetectEngineCtx *de_ctx); +int RunModeIdsPfringWorkers(DetectEngineCtx *de_ctx); void RunModeIdsPfringRegister(void); const char *RunModeIdsPfringGetDefaultMode(void); diff --git a/src/util-runmodes.c b/src/util-runmodes.c index 8ef54c6d06..8668a9c72b 100644 --- a/src/util-runmodes.c +++ b/src/util-runmodes.c @@ -567,41 +567,37 @@ int RunModeSetLiveCaptureAutoFp(DetectEngineCtx *de_ctx, return 0; } -int RunModeSetLiveCaptureSingle(DetectEngineCtx *de_ctx, +static int RunModeSetLiveCaptureWorkersForDevice(DetectEngineCtx *de_ctx, ConfigIfaceParserFunc ConfigParser, ConfigIfaceThreadsCountFunc mod_threads_count, char *recv_mod_name, char *decode_mod_name, char *thread_name, - const char *live_dev) + const char *live_dev, void *aconf, + unsigned char single_mode) { - int nlive = LiveGetDeviceCount(); - void *aconf; - int threads_count; int thread; + int threads_count; - if (nlive > 1) { - SCLogError(SC_ERR_RUNMODE, - "Can't use single runmode with multiple device"); - exit(EXIT_FAILURE); - } - - if (live_dev != NULL) { - aconf = ConfigParser(live_dev); + if (single_mode) { + threads_count = 1; } else { - char *live_dev_c = LiveGetDevice(0); - aconf = ConfigParser(live_dev_c); + threads_count = mod_threads_count(aconf); + SCLogInfo("Going to use %" PRId32 " thread(s)", threads_count); } - threads_count = mod_threads_count(aconf); - SCLogInfo("Going to use %" PRId32 " thread(s)", threads_count); /* create the threads */ for (thread = 0; thread < threads_count; thread++) { - char tname[12]; + char tname[20]; char *n_thread_name = NULL; ThreadVars *tv = NULL; TmModule *tm_module = NULL; - snprintf(tname, sizeof(tname), "%s%"PRIu16, thread_name, thread+1); + if (single_mode) { + snprintf(tname, sizeof(tname), "%s", thread_name); + } else { + snprintf(tname, sizeof(tname), "%s%s%"PRIu16, + thread_name, live_dev, thread+1); + } n_thread_name = SCStrdup(tname); tv = TmThreadCreatePacketHandler(n_thread_name, "packetpool", "packetpool", @@ -650,3 +646,73 @@ int RunModeSetLiveCaptureSingle(DetectEngineCtx *de_ctx, return 0; } + +int RunModeSetLiveCaptureWorkers(DetectEngineCtx *de_ctx, + ConfigIfaceParserFunc ConfigParser, + ConfigIfaceThreadsCountFunc mod_threads_count, + char *recv_mod_name, + char *decode_mod_name, char *thread_name, + const char *live_dev) +{ + int nlive = LiveGetDeviceCount(); + void *aconf; + int ldev; + + for (ldev = 0; ldev < nlive; ldev++) { + char *live_dev_c = NULL; + aconf = ConfigParser(live_dev_c); + if (live_dev != NULL) { + aconf = ConfigParser(live_dev); + live_dev_c = SCStrdup(live_dev); + } else { + live_dev_c = LiveGetDevice(ldev); + aconf = ConfigParser(live_dev_c); + } + RunModeSetLiveCaptureWorkersForDevice(de_ctx, + ConfigParser, + mod_threads_count, + recv_mod_name, + decode_mod_name, + thread_name, + live_dev_c, + aconf, + 0); + } + + return 0; +} + +int RunModeSetLiveCaptureSingle(DetectEngineCtx *de_ctx, + ConfigIfaceParserFunc ConfigParser, + ConfigIfaceThreadsCountFunc mod_threads_count, + char *recv_mod_name, + char *decode_mod_name, char *thread_name, + const char *live_dev) +{ + int nlive = LiveGetDeviceCount(); + void *aconf; + + if (nlive > 1) { + SCLogError(SC_ERR_RUNMODE, + "Can't use single runmode with multiple device"); + exit(EXIT_FAILURE); + } + + if (live_dev != NULL) { + aconf = ConfigParser(live_dev); + } else { + char *live_dev_c = LiveGetDevice(0); + aconf = ConfigParser(live_dev_c); + /* \todo Set threads number in config to 1 */ + } + + return RunModeSetLiveCaptureWorkersForDevice(de_ctx, + ConfigParser, + mod_threads_count, + recv_mod_name, + decode_mod_name, + thread_name, + live_dev, + aconf, + 1); +} diff --git a/src/util-runmodes.h b/src/util-runmodes.h index e16f8d8d8f..d284032598 100644 --- a/src/util-runmodes.h +++ b/src/util-runmodes.h @@ -46,4 +46,12 @@ int RunModeSetLiveCaptureSingle(DetectEngineCtx *de_ctx, char *decode_mod_name, char *thread_name, const char *live_dev); +int RunModeSetLiveCaptureWorkers(DetectEngineCtx *de_ctx, + ConfigIfaceParserFunc configparser, + ConfigIfaceThreadsCountFunc mod_threads_count, + char *recv_mod_name, + char *decode_mod_name, char *thread_name, + const char *live_dev); + + #endif /* __UTIL_RUNMODES_H__ */