/////////////////////////////////////////////////////////////////////////////////////// // File : classif.c // Date : november 2014 // author : Alain Greiner /////////////////////////////////////////////////////////////////////////////////////// // This multi-threaded application takes a stream of ETH/IP/UDP packets, and makes // packets classification, based on the SRC_IP (IP header) and SRC_PORT (UDP header). // // It uses the VciMasterNic peripheral, that can have up to 4 channels. // Each channel implement a private TX-QUEUE, and a private RX queue. // // There is one analyse thread per core. All threads behave as one single server // (i.e. all threads use the same local port number). After each packet analysis, // the SRC and DST IP addresses and port numbers are exchanged and a response // packet is sent to the remote client. // // This application can run on architectures containing up to 16 * 16 clusters, // It requires one shared TTY terminal. // // The main thread exit after launching analyse threads. // It does not use the pthread_join() construct. /////////////////////////////////////////////////////////////////////////////////////// #include "stdio.h" #include "user_lock.h" #include "user_barrier.h" #define X_SIZE_MAX 16 #define Y_SIZE_MAX 16 #define NPROCS_MAX 4 #define NBYTES_MAX 2048 #define SERVER_IP 0x77777777 #define SERVER_PORT 0x8888 #define MAX_PACKETS 25 #define VERBOSE 1 // macro to use a shared TTY #define printf(...); { lock_acquire( &tty_lock ); \ giet_tty_printf(__VA_ARGS__); \ lock_release( &tty_lock ); } /////////////////////////////////////////////////////////////////////////////////////// // Global variables /////////////////////////////////////////////////////////////////////////////////////// // lock protecting shared TTY user_lock_t tty_lock; // barrier for instrumentation giet_sqt_barrier_t barrier; // instrumentation counters unsigned int counter[16]; // threads arguments array unsigned int thread_arg[16][16][4]; ///////////////////////////////////////////////////////////////// __attribute__ ((constructor)) void analyse( unsigned int * arg ) ///////////////////////////////////////////////////////////////// { unsigned char buffer[NBYTES_MAX]; // buffer for one raw packet sockaddr_t server_addr; // local socket address sockaddr_t client_addr; // remote socket address int length; // received packet length int count; // packets counter int error; unsigned int tid = *arg; printf("\n[CLASSIF] analyse thread %x starts at cycle %d\n", tid , giet_proctime() ); // create socket int socket = giet_nic_socket( AF_INET , SOCK_DGRAM , 0 ); if( socket == -1 ) { printf("\n[CLASSIF ERROR] thread %x cannot create socket\n", tid ); giet_pthread_exit( NULL ); } // bind socket server_addr.sin_family = AF_INET; server_addr.sin_addr = HTONL( SERVER_IP ); server_addr.sin_port = HTONS( SERVER_PORT ); error = giet_nic_bind( socket , &server_addr , sizeof(server_addr) ); if( error ) { printf("\n[CLASSIF ERROR] thread %x cannot bind socket\n", tid ); giet_pthread_exit( NULL ); } printf("\n[CLASSIF] socket %x created by thread %x\n", socket , tid ); // reset NIC counters giet_nic_clear_stats(); ///////// loop to receive, analyse, and send packets /////////// for( count = 0 ; count < MAX_PACKETS ; count++ ) { length = sizeof(sockaddr_t); // get one packet from client error = giet_nic_recvfrom( socket, buffer, NBYTES_MAX, 0, &client_addr, &length ); if( error ) { printf("\n[CLASSIF ERROR] thread %x cannot receive packet\n", tid ); giet_pthread_exit( NULL ); } // get type & pktid unsigned int client_ip = client_addr.sin_addr; unsigned short client_port = client_addr.sin_port; unsigned int type = ((client_ip & 0x3) << 2) + (client_port & 0x3); unsigned int pktid = (((unsigned int )buffer[0]) << 24) | (((unsigned int )buffer[1]) << 16) | (((unsigned int )buffer[2]) << 8) | (((unsigned int )buffer[3]) ) ; if( VERBOSE ) { printf("\n[CLASSIF] thread %x receive packet at cycle %d\n" " type = %x / length = %d / pktid = %d\n", tid , giet_proctime() , type , length , pktid ); } atomic_increment( &counter[type], 1 ); // send response packet error = giet_nic_sendto( socket, buffer, length, 0, &client_addr, sizeof(sockaddr_t) ); if( error ) { printf("\n[CLASSIF ERROR] thread %x cannot send packet\n", tid ); giet_pthread_exit( NULL ); } if( VERBOSE ) { printf("\n[CLASSIF] thread %x sent packet at cycle %d\n" " type = %x / length = %d / pktid = %d\n", tid , giet_proctime() , type , length , pktid ); } } // end for // synchro before stats sqt_barrier_wait( &barrier ); if ( tid == 0 ) { // give time to flush the TX pipe-line char byte; printf("\n ###### enter any key to get stats ######\n"); giet_tty_getc( &byte ); // display classification results printf("\nClassification Results\n" " - TYPE 0 : %d packets\n" " - TYPE 1 : %d packets\n" " - TYPE 2 : %d packets\n" " - TYPE 3 : %d packets\n" " - TYPE 4 : %d packets\n" " - TYPE 5 : %d packets\n" " - TYPE 6 : %d packets\n" " - TYPE 7 : %d packets\n" " - TYPE 8 : %d packets\n" " - TYPE 9 : %d packets\n" " - TYPE A : %d packets\n" " - TYPE B : %d packets\n" " - TYPE C : %d packets\n" " - TYPE D : %d packets\n" " - TYPE E : %d packets\n" " - TYPE F : %d packets\n" " TOTAL = %d packets\n", counter[0x0], counter[0x1], counter[0x2], counter[0x3], counter[0x4], counter[0x5], counter[0x6], counter[0x7], counter[0x8], counter[0x9], counter[0xA], counter[0xB], counter[0xC], counter[0xD], counter[0xE], counter[0xF], counter[0x0]+ counter[0x1]+ counter[0x2]+ counter[0x3]+ counter[0x4]+ counter[0x5]+ counter[0x6]+ counter[0x7]+ counter[0x8]+ counter[0x9]+ counter[0xA]+ counter[0xB]+ counter[0xC]+ counter[0xD]+ counter[0xE]+ counter[0xF] ); // display NIC instrumentation counters giet_nic_print_stats(); } giet_pthread_exit( "completed" ); } // end analyse() ////////////////////////////////////////// __attribute__ ((constructor)) void main() ////////////////////////////////////////// { unsigned int x , y , n; unsigned int error; pthread_t trdid; // thread index required by pthread_create() // get plat-form parameters unsigned int x_size; // number of clusters in a row unsigned int y_size; // number of clusters in a column unsigned int nprocs; // number of processors per cluster giet_procs_number( &x_size , &y_size , &nprocs ); // shared TTY allocation giet_tty_alloc( 1 ); lock_init( &tty_lock); giet_pthread_assert( ((x_size >= 1) && (x_size <= 16)), "[CLASSIF ERROR] x_size must be in [1...16]"); giet_pthread_assert( ((y_size >= 1) && (y_size <= 16)), "[CLASSIF ERROR] y_size must be in [1...16]"); printf("\n[CLASSIF] main thread starts at cycle %d\n", giet_proctime() ); // distributed heap[x,y] initialisation for ( x = 0 ; x < x_size ; x++ ) { for ( y = 0 ; y < y_size ; y++ ) { heap_init( x , y ); } } printf("\n[CLASSIF] heap initialized at cycle %d\n", giet_proctime() ); // barrier initialisation sqt_barrier_init( &barrier, x_size , y_size , nprocs ); printf("\n[CLASSIF] barrier initialized at cycle %d\n", giet_proctime() ); // lauch analyse threads for ( x = 0 ; x < x_size ; x++ ) { for ( y = 0 ; y < y_size ; y++ ) { for ( n = 0 ; n < nprocs ; n++ ) { thread_arg[x][y][n] = (x << 16) | (y << 8) | n; error = giet_pthread_create( &trdid, NULL, // no attribute &analyse, &thread_arg[x][y][n] ); if( error ) { printf("\n[CLASSIF ERROR] cannot create thread on core[%d,%d,%d]\n", x, y, n ); giet_pthread_exit( NULL ); } } } } giet_pthread_exit( "completed" ); } // end main()