destination.c File Reference

Routines used to setup a Flowgrind destination for a test. More...

#include "config.h"
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <strings.h>
#include <signal.h>
#include <string.h>
#include <fcntl.h>
#include <math.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/param.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/wait.h>
#include <errno.h>
#include <time.h>
#include <syslog.h>
#include <sys/time.h>
#include <netdb.h>
#include <pthread.h>
#include <float.h>
#include "common.h"
#include "debug.h"
#include "fg_socket.h"
#include "fg_time.h"
#include "fg_math.h"
#include "fg_log.h"
#include "daemon.h"
#include "fg_pcap.h"

Go to the source code of this file.

Functions
int	accept_data (struct flow *flow)
void	add_flow_destination (struct request_add_flow_destination *request)
	To set daemon flow as destination endpoint. More...
static int	create_listen_socket (struct flow *flow, char *bind_addr, unsigned short *listen_port)
int	get_tcp_info (struct flow *flow, struct tcp_info *info)
void	init_flow (struct flow *flow, int is_source)
	To initialize all flows to the default value. More...
void	remove_flow (unsigned i)
void	uninit_flow (struct flow *flow)

Detailed Description

Routines used to setup a Flowgrind destination for a test.

Definition in file destination.c.

Function Documentation

accept_data()

int accept_data

(

struct flow *

flow

)

Bug:

: currently we use portable select() API, which is limited by the number of bits in an fd_set

Definition at line 250 of file destination.c.

{
        struct sockaddr_storage caddr;
        socklen_t addrlen = sizeof(caddr);
        unsigned real_send_buffer_size;
        unsigned real_receive_buffer_size;
 
        flow->fd = accept(flow->listenfd_data, (struct sockaddr *)&caddr,
                          &addrlen);
        if (flow->fd == -1) {
                /* try again later .... */
                if (errno == EINTR || errno == EAGAIN)
                        return 0;
                logging(LOG_ALERT, "accept() failed: %s", strerror(errno));
                return -1;
        }
 
        /* FIXME: currently we use portable select() API, which
         * is limited by the number of bits in an fd_set */
        if (flow->fd >= FD_SETSIZE) {
                logging(LOG_ALERT, "too many file descriptors are "
                        "already in use by this daemon (FD number=%u)", flow->fd);
                flow_error(flow, "failed to add test connection: too many"
                        "file descriptors in use by this daemon");
                close(flow->fd);
                return -1;
        }
 
        if (close(flow->listenfd_data) == -1)
                logging(LOG_WARNING, "close() failed");
        flow->listenfd_data = -1;
 
        logging(LOG_NOTICE, "client %s connected for testing (fd=%u)",
                fg_nameinfo((struct sockaddr *)&caddr, addrlen), flow->fd);
 
#ifdef HAVE_LIBPCAP
        fg_pcap_go(flow);
#endif /* HAVE_LIBPCAP */
 
        real_send_buffer_size =
                set_window_size_directed(flow->fd,
                                         flow->settings.requested_send_buffer_size,
                                         SO_SNDBUF);
        if (flow->requested_server_test_port &&
            flow->real_listen_send_buffer_size != real_send_buffer_size) {
                logging(LOG_WARNING, "failed to set send buffer size of test "
                        "socket to send buffer size size of listen socket "
                        "(listen = %u, test = %u)",
                        flow->real_listen_send_buffer_size, real_send_buffer_size);
                return -1;
        }
        real_receive_buffer_size =
                set_window_size_directed(flow->fd,
                                         flow->settings.requested_read_buffer_size,
                                         SO_RCVBUF);
        if (flow->requested_server_test_port &&
            flow->real_listen_receive_buffer_size != real_receive_buffer_size) {
                logging(LOG_WARNING, "failed to set receive buffer size "
                        "(advertised window) of test socket to receive "
                        "buffer size of listen socket (listen = %u, "
                        "test = %u)", flow->real_listen_receive_buffer_size,
                        real_receive_buffer_size);
                return -1;
        }
        if (set_flow_tcp_options(flow) == -1)
                return -1;
        DEBUG_MSG(LOG_NOTICE, "data socket accepted");
        flow->state = GRIND;
        flow->connect_called = 1;
 
        return 0;
}

add_flow_destination()

void add_flow_destination

(

struct request_add_flow_destination *

request

)

To set daemon flow as destination endpoint.

To set the flow options and settings as destination endpoint. Listening port created and send back to the controller in the same request structure

Parameters

[in,out]

request

contain the test option and parameter for destination source endpoint

Bug:

: currently we use portable select() API, which is limited by the number of bits in an fd_set

Definition at line 158 of file destination.c.

{
        struct flow *flow;
        unsigned short server_data_port;
 
        if (fg_list_size(&flows) >= MAX_FLOWS_DAEMON) {
                logging(LOG_WARNING, "can not accept another flow, already "
                        "handling %zu flows", fg_list_size(&flows));
                request_error(&request->r, "Can not accept another flow, "
                             "already handling %zu flows.", fg_list_size(&flows));
                return;
        }
 
        flow = malloc(sizeof(struct flow));
        if (!flow) {
                logging(LOG_ALERT, "could not allocate memory for flow");
                return;
        }
 
        init_flow(flow, 0);
 
        flow->settings = request->settings;
        flow->write_block = calloc(1, flow->settings.maximum_block_size );
        flow->read_block = calloc(1, flow->settings.maximum_block_size );
        /* Controller flow ID is set in the daemon */
        flow->id=flow->settings.flow_id;
        if (flow->write_block == NULL || flow->read_block == NULL) {
                logging(LOG_ALERT, "could not allocate memory for read/write "
                        "blocks");
                request_error(&request->r, "could not allocate memory "
                              "for read/write blocks");
                uninit_flow(flow);
                return;
        }
 
        if (flow->settings.byte_counting) {
                int byte_idx;
                for (byte_idx = 0; byte_idx < flow->settings.maximum_block_size;
                     byte_idx++)
                        *(flow->write_block + byte_idx) =
                                (unsigned char)(byte_idx & 0xff);
        }
 
        /* Create listen socket for data connection */
        if ((flow->listenfd_data =
                        create_listen_socket(flow,
                                             flow->settings.bind_address[0]
                                                ? flow->settings.bind_address : 0,
                                             &server_data_port)) == -1) {
                logging(LOG_ALERT, "could not create listen socket for "
                        "data connection: %s", flow->error);
                request_error(&request->r, "could not create listen socket "
                              "for data connection: %s", flow->error);
                uninit_flow(flow);
                return;
        } else {
                /* FIXME: currently we use portable select() API, which
                 * is limited by the number of bits in an fd_set */
                if (flow->listenfd_data >= FD_SETSIZE) {
                        logging(LOG_ALERT, "failed to add listen socket: "
                                "fd number too high (fd=%u)", flow->listenfd_data);
                        flow_error(flow, "failed to add listen socket: too many"
                                "file descriptors in use by this daemon");
                        uninit_flow(flow);
                        return;
                }
                DEBUG_MSG(LOG_WARNING, "listening on %s port %u for data "
                          "connection (fd=%u)", flow->settings.bind_address,
                          server_data_port, flow->listenfd_data);
        }
 
        flow->real_listen_send_buffer_size =
                set_window_size_directed(flow->listenfd_data,
                                         flow->settings.requested_send_buffer_size,
                                         SO_SNDBUF);
        flow->real_listen_receive_buffer_size =
                set_window_size_directed(flow->listenfd_data,
                                         flow->settings.requested_read_buffer_size,
                                         SO_RCVBUF);
 
        request->listen_data_port = (int)server_data_port;
        request->real_listen_send_buffer_size =
                flow->real_listen_send_buffer_size;
        request->real_listen_read_buffer_size =
                flow->real_listen_receive_buffer_size;
        request->flow_id = flow->id;
 
        fg_list_push_back(&flows, flow);
 
        return;
}

create_listen_socket()

static int create_listen_socket ( struct flow *  flow, char *  bind_addr, unsigned short *  listen_port  )

static

Definition at line 78 of file destination.c.

{
        int port;
        int rc;
        int fd;
        struct addrinfo hints, *res, *ressave;
 
        bzero(&hints, sizeof(struct addrinfo));
        hints.ai_flags = bind_addr ? 0 : AI_PASSIVE;
        hints.ai_family = AF_UNSPEC;
        hints.ai_socktype = SOCK_STREAM;
 
        /* Any port will be fine */
        if ((rc = getaddrinfo(bind_addr, "0", &hints, &res)) != 0) {
                logging(LOG_ALERT, "getaddrinfo() failed: %s",
                        gai_strerror(rc));
                flow_error(flow, "getaddrinfo() failed: %s",
                           gai_strerror(rc));
                return -1;
        }
 
        ressave = res;
 
        do {
                fd = socket(res->ai_family, res->ai_socktype,
                            res->ai_protocol);
                if (fd < 0)
                        continue;
                if (bind(fd, res->ai_addr, res->ai_addrlen) == 0)
                        break;
                close(fd);
        } while ((res = res->ai_next) != NULL);
 
        if (res == NULL) {
                logging(LOG_ALERT, "failed to create listen socket: %s",
                        strerror(errno));
                flow_error(flow, "failed to create listen socket: %s",
                           strerror(errno));
                freeaddrinfo(ressave);
                return -1;
        }
 
        freeaddrinfo(ressave);
 
        /* we need to set sockopt mtcp before we start listen() */
        if (flow->settings.mtcp)
                set_tcp_mtcp(fd);
 
        if (flow->settings.cc_alg)
                set_congestion_control(fd, flow->settings.cc_alg);
 
        if (listen(fd, 0) < 0) {
                logging(LOG_ALERT, "listen failed: %s", strerror(errno));
                flow_error(flow, "listen failed: %s", strerror(errno));
                return -1;
        }
 
        set_non_blocking(fd);
 
        port = get_port(fd);
        if (port < 0) {
                flow_error(flow, "Could not get port: %s", strerror(errno));
                close(fd);
                return -1;
        }
        DEBUG_MSG(LOG_DEBUG, "listening on port %d", port);
        *listen_port = (unsigned short)port;
 
        return fd;
}

get_tcp_info()

int get_tcp_info	(	struct flow *	flow,
		struct tcp_info *	info
	)

init_flow()

void init_flow	(	struct flow *	flow,
		int	is_source
	)

To initialize all flows to the default value.

The daemon maintain all its data in its flow statistics data structure. These data are initialize to the default value or zero value according to their metrics details.

Parameters

[in,out]	flow	flow structure maintained by a daemon
[in]	is_source	to determine flow endpoint i.e. source or destination

Definition at line 896 of file daemon.c.

{
        memset(flow, 0, sizeof(struct flow));
 
        /* flow id is given by controller */
        flow->id = -1;
        flow->endpoint = is_source ? SOURCE : DESTINATION;
        flow->state = is_source ? GRIND_WAIT_CONNECT : GRIND_WAIT_ACCEPT;
        flow->fd = -1;
        flow->listenfd_data = -1;
 
        flow->current_read_block_size = MIN_BLOCK_SIZE;
        flow->current_write_block_size = MIN_BLOCK_SIZE;
 
        flow->finished[READ] = flow->finished[WRITE] = 0;
 
        flow->addr = 0;
 
        foreach(int *i, INTERVAL, FINAL) {
                flow->statistics[*i].bytes_read = 0;
                flow->statistics[*i].bytes_written = 0;
 
                flow->statistics[*i].request_blocks_read = 0;
                flow->statistics[*i].request_blocks_written = 0;
                flow->statistics[*i].response_blocks_read = 0;
                flow->statistics[*i].response_blocks_written = 0;
 
                flow->statistics[*i].rtt_min = FLT_MAX;
                flow->statistics[*i].rtt_max = FLT_MIN;
                flow->statistics[*i].rtt_sum = 0.0F;
                flow->statistics[*i].iat_min = FLT_MAX;
                flow->statistics[*i].iat_max = FLT_MIN;
                flow->statistics[*i].iat_sum = 0.0F;
                flow->statistics[*i].delay_min = FLT_MAX;
                flow->statistics[*i].delay_max = FLT_MIN;
                flow->statistics[*i].delay_sum = 0.0F;
        }
 
        DEBUG_MSG(LOG_NOTICE, "called init flow %d", flow->id);
}

remove_flow()

void remove_flow

(

unsigned

i

)

uninit_flow()

void uninit_flow

(

struct flow *

flow

)

Definition at line 165 of file daemon.c.

{
        DEBUG_MSG(LOG_DEBUG,"uninit_flow() called for flow %d",flow->id);
        if (flow->fd != -1)
                close(flow->fd);
        if (flow->listenfd_data != -1)
                close(flow->listenfd_data);
#ifdef HAVE_LIBPCAP
        int rc;
        if (flow->settings.traffic_dump && flow->pcap_thread) {
                rc = pthread_cancel(flow->pcap_thread);
                if (rc)
                        logging(LOG_WARNING, "failed to cancel dump thread: %s",
                                strerror(rc));
 
                /* wait for the dump thread to react to the cancellation request */
                rc = pthread_join(flow->pcap_thread, NULL);
                if (rc)
                        logging(LOG_WARNING, "failed to join dump thread: %s",
                                strerror(rc));
        }
#endif /* HAVE_LIBPCAP */
        free_all(flow->read_block, flow->write_block, flow->addr, flow->error);
        free_math_functions(flow);
}