Index: /soft/giet_vm/applications/classif/classif.c =================================================================== --- /soft/giet_vm/applications/classif/classif.c (revision 824) +++ /soft/giet_vm/applications/classif/classif.c (revision 825) @@ -4,88 +4,34 @@ // author : Alain Greiner /////////////////////////////////////////////////////////////////////////////////////// -// This multi-threaded application takes a stream of Gigabit Ethernet packets, -// and makes packet analysis and classification, based on the source MAC address. -// It uses the NIC peripheral, and the distributed kernel chbufs accessed by the CMA -// component to receive and send packets on the Gigabit Ethernet port. -// -// It can run on architectures containing up to 16 * 16 clusters, -// and from 3 to 8 processors per cluster. -// +// 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. // -// This application is described as a TCG (Thread and Communication Graph) -// containing (N+2) threads per cluster. -// - one "load" thread -// - one "store" thread -// - N "analyse" threads -// The containers are distributed (N+2 containers per cluster): -// - one RX container (part of the kernel rx_chbuf), in the kernel heap. -// - one TX container (part of the kernel tx-chbuf), in the kernel heap. -// - N working containers (one per analysis thread), in the user heap. -// In each cluster, the "load", analysis" and "store" threads communicates through -// three local MWMR fifos: -// - fifo_l2a : tranfer a full container from "load" to "analyse" thread. -// - fifo_a2s : transfer a full container from "analyse" to "store" thread. -// - fifo_s2l : transfer an empty container from "store" to "load" thread. -// For each fifo, one item is a 32 bits word defining the index of an -// available working container. -// The pointers on the working containers, and the pointers on the MWMR fifos -// are global arrays stored in cluster[0][0]. -// -// The main thread exit after global initialisation, and launching the other threads: -// It does not use the pthread_join() construct. It is executed on P[0,0,1], -// toavoid overload on P[0,0,0]. -// -// Initialisation is made in two steps: -// -// 1) The global, shared, variables are initialised by the main thread: -// - shared TTY -// - distributed heap (one heap per cluster) -// - distributed rx_barrier (between all "load" threads) -// - distributed tx_barrier (between all "store" threads) -// - RX kernel chbufs (used by "load" threads) -// - TX kernel chbufs (used by "store" threads) -// Then the main thread exit, after launching the "load, "store", and "analyse" -// threads in all clusters. -// -// 2) Each "load" thread allocates containers[x][y][n] from local heap, -// and register containers pointers in the local stack. -// Each "load" thread allocates data buffers & mwmr fifo descriptors -// from local heap, and register pointers in global arrays. -// Each "load" thread initialises the containers as empty in fifo_s2l. -// Then each "load" thread signals mwmr fifos initialisation completion -// to other threads in same cluster, using the local_sync[x][y] variables. -// -// When initialisation is completed, all threads are running in parallel: -// -// 1) The "load" thread get an empty working container from the fifo_s2l, -// transfer one container from the kernel rx_chbuf to this user container, -// and transfer ownership of this container to one "analysis" thread by writing -// into the fifo_l2a. -// 2) The "analyse" thread get one working container from the fifo_l2a, analyse -// each packet header, compute the packet type (depending on the SRC MAC address), -// increment the correspondint classification counter, and transpose the SRC -// and the DST MAC addresses fot TX tranmission. -// 3) The "store" thread transfer get a full working container from the fifo_a2s, -// transfer this user container content to the the kernel tx_chbuf, -// and transfer ownership of this empty container to the "load" thread by writing -// into the fifo_s2l. -// -// Instrumentation results display is done by the "store" thread in cluster[0][0] -// when all "store" threads completed the number of clusters specified by the -// CONTAINERS_MAX parameter. +// 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" -#include "malloc.h" -#include "user_lock.h" -#include "mwmr_channel.h" #define X_SIZE_MAX 16 #define Y_SIZE_MAX 16 -#define NPROCS_MAX 8 -#define CONTAINERS_MAX 5000 -#define VERBOSE_ANALYSE 0 +#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 @@ -97,238 +43,129 @@ /////////////////////////////////////////////////////////////////////////////////////// // Global variables -// The MWMR channels (descriptors and buffers), as well as the working containers -// used by the "analysis" threads are distributed in clusters. -// But the pointers on these distributed structures are stored in cluster[0][0]. -/////////////////////////////////////////////////////////////////////////////////////// - -// pointers on distributed containers -unsigned int* container[X_SIZE_MAX][Y_SIZE_MAX][NPROCS_MAX-2]; - -// pointers on distributed mwmr fifos containing container descriptors -mwmr_channel_t* mwmr_l2a[X_SIZE_MAX][Y_SIZE_MAX]; -mwmr_channel_t* mwmr_a2s[X_SIZE_MAX][Y_SIZE_MAX]; -mwmr_channel_t* mwmr_s2l[X_SIZE_MAX][Y_SIZE_MAX]; - -// local synchros signaling local MWMR fifos initialisation completion -volatile unsigned int local_sync[X_SIZE_MAX][Y_SIZE_MAX]; +/////////////////////////////////////////////////////////////////////////////////////// // lock protecting shared TTY -user_lock_t tty_lock; - -// distributed barrier between "load" threads -giet_sqt_barrier_t rx_barrier; - -// distributed barrier between "store" threads -giet_sqt_barrier_t tx_barrier; +user_lock_t tty_lock; + +// barrier for instrumentation +giet_sqt_barrier_t barrier; // instrumentation counters -unsigned int counter[16]; +unsigned int counter[16]; // threads arguments array unsigned int thread_arg[16][16][4]; -//////////////////////////////////////////////////////////// -__attribute__ ((constructor)) void load( unsigned int* arg ) -//////////////////////////////////////////////////////////// +///////////////////////////////////////////////////////////////// +__attribute__ ((constructor)) void analyse( unsigned int * arg ) +///////////////////////////////////////////////////////////////// { - // 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 ); - - // each "load" thread get processor identifiers - unsigned int x; - unsigned int y; - unsigned int p; - giet_proc_xyp( &x, &y, &p ); - - // each "load" thread allocates containers[x][y][n] (from local heap) - // and register pointers in the local stack - unsigned int n; - unsigned int* cont[NPROCS_MAX-2]; - - for ( n = 0 ; n < (nprocs - 2) ; n++ ) - { - container[x][y][n] = malloc( 4096 ); - cont[n] = container[x][y][n]; - } - - // each "load" thread allocates data buffers for mwmr fifos (from local heap) - unsigned int* data_l2a = malloc( (nprocs - 2)<<2 ); - unsigned int* data_a2s = malloc( (nprocs - 2)<<2 ); - unsigned int* data_s2l = malloc( (nprocs - 2)<<2 ); - - // each "load" thread allocates mwmr fifos descriptors (from local heap) - mwmr_l2a[x][y] = malloc( sizeof(mwmr_channel_t) ); - mwmr_a2s[x][y] = malloc( sizeof(mwmr_channel_t) ); - mwmr_s2l[x][y] = malloc( sizeof(mwmr_channel_t) ); - - // each "load" thread registers local pointers on mwmr fifos in local stack - mwmr_channel_t* fifo_l2a = mwmr_l2a[x][y]; - mwmr_channel_t* fifo_a2s = mwmr_a2s[x][y]; - mwmr_channel_t* fifo_s2l = mwmr_s2l[x][y]; - - // each "load" thread initialises local mwmr fifos descriptors - // ( width = 4 bytes / depth = number of analysis threads ) - mwmr_init( fifo_l2a , data_l2a , 1 , (nprocs - 2) ); - mwmr_init( fifo_a2s , data_a2s , 1 , (nprocs - 2) ); - mwmr_init( fifo_s2l , data_s2l , 1 , (nprocs - 2) ); - - - // each "load" thread initialises local containers as empty in fifo_s2l - for ( n = 0 ; n < (nprocs - 2) ; n++ ) mwmr_write( fifo_s2l , &n , 1 ); - - // each "load" thread signals mwmr fifos initialisation completion - // to other threads in same cluster. - local_sync[x][y] = 1; - - // only "load" thread[0][0] displays status - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] load on P[%d,%d,%d] enters main loop at cycle %d\n" - " &mwmr_l2a = %x / &data_l2a = %x\n" - " &mwmr_a2s = %x / &data_a2s = %x\n" - " &mwmr_s2l = %x / &data_s2l = %x\n" - " &cont[0] = %x\n" - " x_size = %d / y_size = %d / nprocs = %d\n", - x , y , p , giet_proctime(), - (unsigned int)fifo_l2a, (unsigned int)data_l2a, - (unsigned int)fifo_a2s, (unsigned int)data_a2s, - (unsigned int)fifo_s2l, (unsigned int)data_s2l, - (unsigned int)cont[0], - x_size, y_size, nprocs ); - } - - ///////////////////////////////////////////////////////////// - // "load" thread enters the main loop (on containers) - unsigned int count = 0; // loaded containers count - unsigned int index; // available container index - unsigned int* temp; // pointer on available container - - while ( count < CONTAINERS_MAX ) - { - // get one empty container index from fifo_s2l - mwmr_read( fifo_s2l , &index , 1 ); - temp = cont[index]; - - // get one container from kernel rx_chbuf - giet_nic_rx_move( temp ); - - // get packets number - unsigned int npackets = temp[0] & 0x0000FFFF; - unsigned int nwords = temp[0] >> 16; - - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] load on P[%d,%d,%d] get container %d at cycle %d" - " : %d packets / %d words\n", - x, y, p, index, giet_proctime(), npackets, nwords ); - } - - // put the full container index to fifo_l2a - mwmr_write( fifo_l2a, &index , 1 ); - - count++; - } - - // all "load" threads synchronise before stats - sqt_barrier_wait( &rx_barrier ); - - // "load" thread[0][0] displays stats - if ( (x==0) && (y==0) ) giet_nic_rx_stats(); - - // all "load" thread exit - giet_pthread_exit("completed"); - -} // end load() - - -////////////////////////////////////////////////////////////// -__attribute__ ((constructor)) void store( unsigned int * arg ) -////////////////////////////////////////////////////////////// -{ - // 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 ); - - // get processor identifiers - unsigned int x; - unsigned int y; - unsigned int p; - giet_proc_xyp( &x, &y, &p ); - - // each "store" thread wait mwmr channels initialisation - while ( local_sync[x][y] == 0 ) asm volatile ("nop"); - - // each "store" thread registers pointers on working containers in local stack - unsigned int n; - unsigned int* cont[NPROCS_MAX-2]; - - for ( n = 0 ; n < (nprocs - 2) ; n++ ) - { - cont[n] = container[x][y][n]; - } - - // each "store" thread registers pointers on mwmr fifos in local stack - mwmr_channel_t* fifo_l2a = mwmr_l2a[x][y]; - mwmr_channel_t* fifo_a2s = mwmr_a2s[x][y]; - mwmr_channel_t* fifo_s2l = mwmr_s2l[x][y]; - - // only "store" thread[0][0] displays status - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] store on P[%d,%d,%d] enters main loop at cycle %d\n" - " &mwmr_l2a = %x\n" - " &mwmr_a2s = %x\n" - " &mwmr_s2l = %x\n" - " &cont[0] = %x\n", - x , y , p , giet_proctime(), - (unsigned int)fifo_l2a, - (unsigned int)fifo_a2s, - (unsigned int)fifo_s2l, - (unsigned int)cont[0] ); - } - - ///////////////////////////////////////////////////////////// - // "store" thread enter the main loop (on containers) - unsigned int count = 0; // stored containers count - unsigned int index; // empty container index - unsigned int* temp; // pointer on empty container - - while ( count < CONTAINERS_MAX ) - { - // get one working container index from fifo_a2s - mwmr_read( fifo_a2s , &index , 1 ); - temp = cont[index]; - - // put one container to kernel tx_chbuf - giet_nic_tx_move( temp ); - - // get packets number - unsigned int npackets = temp[0] & 0x0000FFFF; - unsigned int nwords = temp[0] >> 16; - - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] store on P[%d,%d,%d] get container %d at cycle %d" - " : %d packets / %d words\n", - x, y, p, index, giet_proctime(), npackets, nwords ); - } - - // put the working container index to fifo_s2l - mwmr_write( fifo_s2l, &index , 1 ); - - count++; - } - - // all "store" threads synchronise before result display - sqt_barrier_wait( &tx_barrier ); - - // "store" thread[0,0] and displays results - if ( (x==0) && (y==0) ) - { + 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" @@ -348,5 +185,5 @@ " - TYPE E : %d packets\n" " - TYPE F : %d packets\n" - " TOTAL = %d packets\n", + " TOTAL = %d packets\n", counter[0x0], counter[0x1], counter[0x2], counter[0x3], counter[0x4], counter[0x5], counter[0x6], counter[0x7], @@ -358,178 +195,13 @@ counter[0xC]+ counter[0xD]+ counter[0xE]+ counter[0xF] ); - giet_nic_tx_stats(); + // display NIC instrumentation counters + giet_nic_print_stats(); } - // all "store" thread exit - giet_pthread_exit("Thread completed"); - -} // end store() - - -/////////////////////////////////////////////////////////////// -__attribute__ ((constructor)) void analyse( unsigned int* arg ) -/////////////////////////////////////////////////////////////// -{ - // get platform parameters - unsigned int x_size; // number of clusters in 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 ); - - // get processor identifiers - unsigned int x; - unsigned int y; - unsigned int p; - giet_proc_xyp( &x, &y, &p ); - - // each "analyse" thread wait mwmr channels initialisation - while ( local_sync[x][y] == 0 ) asm volatile ("nop"); - - // each "analyse" threads register pointers on working containers in local stack - unsigned int n; - unsigned int* cont[NPROCS_MAX-2]; - for ( n = 0 ; n < (nprocs - 2) ; n++ ) - { - cont[n] = container[x][y][n]; - } - - // each "analyse" threads register pointers on mwmr fifos in local stack - mwmr_channel_t* fifo_l2a = mwmr_l2a[x][y]; - mwmr_channel_t* fifo_a2s = mwmr_a2s[x][y]; - - // only "analyse" thread[0][0] display status - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] analyse on P[%d,%d,%d] enters main loop at cycle %d\n" - " &mwmr_l2a = %x\n" - " &mwmr_a2s = %x\n" - " &cont[0] = %x\n", - x, y, p, giet_proctime(), - (unsigned int)fifo_l2a, - (unsigned int)fifo_a2s, - (unsigned int)cont[0] ); - } - - ////////////////////////////////////////////////////////////////////// - // all "analyse" threads enter the main infinite loop (on containers) - unsigned int index; // available container index - unsigned int* temp; // pointer on available container - unsigned int nwords; // number of words in container - unsigned int npackets; // number of packets in container - unsigned int length; // number of bytes in current packet - unsigned int first; // current packet first word in container - unsigned int type; // current packet type - unsigned int pid; // current packet index - -#if VERBOSE_ANALYSE - unsigned int verbose_len[10]; // save length for 10 packets in one container - unsigned long long verbose_dst[10]; // save dest for 10 packets in one container - unsigned long long verbose_src[10]; // save source for 10 packets in one container -#endif - - while ( 1 ) - { - -#if VERBOSE_ANALYSE - for( pid = 0 ; pid < 10 ; pid++ ) - { - verbose_len[pid] = 0; - verbose_dst[pid] = 0; - verbose_src[pid] = 0; - } -#endif - // get one working container index from fifo_l2a - mwmr_read( fifo_l2a , &index , 1 ); - temp = cont[index]; - - // get packets number and words number - npackets = temp[0] & 0x0000FFFF; - nwords = temp[0] >> 16; - - if ( (x==0) && (y==0) ) - { - printf("\n[CLASSIF] analyse on P[%d,%d,%d] get container at cycle %d" - " : %d packets / %d words\n", - x, y, p, giet_proctime(), npackets, nwords ); - } - - // initialize word index in container - first = 34; - - // loop on packets - for( pid = 0 ; pid < npackets ; pid++ ) - { - // get packet length from container header - if ( (pid & 0x1) == 0 ) length = temp[1+(pid>>1)] >> 16; - else length = temp[1+(pid>>1)] & 0x0000FFFF; - - // compute packet DST and SRC MAC addresses - unsigned int word0 = temp[first]; - unsigned int word1 = temp[first + 1]; - unsigned int word2 = temp[first + 2]; - -#if VERBOSE_ANALYSE - unsigned long long dst = ((unsigned long long)(word1 & 0xFFFF0000)>>16) | - (((unsigned long long)word0)<<16); - unsigned long long src = ((unsigned long long)(word1 & 0x0000FFFF)<<32) | - ((unsigned long long)word2); - if ( pid < 10 ) - { - verbose_len[pid] = length; - verbose_dst[pid] = dst; - verbose_src[pid] = src; - } -#endif - // compute type from SRC MAC address and increment counter - type = word1 & 0x0000000F; - atomic_increment( &counter[type], 1 ); - - // exchange SRC & DST MAC addresses for TX - temp[first] = ((word1 & 0x0000FFFF)<<16) | ((word2 & 0xFFFF0000)>>16); - temp[first + 1] = ((word2 & 0x0000FFFF)<<16) | ((word0 & 0xFFFF0000)>>16); - temp[first + 2] = ((word0 & 0x0000FFFF)<<16) | ((word1 & 0xFFFF0000)>>16); - - // update first word index - if ( length & 0x3 ) first += (length>>2)+1; - else first += (length>>2); - } - -#if VERBOSE_ANALYSE - if ( (x==0) && (y==0) ) - { - printf("\n*** Thread analyse on P[%d,%d,%d] / container %d at cycle %d\n" - " - Packet 0 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 1 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 2 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 3 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 4 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 5 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 6 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 7 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 8 : plen = %d / dst_mac = %l / src_mac = %l\n" - " - Packet 9 : plen = %d / dst_mac = %l / src_mac = %l\n", - x , y , p , index , giet_proctime() , - verbose_len[0] , verbose_dst[0] , verbose_src[0] , - verbose_len[1] , verbose_dst[1] , verbose_src[1] , - verbose_len[2] , verbose_dst[2] , verbose_src[2] , - verbose_len[3] , verbose_dst[3] , verbose_src[3] , - verbose_len[4] , verbose_dst[4] , verbose_src[4] , - verbose_len[5] , verbose_dst[5] , verbose_src[5] , - verbose_len[6] , verbose_dst[6] , verbose_src[6] , - verbose_len[7] , verbose_dst[7] , verbose_src[7] , - verbose_len[8] , verbose_dst[8] , verbose_src[8] , - verbose_len[9] , verbose_dst[9] , verbose_src[9] ); - } -#endif - - // pseudo-random delay - unsigned int delay = giet_rand()>>3; - unsigned int time; - for( time = 0 ; time < delay ; time++ ) asm volatile ("nop"); - - // put the working container index to fifo_a2s - mwmr_write( fifo_a2s , &index , 1 ); - } + giet_pthread_exit( "completed" ); + } // end analyse() + + ////////////////////////////////////////// @@ -537,12 +209,7 @@ ////////////////////////////////////////// { - // indexes for loops unsigned int x , y , n; - - // get identifiers for proc executing main - unsigned int x_id; // x cluster coordinate - unsigned int y_id; // y cluster coordinate - unsigned int p_id; // local processor index - giet_proc_xyp( &x_id , &y_id , &p_id ); + unsigned int error; + pthread_t trdid; // thread index required by pthread_create() // get plat-form parameters @@ -556,8 +223,4 @@ lock_init( &tty_lock); - // check plat-form parameters - giet_pthread_assert( ((nprocs >= 3) && (nprocs <= 8)), - "[CLASSIF ERROR] number of procs per cluster must in [3...8]"); - giet_pthread_assert( ((x_size >= 1) && (x_size <= 16)), "[CLASSIF ERROR] x_size must be in [1...16]"); @@ -566,8 +229,10 @@ "[CLASSIF ERROR] y_size must be in [1...16]"); - // distributed heap initialisation - for ( x = 0 ; x < x_size ; x++ ) - { - for ( y = 0 ; y < y_size ; y++ ) + 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 ); @@ -575,30 +240,12 @@ } - printf("\n[CLASSIF] start at cycle %d on %d cores\n", - giet_proctime(), (x_size * y_size * nprocs) ); - - // thread index - // required by pthread_create() - // unused in this appli because no pthread_join() - pthread_t trdid; - - // rx_barrier initialisation - sqt_barrier_init( &rx_barrier, x_size , y_size , 1 ); - - // tx_barrier initialisation - sqt_barrier_init( &tx_barrier, x_size , y_size , 1 ); - - // allocate and start RX NIC and CMA channels - giet_nic_rx_alloc( x_size , y_size ); - giet_nic_rx_start(); - - // allocate and start TX NIC and CMA channels - giet_nic_tx_alloc( x_size , y_size ); - giet_nic_tx_start(); - - // Initialisation completed - printf("\n[CLASSIF] initialisation completed at cycle %d\n", giet_proctime() ); - - // launch load, store and analyse threads + 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++ ) { @@ -607,51 +254,15 @@ for ( n = 0 ; n < nprocs ; n++ ) { - // compute argument value - thread_arg[x][y][n] = (x<<8) | (y<<4) | n; - - if ( n == 0 ) // "load" thread + 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 ) { - if ( giet_pthread_create( &trdid, - NULL, // no attribute - &load, - &thread_arg[x][y][n] ) ) - { - printf("\n[CLASSIF ERROR] launching thread load\n" ); - giet_pthread_exit( NULL ); - } - else - { - printf("\n[CLASSIF] thread load activated : trdid = %x\n", trdid ); - } - } - else if ( n == 1 ) // "store" thread - { - if ( giet_pthread_create( &trdid, - NULL, // no attribute - &store, - &thread_arg[x][y][n] ) ) - { - printf("\n[CLASSIF ERROR] launching thread store\n" ); - giet_pthread_exit( NULL ); - } - else - { - printf("\n[CLASSIF] thread store activated : trdid = %x\n", trdid ); - } - } - else // "analyse" threads - { - if ( giet_pthread_create( &trdid, - NULL, // no attribute - &analyse, - &thread_arg[x][y][n] ) ) - { - printf("\n[CLASSIF ERROR] launching thread analyse\n" ); - giet_pthread_exit( NULL ); - } - else - { - printf("\n[CLASSIF] thread analyse activated : trdid = %x\n", trdid ); - } + printf("\n[CLASSIF ERROR] cannot create thread on core[%d,%d,%d]\n", + x, y, n ); + giet_pthread_exit( NULL ); } } Index: /soft/giet_vm/applications/classif/classif.py =================================================================== --- /soft/giet_vm/applications/classif/classif.py (revision 824) +++ /soft/giet_vm/applications/classif/classif.py (revision 825) @@ -12,7 +12,5 @@ # The mapping of threads on processors is the following: # - the "main" on cluster[0][0] -# - one "load" thread per cluster containing processors, -# - one "store" thread per cluster containing processors, -# - (nprocs-2) "analyse" thread per cluster containing processors. +# - one "analyse" thread per processor, # The mapping of virtual segments is the following: # - There is one shared data vseg in cluster[0][0] @@ -26,7 +24,4 @@ # - y_width : number of bits for y field # - nprocs : number of processors per cluster -# -# WARNING: The target architecture cannot contain less -# than 3 processors per cluster. ################################################################################## @@ -40,6 +35,4 @@ y_width = mapping.y_width - assert (nprocs >= 3) and (nprocs <= 8) - # define vsegs base & size code_base = 0x10000000 @@ -49,14 +42,14 @@ data_size = 0x00010000 # 64 Kbytes (non replicated) - heap_base = 0x30000000 - heap_size = 0x00200000 # 2M bytes (per cluster) + stack_base = 0x30000000 + stack_size = 0x00010000 # 64 Kbytes (per thread) - stack_base = 0x40000000 - stack_size = 0x00010000 # 64 Kbytes (per thread) + heap_base = 0x40000000 + heap_size = 0x00200000 # 2 Mbytes (per cluster) # create vspace vspace = mapping.addVspace( name = 'classif', startname = 'classif_data', - active = False ) + active = True ) # data vseg : shared / cluster[0][0] @@ -65,16 +58,4 @@ binpath = 'bin/classif/appli.elf', local = False ) - - # heap vsegs : shared (one per cluster) - for x in xrange (x_size): - for y in xrange (y_size): - cluster_id = (x * y_size) + y - if ( mapping.clusters[cluster_id].procs ): - size = heap_size - base = heap_base + (cluster_id * size) - - mapping.addVseg( vspace, 'classif_heap_%d_%d' %(x,y), base , size, - 'C_WU', vtype = 'HEAP', x = x, y = y, pseg = 'RAM', - local = False, big = True ) # code vsegs : local (one copy per cluster) @@ -110,9 +91,19 @@ local = True ) + # heap vsegs : distributed but non local (any heap can be accessed by any thread) + for x in xrange (x_size): + for y in xrange (y_size): + cluster_id = (x * y_size) + y + if ( mapping.clusters[cluster_id].procs ): + base = heap_base + (cluster_id * heap_size) + + mapping.addVseg( vspace, 'classif_heap_%d_%d' % (x,y), base, heap_size, + 'C_WU' , vtype = 'HEAP' , x = x , y = y , pseg = 'RAM', + local = False, big = True ) + # distributed threads / one thread per processor # ... plus main on P[0][0][0] - mapping.addThread( vspace, 'main', True, 0, 0, 1, - 'main_stack', - 'classif_heap_0_0', + mapping.addThread( vspace, 'main', True, 0, 0, 0, + 'main_stack', '' , 0 ) # index in start_vector @@ -122,18 +113,11 @@ if ( mapping.clusters[cluster_id].procs ): for p in xrange( nprocs ): - if ( p== 0 ): # thread load - start_index = 3 - thread_name = 'load_%d_%d_%d' %(x,y,p) - elif ( p== 1 ): # thread store - start_index = 2 - thread_name = 'stor_%d_%d_%d' %(x,y,p) - else : # thread analyse - start_index = 1 - thread_name = 'anal_%d_%d_%d' % (x,y,p) + start_index = 1 + thread_name = 'analyse_%d_%d_%d' % (x,y,p) mapping.addThread( vspace, thread_name, False , x, y, p, - 'classif_stack_%d_%d_%d' % (x,y,p), + 'classif_stack_%d_%d_%d' % (x,y,p), 'classif_heap_%d_%d' % (x,y), - start_index ) + 1 ) # index in start_vector # extend mapping name Index: /soft/giet_vm/applications/convol/convol.py =================================================================== --- /soft/giet_vm/applications/convol/convol.py (revision 824) +++ /soft/giet_vm/applications/convol/convol.py (revision 825) @@ -3,9 +3,9 @@ from mapping import * -###################################################################################### +##################################################################################### # file : convol.py # date : may 2014 # author : Alain Greiner -####################################################################################### +##################################################################################### # This file describes the mapping of the multi-threaded "convol" # application on a multi-clusters, multi-processors architecture. @@ -94,5 +94,5 @@ mapping.addVseg( vspace, 'conv_heap_%d_%d' % (x,y), base, size, - 'C_WU', vtype = 'BUFFER', x = x , y = y , pseg = 'RAM', + 'C_WU', vtype = 'HEAP', x = x , y = y , pseg = 'RAM', local = False, big = True ) Index: /soft/giet_vm/applications/fft/fft.py =================================================================== --- /soft/giet_vm/applications/fft/fft.py (revision 824) +++ /soft/giet_vm/applications/fft/fft.py (revision 825) @@ -89,5 +89,5 @@ mapping.addVseg( vspace, 'fft_heap_%d_%d' % (x,y), base, size, - 'C_WU', vtype = 'BUFFER', x = x , y = y , pseg = 'RAM', + 'C_WU', vtype = 'HEAP', x = x , y = y , pseg = 'RAM', local = False, big = True ) Index: /soft/giet_vm/applications/shell/shell.py =================================================================== --- /soft/giet_vm/applications/shell/shell.py (revision 824) +++ /soft/giet_vm/applications/shell/shell.py (revision 825) @@ -59,5 +59,5 @@ # heap vseg mapping.addVseg( vspace, 'shell_heap', heap_base, heap_size, - 'C_WU', vtype = 'BUFFER', x = xmap , y = ymap , pseg = 'RAM', + 'C_WU', vtype = 'HEAP', x = xmap , y = ymap , pseg = 'RAM', local = False ) Index: /soft/giet_vm/applications/sort/sort.py =================================================================== --- /soft/giet_vm/applications/sort/sort.py (revision 824) +++ /soft/giet_vm/applications/sort/sort.py (revision 825) @@ -88,5 +88,5 @@ mapping.addVseg( vspace, 'sort_heap_%d_%d' % (x,y), base, size, - 'C_WU', vtype = 'BUFFER', x = x, y = y, pseg = 'RAM', + 'C_WU', vtype = 'HEAP', x = x, y = y, pseg = 'RAM', local = False, big = True )