/* * Copyright (c) Washington University in St. Louis. * All rights reserved * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * 1. Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * 3. The name of the author or Washington University may not be used * to endorse or promote products derived from this source code * without specific prior written permission. * 4. Conditions of any other entities that contributed to this are also * met. If a copyright notice is present from another entity, it must * be maintained in redistributions of the source code. * * THIS INTELLECTUAL PROPERTY (WHICH MAY INCLUDE BUT IS NOT LIMITED TO SOFTWARE, * FIRMWARE, VHDL, etc) IS PROVIDED BY THE AUTHOR AND WASHINGTON UNIVERSITY * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR WASHINGTON UNIVERSITY * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS INTELLECTUAL PROPERTY, EVEN IF * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * * */ /* * File: flowcount.c * Author: Eric Peter * Email: epeter@wustl.edu * Organization: Applied Research Laboratory * * Derived from: NONE * * Date Created: Spring 2009 * * Description: Simple flow counter * */ //flow ID representation for calculation of hash typedef __declspec(packed) union t_cdv_flow_id_u { struct { unsigned int srcIP_a: 5; //used to be 5 unsigned int srcIP_b: 8; unsigned int dstIP_a: 5; //used to be 5 unsigned int dstIP_b: 8; unsigned int sport: 3; //was 3 unsigned int dport: 3; //was 3 }; unsigned int value[2]; } t_cdv_flow_id; //countdown vector typedef __declspec(packed) union t_cdv_vector_u { struct t_int_vector { unsigned int left: 4; unsigned int right: 4; }; unsigned char raw; } t_cdv_vector; //representation of IP address typedef __declspec(packed) union t_cdv_ip_u { struct { unsigned int part1: 8; unsigned int part2: 8; unsigned int part3: 8; unsigned int part4: 8; }; unsigned int value; } t_cdv_ip; //representation of ports typedef __declspec(packed) union t_cdv_port_u { struct { unsigned int whocares1: 13; unsigned int sport_right: 3; unsigned int whocares2: 13; unsigned int dport_right: 3; }; struct { unsigned int sport: 16; unsigned int dport: 16; }; unsigned int value; } t_cdv_port; #include #include "plugin_api.h" //----------------------------------------------------------- // Global variables/Registers //----------------------------------------------------------- //CDV globals const unsigned int b = 18000; //size of countdown vector (cdv) const unsigned int c = 15; //max size of element in the cdv const unsigned int TIME_WINDOW_SLEEP = 2100; //sleep interval for callback (.4 / (b(c-.5)) bc timewindow = 400 ms = .4seconds. (2100 cycles is about 1.5 usec). __declspec(shared sram) t_cdv_vector cdVector[9000]; //CDV - size set to half of b __declspec(shared sram) unsigned int z; //"z" parameter __declspec(shared sram) unsigned int cdvCounter; //counter to keep track of decrementing the CDVs __declspec(local_mem) char dbgmsg[28] = ""; //debug message tmp __declspec(shared gp_reg) unsigned int debug_on; __declspec(shared gp_reg) unsigned int n; //number of flows #define UNLOCKED 0 #define LOCKED 1 __declspec(shared gp_reg) unsigned int queue_lock = UNLOCKED; /* __declspec(sram) unsigned long even; __declspec(sram) unsigned long evenD; __declspec(sram) unsigned long odd; __declspec(sram) unsigned long oddD; //debug code */ // thread-specific globals __declspec(gp_reg) int dlNextBlock; // where to send packets to next __declspec(gp_reg) int dlFromBlock; // where to get packets from __declspec(gp_reg) int msgNextBlock; // where to send control messages to next __declspec(gp_reg) int msgFromBlock; // where to get control messages from // see ring_formats.h for struct definitions volatile __declspec(gp_reg) plc_plugin_data ring_in; // ring data from PLC volatile __declspec(gp_reg) plugin_out_data ring_out; // ring data to next block __declspec(local_mem) unsigned int timeout; // time (cycles) between callbacks __declspec(gp_reg) unsigned int pluginId; // plugin id (0...7) #include //----------------------------------------------------------- // Function prototypes //----------------------------------------------------------- /* functions */ void handle_packet_count(); void callback_user(); void plugin_init_user(); void default_format_out_data(unsigned int); //helper function to set a counter to a specific value static __forceinline void helper_plugin_cntr_set_value( int counter_id, int new_value ) { get_pcntr( pluginId, counter_id ); WU_loadGlobalRegister( stats_regnum, new_value, stats_cerr ); } // output control msg with 1 unsigned long // taken from erd++'s plugin_helpers.h _forceinline void helper_sram_outmsg_str_1ul( char *cstr, unsigned long x0, __declspec(local_mem) char *outmsgstr ) { __declspec(sram) char sram_msg_buf[28]; __declspec(gp_reg) int n; n = strlen_sram( cstr ); strcpy_sram( sram_msg_buf, cstr ); sram_msg_buf[n] = ' '; sram_msg_buf[n+1] = '\0'; ++n; helper_ultoa_sram( x0, sram_msg_buf+n, 28-n ); memcpy_lmem_sram( outmsgstr, sram_msg_buf, 28 ); } #define FormRawQid(out_port,qid) ((((out_port)+1) << 13) | qid) // approximatin of the formula b * ln(b/z). // used to get the count of flows from the CDV. // // Author: Ken Wong // _forceinline unsigned int nflows( unsigned int b, __declspec(shared sram) int z ) { __declspec(gp_reg) unsigned int y1 = (b-z); __declspec(gp_reg) unsigned int y2 = (b+z); __declspec(gp_reg) unsigned long acc; __declspec(gp_reg) unsigned long term; __declspec(local_mem) unsigned int k; term = (y1<<12)/y2; acc = 0; for( k=0; k<2; k++ ) { acc = acc + term/(2*k+1); term = (term*y1/y2)*y1/y2; } return (b * 2 * acc)>>12; } void handle_pkt_user() { __declspec(local_mem) t_cdv_flow_id flowInfo; //temporary flow tracking information __declspec(local_mem) t_cdv_ip ipInfo; //temporary flow tracking information __declspec(local_mem) t_cdv_port portInfo; //temporary flow tracking information __declspec(local_mem) unsigned long hashValue; //calculated hash of the flow __declspec(local_mem) unsigned long* __declspec(local_mem) hashTemp; //used for calculation of hashValue __declspec(local_mem) unsigned int hashPosition; //position in the CDV (from the flow hash) //basic packet header information __declspec(gp_reg) onl_api_ip_hdr ipv4_hdr; __declspec(local_mem) unsigned int ipv4HdrPtr; __declspec(gp_reg) buf_handle_t buf_handle; __declspec(gp_reg) onl_api_buf_desc bufDescriptor; __declspec(sram) unsigned int bufDescPtr; __declspec(sram) unsigned int dramBufferPtr; __declspec(local_mem) unsigned int udpHdrPtr; __declspec(local_mem) unsigned int tcpHdrPtr; __declspec(gp_reg) onl_api_udp_hdr udp_hdr; __declspec(gp_reg) onl_api_tcp_hdr tcp_hdr; //zero all tracking variables //flow info flowInfo.value[0] = 0; flowInfo.value[1] = 0; //port info portInfo.value = 0; //ip info ipInfo.value = 0; //pull in information about the current packet onl_api_get_buf_handle(&buf_handle); // Read the buf handle from the input Ring bufDescPtr = onl_api_getBufferDescriptorPtr(buf_handle); onl_api_readBufferDescriptor(bufDescPtr, &bufDescriptor); dramBufferPtr = onl_api_getBufferPtr(buf_handle); //get pointer to dram buffer ipv4HdrPtr = onl_api_getIpv4HdrPtr(dramBufferPtr, bufDescriptor.offset); //get a pointer to ipv4 header onl_api_readIpv4Hdr(ipv4HdrPtr, &ipv4_hdr); //read the ipV4 header //get src IP from IPv4 Header ipInfo.value = ipv4_hdr.ip_src; flowInfo.srcIP_a = ipInfo.part3; flowInfo.srcIP_b = ipInfo.part4; //DEBUG helper_sram_dbgmsg_1ul(ipInfo.value); //get dst IP from IPv4 Header ipInfo.value = 0; ipInfo.value = ipv4_hdr.ip_dst; flowInfo.dstIP_a = ipInfo.part3; flowInfo.dstIP_b = ipInfo.part4; //DEBUG helper_sram_dbgmsg_1ul(ipInfo.value); //determine if we have a TCP or UDP packet and get appropiate information. if(ipv4_hdr.ip_proto == PROTO_TCP) { tcpHdrPtr = onl_api_getTcpHdrPtr(ipv4HdrPtr, ipv4_hdr.ip_hl); onl_api_readTcpHdr(tcpHdrPtr, &tcp_hdr); //get port #s for hashing portInfo.sport = tcp_hdr.th_sport; portInfo.dport = tcp_hdr.th_dport; flowInfo.sport = portInfo.sport_right; flowInfo.dport = portInfo.dport_right; } else if(ipv4_hdr.ip_proto == PROTO_UDP) { udpHdrPtr = onl_api_getUdpHdrPtr(ipv4HdrPtr, ipv4_hdr.ip_hl); onl_api_readUdpHdr(udpHdrPtr, &udp_hdr); //get port #s for hashing portInfo.sport = udp_hdr.uh_sport; portInfo.dport = udp_hdr.uh_dport; flowInfo.sport = portInfo.sport_right; flowInfo.dport = portInfo.dport_right; } //DEBUG helper_sram_dbgmsg_1ul(portInfo.sport); //DEBUG helper_sram_dbgmsg_1ul(portInfo.dport); //hash the flow and get a position into the vector hashTemp = (__declspec(local_mem) unsigned long *) &flowInfo; hashValue = ((__declspec(local_mem) unsigned int) *hashTemp % b); hashValue = hashValue + portInfo.sport + portInfo.dport; hashValue = hashValue % b; hashPosition = hashValue/2; //DEBUG helper_sram_dbgmsg_1ul(*hashTemp); //DEBUG helper_sram_outmsg_3ul(*hashTemp, hashValue, b, dbgmsg); //DEBUG onl_api_debug_message(msgNextBlock, dbgmsg); //start updating the countdown vector while( queue_lock == LOCKED ) { ctx_swap(); } queue_lock = LOCKED; if(hashValue % 2 == 0) { //even = left 4 bytes of the char if(cdVector[hashPosition].left == 0) { //decrement 0 counter z--; } if(cdVector[hashPosition].left < c) { //don't increment if the position is already c cdVector[hashPosition].left++; } } else { //odd = right 4 bytes of the char if(cdVector[hashPosition].right == 0) { //decrement 0 counter z--; } if(cdVector[hashPosition].right < c) { //don't increment if the position is already c cdVector[hashPosition].right++; } } queue_lock = UNLOCKED; //DEBUG helper_sram_dbgmsg_1ul(cdVector[hashPosition].right); //DEBUG helper_sram_dbgmsg_1ul(cdVector[hashPosition].left); //debug the updating process if(debug_on) { helper_sram_outmsg_3ul(hashValue, *hashTemp, z, dbgmsg); onl_api_debug_message(msgNextBlock, dbgmsg); } //debug values of Z after cdv update //DEBUG helper_sram_dbgmsg_1ul(z); } // handle control messages // 'z' zero counters // 'd' toggle debug_on flag // 'g' get pkt_count // void handle_msg() { __declspec(gp_reg) unsigned int i; // assume messages are at most 8 words for now __declspec(gp_reg) unsigned int message[8]; __declspec(gp_reg) onl_api_ctrl_msg_hdr hdr; __declspec(local_mem) char inmsgstr[28]; // inbound __declspec(local_mem) char outmsgstr[28]; // outbound __declspec(local_mem) char lmem_tmpstr[8]; __declspec(sram) char sram_inmsgstr[28]; char GET_flows[8] = "=flows"; //return Rn Dn char GET_zvalue[8] = "=z"; //return Rz Dz char GET_debug[8] = "=debug"; //return debug_on char SET_debug[8] = "debug="; char BAD_OP_msg[8] = "BAD OP"; char RESET_msg[8] = "RD RST"; char BAD_NARGS_msg[12]= "BAD #ARGS ="; // expand for-loop to get rid of the compiler error: // "Incorrect use of register variable message[0] = 0; message[1] = 0; message[2] = 0; message[3] = 0; message[4] = 0; message[5] = 0; message[6] = 0; message[7] = 0; dl_source_message(msgFromBlock, message); hdr.value = message[0]; if( hdr.type != CM_CONTROLMSG ) return; if( hdr.response_requested != 1 ) return; onl_api_intarr2str( &message[1], inmsgstr ); outmsgstr[0] = '\0'; memcpy_sram_lmem( sram_inmsgstr, inmsgstr, 28 ); if( strncmp_sram(sram_inmsgstr, GET_flows, 6) == 0 ) { helper_sram_outmsg_1ul(n, outmsgstr); } else if( strncmp_sram(sram_inmsgstr, GET_zvalue, 2) == 0 ) { helper_sram_outmsg_1ul(z, outmsgstr); } else if( strncmp_sram(sram_inmsgstr, GET_debug, 6) == 0 ) { helper_sram_outmsg_1ul(debug_on, outmsgstr); } else if( strncmp_sram(sram_inmsgstr, SET_debug, 6) == 0 ) { char *cmnd_word; char *debug_on_word; unsigned int nwords; nwords = helper_count_words(sram_inmsgstr); if( nwords != 2 ) { helper_sram_outmsg_str_1ul( BAD_NARGS_msg, nwords, outmsgstr ); } else { cmnd_word = helper_tokenize( sram_inmsgstr ); // get command debug_on_word = helper_tokenize( cmnd_word+strlen(cmnd_word)+1 ); debug_on = helper_atou_sram(debug_on_word); helper_sram_outmsg_1ul(debug_on, outmsgstr); } } else { memcpy_lmem_sram( outmsgstr, BAD_OP_msg, 8 ); } if( onl_api_str2intarr(outmsgstr, &message[1]) < 0 ) return; hdr.type = CM_CONTROLMSGRSP; hdr.response_requested = 0; hdr.num_words = 7; message[0] = hdr.value; dl_sink_message(msgNextBlock, message); } void callback_user() { __declspec(local_mem) unsigned int hashPosition; //position into the vector //unsigned int n = 0; //DEBUG __declspec(local_mem) char dbgmsg2[16] = "CALLBACK CALLED"; //implement log algorithm //flow counter update //countdown algorithm //set queue params (when we have 2 flows) sleep(TIME_WINDOW_SLEEP); //DEBUG onl_api_debug_message(msgNextBlock, dbgmsg2); //DECREMENT CDV COUNTERS hashPosition = cdvCounter/2; //find position into vector given current counter while( queue_lock == LOCKED ) { ctx_swap(); } queue_lock = LOCKED; if(cdvCounter % 2 == 0) { //even = left 4 bits of the char if(cdVector[hashPosition].left == 1) { //increase 0 count if position will change to 0 z++; cdVector[hashPosition].left = 0; } else if(cdVector[hashPosition].left != 0) { //decrement if not already 0 cdVector[hashPosition].left = cdVector[hashPosition].left - 1; } } else { //odd = right 4 bits of the char if(cdVector[hashPosition].right == 1) { //increase 0 count if position will change to 0 z++; cdVector[hashPosition].right = 0; } else if(cdVector[hashPosition].right != 0) { //decrement if not already 0 cdVector[hashPosition].right = cdVector[hashPosition].right - 1; } } queue_lock = UNLOCKED; //END decrementing CDV //DEBUG /* if(debug_on) { helper_sram_outmsg_3ul(cdvCounter, Dz, Rz, dbgmsg); onl_api_debug_message(msgNextBlock, dbgmsg); helper_sram_outmsg_5ul(cdVectorD[hashPosition].left, cdVectorD[hashPosition].right, cdVectorR[hashPosition].left, cdVectorR[hashPosition].right , cdvCounter, dbgmsg); onl_api_debug_message(msgNextBlock, dbgmsg); } */ cdvCounter = (cdvCounter + 1) % b; //increase counter for next time //calculate the number of flows n = nflows(b, z); helper_plugin_cntr_set_value(0,n); //DEBUG if(debug_on) { helper_sram_dbgmsg_1ul(n); } } void plugin_init_user() { unsigned int i; if(ctx() == 0) // only thread 0 should do the initialization { z = b; //set "Z" in algorithm cdvCounter = 0; while( queue_lock == LOCKED ) { ctx_swap(); } queue_lock = LOCKED; for(i = 0; i <= b/2; i++) { //set countdown vector to all 0's cdVector[i].left = 0; cdVector[i].right = 0; } queue_lock = UNLOCKED; //helper_sram_dbgmsg_1ul(Dz); debug_on = 0; } } /** ---------------------------------------------------------------- @User: YOU SHOULD NOT NEED TO MAKE ANY CHANGES TO THE REST OF THIS FILE ---------------------------------------------------------------- */ void default_format_out_data(unsigned int nextblock) { __declspec(gp_reg) int out_port; if(nextblock == QM) { __declspec(gp_reg) dl_buf_handle_t buf_handle; __declspec(sram) unsigned int *buf_desc_ptr; __declspec(sram_write_reg) unsigned int sram_wr_regs[3]; SIGNAL sram_sig; // assume unicast out_port = (ring_in.uc_mc_bits >> 3) & 0x7; ring_out.plugin_qm_data_out.out_port = out_port; ring_out.plugin_qm_data_out.qid = ((out_port+1) << 13) | ring_in.qid; ring_out.plugin_qm_data_out.l3_pkt_len = ring_in.l3_pkt_len; ring_out.plugin_qm_data_out.buf_handle_lo24 = ring_in.buf_handle_lo24; /* // also need to write the stats index, nh mac info and ethertype info into the buffer // descriptor onl_api_get_buf_handle(&buf_handle); buf_desc_ptr = (__declspec(sram) unsigned int*) (Dl_BufGetDesc(buf_handle) + 12); sram_wr_regs[0] = (ring_in.stats_index << 16) | (ring_in.nh_eth_daddr_hi32 >> 16); sram_wr_regs[1] = (ring_in.nh_eth_daddr_hi32 << 16) | (ring_in.nh_eth_daddr_lo16); sram_wr_regs[2] = ring_in.eth_type << 16; // write data to payload buf desc sram_write(sram_wr_regs, buf_desc_ptr, 3, ctx_swap, &sram_sig); */ } else // assume MUX and ignore xscale data packets for now { // assume unicast out_port = (ring_in.uc_mc_bits >> 3) & 0x7; ring_out.plugin_mux_data_out.out_port = out_port; ring_out.plugin_mux_data_out.qid = ((out_port+1) << 13) | ring_in.qid; // assume pass-through ring_out.plugin_mux_data_out.flags = 1; ring_out.plugin_mux_data_out.plugin_tag = ring_in.plugin_tag; ring_out.plugin_mux_data_out.in_port = ring_in.in_port; ring_out.plugin_mux_data_out.stats_index = ring_in.stats_index; ring_out.plugin_mux_data_out.l3_pkt_len = ring_in.l3_pkt_len; ring_out.plugin_mux_data_out.buf_handle_lo24 = ring_in.buf_handle_lo24; } } /* handle packets */ void handle_pkt() { dl_source_packet(dlFromBlock); // format ring_out data based only on ring_in data default_format_out_data(dlNextBlock); handle_pkt_user(); // format ring_out data based only on ring_in data default_format_out_data(dlNextBlock); dl_sink_packet(dlNextBlock); } /* handle periodic functionality */ void callback() { callback_user(); // for now, the timeout value is in cycles (MEs operate at 1.4 GHz) // also, need to think about very large values for timeout // sleep(timeout); } /* take care of any setup that needs to be done before processing begins */ void plugin_init() { /* set a default timeout value for callback of 10000 cycles */ timeout = 10000; /* set the default next block to be the Queue Manager */ dlNextBlock = QM; /* by default, get packets and get and put control messages from input rings * based on which microengine we are currently running on; this assumes a * default one to one mapping */ switch(__ME()) { case 0x7: pluginId = 0; dlFromBlock = PACKET_IN_RING_0; msgFromBlock = MESSAGE_IN_RING_0; msgNextBlock = MESSAGE_OUT_RING_0; break; case 0x10: pluginId = 1; dlFromBlock = PACKET_IN_RING_1; msgFromBlock = MESSAGE_IN_RING_1; msgNextBlock = MESSAGE_OUT_RING_1; break; case 0x11: pluginId = 2; dlFromBlock = PACKET_IN_RING_2; msgFromBlock = MESSAGE_IN_RING_2; msgNextBlock = MESSAGE_OUT_RING_2; break; case 0x12: pluginId = 3; dlFromBlock = PACKET_IN_RING_3; msgFromBlock = MESSAGE_IN_RING_3; msgNextBlock = MESSAGE_OUT_RING_3; break; case 0x13: pluginId = 4; dlFromBlock = PACKET_IN_RING_4; msgFromBlock = MESSAGE_IN_RING_4; msgNextBlock = MESSAGE_OUT_RING_4; break; default: // keep the compiler happy pluginId = 0; dlFromBlock = PACKET_IN_RING_0; msgFromBlock = MESSAGE_IN_RING_0; msgNextBlock = MESSAGE_OUT_RING_0; break; } plugin_init_user(); // user hook } /* entry point */ void main() { int c; /* do initialization */ plugin_init(); dl_sink_init(); dl_source_init(); /* get the current thread's context number (0-7) */ c = ctx(); if(c >= FIRST_PACKET_THREAD && c <= LAST_PACKET_THREAD) { while(1) { handle_pkt(); } } #ifdef MESSAGE_THREAD else if(c == MESSAGE_THREAD) { while(1) { handle_msg(); } } #endif #ifdef CALLBACK_THREAD else if(c == CALLBACK_THREAD) { while(1) { callback(); } } #endif }