libipq and threads, problems, very annoying problem

libipq and threads, problems, very annoying problem

[Oumer Teyeb]

Hi,

Here is a sample code where I try to delay every packet by some time (in 
the code I posted here it is fixed to 100 msec.). Inorder to do so, I 
start a pthread for every packet that is received by libipq . I have 
iptables rules that queues tcp packets that are coming and going to some 
other machine. I then ftp to that machine. The program works, the 
problem is with the dumping. I have two files, one is supposed to 
capture the files when they reach the queue (new_file_2) and the other 
just before the set verdict is called (new_file_3). The new_file_2 works 
properly. The problem is with new_file_3. I couldn't get how, but 
somehow, at some location, some erroneous entries are put into the file, 
the end effect being tcpdump not able to recognize the file at some 
point. To check, I commented out the only line which makes the two files 
differ, i.e the one that changes the time stamp of the packet header (as 
indicated by *********), and the two files are not the same. First I 
thought maybe one thread was writing to the file, and then another 
thread becomes active and writes also. So I introduced a mutex_variable 
that will avoid such
problem. But it doesn't help at all. Could you please tell me what I am 
doing wrong? I have to save the files myself as tcpdump get the packets 
before the firewall for incoming packets  

Thanks in advance,
Oumer

#include <sys/time.h>
#include <linux/netfilter.h>
#include <netinet/in.h>
#include <libipq.h>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include "errors.h"
#include <mcheck.h>

#define BUFSIZE 2048

struct ipq_handle *h;
ipq_packet_msg_t *m;

FILE* new_file_2;
FILE* new_file_3;

pthread_mutex_t mutex_write;

/* char src_mac[6] = {0,0,0xb4,0x4b,0x2c, 0xe1}; */
/* char dst_mac[6] = {0,0xb,0x46, 0x10, 0x57, 0x80}; */
/* char ether_type[2] = {8,0}; */

char src_dst_ether [14] = {0,0,0xb4,0x4b,0x2c, 0xe1,0,0xb,0x46, 0x10, 
0x57, 0x80,8,0};

//stolen from pcap
struct PCAP_file_header {
    unsigned int magic;
    short major;
    short minor;
    int thiszone;    /* gmt to local correction */
    unsigned int sigfigs;    /* accuracy of timestamps */
    unsigned int snaplen;    /* max length saved portion of each pkt */
    unsigned int linktype;    /* data link type (LINKTYPE_*) */
};

struct PCAP_packet_header {
    struct timeval ts;    /* time stamp */
    int caplen;    /* length of portion present */
    int len;    /* length this packet (off wire) */
};

typedef struct {
  int         sleep_time;
  ipq_packet_msg_t *packet;
//  unsigned char *data;
  struct PCAP_packet_header pkt_header;
  int thread_id;
} params;

static void die(struct ipq_handle *h)
{
        ipq_perror("ipq_error::");
        ipq_destroy_handle(h);
        exit(1);
}

void *delay_packet (void *param_packet)
{
    int status;
    params *myparams= (params*) param_packet;
    status = pthread_detach (pthread_self ());
    if (status != 0)
        err_abort (status, "Detach thread");

   usleep(100000);
   pthread_mutex_lock (&mutex_write);
   //  gettimeofday(&(myparams->pkt_header.ts), NULL);   
 ************************
   fwrite(&(myparams->pkt_header), sizeof(struct PCAP_packet_header),1, 
new_file_3);
   fwrite(src_dst_ether,1, 14, new_file_3);
   fwrite((unsigned int*)(myparams->packet+1), 1, 
myparams->packet->data_len, new_file_3);
   //free(myparams->packet);
   pthread_mutex_unlock (&mutex_write);
   status = ipq_set_verdict(h, myparams->packet->packet_id,NF_ACCEPT, 0, 
NULL);
   if (status < 0)
      die(h);
   free(myparams);
   return NULL;
}

void signal_handler(int sig)
{
  if (h)
    ipq_destroy_handle(h);
  fclose(new_file_2);
  fclose(new_file_3);
  exit(0);
}

int main(int argc, char **argv)
{
   mtrace();
   int status;
   unsigned char buf[BUFSIZE];
   int thread_id=0;
   
   new_file_2= fopen("dump_tcp_data.txt", "w+b");
   new_file_3 = fopen("dump_tcp_data_after.txt", "w+b");
   
   pthread_mutex_init(&mutex_write, NULL);
   struct PCAP_file_header *file_header =   (struct 
PCAP_file_header*)(malloc(sizeof(struct PCAP_file_header)));
   
   file_header->magic=0xa1b2c3d4;
   file_header->major=2;
   file_header->minor=4;
   file_header->thiszone=0;
   file_header->sigfigs=0;
   file_header->snaplen=65535;
   file_header->linktype=1;
   
   fwrite(file_header, sizeof(struct PCAP_file_header),1, new_file_2);
   fwrite(file_header, sizeof(struct PCAP_file_header),1, new_file_3);
   
   free(file_header);
   pthread_t thread;
   
    h = ipq_create_handle(0, PF_INET);
    sigset(SIGINT, signal_handler);
    if (!h)
                die(h);
    status = ipq_set_mode(h, IPQ_COPY_PACKET, BUFSIZE);
    if (status < 0)
      die(h);
    printf("PACKET_ID\t\tDATA_LENGTH\tSTATUS\n");
    printf("=========\t\t============\t=======\n");
    while (1) {
      status = ipq_read(h, buf, BUFSIZE, 0);
      if (status < 0)
        die(h);
      switch (ipq_message_type(buf)) {
         case NLMSG_ERROR:
         {
           fprintf(stderr, "Error! %s\n", ipq_errstr());
           break;
         }
         case IPQM_PACKET: {
            params *packet_param = (params*)malloc (sizeof (params));;
           if (!packet_param)
          {
           printf("memory allocatin problem\n");
           die(h);
         }
         struct PCAP_packet_header *pkt_header=
          (struct PCAP_packet_header*)(malloc(sizeof(struct 
PCAP_packet_header)));
         ipq_packet_msg_t *m = ipq_get_packet(buf);
         gettimeofday(&(pkt_header->ts), NULL);
         pkt_header->caplen=m->data_len+14;
         pkt_header->len=m->data_len+14;
           
         unsigned char *data = (unsigned char*)(m+1);           
         printf("%lud\t\t%u\t\t", m->packet_id, m->data_len);
           
        packet_param-> packet =  m;
        packet_param->pkt_header = *(pkt_header);
        if (m->data_len > 0)
           {
         fwrite(pkt_header, sizeof(struct PCAP_packet_header),1, 
new_file_2);
         fwrite(src_dst_ether,1,14,new_file_2);
      /*   packet_param->data =(unsigned char*) malloc (sizeof(unsigned 
char*) * m->data_len);

         for (x=0; x < m->data_len; x++)
            {
             packet_param->data[x] = *(data+x);
           }
         */
         fwrite(data, 1, m->data_len, new_file_2);

         status = pthread_create ( &thread, NULL, delay_packet, 
packet_param);
         if (status != 0)
            err_abort (status, "Create a new thread");
         break;
           }
         }

         default:{
           fprintf(stderr, "Unknown message type!\n");
           break;
         }
      }
    }
    ipq_destroy_handle(h);
    return 0;
}

Re: libipq and threads, problems, very annoying problem

[Oumer Teyeb]

I am sorry, but I forgot to mention that I am aware that libipq is not 
thread safe (I read it in one of the message boards, don't remember 
where). Can this problem be attributed to that ? And if so is there a 
way around it? I am kind of stuck with my project unless I do this 
properly, so help is really really appreciated.

Thanks again

Oumer Teyeb wrote:

> Hi,
>
> Here is a sample code where I try to delay every packet by some time 
> (in the code I posted here it is fixed to 100 msec.). Inorder to do 
> so, I start a pthread for every packet that is received by libipq . I 
> have iptables rules that queues tcp packets that are coming and going 
> to some other machine. I then ftp to that machine. The program works, 
> the problem is with the dumping. I have two files, one is supposed to 
> capture the files when they reach the queue (new_file_2) and the other 
> just before the set verdict is called (new_file_3). The new_file_2 
> works properly. The problem is with new_file_3. I couldn't get how, 
> but somehow, at some location, some erroneous entries are put into the 
> file, the end effect being tcpdump not able to recognize the file at 
> some point. To check, I commented out the only line which makes the 
> two files differ, i.e the one that changes the time stamp of the 
> packet header (as indicated by *********), and the two files are not 
> the same. First I thought maybe one thread was writing to the file, 
> and then another thread becomes active and writes also. So I 
> introduced a mutex_variable that will avoid such
> problem. But it doesn't help at all. Could you please tell me what I 
> am doing wrong? I have to save the files myself as tcpdump get the 
> packets before the firewall for incoming packets 
> Thanks in advance,
> Oumer
>
> #include <sys/time.h>
> #include <linux/netfilter.h>
> #include <netinet/in.h>
> #include <libipq.h>
> #include <pthread.h>
> #include <unistd.h>
> #include <signal.h>
> #include <string.h>
> #include <stdio.h>
> #include <stdlib.h>
> #include "errors.h"
> #include <mcheck.h>
>
> #define BUFSIZE 2048
>
> struct ipq_handle *h;
> ipq_packet_msg_t *m;
>
> FILE* new_file_2;
> FILE* new_file_3;
>
> pthread_mutex_t mutex_write;
>
> /* char src_mac[6] = {0,0,0xb4,0x4b,0x2c, 0xe1}; */
> /* char dst_mac[6] = {0,0xb,0x46, 0x10, 0x57, 0x80}; */
> /* char ether_type[2] = {8,0}; */
>
> char src_dst_ether [14] = {0,0,0xb4,0x4b,0x2c, 0xe1,0,0xb,0x46, 0x10, 
> 0x57, 0x80,8,0};
>
> //stolen from pcap
> struct PCAP_file_header {
>    unsigned int magic;
>    short major;
>    short minor;
>    int thiszone;    /* gmt to local correction */
>    unsigned int sigfigs;    /* accuracy of timestamps */
>    unsigned int snaplen;    /* max length saved portion of each pkt */
>    unsigned int linktype;    /* data link type (LINKTYPE_*) */
> };
>
> struct PCAP_packet_header {
>    struct timeval ts;    /* time stamp */
>    int caplen;    /* length of portion present */
>    int len;    /* length this packet (off wire) */
> };
>
> typedef struct {
>  int         sleep_time;
>  ipq_packet_msg_t *packet;
> //  unsigned char *data;
>  struct PCAP_packet_header pkt_header;
>  int thread_id;
> } params;
>
> static void die(struct ipq_handle *h)
> {
>        ipq_perror("ipq_error::");
>        ipq_destroy_handle(h);
>        exit(1);
> }
>
> void *delay_packet (void *param_packet)
> {
>    int status;
>    params *myparams= (params*) param_packet;
>    status = pthread_detach (pthread_self ());
>    if (status != 0)
>        err_abort (status, "Detach thread");
>
>   usleep(100000);
>   pthread_mutex_lock (&mutex_write);
>   //  gettimeofday(&(myparams->pkt_header.ts), NULL);   
> ************************
>   fwrite(&(myparams->pkt_header), sizeof(struct PCAP_packet_header),1, 
> new_file_3);
>   fwrite(src_dst_ether,1, 14, new_file_3);
>   fwrite((unsigned int*)(myparams->packet+1), 1, 
> myparams->packet->data_len, new_file_3);
>   //free(myparams->packet);
>   pthread_mutex_unlock (&mutex_write);
>   status = ipq_set_verdict(h, myparams->packet->packet_id,NF_ACCEPT, 
> 0, NULL);
>   if (status < 0)
>      die(h);
>   free(myparams);
>   return NULL;
> }
>
> void signal_handler(int sig)
> {
>  if (h)
>    ipq_destroy_handle(h);
>  fclose(new_file_2);
>  fclose(new_file_3);
>  exit(0);
> }
>
> int main(int argc, char **argv)
> {
>   mtrace();
>   int status;
>   unsigned char buf[BUFSIZE];
>   int thread_id=0;
>     new_file_2= fopen("dump_tcp_data.txt", "w+b");
>   new_file_3 = fopen("dump_tcp_data_after.txt", "w+b");
>     pthread_mutex_init(&mutex_write, NULL);
>   struct PCAP_file_header *file_header =   (struct 
> PCAP_file_header*)(malloc(sizeof(struct PCAP_file_header)));
>     file_header->magic=0xa1b2c3d4;
>   file_header->major=2;
>   file_header->minor=4;
>   file_header->thiszone=0;
>   file_header->sigfigs=0;
>   file_header->snaplen=65535;
>   file_header->linktype=1;
>     fwrite(file_header, sizeof(struct PCAP_file_header),1, new_file_2);
>   fwrite(file_header, sizeof(struct PCAP_file_header),1, new_file_3);
>     free(file_header);
>   pthread_t thread;
>      h = ipq_create_handle(0, PF_INET);
>    sigset(SIGINT, signal_handler);
>    if (!h)
>                die(h);
>    status = ipq_set_mode(h, IPQ_COPY_PACKET, BUFSIZE);
>    if (status < 0)
>      die(h);
>    printf("PACKET_ID\t\tDATA_LENGTH\tSTATUS\n");
>    printf("=========\t\t============\t=======\n");
>    while (1) {
>      status = ipq_read(h, buf, BUFSIZE, 0);
>      if (status < 0)
>        die(h);
>      switch (ipq_message_type(buf)) {
>         case NLMSG_ERROR:
>         {
>           fprintf(stderr, "Error! %s\n", ipq_errstr());
>           break;
>         }
>         case IPQM_PACKET: {
>            params *packet_param = (params*)malloc (sizeof (params));;
>           if (!packet_param)
>          {
>           printf("memory allocatin problem\n");
>           die(h);
>         }
>         struct PCAP_packet_header *pkt_header=
>          (struct PCAP_packet_header*)(malloc(sizeof(struct 
> PCAP_packet_header)));
>         ipq_packet_msg_t *m = ipq_get_packet(buf);
>         gettimeofday(&(pkt_header->ts), NULL);
>         pkt_header->caplen=m->data_len+14;
>         pkt_header->len=m->data_len+14;
>                   unsigned char *data = (unsigned 
> char*)(m+1);                   printf("%lud\t\t%u\t\t", m->packet_id, 
> m->data_len);
>                  packet_param-> packet =  m;
>        packet_param->pkt_header = *(pkt_header);
>        if (m->data_len > 0)
>           {
>         fwrite(pkt_header, sizeof(struct PCAP_packet_header),1, 
> new_file_2);
>         fwrite(src_dst_ether,1,14,new_file_2);
>      /*   packet_param->data =(unsigned char*) malloc (sizeof(unsigned 
> char*) * m->data_len);
>
>         for (x=0; x < m->data_len; x++)
>            {
>             packet_param->data[x] = *(data+x);
>           }
>         */
>         fwrite(data, 1, m->data_len, new_file_2);
>
>         status = pthread_create ( &thread, NULL, delay_packet, 
> packet_param);
>         if (status != 0)
>            err_abort (status, "Create a new thread");
>         break;
>           }
>         }
>
>         default:{
>           fprintf(stderr, "Unknown message type!\n");
>           break;
>         }
>      }
>    }
>    ipq_destroy_handle(h);
>    return 0;
> }
>
>
>
>

Re: libipq and threads, problems, very annoying problem

[Scott MacKay]

Is there a particular reason why you need to generate
a thread per packet?  Threads can have a moderate
overhead, expecially if you are creating them in the
hundreds (depending on your traffic levels).
If you are just looking to delay, I would suggest
this:

Create a fifo, linked list, or such with 2 structure
elements: a packet buffer (for the ipq_read data) and
a timestamp.  Optional: I believe the packet has a
timestamp marked on it but I have never used that.
in a loop do something like:
while (some loop condition)  {
   get the current time
   if there are 1+ packets queued
       get the timestamp of the 1st packet 
       set 'timeout':subtract first packet time from
now
   else
       'timeout' = 0
   call ipq_read with the timeout
   if we got a packet
       set its time += 100ms
       add it to the end of the queue
   get the current time
   if first packet time - now <= 0
      release packet
} 
Basically, the first packet will always have the
shortest delay.  By setting a timeout, you can exit
the ipq_read and handle a packet due for release
without the need for threads.
Hope it helps!

--- Oumer Teyeb <oumer@kom.auc.dk> wrote:
> I am sorry, but I forgot to mention that I am aware
> that libipq is not 
> thread safe (I read it in one of the message boards,
> don't remember 
> where). Can this problem be attributed to that ? And
> if so is there a 
> way around it? I am kind of stuck with my project
> unless I do this 
> properly, so help is really really appreciated.
> 
> Thanks again
> 
> Oumer Teyeb wrote:
> 
> > Hi,
> >
> > Here is a sample code where I try to delay every
> packet by some time 
> > (in the code I posted here it is fixed to 100
> msec.). Inorder to do 
> > so, I start a pthread for every packet that is
> received by libipq . I 
> > have iptables rules that queues tcp packets that
> are coming and going 
> > to some other machine. I then ftp to that machine.
> The program works, 
> > the problem is with the dumping. I have two files,
> one is supposed to 
> > capture the files when they reach the queue
> (new_file_2) and the other 
> > just before the set verdict is called
> (new_file_3). The new_file_2 
> > works properly. The problem is with new_file_3. I
> couldn't get how, 
> > but somehow, at some location, some erroneous
> entries are put into the 
> > file, the end effect being tcpdump not able to
> recognize the file at 
> > some point. To check, I commented out the only
> line which makes the 
> > two files differ, i.e the one that changes the
> time stamp of the 
> > packet header (as indicated by *********), and the
> two files are not 
> > the same. First I thought maybe one thread was
> writing to the file, 
> > and then another thread becomes active and writes
> also. So I 
> > introduced a mutex_variable that will avoid such
> > problem. But it doesn't help at all. Could you
> please tell me what I 
> > am doing wrong? I have to save the files myself as
> tcpdump get the 
> > packets before the firewall for incoming packets 
> > Thanks in advance,
> > Oumer
> >
> > #include <sys/time.h>
> > #include <linux/netfilter.h>
> > #include <netinet/in.h>
> > #include <libipq.h>
> > #include <pthread.h>
> > #include <unistd.h>
> > #include <signal.h>
> > #include <string.h>
> > #include <stdio.h>
> > #include <stdlib.h>
> > #include "errors.h"
> > #include <mcheck.h>
> >
> > #define BUFSIZE 2048
> >
> > struct ipq_handle *h;
> > ipq_packet_msg_t *m;
> >
> > FILE* new_file_2;
> > FILE* new_file_3;
> >
> > pthread_mutex_t mutex_write;
> >
> > /* char src_mac[6] = {0,0,0xb4,0x4b,0x2c, 0xe1};
> */
> > /* char dst_mac[6] = {0,0xb,0x46, 0x10, 0x57,
> 0x80}; */
> > /* char ether_type[2] = {8,0}; */
> >
> > char src_dst_ether [14] = {0,0,0xb4,0x4b,0x2c,
> 0xe1,0,0xb,0x46, 0x10, 
> > 0x57, 0x80,8,0};
> >
> > //stolen from pcap
> > struct PCAP_file_header {
> >    unsigned int magic;
> >    short major;
> >    short minor;
> >    int thiszone;    /* gmt to local correction */
> >    unsigned int sigfigs;    /* accuracy of
> timestamps */
> >    unsigned int snaplen;    /* max length saved
> portion of each pkt */
> >    unsigned int linktype;    /* data link type
> (LINKTYPE_*) */
> > };
> >
> > struct PCAP_packet_header {
> >    struct timeval ts;    /* time stamp */
> >    int caplen;    /* length of portion present */
> >    int len;    /* length this packet (off wire) */
> > };
> >
> > typedef struct {
> >  int         sleep_time;
> >  ipq_packet_msg_t *packet;
> > //  unsigned char *data;
> >  struct PCAP_packet_header pkt_header;
> >  int thread_id;
> > } params;
> >
> > static void die(struct ipq_handle *h)
> > {
> >        ipq_perror("ipq_error::");
> >        ipq_destroy_handle(h);
> >        exit(1);
> > }
> >
> > void *delay_packet (void *param_packet)
> > {
> >    int status;
> >    params *myparams= (params*) param_packet;
> >    status = pthread_detach (pthread_self ());
> >    if (status != 0)
> >        err_abort (status, "Detach thread");
> >
> >   usleep(100000);
> >   pthread_mutex_lock (&mutex_write);
> >   //  gettimeofday(&(myparams->pkt_header.ts),
> NULL);   
> > ************************
> >   fwrite(&(myparams->pkt_header), sizeof(struct
> PCAP_packet_header),1, 
> > new_file_3);
> >   fwrite(src_dst_ether,1, 14, new_file_3);
> >   fwrite((unsigned int*)(myparams->packet+1), 1, 
> > myparams->packet->data_len, new_file_3);
> >   //free(myparams->packet);
> >   pthread_mutex_unlock (&mutex_write);
> >   status = ipq_set_verdict(h,
> myparams->packet->packet_id,NF_ACCEPT, 
> > 0, NULL);
> >   if (status < 0)
> >      die(h);
> >   free(myparams);
> >   return NULL;
> > }
> >
> > void signal_handler(int sig)
> > {
> >  if (h)
> >    ipq_destroy_handle(h);
> >  fclose(new_file_2);
> >  fclose(new_file_3);
> >  exit(0);
> > }
> >
> > int main(int argc, char **argv)
> > {
> >   mtrace();
> >   int status;
> >   unsigned char buf[BUFSIZE];
> >   int thread_id=0;
> >     new_file_2= fopen("dump_tcp_data.txt", "w+b");
> >   new_file_3 = fopen("dump_tcp_data_after.txt",
> "w+b");
> >     pthread_mutex_init(&mutex_write, NULL);
> >   struct PCAP_file_header *file_header =   (struct
> 
> > PCAP_file_header*)(malloc(sizeof(struct
> PCAP_file_header)));
> >     file_header->magic=0xa1b2c3d4;
> >   file_header->major=2;
> >   file_header->minor=4;
> >   file_header->thiszone=0;
> >   file_header->sigfigs=0;
> >   file_header->snaplen=65535;
> >   file_header->linktype=1;
> >     fwrite(file_header, sizeof(struct
> PCAP_file_header),1, new_file_2);
> >   fwrite(file_header, sizeof(struct
> PCAP_file_header),1, new_file_3);
> >     free(file_header);
> >   pthread_t thread;
> >      h = ipq_create_handle(0, PF_INET);
> >    sigset(SIGINT, signal_handler);
> >    if (!h)
> >                die(h);
> >    status = ipq_set_mode(h, IPQ_COPY_PACKET,
> BUFSIZE);
> >    if (status < 0)
> >      die(h);
> >    printf("PACKET_ID\t\tDATA_LENGTH\tSTATUS\n");
> >    printf("=========\t\t============\t=======\n");
> 
=== message truncated ===


__________________________________
Do you Yahoo!?
The New Yahoo! Shopping - with improved product search
http://shopping.yahoo.com

Re: libipq and threads, problems, very annoying problem

[oumer]

The code snippet I posted is just a sample, in my real code I have only two
threads, one thread using libipq to get packets and notify another thread about
that, and that thread will be emulate a certain link layer protocol, and when
the processing is done, the thread that is doing the libipq staff will get the
verdict and it will use the set_verdict to let the packet go or drop it. As I am
interested on the exact arrival times of the packets, I wanted to dump the data
into two files, one when it reaches the queue and once just before the verdict
is passed on the data. As there are more than twenty parameters defining my link
layer protocol, and it is statistical, I don't think I can use the method you
specified. 


Quoting Scott MacKay <scottmackay@yahoo.com>:

> Is there a particular reason why you need to generate
> a thread per packet?  Threads can have a moderate
> overhead, expecially if you are creating them in the
> hundreds (depending on your traffic levels).
> If you are just looking to delay, I would suggest
> this:
> 
> Create a fifo, linked list, or such with 2 structure
> elements: a packet buffer (for the ipq_read data) and
> a timestamp.  Optional: I believe the packet has a
> timestamp marked on it but I have never used that.
> in a loop do something like:
> while (some loop condition)  {
>    get the current time
>    if there are 1+ packets queued
>        get the timestamp of the 1st packet 
>        set 'timeout':subtract first packet time from
> now
>    else
>        'timeout' = 0
>    call ipq_read with the timeout
>    if we got a packet
>        set its time += 100ms
>        add it to the end of the queue
>    get the current time
>    if first packet time - now <= 0
>       release packet
> } 
> Basically, the first packet will always have the
> shortest delay.  By setting a timeout, you can exit
> the ipq_read and handle a packet due for release
> without the need for threads.
> Hope it helps!
> 
> --- Oumer Teyeb <oumer@kom.auc.dk> wrote:
> > I am sorry, but I forgot to mention that I am aware
> > that libipq is not 
> > thread safe (I read it in one of the message boards,
> > don't remember 
> > where). Can this problem be attributed to that ? And
> > if so is there a 
> > way around it? I am kind of stuck with my project
> > unless I do this 
> > properly, so help is really really appreciated.
> > 
> > Thanks again
> > 
> > Oumer Teyeb wrote:
> > 
> > > Hi,
> > >
> > > Here is a sample code where I try to delay every
> > packet by some time 
> > > (in the code I posted here it is fixed to 100
> > msec.). Inorder to do 
> > > so, I start a pthread for every packet that is
> > received by libipq . I 
> > > have iptables rules that queues tcp packets that
> > are coming and going 
> > > to some other machine. I then ftp to that machine.
> > The program works, 
> > > the problem is with the dumping. I have two files,
> > one is supposed to 
> > > capture the files when they reach the queue
> > (new_file_2) and the other 
> > > just before the set verdict is called
> > (new_file_3). The new_file_2 
> > > works properly. The problem is with new_file_3. I
> > couldn't get how, 
> > > but somehow, at some location, some erroneous
> > entries are put into the 
> > > file, the end effect being tcpdump not able to
> > recognize the file at 
> > > some point. To check, I commented out the only
> > line which makes the 
> > > two files differ, i.e the one that changes the
> > time stamp of the 
> > > packet header (as indicated by *********), and the
> > two files are not 
> > > the same. First I thought maybe one thread was
> > writing to the file, 
> > > and then another thread becomes active and writes
> > also. So I 
> > > introduced a mutex_variable that will avoid such
> > > problem. But it doesn't help at all. Could you
> > please tell me what I 
> > > am doing wrong? I have to save the files myself as
> > tcpdump get the 
> > > packets before the firewall for incoming packets 
> > > Thanks in advance,
> > > Oumer
> > >
> > > #include <sys/time.h>
> > > #include <linux/netfilter.h>
> > > #include <netinet/in.h>
> > > #include <libipq.h>
> > > #include <pthread.h>
> > > #include <unistd.h>
> > > #include <signal.h>
> > > #include <string.h>
> > > #include <stdio.h>
> > > #include <stdlib.h>
> > > #include "errors.h"
> > > #include <mcheck.h>
> > >
> > > #define BUFSIZE 2048
> > >
> > > struct ipq_handle *h;
> > > ipq_packet_msg_t *m;
> > >
> > > FILE* new_file_2;
> > > FILE* new_file_3;
> > >
> > > pthread_mutex_t mutex_write;
> > >
> > > /* char src_mac[6] = {0,0,0xb4,0x4b,0x2c, 0xe1};
> > */
> > > /* char dst_mac[6] = {0,0xb,0x46, 0x10, 0x57,
> > 0x80}; */
> > > /* char ether_type[2] = {8,0}; */
> > >
> > > char src_dst_ether [14] = {0,0,0xb4,0x4b,0x2c,
> > 0xe1,0,0xb,0x46, 0x10, 
> > > 0x57, 0x80,8,0};
> > >
> > > //stolen from pcap
> > > struct PCAP_file_header {
> > >    unsigned int magic;
> > >    short major;
> > >    short minor;
> > >    int thiszone;    /* gmt to local correction */
> > >    unsigned int sigfigs;    /* accuracy of
> > timestamps */
> > >    unsigned int snaplen;    /* max length saved
> > portion of each pkt */
> > >    unsigned int linktype;    /* data link type
> > (LINKTYPE_*) */
> > > };
> > >
> > > struct PCAP_packet_header {
> > >    struct timeval ts;    /* time stamp */
> > >    int caplen;    /* length of portion present */
> > >    int len;    /* length this packet (off wire) */
> > > };
> > >
> > > typedef struct {
> > >  int         sleep_time;
> > >  ipq_packet_msg_t *packet;
> > > //  unsigned char *data;
> > >  struct PCAP_packet_header pkt_header;
> > >  int thread_id;
> > > } params;
> > >
> > > static void die(struct ipq_handle *h)
> > > {
> > >        ipq_perror("ipq_error::");
> > >        ipq_destroy_handle(h);
> > >        exit(1);
> > > }
> > >
> > > void *delay_packet (void *param_packet)
> > > {
> > >    int status;
> > >    params *myparams= (params*) param_packet;
> > >    status = pthread_detach (pthread_self ());
> > >    if (status != 0)
> > >        err_abort (status, "Detach thread");
> > >
> > >   usleep(100000);
> > >   pthread_mutex_lock (&mutex_write);
> > >   //  gettimeofday(&(myparams->pkt_header.ts),
> > NULL);   
> > > ************************
> > >   fwrite(&(myparams->pkt_header), sizeof(struct
> > PCAP_packet_header),1, 
> > > new_file_3);
> > >   fwrite(src_dst_ether,1, 14, new_file_3);
> > >   fwrite((unsigned int*)(myparams->packet+1), 1, 
> > > myparams->packet->data_len, new_file_3);
> > >   //free(myparams->packet);
> > >   pthread_mutex_unlock (&mutex_write);
> > >   status = ipq_set_verdict(h,
> > myparams->packet->packet_id,NF_ACCEPT, 
> > > 0, NULL);
> > >   if (status < 0)
> > >      die(h);
> > >   free(myparams);
> > >   return NULL;
> > > }
> > >
> > > void signal_handler(int sig)
> > > {
> > >  if (h)
> > >    ipq_destroy_handle(h);
> > >  fclose(new_file_2);
> > >  fclose(new_file_3);
> > >  exit(0);
> > > }
> > >
> > > int main(int argc, char **argv)
> > > {
> > >   mtrace();
> > >   int status;
> > >   unsigned char buf[BUFSIZE];
> > >   int thread_id=0;
> > >     new_file_2= fopen("dump_tcp_data.txt", "w+b");
> > >   new_file_3 = fopen("dump_tcp_data_after.txt",
> > "w+b");
> > >     pthread_mutex_init(&mutex_write, NULL);
> > >   struct PCAP_file_header *file_header =   (struct
> > 
> > > PCAP_file_header*)(malloc(sizeof(struct
> > PCAP_file_header)));
> > >     file_header->magic=0xa1b2c3d4;
> > >   file_header->major=2;
> > >   file_header->minor=4;
> > >   file_header->thiszone=0;
> > >   file_header->sigfigs=0;
> > >   file_header->snaplen=65535;
> > >   file_header->linktype=1;
> > >     fwrite(file_header, sizeof(struct
> > PCAP_file_header),1, new_file_2);
> > >   fwrite(file_header, sizeof(struct
> > PCAP_file_header),1, new_file_3);
> > >     free(file_header);
> > >   pthread_t thread;
> > >      h = ipq_create_handle(0, PF_INET);
> > >    sigset(SIGINT, signal_handler);
> > >    if (!h)
> > >                die(h);
> > >    status = ipq_set_mode(h, IPQ_COPY_PACKET,
> > BUFSIZE);
> > >    if (status < 0)
> > >      die(h);
> > >    printf("PACKET_ID\t\tDATA_LENGTH\tSTATUS\n");
> > >    printf("=========\t\t============\t=======\n");
> > 
> === message truncated ===
> 
> 
> __________________________________
> Do you Yahoo!?
> The New Yahoo! Shopping - with improved product search
> http://shopping.yahoo.com
> 
> 

Re: libipq and threads, problems, very annoying problem

[oumer]

I asked some guys at experts-exchange, and here are some of the correspondance
so far, may be some one from here, pick from that discussion? 
http://www.experts-exchange.com/Programming/Programming_Platforms/Linux_Programming/Q_20766491.html

Quoting Scott MacKay <scottmackay@yahoo.com>:

> Is there a particular reason why you need to generate
> a thread per packet?  Threads can have a moderate
> overhead, expecially if you are creating them in the
> hundreds (depending on your traffic levels).
> If you are just looking to delay, I would suggest
> this:
> 
> Create a fifo, linked list, or such with 2 structure
> elements: a packet buffer (for the ipq_read data) and
> a timestamp.  Optional: I believe the packet has a
> timestamp marked on it but I have never used that.
> in a loop do something like:
> while (some loop condition)  {
>    get the current time
>    if there are 1+ packets queued
>        get the timestamp of the 1st packet 
>        set 'timeout':subtract first packet time from
> now
>    else
>        'timeout' = 0
>    call ipq_read with the timeout
>    if we got a packet
>        set its time += 100ms
>        add it to the end of the queue
>    get the current time
>    if first packet time - now <= 0
>       release packet
> } 
> Basically, the first packet will always have the
> shortest delay.  By setting a timeout, you can exit
> the ipq_read and handle a packet due for release
> without the need for threads.
> Hope it helps!
> 
> --- Oumer Teyeb <oumer@kom.auc.dk> wrote:
> > I am sorry, but I forgot to mention that I am aware
> > that libipq is not 
> > thread safe (I read it in one of the message boards,
> > don't remember 
> > where). Can this problem be attributed to that ? And
> > if so is there a 
> > way around it? I am kind of stuck with my project
> > unless I do this 
> > properly, so help is really really appreciated.
> > 
> > Thanks again
> > 
> > Oumer Teyeb wrote:
> > 
> > > Hi,
> > >
> > > Here is a sample code where I try to delay every
> > packet by some time 
> > > (in the code I posted here it is fixed to 100
> > msec.). Inorder to do 
> > > so, I start a pthread for every packet that is
> > received by libipq . I 
> > > have iptables rules that queues tcp packets that
> > are coming and going 
> > > to some other machine. I then ftp to that machine.
> > The program works, 
> > > the problem is with the dumping. I have two files,
> > one is supposed to 
> > > capture the files when they reach the queue
> > (new_file_2) and the other 
> > > just before the set verdict is called
> > (new_file_3). The new_file_2 
> > > works properly. The problem is with new_file_3. I
> > couldn't get how, 
> > > but somehow, at some location, some erroneous
> > entries are put into the 
> > > file, the end effect being tcpdump not able to
> > recognize the file at 
> > > some point. To check, I commented out the only
> > line which makes the 
> > > two files differ, i.e the one that changes the
> > time stamp of the 
> > > packet header (as indicated by *********), and the
> > two files are not 
> > > the same. First I thought maybe one thread was
> > writing to the file, 
> > > and then another thread becomes active and writes
> > also. So I 
> > > introduced a mutex_variable that will avoid such
> > > problem. But it doesn't help at all. Could you
> > please tell me what I 
> > > am doing wrong? I have to save the files myself as
> > tcpdump get the 
> > > packets before the firewall for incoming packets 
> > > Thanks in advance,
> > > Oumer
> > >
> > > #include <sys/time.h>
> > > #include <linux/netfilter.h>
> > > #include <netinet/in.h>
> > > #include <libipq.h>
> > > #include <pthread.h>
> > > #include <unistd.h>
> > > #include <signal.h>
> > > #include <string.h>
> > > #include <stdio.h>
> > > #include <stdlib.h>
> > > #include "errors.h"
> > > #include <mcheck.h>
> > >
> > > #define BUFSIZE 2048
> > >
> > > struct ipq_handle *h;
> > > ipq_packet_msg_t *m;
> > >
> > > FILE* new_file_2;
> > > FILE* new_file_3;
> > >
> > > pthread_mutex_t mutex_write;
> > >
> > > /* char src_mac[6] = {0,0,0xb4,0x4b,0x2c, 0xe1};
> > */
> > > /* char dst_mac[6] = {0,0xb,0x46, 0x10, 0x57,
> > 0x80}; */
> > > /* char ether_type[2] = {8,0}; */
> > >
> > > char src_dst_ether [14] = {0,0,0xb4,0x4b,0x2c,
> > 0xe1,0,0xb,0x46, 0x10, 
> > > 0x57, 0x80,8,0};
> > >
> > > //stolen from pcap
> > > struct PCAP_file_header {
> > >    unsigned int magic;
> > >    short major;
> > >    short minor;
> > >    int thiszone;    /* gmt to local correction */
> > >    unsigned int sigfigs;    /* accuracy of
> > timestamps */
> > >    unsigned int snaplen;    /* max length saved
> > portion of each pkt */
> > >    unsigned int linktype;    /* data link type
> > (LINKTYPE_*) */
> > > };
> > >
> > > struct PCAP_packet_header {
> > >    struct timeval ts;    /* time stamp */
> > >    int caplen;    /* length of portion present */
> > >    int len;    /* length this packet (off wire) */
> > > };
> > >
> > > typedef struct {
> > >  int         sleep_time;
> > >  ipq_packet_msg_t *packet;
> > > //  unsigned char *data;
> > >  struct PCAP_packet_header pkt_header;
> > >  int thread_id;
> > > } params;
> > >
> > > static void die(struct ipq_handle *h)
> > > {
> > >        ipq_perror("ipq_error::");
> > >        ipq_destroy_handle(h);
> > >        exit(1);
> > > }
> > >
> > > void *delay_packet (void *param_packet)
> > > {
> > >    int status;
> > >    params *myparams= (params*) param_packet;
> > >    status = pthread_detach (pthread_self ());
> > >    if (status != 0)
> > >        err_abort (status, "Detach thread");
> > >
> > >   usleep(100000);
> > >   pthread_mutex_lock (&mutex_write);
> > >   //  gettimeofday(&(myparams->pkt_header.ts),
> > NULL);   
> > > ************************
> > >   fwrite(&(myparams->pkt_header), sizeof(struct
> > PCAP_packet_header),1, 
> > > new_file_3);
> > >   fwrite(src_dst_ether,1, 14, new_file_3);
> > >   fwrite((unsigned int*)(myparams->packet+1), 1, 
> > > myparams->packet->data_len, new_file_3);
> > >   //free(myparams->packet);
> > >   pthread_mutex_unlock (&mutex_write);
> > >   status = ipq_set_verdict(h,
> > myparams->packet->packet_id,NF_ACCEPT, 
> > > 0, NULL);
> > >   if (status < 0)
> > >      die(h);
> > >   free(myparams);
> > >   return NULL;
> > > }
> > >
> > > void signal_handler(int sig)
> > > {
> > >  if (h)
> > >    ipq_destroy_handle(h);
> > >  fclose(new_file_2);
> > >  fclose(new_file_3);
> > >  exit(0);
> > > }
> > >
> > > int main(int argc, char **argv)
> > > {
> > >   mtrace();
> > >   int status;
> > >   unsigned char buf[BUFSIZE];
> > >   int thread_id=0;
> > >     new_file_2= fopen("dump_tcp_data.txt", "w+b");
> > >   new_file_3 = fopen("dump_tcp_data_after.txt",
> > "w+b");
> > >     pthread_mutex_init(&mutex_write, NULL);
> > >   struct PCAP_file_header *file_header =   (struct
> > 
> > > PCAP_file_header*)(malloc(sizeof(struct
> > PCAP_file_header)));
> > >     file_header->magic=0xa1b2c3d4;
> > >   file_header->major=2;
> > >   file_header->minor=4;
> > >   file_header->thiszone=0;
> > >   file_header->sigfigs=0;
> > >   file_header->snaplen=65535;
> > >   file_header->linktype=1;
> > >     fwrite(file_header, sizeof(struct
> > PCAP_file_header),1, new_file_2);
> > >   fwrite(file_header, sizeof(struct
> > PCAP_file_header),1, new_file_3);
> > >     free(file_header);
> > >   pthread_t thread;
> > >      h = ipq_create_handle(0, PF_INET);
> > >    sigset(SIGINT, signal_handler);
> > >    if (!h)
> > >                die(h);
> > >    status = ipq_set_mode(h, IPQ_COPY_PACKET,
> > BUFSIZE);
> > >    if (status < 0)
> > >      die(h);
> > >    printf("PACKET_ID\t\tDATA_LENGTH\tSTATUS\n");
> > >    printf("=========\t\t============\t=======\n");
> > 
> === message truncated ===
> 
> 
> __________________________________
> Do you Yahoo!?
> The New Yahoo! Shopping - with improved product search
> http://shopping.yahoo.com
> 
>