/*
	Copyright (C) Slava Astashonok <sla@0n.ru>

	This program is free software; you can redistribute it and/or
	modify it under the terms of the GNU General Public License.

	$Id: fprobe-ulog.c,v 1.1.2.4 2005/01/30 09:06:19 sla Exp $

			Sapan Bhatia <sapanb@cs.princeton.edu> 
			
	7/11/2007	Added data collection (-f) functionality, slice_id support in the header and log file
			rotation.

	15/11/2007	Added check to make sure fprobe doesn't overflow the disk. Also added a test facility.

*/

#include <common.h>

/* stdout, stderr, freopen() */
#include <stdio.h>

/* atoi(), exit() */
#include <stdlib.h>

/* getopt(), alarm(), getpid(), sedsid(), chdir() */
#include <unistd.h>

/* strerror() */
#include <string.h>

/* sig*() */
#include <signal.h>

/* statfs() */

#include <sys/vfs.h>

#include <libipulog/libipulog.h>
/* #include "vserver.h" */

struct ipulog_handle {
	int fd;
	u_int8_t blocking;
	struct sockaddr_nl local;
	struct sockaddr_nl peer;
	struct nlmsghdr* last_nlhdr;
};

/* inet_*() (Linux, FreeBSD, Solaris), getpid() */
#include <sys/types.h>
#include <netinet/in_systm.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netinet/ip.h>
#include <netinet/tcp.h>
#include <netinet/udp.h>
#include <netinet/ip_icmp.h>
#include <net/if.h>

#include <sys/param.h>
#include <pwd.h>
#ifdef OS_LINUX
#include <grp.h>
#endif

/* pthread_*() */
#include <pthread.h>

/* errno */
#include <errno.h>

/* getaddrinfo() */
#include <netdb.h>

/* nanosleep() */
#include <time.h>

/* gettimeofday() */
#include <sys/time.h>

/* scheduling */
#include <sched.h>

/* select() (POSIX)*/
#include <sys/select.h>

/* open() */
#include <sys/stat.h>
#include <fcntl.h>

#include <fprobe-ulog.h>
#include <my_log.h>
#include <my_getopt.h>
#include <netflow.h>
#include <hash.h>
#include <mem.h>

#define PIDFILE 		"/var/log/fprobe-ulog.pid"
#define LAST_EPOCH_FILE 	"/var/log/fprobe_last_epoch"
#define MAX_EPOCH_SIZE		sizeof("32767")
#define STD_NETFLOW_PDU

enum {
	aflag,
	Bflag,
	bflag,
	cflag,
	dflag,
	Dflag,
	eflag,
	Eflag,
	fflag,
	gflag,
	hflag,
	lflag,
	mflag,
	Mflag,
	nflag,
	qflag,
	rflag,
	sflag,
	tflag,
	Tflag,
	Uflag,
	uflag,
	vflag,
	Wflag,
	Xflag,
};

static struct getopt_parms parms[] = {
	{'a', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'B', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'b', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'c', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'d', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'D', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'e', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'E', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'f', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'g', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'h', 0, 0, 0},
	{'l', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'m', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'M', 0, 0, 0},
	{'n', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'q', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'r', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'s', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'t', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'T', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'U', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'u', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'v', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{'X', MY_GETOPT_ARG_REQUIRED, 0, 0},
	{0, 0, 0, 0}
};

extern char *optarg;
extern int optind, opterr, optopt;
extern int errno;

extern struct NetFlow NetFlow1;
extern struct NetFlow NetFlow5;
extern struct NetFlow NetFlow7;

#define START_DATA_FD -5
#define mark_is_tos parms[Mflag].count
static unsigned scan_interval = 5;
static unsigned int min_free = 0;
static int frag_lifetime = 30;
static int inactive_lifetime = 60;
static int active_lifetime = 300;
static int sockbufsize;
#define BULK_QUANTITY_MAX (unsigned)(mem_index_t)(-1)
#if (MEM_BITS == 0) || (MEM_BITS == 16)
#define BULK_QUANTITY 10000
#else
#define BULK_QUANTITY 200
#endif

static unsigned epoch_length=60, log_epochs=1; 
static unsigned cur_epoch=0,prev_uptime=0,last_peak=0;

static unsigned bulk_quantity = BULK_QUANTITY;
static unsigned pending_queue_length = 100;
static struct NetFlow *netflow = &NetFlow5;
static unsigned verbosity = 6;
static unsigned log_dest = MY_LOG_SYSLOG;
static struct Time start_time;
static long start_time_offset;
static int off_tl;
/* From mem.c */
extern unsigned total_elements;
extern unsigned free_elements;
extern unsigned total_memory;
#if ((DEBUG) & DEBUG_I)
static unsigned emit_pkts, emit_queue;
static uint64_t size_total;
static unsigned pkts_total, pkts_total_fragmented;
static unsigned pkts_ignored, pkts_lost_capture, pkts_lost_unpending;
static unsigned pkts_pending, pkts_pending_done;
static unsigned pending_queue_trace, pending_queue_trace_candidate;
static unsigned flows_total, flows_fragmented;
#endif
static unsigned emit_count;
static uint32_t emit_sequence;
static unsigned emit_rate_bytes, emit_rate_delay;
static struct Time emit_time;
static uint8_t emit_packet[NETFLOW_MAX_PACKET];
static pthread_t thid;
static sigset_t sig_mask;
static struct sched_param schedp;
static int sched_min, sched_max;
static int npeers, npeers_rot;
static struct peer *peers;
static int sigs;
static char cur_output_file[MAX_PATH_LEN];

static struct Flow *flows[1 << HASH_BITS];
static pthread_mutex_t flows_mutex[1 << HASH_BITS];

static pthread_mutex_t unpending_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t unpending_cond = PTHREAD_COND_INITIALIZER;

static pthread_mutex_t scan_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t scan_cond = PTHREAD_COND_INITIALIZER;
static struct Flow *pending_head, *pending_tail;
static struct Flow *scan_frag_dreg;

static pthread_mutex_t emit_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t emit_cond = PTHREAD_COND_INITIALIZER;
static struct Flow *flows_emit;

static char ident[256] = "fprobe-ulog";
static FILE *pidfile;
static char *pidfilepath;
static pid_t pid;
static int killed;
static int emit_timeout = EMIT_TIMEOUT, unpending_timeout = UNPENDING_TIMEOUT;
static struct ipulog_handle *ulog_handle;
static uint32_t ulog_gmask = 1;
static char *cap_buf;
static int nsnmp_rules;
static struct snmp_rule *snmp_rules;
static struct passwd *pw = 0;

void usage()
{
	fprintf(stdout,
			"fprobe-ulog: a NetFlow probe. Version %s\n"
			"Usage: fprobe-ulog [options] remote:port[/[local][/type]] ...\n"
			"\n"
			"-h\t\tDisplay this help\n"
			"-U <mask>\tULOG group bitwise mask [1]\n"
			"-s <seconds>\tHow often scan for expired flows [5]\n"
			"-g <seconds>\tFragmented flow lifetime [30]\n"
			"-e <seconds>\tActive flow lifetime (active timer) [300]\n"
			"-f <filename>\tLog flow data in a file\n"
			"-G <graduation period>\tRotate logs on an 0-hourly, 1-daily basis\n"
			"-n <version>\tNetFlow version for use (1, 5 or 7) [5]\n"
			"-a <address>\tUse <address> as source for NetFlow flow\n"
			"-X <rules>\tInterface name to SNMP-index conversion rules\n"
			"-M\t\tUse netfilter mark value as ToS flag\n"
			"-b <flows>\tMemory bulk size (1..%u) [%u]\n"
			"-m <kilobytes>\tMemory limit (0=no limit) [0]\n"
			"-q <flows>\tPending queue length [100]\n"
			"-B <kilobytes>\tKernel capture buffer size [0]\n"
			"-r <priority>\tReal-time priority (0=disabled, %d..%d) [0]\n"
			"-t <B:N>\tProduce <N> nanosecond delay after each <B> bytes sent [0:0]\n"
			"-c <directory>\tDirectory to chroot to\n"
			"-u <user>\tUser to run as\n"
			"-v <level>\tMaximum log level (0=EMERG, ..., 6=INFO, 7=DEBUG) [6]\n"
			"-l <[dst][:id]>\tLog destination and log/pidfile idetifier [1]\n"
			"-y <remote:port>\tAddress of the NetFlow collector\n"
			"-f <writable file>\tFile to write data into\n"
			"-T <n>\tRotate log file every n epochs\n"
			"-W <n>\tSet current epoch to n. Useful when restarting fprobe\n"
			"-E <[1..60]>\tSize of an epoch in minutes\n"
			"-D <number of blocks>\tNumber of disk blocks to preserve as free space\n"
			,
		VERSION, BULK_QUANTITY_MAX, bulk_quantity, sched_min, sched_max);
	exit(0);
}

#if ((DEBUG) & DEBUG_I)
void info_debug()
{
	my_log(LOG_DEBUG, "I: received:%d/%d (%lld) pending:%d/%d",
			pkts_total, pkts_total_fragmented, size_total,
			pkts_pending - pkts_pending_done, pending_queue_trace);
	my_log(LOG_DEBUG, "I: ignored:%d lost:%d+%d",
			pkts_ignored, pkts_lost_capture, pkts_lost_unpending);
	my_log(LOG_DEBUG, "I: cache:%d/%d emit:%d/%d/%d",
			flows_total, flows_fragmented, emit_sequence, emit_pkts, emit_queue);
	my_log(LOG_DEBUG, "I: memory:%d/%d (%d)",
			total_elements, free_elements, total_memory);
}
#endif

void sighandler(int sig)
{
	switch (sig) {
		case SIGTERM:
			sigs |= SIGTERM_MASK;
			break;
#if ((DEBUG) & DEBUG_I)
		case SIGUSR1:
			sigs |= SIGUSR1_MASK;
			break;
#endif
	}
}

void gettime(struct Time *now)
{
	struct timeval t;

	gettimeofday(&t, 0);
	now->sec = t.tv_sec;
	now->usec = t.tv_usec;
}


inline time_t cmpMtime(struct Time *t1, struct Time *t2)
{
	return (t1->sec - t2->sec)/60;
}

inline time_t cmpmtime(struct Time *t1, struct Time *t2)
{
	return (t1->sec - t2->sec) * 1000 + (t1->usec - t2->usec) / 1000;
}

/* Uptime in miliseconds */
uint32_t getuptime(struct Time *t)
{
	/* Maximum uptime is about 49/2 days */
	return cmpmtime(t, &start_time);
}

/* Uptime in minutes */
uint32_t getuptime_minutes(struct Time *t)
{
	/* Maximum uptime is about 49/2 days */
	return cmpMtime(t, &start_time);
}

hash_t hash_flow(struct Flow *flow)
{
	if (flow->flags & FLOW_FRAG) return hash(flow, sizeof(struct Flow_F));
	else return hash(flow, sizeof(struct Flow_TL));
}

uint16_t snmp_index(char *name) {
	uint32_t i;

	if (!*name) return 0;

	for (i = 0; (int) i < nsnmp_rules; i++) {
		if (strncmp(snmp_rules[i].basename, name, snmp_rules[i].len)) continue;
		return atoi(&name[snmp_rules[i].len]) + snmp_rules[i].base;
	}

	if ((i = if_nametoindex(name))) return i;

	return -1;
}

inline void copy_flow(struct Flow *src, struct Flow *dst)
{
	dst->iif = src->iif;
	dst->oif = src->oif;
	dst->sip = src->sip;
	dst->dip = src->dip;
	dst->tos = src->tos;
	dst->slice_id = src->slice_id;
	dst->proto = src->proto;
	dst->tcp_flags = src->tcp_flags;
	dst->id = src->id;
	dst->sp = src->sp;
	dst->dp = src->dp;
	dst->pkts = src->pkts;
	dst->size = src->size;
	dst->sizeF = src->sizeF;
	dst->sizeP = src->sizeP;
	dst->ctime = src->ctime;
	dst->mtime = src->mtime;
	dst->flags = src->flags;
}

void read_cur_epoch() {
	int fd;
	/* Reset to -1 in case the read fails */
	cur_epoch=-1;
	fd = open(LAST_EPOCH_FILE, O_RDONLY);
	if (fd != -1) {
		char snum[MAX_EPOCH_SIZE];
		ssize_t len;
		len = read(fd, snum, MAX_EPOCH_SIZE-1);
		if (len != -1) {
			snum[len]='\0';
			sscanf(snum,"%d",&cur_epoch);
			cur_epoch++; /* Let's not stone the last epoch */
			close(fd);
		}
	}
	return;
}


/* Dumps the current epoch in a file to cope with
 * reboots and killings of fprobe */

void update_cur_epoch_file(int n) {
	int fd, len;
	char snum[MAX_EPOCH_SIZE];
	len=snprintf(snum, MAX_EPOCH_SIZE-1,"%d", n);
	fd = open(LAST_EPOCH_FILE, O_RDWR|O_CREAT|O_TRUNC,S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH);
	if (fd == -1) {
		my_log(LOG_ERR, "open() failed: %s.The next restart will resume logging from epoch id 0.",LAST_EPOCH_FILE);
		return;
	}
	write(fd, snum, len);
	close(fd);
}

/* Get the file descriptor corresponding to the current file.
 * The kludgy implementation is to abstract away the 'current
 * file descriptor', which may also be a socket. 
 */

unsigned get_data_file_fd(char *fname, int cur_fd) {
	struct Time now;
	unsigned cur_uptime;

	struct statfs statfs;
	int ret_fd;

	/* We check if the amount of space left on the disk < some threshold and start reusing logs, or bail out if that
	 * doesn't solve the problem */
	gettime(&now);
	cur_uptime = getuptime_minutes(&now);

	if (cur_fd != START_DATA_FD) {
		if (fstatfs(cur_fd, &statfs) == -1) {
			my_log(LOG_ERR, "PANIC! Can't stat disk to calculate free blocks");
		}
		else {
			if (min_free && (statfs.f_bavail < min_free)
					&& (cur_epoch==last_peak))
			{
				my_log(LOG_INFO, "Disk almost full (%u free blocks). I'm going to drop data. Max epochs = %d\n",statfs.f_bavail,cur_epoch);
				cur_epoch = -1;
			}
			/*
			   else 
			   assume that we can reclaim space by overwriting our own files
			   and that the difference in size will not fill the disk - sapan
			   */
		}
	}

	/* If epoch length has been exceeded, 
	 * or we're starting up 
	 * or we're going back to the first epoch */
	if (((cur_uptime - prev_uptime) > epoch_length) || (cur_fd < 0) || (cur_epoch==-1)) {
		int write_fd;
		prev_uptime = cur_uptime;
		cur_epoch = (cur_epoch + 1) % log_epochs;
		if (cur_epoch>last_peak) last_peak = cur_epoch;
		if (cur_fd>0) {
			close(cur_fd);
            /* Compress the finished file */
            char gzip_cmd[MAX_PATH_LEN+sizeof("gzip -f ")]; 
            snprintf(gzip_cmd, MAX_PATH_LEN+sizeof("gzip -f "),"gzip -f %s",cur_output_file);
            system(gzip_cmd);
        }
		snprintf(cur_output_file,MAX_PATH_LEN,"%s.%d",fname,cur_epoch);
		if ((write_fd = open(cur_output_file, O_RDWR|O_CREAT|O_TRUNC,S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH)) < 0) {
			my_log(LOG_ERR, "open(): %s (%s)\n", cur_output_file, strerror(errno));
			exit(1);
		}
		if (fchmod(write_fd,S_IRUSR|S_IWUSR|S_IROTH|S_IRGRP) == -1) {
			my_log(LOG_ERR, "fchmod() failed: %s (%s). Continuing...\n", cur_output_file, strerror(errno));
		}
		update_cur_epoch_file(cur_epoch);
		ret_fd = write_fd;
	}
	else {
		ret_fd = cur_fd;
	}
	return(ret_fd);
}

struct Flow *find(struct Flow *where, struct Flow *what, struct Flow ***prev)
{
	struct Flow **flowpp;

#ifdef WALL
	flowpp = 0;
#endif

	if (prev) flowpp = *prev;

	while (where) {
		if (where->sip.s_addr == what->sip.s_addr
				&& where->dip.s_addr == what->dip.s_addr
				&& where->proto == what->proto) {
			switch ((what->flags + where->flags) & FLOW_FRAGMASK) {
				case 0:
					/* Both unfragmented */
					if ((what->sp == where->sp)
							&& (what->dp == where->dp)) goto done;
					break;
				case 2:
					/* Both fragmented */
					if (where->id == what->id) goto done;
					break;
			}
		}
		flowpp = &where->next;
		where = where->next;
	}
done:
	if (prev) *prev = flowpp;
	return where;
}

	int put_into(struct Flow *flow, int flag
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
			, char *logbuf
#endif
		    )
{
	int ret = 0;
	hash_t h;
	struct Flow *flown, **flowpp;
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	char buf[64];
#endif

	h = hash_flow(flow);
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	sprintf(buf, " %x H:%04x", (unsigned) flow, h);
	strcat(logbuf, buf);
#endif
	pthread_mutex_lock(&flows_mutex[h]);
	flowpp = &flows[h];
	if (!(flown = find(flows[h], flow, &flowpp))) {
		/* No suitable flow found - add */
		if (flag == COPY_INTO) {
			if ((flown = mem_alloc())) {
				copy_flow(flow, flown);
				flow = flown;
			} else {
#if ((DEBUG) & (DEBUG_S | DEBUG_U)) || defined MESSAGES
				my_log(LOG_ERR, "%s %s. %s",
						"mem_alloc():", strerror(errno), "packet lost");
#endif
				return -1;
			}
		}
		flow->next = flows[h];
		flows[h] = flow;
#if ((DEBUG) & DEBUG_I)
		flows_total++;
		if (flow->flags & FLOW_FRAG) flows_fragmented++;
#endif
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
		if (flown) {
			sprintf(buf, " => %x, flags: %x", (unsigned) flown, flown->flags);
			strcat(logbuf, buf);
		}
#endif
	} else {
		/* Found suitable flow - update */
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
		sprintf(buf, " +> %x", (unsigned) flown);
		strcat(logbuf, buf);
#endif
		if (cmpmtime(&flow->mtime, &flown->mtime) > 0)
			flown->mtime = flow->mtime;
		if (cmpmtime(&flow->ctime, &flown->ctime) < 0)
			flown->ctime = flow->ctime;
		flown->tcp_flags |= flow->tcp_flags;
		flown->size += flow->size;
		flown->pkts += flow->pkts;

		/* The slice_id of the first slice_id of a flow is misleading. Reset the slice_id of the flow
		 * if a better value comes along. A good example of this is that by the time CoDemux sets the
		 * peercred of a flow, it has already been accounted for here and attributed to root. */

		if (flown->slice_id<1)
				flown->slice_id = flow->slice_id;


		if (flow->flags & FLOW_FRAG) {
			/* Fragmented flow require some additional work */
			if (flow->flags & FLOW_TL) {
				/*
				   ?FIXME?
				   Several packets with FLOW_TL (attack)
				   */
				flown->sp = flow->sp;
				flown->dp = flow->dp;
			}
			if (flow->flags & FLOW_LASTFRAG) {
				/*
				   ?FIXME?
				   Several packets with FLOW_LASTFRAG (attack)
				   */
				flown->sizeP = flow->sizeP;
			}
			flown->flags |= flow->flags;
			flown->sizeF += flow->sizeF;
			if ((flown->flags & FLOW_LASTFRAG)
					&& (flown->sizeF >= flown->sizeP)) {
				/* All fragments received - flow reassembled */
				*flowpp = flown->next;
				pthread_mutex_unlock(&flows_mutex[h]);
#if ((DEBUG) & DEBUG_I)
				flows_total--;
				flows_fragmented--;
#endif
				flown->id = 0;
				flown->flags &= ~FLOW_FRAG;
#if ((DEBUG) & (DEBUG_U | DEBUG_S))
				strcat(logbuf," R");
#endif
				ret = put_into(flown, MOVE_INTO
#if ((DEBUG) & (DEBUG_U | DEBUG_S))
						, logbuf
#endif
					      );
			}
		}
		if (flag == MOVE_INTO) mem_free(flow);
	}
	pthread_mutex_unlock(&flows_mutex[h]);
	return ret;
}

int onlyonce=0;

void *fill(int fields, uint16_t *format, struct Flow *flow, void *p)
{
	int i;

	for (i = 0; i < fields; i++) {
#if ((DEBUG) & DEBUG_F)
		my_log(LOG_DEBUG, "F: field %04d at %x", format[i], (unsigned) p);
#endif
		switch (format[i]) {
			case NETFLOW_IPV4_SRC_ADDR:
				((struct in_addr *) p)->s_addr = flow->sip.s_addr;
				p += NETFLOW_IPV4_SRC_ADDR_SIZE;
				break;

			case NETFLOW_IPV4_DST_ADDR:
				((struct in_addr *) p)->s_addr = flow->dip.s_addr;
				if ((flow->dip.s_addr == inet_addr("10.0.0.8"))) {
					my_log(LOG_INFO, "Created records for test flow. No. of packets=%d",flow->pkts);
				}
				p += NETFLOW_IPV4_DST_ADDR_SIZE;
				break;

			case NETFLOW_INPUT_SNMP:
				*((uint16_t *) p) = htons(flow->iif);
				p += NETFLOW_INPUT_SNMP_SIZE;
				break;

			case NETFLOW_OUTPUT_SNMP:
				*((uint16_t *) p) = htons(flow->oif);
				p += NETFLOW_OUTPUT_SNMP_SIZE;
				break;

			case NETFLOW_PKTS_32:
				*((uint32_t *) p) = htonl(flow->pkts);
				p += NETFLOW_PKTS_32_SIZE;
				break;

			case NETFLOW_BYTES_32:
				*((uint32_t *) p) = htonl(flow->size);
				p += NETFLOW_BYTES_32_SIZE;
				break;

			case NETFLOW_FIRST_SWITCHED:
				*((uint32_t *) p) = htonl(getuptime(&flow->ctime));
				p += NETFLOW_FIRST_SWITCHED_SIZE;
				break;

			case NETFLOW_LAST_SWITCHED:
				*((uint32_t *) p) = htonl(getuptime(&flow->mtime));
				p += NETFLOW_LAST_SWITCHED_SIZE;
				break;

			case NETFLOW_L4_SRC_PORT:
				*((uint16_t *) p) = flow->sp;
				p += NETFLOW_L4_SRC_PORT_SIZE;
				break;

			case NETFLOW_L4_DST_PORT:
				*((uint16_t *) p) = flow->dp;
				p += NETFLOW_L4_DST_PORT_SIZE;
				break;

			case NETFLOW_PROT:
				*((uint8_t *) p) = flow->proto;
				p += NETFLOW_PROT_SIZE;
				break;

			case NETFLOW_SRC_TOS:
				*((uint8_t *) p) = flow->tos;
				p += NETFLOW_SRC_TOS_SIZE;
				break;

			case NETFLOW_TCP_FLAGS:
				*((uint8_t *) p) = flow->tcp_flags;
				p += NETFLOW_TCP_FLAGS_SIZE;
				break;

			case NETFLOW_VERSION:
				*((uint16_t *) p) = htons(netflow->Version);
				p += NETFLOW_VERSION_SIZE;
				break;

			case NETFLOW_COUNT:
				*((uint16_t *) p) = htons(emit_count);
				p += NETFLOW_COUNT_SIZE;
				break;

			case NETFLOW_UPTIME:
				*((uint32_t *) p) = htonl(getuptime(&emit_time));
				p += NETFLOW_UPTIME_SIZE;
				break;

			case NETFLOW_UNIX_SECS:
				*((uint32_t *) p) = htonl(emit_time.sec);
				p += NETFLOW_UNIX_SECS_SIZE;
				break;

			case NETFLOW_UNIX_NSECS:
				*((uint32_t *) p) = htonl(emit_time.usec * 1000);
				p += NETFLOW_UNIX_NSECS_SIZE;
				break;

			case NETFLOW_FLOW_SEQUENCE:
				//*((uint32_t *) p) = htonl(emit_sequence);
				*((uint32_t *) p) = 0;
				p += NETFLOW_FLOW_SEQUENCE_SIZE;
				break;

			case NETFLOW_PAD8:
				/* Unsupported (uint8_t) */
			case NETFLOW_ENGINE_TYPE:
			case NETFLOW_ENGINE_ID:
			case NETFLOW_FLAGS7_1:
			case NETFLOW_SRC_MASK:
			case NETFLOW_DST_MASK:
				if (onlyonce) {
					my_log(LOG_CRIT, "Adding SRC/DST masks: this version of fprobe is seriously broken\n");
					onlyonce=1;
				}
				*((uint8_t *) p) = 0;
				p += NETFLOW_PAD8_SIZE;
				break;
			case NETFLOW_SLICE_ID:
				*((uint32_t *) p) = flow->slice_id;
				p += NETFLOW_SLICE_ID_SIZE;
				break;
			case NETFLOW_PAD16:
				/* Unsupported (uint16_t) */
			case NETFLOW_SRC_AS:
			case NETFLOW_DST_AS:
			case NETFLOW_FLAGS7_2:
				*((uint16_t *) p) = 0;
				p += NETFLOW_PAD16_SIZE;
				break;

			case NETFLOW_PAD32:
				/* Unsupported (uint32_t) */
			case NETFLOW_IPV4_NEXT_HOP:
			case NETFLOW_ROUTER_SC:
				*((uint32_t *) p) = 0;
				p += NETFLOW_PAD32_SIZE;
				break;

			default:
				my_log(LOG_CRIT, "fill(): Unknown format at %x[%d]: %d",
						format, i, format[i]);
				exit(1);
		}
	}
#if ((DEBUG) & DEBUG_F)
	my_log(LOG_DEBUG, "F: return %x", (unsigned) p);
#endif
	return p;
}

void setuser() {
	/*
	   Workaround for clone()-based threads
	   Try to change EUID independently of main thread
	   */
	if (pw) {
		setgroups(0, NULL);
		setregid(pw->pw_gid, pw->pw_gid);
		setreuid(pw->pw_uid, pw->pw_uid);
	}
}

void *emit_thread()
{
	struct Flow *flow;
	void *p;
	struct timeval now;
	struct timespec timeout;
	int i, ret, sent = 0, size, delay, peer_rot_cur, peer_rot_work = 0;

	p = (void *) &emit_packet + netflow->HeaderSize;
	timeout.tv_nsec = 0;

	setuser();
        
	for (;;) {
		pthread_mutex_lock(&emit_mutex);
		while (!flows_emit) {
			gettimeofday(&now, 0);
			timeout.tv_sec = now.tv_sec + emit_timeout;
			/* Do not wait until emit_packet will filled - it may be too long */
            int res=-1;
            while ((res=pthread_cond_timedwait(&emit_cond, &emit_mutex, &timeout))==-1) continue;

			if (res && emit_count) {
                //my_log(LOG_INFO,"Timeout: %d, %d",emit_count, timeout.tv_sec);
				pthread_mutex_unlock(&emit_mutex);
				goto sendit;
			}
		}
		flow = flows_emit;
		flows_emit = flows_emit->next;
#if ((DEBUG) & DEBUG_I)
		emit_queue--;
#endif		
		pthread_mutex_unlock(&emit_mutex);

#ifdef UPTIME_TRICK
		if (!emit_count) {
			gettime(&start_time);
			start_time.sec -= start_time_offset;
		}
#endif
		p = fill(netflow->FlowFields, netflow->FlowFormat, flow, p);
		mem_free(flow);
		emit_count++;
#ifdef PF2_DEBUG
		printf("Emit count = %d\n", emit_count);
		fflush(stdout);
#endif
		if (emit_count == netflow->MaxFlows) {
sendit:
			gettime(&emit_time);
			p = fill(netflow->HeaderFields, netflow->HeaderFormat, 0, &emit_packet);
			size = netflow->HeaderSize + emit_count * netflow->FlowSize;
			/* Netflow PDUs need to be padded to 1464 bytes - Sapan */
#ifdef STD_NETFLOW_PDU
			if (size < NETFLOW_PDU_SIZE) size = NETFLOW_PDU_SIZE;
#endif
			peer_rot_cur = 0;
			for (i = 0; i < npeers; i++) {
				if (peers[i].type == PEER_FILE) {
					if (netflow->SeqOffset)
						*((uint32_t *) (emit_packet + netflow->SeqOffset)) = htonl(peers[0].seq);
					peers[i].write_fd = get_data_file_fd(peers[i].fname, peers[i].write_fd);
					ret = write(peers[i].write_fd, emit_packet, size);
					if (ret < size) {

#if ((DEBUG) & DEBUG_E) || defined MESSAGES
						my_log(LOG_ERR, "write(to #%d, seq %d, flows %d, size %d) == %d: %s",
								i + 1, peers[i].seq, emit_count, size, ret, strerror(errno));
#endif
#undef MESSAGES
					}
#if ((DEBUG) & DEBUG_E)
					else {
						my_log(LOG_DEBUG, "E: Emitted %d flow(s) to #%d, seq %d",
								emit_count, i + 1, peers[i].seq);
					}
#endif
					peers[i].seq += emit_count;

					/* Rate limit */
					if (emit_rate_bytes) {
						sent += size;
						delay = sent / emit_rate_bytes;
						if (delay) {
							sent %= emit_rate_bytes;
							timeout.tv_sec = 0;
							timeout.tv_nsec = emit_rate_delay * delay;
							while (nanosleep(&timeout, &timeout) == -1 && errno == EINTR);
						}
					}
				}
				else
					if (peers[i].type == PEER_MIRROR) goto sendreal;
					else
						if (peers[i].type == PEER_ROTATE) 
							if (peer_rot_cur++ == peer_rot_work) {
sendreal:
								if (netflow->SeqOffset)
									*((uint32_t *) (emit_packet + netflow->SeqOffset)) = htonl(peers[i].seq);
								ret = send(peers[i].write_fd, emit_packet, size, 0);
								if (ret < size) {
#if ((DEBUG) & DEBUG_E) || defined MESSAGES
									my_log(LOG_ERR, "send(to #%d, seq %d, flows %d, size %d) == %d: %s",
											i + 1, peers[i].seq, emit_count, size, ret, strerror(errno));
#endif
								}
#if ((DEBUG) & DEBUG_E)
								else {
									my_log(LOG_DEBUG, "E: Emitted %d flow(s) to #%d, seq %d",
											emit_count, i + 1, peers[i].seq);
								}
#endif
								peers[i].seq += emit_count;

								/* Rate limit */
								if (emit_rate_bytes) {
									sent += size;
									delay = sent / emit_rate_bytes;
									if (delay) {
										sent %= emit_rate_bytes;
										timeout.tv_sec = 0;
										timeout.tv_nsec = emit_rate_delay * delay;
										while (nanosleep(&timeout, &timeout) == -1 && errno == EINTR);
									}
								}
							}
			}
			if (npeers_rot) peer_rot_work = (peer_rot_work + 1) % npeers_rot;
			emit_sequence += emit_count;
			emit_count = 0;
#if ((DEBUG) & DEBUG_I)
			emit_pkts++;
#endif
		}
	}
}	

void *unpending_thread()
{
	struct timeval now;
	struct timespec timeout;
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	char logbuf[256];
#endif

	setuser();

	timeout.tv_nsec = 0;
	pthread_mutex_lock(&unpending_mutex);

	for (;;) {
		while (!(pending_tail->flags & FLOW_PENDING)) {
			gettimeofday(&now, 0);
			timeout.tv_sec = now.tv_sec + unpending_timeout;
			while (pthread_cond_timedwait(&unpending_cond, &unpending_mutex, &timeout)==-1)
                continue;
		}

#if ((DEBUG) & (DEBUG_S | DEBUG_U))
		*logbuf = 0;
#endif
		if (put_into(pending_tail, COPY_INTO
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
					, logbuf
#endif
			    ) < 0) {
#if ((DEBUG) & DEBUG_I)
			pkts_lost_unpending++;
#endif				
		}

#if ((DEBUG) & DEBUG_U)
		my_log(LOG_DEBUG, "%s%s", "U:", logbuf);
#endif

		pending_tail->flags = 0;
		pending_tail = pending_tail->next;
#if ((DEBUG) & DEBUG_I)
		pkts_pending_done++;
#endif
	}
}

void *scan_thread()
{
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	char logbuf[256];
#endif
	int i;
	struct Flow *flow, **flowpp;
	struct Time now;
	struct timespec timeout;

	setuser();

	timeout.tv_nsec = 0;
	pthread_mutex_lock(&scan_mutex);

	for (;;) {
		gettime(&now);
		timeout.tv_sec = now.sec + scan_interval;
		while (pthread_cond_timedwait(&scan_cond, &scan_mutex, &timeout)==-1)
            continue;

		gettime(&now);
#if ((DEBUG) & DEBUG_S)
		my_log(LOG_DEBUG, "S: %d", now.sec);
#endif
		for (i = 0; i < 1 << HASH_BITS ; i++) {
			pthread_mutex_lock(&flows_mutex[i]);
			flow = flows[i];
			flowpp = &flows[i];
			while (flow) {
				if (flow->flags & FLOW_FRAG) {
					/* Process fragmented flow */
					if ((now.sec - flow->mtime.sec) > frag_lifetime) {
						/* Fragmented flow expired - put it into special chain */
#if ((DEBUG) & DEBUG_I)
						flows_fragmented--;
						flows_total--;
#endif
						*flowpp = flow->next;
						flow->id = 0;
						flow->flags &= ~FLOW_FRAG;
						flow->next = scan_frag_dreg;
						scan_frag_dreg = flow;
						flow = *flowpp;
						continue;
					}
				} else {
					/* Flow is not frgamented */
					if ((now.sec - flow->mtime.sec) > inactive_lifetime
							|| (flow->mtime.sec - flow->ctime.sec) > active_lifetime) {
						/* Flow expired */
#if ((DEBUG) & DEBUG_S)
						my_log(LOG_DEBUG, "S: E %x", flow);
#endif
#if ((DEBUG) & DEBUG_I)
						flows_total--;
#endif
						*flowpp = flow->next;
						pthread_mutex_lock(&emit_mutex);
						flow->next = flows_emit;
						flows_emit = flow;
#if ((DEBUG) & DEBUG_I)
						emit_queue++;
#endif				
						pthread_mutex_unlock(&emit_mutex);
						flow = *flowpp;
						continue;
					}
				}
				flowpp = &flow->next;
				flow = flow->next;
			} /* chain loop */
			pthread_mutex_unlock(&flows_mutex[i]);
		} /* hash loop */
		if (flows_emit) pthread_cond_signal(&emit_cond);

		while (scan_frag_dreg) {
			flow = scan_frag_dreg;
			scan_frag_dreg = flow->next;
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
			*logbuf = 0;
#endif
			put_into(flow, MOVE_INTO
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
					, logbuf
#endif
				);
#if ((DEBUG) & DEBUG_S)
			my_log(LOG_DEBUG, "%s%s", "S: FE", logbuf);
#endif
		}
	}
}

void *cap_thread()
{
	struct ulog_packet_msg *ulog_msg;
	struct ip *nl;
	void *tl;
	struct Flow *flow;
	int len, off_frag, psize;
#if ((DEBUG) & DEBUG_C)
	char buf[64];
	char logbuf[256];
#endif
	int challenge;

	setuser();

	while (!killed) {
		len = ipulog_read(ulog_handle, cap_buf, CAPTURE_SIZE, 1);
		if (len <= 0) {
			my_log(LOG_ERR, "ipulog_read(): %s", ipulog_strerror(ipulog_errno));
			continue;
		}
		while ((ulog_msg = ipulog_get_packet(ulog_handle, cap_buf, len))) {

#if ((DEBUG) & DEBUG_C)
			sprintf(logbuf, "C: %d", ulog_msg->data_len);
#endif

			nl = (void *) &ulog_msg->payload;
			psize = ulog_msg->data_len;

			/* Sanity check */
			if (psize < (signed) sizeof(struct ip) || nl->ip_v != 4) {
#if ((DEBUG) & DEBUG_C)
				strcat(logbuf, " U");
				my_log(LOG_DEBUG, "%s", logbuf);
#endif
#if ((DEBUG) & DEBUG_I)
				pkts_ignored++;
#endif
				continue;
			}

			if (pending_head->flags) {
#if ((DEBUG) & DEBUG_C) || defined MESSAGES
				my_log(LOG_ERR,
# if ((DEBUG) & DEBUG_C)
						"%s %s %s", logbuf,
# else
						"%s %s",
# endif
						"pending queue full:", "packet lost");
#endif
#if ((DEBUG) & DEBUG_I)
				pkts_lost_capture++;
#endif
				goto done;
			}

#if ((DEBUG) & DEBUG_I)
			pkts_total++;
#endif

			flow = pending_head;

			/* ?FIXME? Add sanity check for ip_len? */
			flow->size = ntohs(nl->ip_len);
#if ((DEBUG) & DEBUG_I)
			size_total += flow->size;
#endif

			flow->sip = nl->ip_src;
			flow->dip = nl->ip_dst;
			flow->tos = mark_is_tos ? ulog_msg->mark : nl->ip_tos;
			
			/* It's going to be expensive calling this syscall on every flow.
			 * We should keep a local hash table, for now just bear the overhead... - Sapan*/

			flow->slice_id = ulog_msg->mark;

			/*if (flow->slice_id < 1)
				flow->slice_id = ulog_msg->mark; // Couldn't look up the slice id, let's at least store the local xid*/


			if ((flow->dip.s_addr == inet_addr("10.0.0.8")) || (flow->sip.s_addr == inet_addr("10.0.0.8"))) {
				my_log(LOG_INFO, "Received test flow to corewars.org from slice %d ",flow->slice_id);
			}
			flow->iif = snmp_index(ulog_msg->indev_name);
			flow->oif = snmp_index(ulog_msg->outdev_name);
			flow->proto = nl->ip_p;
			flow->id = 0;
			flow->tcp_flags = 0;
			flow->pkts = 1;
			flow->sizeF = 0;
			flow->sizeP = 0;
			/* Packets captured from OUTPUT table didn't contains valid timestamp */
			if (ulog_msg->timestamp_sec) {
				flow->ctime.sec = ulog_msg->timestamp_sec;
				flow->ctime.usec = ulog_msg->timestamp_usec;
			} else gettime(&flow->ctime);
			flow->mtime = flow->ctime;

			off_frag = (ntohs(nl->ip_off) & IP_OFFMASK) << 3;

			/*
			   Offset (from network layer) to transport layer header/IP data
			   IOW IP header size ;-)

			   ?FIXME?
			   Check ip_hl for valid value (>=5)? Maybe check CRC? No, thanks...
			   */
			off_tl = nl->ip_hl << 2;
			tl = (void *) nl + off_tl;

			/* THIS packet data size: data_size = total_size - ip_header_size*4 */
			flow->sizeF = ntohs(nl->ip_len) - off_tl;
			psize -= off_tl;
			if ((signed) flow->sizeF < 0) flow->sizeF = 0;
			if (psize > (signed) flow->sizeF) psize = flow->sizeF;

			if (ntohs(nl->ip_off) & (IP_MF | IP_OFFMASK)) {
				/* Fragmented packet (IP_MF flag == 1 or fragment offset != 0) */
#if ((DEBUG) & DEBUG_C)
				strcat(logbuf, " F");
#endif
#if ((DEBUG) & DEBUG_I)
				pkts_total_fragmented++;
#endif
				flow->flags |= FLOW_FRAG;
				flow->id = nl->ip_id;

				if (!(ntohs(nl->ip_off) & IP_MF)) {
					/* Packet whith IP_MF contains information about whole datagram size */
					flow->flags |= FLOW_LASTFRAG;
					/* size = frag_offset*8 + data_size */
					flow->sizeP = off_frag + flow->sizeF;
				}
			}

#if ((DEBUG) & DEBUG_C)
			sprintf(buf, " %s@%u>", inet_ntoa(flow->sip), flow->iif);
			strcat(logbuf, buf);
			sprintf(buf, "%s@%u P:%x", inet_ntoa(flow->dip), flow->oif, flow->proto);
			strcat(logbuf, buf);
#endif

			/*
			   Fortunately most interesting transport layer information fit
			   into first 8 bytes of IP data field (minimal nonzero size).
			   Thus we don't need actual packet reassembling to build whole
			   transport layer data. We only check the fragment offset for
			   zero value to find packet with this information.
			   */
			if (!off_frag && psize >= 8) {
				switch (flow->proto) {
					case IPPROTO_TCP:
					case IPPROTO_UDP:
						flow->sp = ((struct udphdr *)tl)->uh_sport;
						flow->dp = ((struct udphdr *)tl)->uh_dport;
						goto tl_known;

#ifdef ICMP_TRICK
					case IPPROTO_ICMP:
						flow->sp = htons(((struct icmp *)tl)->icmp_type);
						flow->dp = htons(((struct icmp *)tl)->icmp_code);
						goto tl_known;
#endif
#ifdef ICMP_TRICK_CISCO
					case IPPROTO_ICMP:
						flow->dp = *((int32_t *) tl);
						goto tl_known;
#endif

					default:
						/* Unknown transport layer */
#if ((DEBUG) & DEBUG_C)
						strcat(logbuf, " U");
#endif
						flow->sp = 0;
						flow->dp = 0;
						break;

tl_known:
#if ((DEBUG) & DEBUG_C)
						sprintf(buf, " %d>%d", ntohs(flow->sp), ntohs(flow->dp));
						strcat(logbuf, buf);
#endif
						flow->flags |= FLOW_TL;
				}
			}

			/* Check for tcp flags presence (including CWR and ECE). */
			if (flow->proto == IPPROTO_TCP
					&& off_frag < 16
					&& psize >= 16 - off_frag) {
				flow->tcp_flags = *((uint8_t *)(tl + 13 - off_frag));
#if ((DEBUG) & DEBUG_C)
				sprintf(buf, " TCP:%x", flow->tcp_flags);
				strcat(logbuf, buf);
#endif
			}

#if ((DEBUG) & DEBUG_C)
			sprintf(buf, " => %x", (unsigned) flow);
			strcat(logbuf, buf);
			my_log(LOG_DEBUG, "%s", logbuf);
#endif

#if ((DEBUG) & DEBUG_I)
			pkts_pending++;
			pending_queue_trace_candidate = pkts_pending - pkts_pending_done;
			if (pending_queue_trace < pending_queue_trace_candidate)
				pending_queue_trace = pending_queue_trace_candidate;
#endif

			/* Flow complete - inform unpending_thread() about it */
			pending_head->flags |= FLOW_PENDING;
			pending_head = pending_head->next;
done:
			pthread_cond_signal(&unpending_cond);
		}
	}
	return 0;
}

/* Copied out of CoDemux */

static int init_daemon() {
	pid_t pid;
	FILE *pidfile;

	pidfile = fopen(PIDFILE, "w");
	if (pidfile == NULL) {
		my_log(LOG_ERR, "%s creation failed\n", PIDFILE);
	}

	if ((pid = fork()) < 0) {
		fclose(pidfile);
		my_log(LOG_ERR, "Could not fork!\n");
		return(-1);
	}
	else if (pid != 0) {
		/* i'm the parent, writing down the child pid  */
		fprintf(pidfile, "%u\n", pid);
		fclose(pidfile);
		exit(0);
	}

	/* close the pid file */
	fclose(pidfile);

	/* routines for any daemon process
	   1. create a new session 
	   2. change directory to the root
	   3. change the file creation permission 
	   */
	setsid();
	chdir("/var/local/fprobe");
	umask(0);

	return(0);
}

int main(int argc, char **argv)
{
	char errpbuf[512];
	char *dhost, *dport, *lhost, *type = 0, *log_suffix = 0, *rule;
	int c, i, write_fd, memory_limit = 0;
	struct addrinfo hints, *res;
	struct sockaddr_in saddr;
	pthread_attr_t tattr;
	struct sigaction sigact;
	static void *threads[THREADS - 1] = {&emit_thread, &scan_thread, &unpending_thread, &cap_thread};
	struct timeval timeout;

	sched_min = sched_get_priority_min(SCHED);
	sched_max = sched_get_priority_max(SCHED);

	memset(&saddr, 0 , sizeof(saddr));
	memset(&hints, 0 , sizeof(hints));
	hints.ai_flags = AI_PASSIVE;
	hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_DGRAM;

	/* Process command line options */

	opterr = 0;
	while ((c = my_getopt(argc, argv, parms)) != -1) {
		switch (c) {
			case '?':
				usage();

			case 'h':
				usage();
		}
	}

	if (parms[Uflag].count) ulog_gmask = atoi(parms[Uflag].arg);
	if (parms[Wflag].count) cur_epoch = atoi(parms[Wflag].arg);
	if (parms[Tflag].count) log_epochs = atoi(parms[Tflag].arg);
	if (parms[Eflag].count) epoch_length = atoi(parms[Eflag].arg);
	if (parms[sflag].count) scan_interval = atoi(parms[sflag].arg);
	if (parms[gflag].count) frag_lifetime = atoi(parms[gflag].arg);
	if (parms[dflag].count) inactive_lifetime = atoi(parms[dflag].arg);
	if (parms[Dflag].count) min_free = atoi(parms[Dflag].arg);
	if (parms[eflag].count) active_lifetime = atoi(parms[eflag].arg);
	if (parms[nflag].count) {
		switch (atoi(parms[nflag].arg)) {
			case 1:
				netflow = &NetFlow1;
				break;

			case 5:
				break;

			case 7:
				netflow = &NetFlow7;
				break;

			default:
				fprintf(stderr, "Illegal %s\n", "NetFlow version");
				exit(1);
		}
	}
	if (parms[vflag].count) verbosity = atoi(parms[vflag].arg);
	if (parms[lflag].count) {
		if ((log_suffix = strchr(parms[lflag].arg, ':'))) {
			*log_suffix++ = 0;
			if (*log_suffix) {
				sprintf(errpbuf, "[%s]", log_suffix);
				strcat(ident, errpbuf);
			}
		}
		if (*parms[lflag].arg) log_dest = atoi(parms[lflag].arg);
		if (log_suffix) *--log_suffix = ':';
	}
	if (!(pidfilepath = malloc(sizeof(PID_DIR) + 1 + strlen(ident) + 1 + 3 + 1))) {
err_malloc:
		fprintf(stderr, "malloc(): %s\n", strerror(errno));
		exit(1);
	}
	sprintf(pidfilepath, "%s/%s.pid", PID_DIR, ident);
	if (parms[qflag].count) {
		pending_queue_length = atoi(parms[qflag].arg);
		if (pending_queue_length < 1) {
			fprintf(stderr, "Illegal %s\n", "pending queue length");
			exit(1);
		}
	}
	if (parms[rflag].count) {
		schedp.sched_priority = atoi(parms[rflag].arg);
		if (schedp.sched_priority
				&& (schedp.sched_priority < sched_min
					|| schedp.sched_priority > sched_max)) {
			fprintf(stderr, "Illegal %s\n", "realtime priority");
			exit(1);
		}
	}
	if (parms[Bflag].count) {
		sockbufsize = atoi(parms[Bflag].arg) << 10;
	}
	if (parms[bflag].count) {
		bulk_quantity = atoi(parms[bflag].arg);
		if (bulk_quantity < 1 || bulk_quantity > BULK_QUANTITY_MAX) {
			fprintf(stderr, "Illegal %s\n", "bulk size");
			exit(1);
		}
	}
	if (parms[mflag].count) memory_limit = atoi(parms[mflag].arg) << 10;
	if (parms[Xflag].count) {
		for(i = 0; parms[Xflag].arg[i]; i++)
			if (parms[Xflag].arg[i] == ':') nsnmp_rules++;
		if (!(snmp_rules = malloc(nsnmp_rules * sizeof(struct snmp_rule))))
			goto err_malloc;
		rule = strtok(parms[Xflag].arg, ":");
		for (i = 0; rule; i++) {
			snmp_rules[i].len = strlen(rule);
			if (snmp_rules[i].len > IFNAMSIZ) {
				fprintf(stderr, "Illegal %s\n", "interface basename");
				exit(1);
			}
			strncpy(snmp_rules[i].basename, rule, snmp_rules[i].len);
			if (!*(rule - 1)) *(rule - 1) = ',';
			rule = strtok(NULL, ",");
			if (!rule) {
				fprintf(stderr, "Illegal %s\n", "SNMP rule");
				exit(1);
			}
			snmp_rules[i].base = atoi(rule);
			*(rule - 1) = ':';
			rule = strtok(NULL, ":");
		}
		nsnmp_rules = i;
	}
	if (parms[tflag].count)
		sscanf(parms[tflag].arg, "%d:%d", &emit_rate_bytes, &emit_rate_delay);
	if (parms[aflag].count) {
		if (getaddrinfo(parms[aflag].arg, 0, &hints, &res)) {
bad_lhost:
			fprintf(stderr, "Illegal %s\n", "source address");
			exit(1);
		} else {
			saddr = *((struct sockaddr_in *) res->ai_addr);
			freeaddrinfo(res);
		}
	}
	if (parms[uflag].count) 
		if ((pw = getpwnam(parms[uflag].arg)) == NULL) {
			fprintf(stderr, "getpwnam(%s): %s\n", parms[uflag].arg, errno ? strerror(errno) : "Unknown user");
			exit(1);
		}


	/* Process collectors parameters. Brrrr... :-[ */

	npeers = argc - optind;
	if (npeers >= 1) {
		/* Send to remote Netflow collector */
		if (!(peers = malloc(npeers * sizeof(struct peer)))) goto err_malloc;
		for (i = optind, npeers = 0; i < argc; i++, npeers++) {
			dhost = argv[i];
			if (!(dport = strchr(dhost, ':'))) goto bad_collector;
			*dport++ = 0;
			if ((write_fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
				fprintf(stderr, "socket(): %s\n", strerror(errno));
				exit(1);
			}
			peers[npeers].write_fd = write_fd;
			peers[npeers].type = PEER_MIRROR;
			peers[npeers].laddr = saddr;
			peers[npeers].seq = 0;
			if ((lhost = strchr(dport, '/'))) {
				*lhost++ = 0;
				if ((type = strchr(lhost, '/'))) {
					*type++ = 0;
					switch (*type) {
						case 0:
						case 'm':
							break;

						case 'r':
							peers[npeers].type = PEER_ROTATE;
							npeers_rot++;
							break;

						default:
							goto bad_collector;
					}
				}
				if (*lhost) {
					if (getaddrinfo(lhost, 0, &hints, &res)) goto bad_lhost;
					peers[npeers].laddr = *((struct sockaddr_in *) res->ai_addr);
					freeaddrinfo(res);
				}
			}
			if (bind(write_fd, (struct sockaddr *) &peers[npeers].laddr,
						sizeof(struct sockaddr_in))) {
				fprintf(stderr, "bind(): %s\n", strerror(errno));
				exit(1);
			}
			if (getaddrinfo(dhost, dport, &hints, &res)) {
bad_collector:
				fprintf(stderr, "Error in collector #%d parameters\n", npeers + 1);
				exit(1);
			}
			peers[npeers].addr = *((struct sockaddr_in *) res->ai_addr);
			freeaddrinfo(res);
			if (connect(write_fd, (struct sockaddr *) &peers[npeers].addr,
						sizeof(struct sockaddr_in))) {
				fprintf(stderr, "connect(): %s\n", strerror(errno));
				exit(1);
			}

			/* Restore command line */
			if (type) *--type = '/';
			if (lhost) *--lhost = '/';
			*--dport = ':';
		}
	}
	else if (parms[fflag].count) {
		// log into a file
		if (!(peers = malloc(sizeof(struct peer)))) goto err_malloc;
		if (!(peers[npeers].fname = strndup(parms[fflag].arg, MAX_PATH_LEN))) goto err_malloc;
		peers[npeers].write_fd = START_DATA_FD;
		peers[npeers].type = PEER_FILE;
		peers[npeers].seq = 0;

		read_cur_epoch();
		npeers++;
	}
	else 
		usage();


	if (!(cap_buf = malloc(CAPTURE_SIZE))) goto err_malloc;
	ulog_handle = ipulog_create_handle(ulog_gmask, CAPTURE_SIZE);
	if (!ulog_handle) {
		fprintf(stderr, "libipulog initialization error: %s",
				ipulog_strerror(ipulog_errno));
		exit(1);
	}
	if (sockbufsize)
		if (setsockopt(ulog_handle->fd, SOL_SOCKET, SO_RCVBUF,
					&sockbufsize, sizeof(sockbufsize)) < 0)
			fprintf(stderr, "setsockopt(): %s", strerror(errno));

	/* Daemonize (if log destination stdout-free) */

	my_log_open(ident, verbosity, log_dest);

	init_daemon();

	if (!(log_dest & 2)) {
		/* Crash-proofing - Sapan*/
		while (1) {
			int pid=fork();
			if (pid==-1) {
				fprintf(stderr, "fork(): %s", strerror(errno));
				exit(1);
			}
			else if (pid==0) {
				setsid();
				freopen("/dev/null", "r", stdin);
				freopen("/dev/null", "w", stdout);
				freopen("/dev/null", "w", stderr);
				break;
			}
			else {
				while (wait3(NULL,0,NULL) < 1);
			}
		}
	} else {
		setvbuf(stdout, (char *)0, _IONBF, 0);
		setvbuf(stderr, (char *)0, _IONBF, 0);
	}

	pid = getpid();
	sprintf(errpbuf, "[%ld]", (long) pid);
	strcat(ident, errpbuf);

	/* Initialization */

    // init_slice_id_hash();
	hash_init(); /* Actually for crc16 only */
	mem_init(sizeof(struct Flow), bulk_quantity, memory_limit);
	for (i = 0; i < 1 << HASH_BITS; i++) pthread_mutex_init(&flows_mutex[i], 0);

#ifdef UPTIME_TRICK
	/* Hope 12 days is enough :-/ */
	start_time_offset = 1 << 20;

	/* start_time_offset = active_lifetime + inactive_lifetime + scan_interval; */
#endif
	gettime(&start_time);

	/*
	   Build static pending queue as circular buffer.
	   */
	if (!(pending_head = mem_alloc())) goto err_mem_alloc;
	pending_tail = pending_head;
	for (i = pending_queue_length - 1; i--;) {
		if (!(pending_tail->next = mem_alloc())) {
err_mem_alloc:
			my_log(LOG_CRIT, "mem_alloc(): %s", strerror(errno));
			exit(1);
		}
		pending_tail = pending_tail->next;
	}
	pending_tail->next = pending_head;
	pending_tail = pending_head;

	sigemptyset(&sig_mask);
	sigact.sa_handler = &sighandler;
	sigact.sa_mask = sig_mask;
	sigact.sa_flags = 0;
	sigaddset(&sig_mask, SIGTERM);
	sigaction(SIGTERM, &sigact, 0);
#if ((DEBUG) & DEBUG_I)
	sigaddset(&sig_mask, SIGUSR1);
	sigaction(SIGUSR1, &sigact, 0);
#endif
	if (pthread_sigmask(SIG_BLOCK, &sig_mask, 0)) {
		my_log(LOG_CRIT, "pthread_sigmask(): %s", strerror(errno));
		exit(1);
	}

	my_log(LOG_INFO, "Starting %s...", VERSION);

	if (parms[cflag].count) {
		if (chdir(parms[cflag].arg) || chroot(".")) {
			my_log(LOG_CRIT, "could not chroot to %s: %s", parms[cflag].arg, strerror(errno));
			exit(1);
		}
	}

	schedp.sched_priority = schedp.sched_priority - THREADS + 2;
	pthread_attr_init(&tattr);
	for (i = 0; i < THREADS - 1; i++) {
		if (schedp.sched_priority > 0) {
			if ((pthread_attr_setschedpolicy(&tattr, SCHED)) ||
					(pthread_attr_setschedparam(&tattr, &schedp))) {
				my_log(LOG_CRIT, "pthread_attr_setschedpolicy(): %s", strerror(errno));
				exit(1);
			}
		}
		if (pthread_create(&thid, &tattr, threads[i], 0)) {
			my_log(LOG_CRIT, "pthread_create(): %s", strerror(errno));
			exit(1);
		}
		pthread_detach(thid);
		schedp.sched_priority++;
	}

	if (pw) {
		if (setgroups(0, NULL)) {
			my_log(LOG_CRIT, "setgroups(): %s", strerror(errno));
			exit(1);
		}
		if (setregid(pw->pw_gid, pw->pw_gid)) {
			my_log(LOG_CRIT, "setregid(%u): %s", pw->pw_gid, strerror(errno));
			exit(1);
		}
		if (setreuid(pw->pw_uid, pw->pw_uid)) {
			my_log(LOG_CRIT, "setreuid(%u): %s", pw->pw_uid, strerror(errno));
			exit(1);
		}
	}

	if (!(pidfile = fopen(pidfilepath, "w")))
		my_log(LOG_ERR, "Can't create pid file. fopen(): %s", strerror(errno));
	else {
		fprintf(pidfile, "%ld\n", (long) pid);
		fclose(pidfile);
	}

	my_log(LOG_INFO, "pid: %d", pid);
	my_log(LOG_INFO, "options: u=%u s=%u g=%u d=%u e=%u n=%u a=%s "
			"M=%d b=%u m=%u q=%u B=%u r=%u t=%u:%u c=%s u=%s v=%u l=%u%s",
			ulog_gmask, scan_interval, frag_lifetime, inactive_lifetime, active_lifetime,
			netflow->Version, inet_ntoa(saddr.sin_addr), mark_is_tos, bulk_quantity,
			memory_limit >> 10, pending_queue_length, sockbufsize >> 10, schedp.sched_priority - 1,
			emit_rate_bytes, emit_rate_delay, parms[cflag].count ? parms[cflag].arg : "",
			parms[uflag].count ? parms[uflag].arg : "", verbosity, log_dest, log_suffix ? log_suffix : "");
	for (i = 0; i < nsnmp_rules; i++) {
		my_log(LOG_INFO, "SNMP rule #%d %s:%d",
				i + 1, snmp_rules[i].basename, snmp_rules[i].base);
	}
	for (i = 0; i < npeers; i++) {
		switch (peers[i].type) {
			case PEER_MIRROR:
				c = 'm';
				break;
			case PEER_ROTATE:
				c = 'r';
				break;
		}
		snprintf(errpbuf, sizeof(errpbuf), "%s", inet_ntoa(peers[i].laddr.sin_addr));
		my_log(LOG_INFO,"collector #%d: %s:%u/%s/%c", i + 1,
				inet_ntoa(peers[i].addr.sin_addr), ntohs(peers[i].addr.sin_port), errpbuf, c);
	}

	pthread_sigmask(SIG_UNBLOCK, &sig_mask, 0);

	timeout.tv_usec = 0;
	while (!killed
			|| (total_elements - free_elements - pending_queue_length)
			|| emit_count
			|| pending_tail->flags) {

		if (!sigs) {
			timeout.tv_sec = scan_interval;
			select(0, 0, 0, 0, &timeout);
		}

		if (sigs & SIGTERM_MASK && !killed) {
			sigs &= ~SIGTERM_MASK;
			my_log(LOG_INFO, "SIGTERM received. Emitting flows cache...");
			scan_interval = 1;
			frag_lifetime = -1;
			active_lifetime = -1;
			inactive_lifetime = -1;
			emit_timeout = 1;
			unpending_timeout = 1;
			killed = 1;
			pthread_cond_signal(&scan_cond);
			pthread_cond_signal(&unpending_cond);
		}

#if ((DEBUG) & DEBUG_I)
		if (sigs & SIGUSR1_MASK) {
			sigs &= ~SIGUSR1_MASK;
			info_debug();
		}
#endif
	}
	remove(pidfilepath);
#if ((DEBUG) & DEBUG_I)
	info_debug();
#endif
	my_log(LOG_INFO, "Done.");
#ifdef WALL
	return 0;
#endif
}