Changeset dd6e168 for lib

Timestamp:

07/28/05 13:45:54 (16 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, getfragoff, help, libtrace4, master, ndag_format, pfring, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes

Children:

4dedc28

Parents:

5eae97a

Message:

gzerf and rtserver output added, no fifo ack though

File:

• lib/trace.c (modified) (9 diffs)

lib/trace.c

@@ -176 +176 @@
 
         struct fifo_t *fifo;
+
+        // rtserver stuff
+        
+        /** fd that watches for incoming connections */
+        int connect_fd;
+        /** fds that represent connected clients */
+        fd_set rt_fds;
+        /** fds that are listening for connections (this should only contain connect_fd) */
+        fd_set listen;
+        int max_rtfds;
+        /** Used for creating new fds as required */
+        struct sockaddr_in * remote;
 };
 
@@ -677 +689 @@
  *
  * @param uri   the uri string describing the output format and the destination
- * @returns the newly created libtrace_out_t structure
+ * @returns the newly created libtrace_out_t structure, or NULL if an error occurs
  * 
  * @author Shane Alcock
@@ -683 +695 @@
 struct libtrace_out_t *trace_output_create(char *uri) {
         struct libtrace_out_t *libtrace = malloc(sizeof(struct libtrace_out_t));
-        struct sockaddr_in remote, client;
-        int client_fd, clilen;
+        struct sockaddr_in listener;
         struct hostent *he;
+        int yes = 1;
         
         if (init_output(&libtrace, uri) == 0)
@@ -692 +704 @@
         switch(libtrace->outputformat) {
                 case RTSERVER:
+                        FD_ZERO(&libtrace->listen);
+                        FD_ZERO(&libtrace->rt_fds);
+                        
                         if ((he=gethostbyname(libtrace->conn_info.rt.hostname)) == NULL) {
                                 perror("gethostbyname");
                                 return 0;
                         }
-                        if ((libtrace->output.fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
+                        if ((libtrace->connect_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
                                 perror("socket");
                                 return 0;
                         }
+                        if (setsockopt(libtrace->connect_fd,SOL_SOCKET,SO_REUSEADDR,&yes,sizeof(int)) == -1) {
+                                perror("setsockopt");
+                                return 0;
+                        }
+                        libtrace->remote = calloc(1,sizeof(struct sockaddr_in));
                         // Need to set up a listening server here
-                        bzero((char *) &remote, sizeof(remote)); 
-                        remote.sin_family = AF_INET;
-                        remote.sin_addr.s_addr = INADDR_ANY;
-                        remote.sin_port = htons(libtrace->conn_info.rt.port);
-
-                        if (bind(libtrace->output.fd, (struct sockaddr *) &remote, sizeof(remote)) < 0) {
+                        bzero((char *) libtrace->remote, sizeof(libtrace->remote)); 
+                        libtrace->remote->sin_family = AF_INET;
+                        libtrace->remote->sin_addr.s_addr = INADDR_ANY;
+                        libtrace->remote->sin_port = htons(libtrace->conn_info.rt.port);
+                        
+                        if (bind(libtrace->connect_fd, (struct sockaddr *) libtrace->remote, sizeof(struct sockaddr_in)) < 0) {
                                 perror("bind");
                                 return 0;
                         }
-                        fprintf(stderr, "Waiting for client to connect\n");
-
-                        listen(libtrace->output.fd, 5);
+                        // fprintf(stderr, "Waiting for client to connect\n");
+
+                        if (listen(libtrace->connect_fd, 10) == -1) {
+                                perror("listen");
+                                return 0;
+                        }
+                                
+                        FD_SET(libtrace->connect_fd, &libtrace->listen);
+                        libtrace->max_rtfds = libtrace->connect_fd;
+                        /*
                         clilen = sizeof(client);
                         if ((client_fd = accept(libtrace->output.fd, (struct sockaddr *) &client, &clilen)) < 0) {
@@ -720 +747 @@
                         libtrace->output.fd = client_fd;
                         fprintf(stderr, "Client connected\n");                       
+                        */
                         break;
 
@@ -798 +826 @@
         }
         destroy_fifo(libtrace->fifo);
+        free(libtrace->remote);
         free(libtrace);
 }
@@ -1173 +1202 @@
         void *buffer = &buf[intsize];
         int write_required = 0;
-
+        struct timeval tv;
+        int i;
+        int rt_fd;
+        int sin_size = sizeof(struct sockaddr_in);
+        
+        fd_set current;
+        
         assert(libtrace);
         assert(packet); 
@@ -1189 +1224 @@
         if (libtrace->outputformat == RTSERVER) {
                 do {
+                        // check for incoming connections  NOTE: Should this be inside the do { }
+                        tv.tv_sec = 0;
+                        tv.tv_usec = 10;
+                        current = libtrace->listen;
+                        do {
+                                if (select(libtrace->max_rtfds + 1, &current, NULL, NULL,&tv) >=0 ) {
+                                        break;
+                                }
+                        }
+                        while (errno == EINTR);
+                        for (i = 0; i <= libtrace->max_rtfds; i++) {
+                                if (FD_ISSET(i, &current)) {
+                                        // Got something on the listening socket
+                                        if (i == libtrace->connect_fd) {
+                                                if ((rt_fd = accept(i, (struct sockaddr *) libtrace->remote, 
+                                                                                &sin_size)) == -1) {
+                                                        perror("accept");
+                                                } else {
+                                                        printf("Client connected\n");
+                                                        FD_SET(rt_fd, &libtrace->rt_fds);
+                                                        if (rt_fd > libtrace->max_rtfds)
+                                                                libtrace->max_rtfds = rt_fd;
+                                                }
+                                        }
+                                }
+                        }
+                        
                         assert(libtrace->fifo);
 
@@ -1217 +1279 @@
                                 continue;
                         }
-                        fifo_out_update(libtrace->fifo, size);  
+                        current = libtrace->rt_fds;
                         // Sort out the protocol header
                         memcpy(buf, &packet->status, intsize);
-                                                
-
-                        // Send the buffer out on the wire
-                        if ((numbytes = trace_write(libtrace, buf, size + sizeof(int))) <=0 ) {
-                                return numbytes;
-                        }
-
+                        if (select(libtrace->max_rtfds + 1, NULL, &current, NULL, &tv) == -1 ) {
+                                perror("select");
+                                write_required = 0;
+                                continue;
+                        }
+                        
+                        // Send the data to each ready client
+                        for (i = 0; i <= libtrace->max_rtfds; i++) {
+                                if (FD_ISSET(i, &current)) {
+                                        libtrace->output.fd = i;
+                                        if ((numbytes = trace_write(libtrace, buf, size + sizeof(int))) <=0 ) {
+                                                // close rt_client
+                                                FD_CLR(i, &libtrace->rt_fds);
+                                                close(i);
+                                                numbytes = 0;
+                                                continue;
+                                        }
+                                }
+                        }
+
+                        fifo_out_update(libtrace->fifo, size);
                         // Need an ack to come back
                         // TODO: Obviously this is a little unfinished