source: soft/giet_vm/applications/classif/classif.c @ 825

///////////////////////////////////////////////////////////////////////////////////////
// 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()