main.cpp

00001 #ifndef MBED_EXTENDED_TESTS
00002     #error [NOT_SUPPORTED] Parallel pressure tests are not supported by default
00003 #endif
00004 
00005 #include "mbed.h"
00006 #include "ESP8266Interface.h"
00007 #include "UDPSocket.h"
00008 #include "greentea-client/test_env.h"
00009 #include "unity/unity.h"
00010 #include "utest.h"
00011 
00012 using namespace utest::v1;
00013 
00014 
00015 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN
00016 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN 64
00017 #endif
00018 
00019 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX
00020 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX 0x80000
00021 #endif
00022 
00023 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_TIMEOUT
00024 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_TIMEOUT 100
00025 #endif
00026 
00027 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_SEED
00028 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_SEED 0x6d626564
00029 #endif
00030 
00031 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS
00032 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS 3
00033 #endif
00034 
00035 #ifndef MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG
00036 #define MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG false
00037 #endif
00038 
00039 #ifndef MBED_CFG_ESP8266_TX
00040 #define MBED_CFG_ESP8266_TX D1
00041 #endif
00042 
00043 #ifndef MBED_CFG_ESP8266_RX
00044 #define MBED_CFG_ESP8266_RX D0
00045 #endif
00046 
00047 #ifndef MBED_CFG_ESP8266_DEBUG
00048 #define MBED_CFG_ESP8266_DEBUG false
00049 #endif
00050 
00051 #define STRINGIZE(x) STRINGIZE2(x)
00052 #define STRINGIZE2(x) #x
00053 
00054 
00055 // Simple xorshift pseudorandom number generator
00056 class RandSeq {
00057 private:
00058     uint32_t x;
00059     uint32_t y;
00060     static const int A = 15;
00061     static const int B = 18;
00062     static const int C = 11;
00063 
00064 public:
00065     RandSeq(uint32_t seed=MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_SEED)
00066         : x(seed), y(seed) {}
00067 
00068     uint32_t next(void) {
00069         x ^= x << A;
00070         x ^= x >> B;
00071         x ^= y ^ (y >> C);
00072         return x + y;
00073     }
00074 
00075     void skip(size_t size) {
00076         for (size_t i = 0; i < size; i++) {
00077             next();
00078         }
00079     }
00080 
00081     void buffer(uint8_t *buffer, size_t size) {
00082         RandSeq lookahead = *this;
00083 
00084         for (size_t i = 0; i < size; i++) {
00085             buffer[i] = lookahead.next() & 0xff;
00086         }
00087     }
00088 
00089     int cmp(uint8_t *buffer, size_t size) {
00090         RandSeq lookahead = *this;
00091 
00092         for (size_t i = 0; i < size; i++) {
00093             int diff = buffer[i] - (lookahead.next() & 0xff);
00094             if (diff != 0) {
00095                 return diff;
00096             }
00097         }
00098         return 0;
00099     }
00100 };
00101 
00102 // Tries to get the biggest buffer possible on the device. Exponentially
00103 // grows a buffer until heap runs out of space, and uses half to leave
00104 // space for the rest of the program
00105 void generate_buffer(uint8_t **buffer, size_t *size, size_t min, size_t max) {
00106     size_t i = min;
00107     while (i < max) {
00108         void *b = malloc(i);
00109         if (!b) {
00110             i /= 8;
00111             if (i < min) {
00112                 i = min;
00113             }
00114             break;
00115         }
00116         free(b);
00117         i *= 2;
00118     }
00119 
00120     *buffer = (uint8_t *)malloc(i);
00121     *size = i;
00122     TEST_ASSERT(buffer);
00123 }
00124 
00125 
00126 // Global variables shared between pressure tests
00127 ESP8266Interface net(MBED_CFG_ESP8266_TX, MBED_CFG_ESP8266_RX, MBED_CFG_ESP8266_DEBUG);
00128 SocketAddress udp_addr;
00129 Timer timer;
00130 Mutex iomutex;
00131 
00132 // Single instance of a pressure test
00133 class PressureTest {
00134 private:
00135     uint8_t *buffer;
00136     size_t buffer_size;
00137 
00138     UDPSocket sock;
00139     Thread thread;
00140 
00141 public:
00142     PressureTest(uint8_t *buffer, size_t buffer_size)
00143         : buffer(buffer), buffer_size(buffer_size) {
00144     }
00145 
00146     void start() {
00147         osStatus status = thread.start(callback(this, &PressureTest::run));
00148         TEST_ASSERT_EQUAL(osOK, status);
00149     }
00150 
00151     void join() {
00152         osStatus status = thread.join();
00153         TEST_ASSERT_EQUAL(osOK, status);
00154     }
00155 
00156     void run() {
00157         // Tests exponentially growing sequences
00158         for (size_t size = MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN;
00159              size < MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX;
00160              size *= 2) {
00161             int err = sock.open(&net);
00162             TEST_ASSERT_EQUAL(0, err);
00163             iomutex.lock();
00164             printf("UDP: %s:%d streaming %d bytes\r\n",
00165                 udp_addr.get_ip_address(), udp_addr.get_port(), size);
00166             iomutex.unlock();
00167 
00168             sock.set_blocking(false);
00169 
00170             // Loop to send/recv all data
00171             RandSeq tx_seq;
00172             RandSeq rx_seq;
00173             size_t rx_count = 0;
00174             size_t tx_count = 0;
00175             int known_time = timer.read_ms();
00176             size_t window = buffer_size;
00177 
00178             while (tx_count < size || rx_count < size) {
00179                 // Send out packets
00180                 if (tx_count < size) {
00181                     size_t chunk_size = size - tx_count;
00182                     if (chunk_size > window) {
00183                         chunk_size = window;
00184                     }
00185 
00186                     tx_seq.buffer(buffer, chunk_size);
00187                     int td = sock.sendto(udp_addr, buffer, chunk_size);
00188 
00189                     if (td > 0) {
00190                         if (MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG) {
00191                             iomutex.lock();
00192                             printf("UDP: tx -> %d\r\n", td);
00193                             iomutex.unlock();
00194                         }
00195                         tx_seq.skip(td);
00196                         tx_count += td;
00197                     } else if (td != NSAPI_ERROR_WOULD_BLOCK) {
00198                         // We may fail to send because of buffering issues, revert to
00199                         // last good sequence and cut buffer in half
00200                         if (window > MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN) {
00201                             window /= 2;
00202                         }
00203 
00204                         if (MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG) {
00205                             iomutex.lock();
00206                             printf("UDP: Not sent (%d), window = %d\r\n", td, window);
00207                             iomutex.unlock();
00208                         }
00209                     }
00210                 }
00211 
00212                 // Prioritize recieving over sending packets to avoid flooding
00213                 // the network while handling erronous packets
00214                 while (rx_count < size) {
00215                     int rd = sock.recvfrom(NULL, buffer, buffer_size);
00216                     TEST_ASSERT(rd > 0 || rd == NSAPI_ERROR_WOULD_BLOCK);
00217 
00218                     if (rd > 0) {
00219                         if (MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG) {
00220                             iomutex.lock();
00221                             printf("UDP: rx <- %d\r\n", rd);
00222                             iomutex.unlock();
00223                         }
00224 
00225                         if (rx_seq.cmp(buffer, rd) == 0) {
00226                             rx_seq.skip(rd);
00227                             rx_count += rd;
00228                             known_time = timer.read_ms();
00229                             if (window < MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX) {
00230                                 window += MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN;
00231                             }
00232                         }
00233                     } else if (timer.read_ms() - known_time >
00234                             MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_TIMEOUT) {
00235                         // Dropped packet or out of order, revert to last good sequence
00236                         // and cut buffer in half
00237                         tx_seq = rx_seq;
00238                         tx_count = rx_count;
00239                         known_time = timer.read_ms();
00240                         if (window > MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN) {
00241                             window /= 2;
00242                         }
00243 
00244                         if (MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_DEBUG) {
00245                             iomutex.lock();
00246                             printf("UDP: Dropped, window = %d\r\n", window);
00247                             iomutex.unlock();
00248                         }
00249                     } else if (rd == NSAPI_ERROR_WOULD_BLOCK) {
00250                         break;
00251                     }
00252                 }
00253             }
00254 
00255             err = sock.close();
00256             TEST_ASSERT_EQUAL(0, err);
00257         }
00258     }
00259 };
00260 
00261 PressureTest *pressure_tests[MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS];
00262 
00263 
00264 void test_udp_packet_pressure_parallel() {
00265     uint8_t *buffer;
00266     size_t buffer_size;
00267     generate_buffer(&buffer, &buffer_size,
00268         MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN,
00269         MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX);
00270 
00271     size_t buffer_subsize = buffer_size / MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS;
00272     printf("MBED: Generated buffer %d\r\n", buffer_size);
00273     printf("MBED: Split into %d buffers %d\r\n",
00274             MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS,
00275             buffer_subsize);
00276 
00277     int err = net.connect(STRINGIZE(MBED_CFG_ESP8266_SSID), STRINGIZE(MBED_CFG_ESP8266_PASS));
00278     TEST_ASSERT_EQUAL(0, err);
00279 
00280     printf("MBED: UDPClient IP address is '%s'\n", net.get_ip_address());
00281     printf("MBED: UDPClient waiting for server IP and port...\n");
00282 
00283     greentea_send_kv("target_ip", net.get_ip_address());
00284 
00285     char recv_key[] = "host_port";
00286     char ipbuf[60] = {0};
00287     char portbuf[16] = {0};
00288     unsigned int port = 0;
00289 
00290     greentea_send_kv("host_ip", " ");
00291     greentea_parse_kv(recv_key, ipbuf, sizeof(recv_key), sizeof(ipbuf));
00292 
00293     greentea_send_kv("host_port", " ");
00294     greentea_parse_kv(recv_key, portbuf, sizeof(recv_key), sizeof(ipbuf));
00295     sscanf(portbuf, "%u", &port);
00296 
00297     printf("MBED: Server IP address received: %s:%d \n", ipbuf, port);
00298     udp_addr.set_ip_address(ipbuf);
00299     udp_addr.set_port(port);
00300 
00301     timer.start();
00302 
00303     // Startup pressure tests in parallel
00304     for (int i = 0; i < MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS; i++) {
00305         pressure_tests[i] = new PressureTest(&buffer[i*buffer_subsize], buffer_subsize);
00306         pressure_tests[i]->start();
00307     }
00308 
00309     for (int i = 0; i < MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS; i++) {
00310         pressure_tests[i]->join();
00311         delete pressure_tests[i];
00312     }
00313 
00314     timer.stop();
00315     printf("MBED: Time taken: %fs\r\n", timer.read());
00316     printf("MBED: Speed: %.3fkb/s\r\n",
00317             MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_THREADS*
00318             8*(2*MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MAX - 
00319             MBED_CFG_UDP_CLIENT_PACKET_PRESSURE_MIN) / (1000*timer.read()));
00320 
00321     net.disconnect();
00322 }
00323 
00324 
00325 // Test setup
00326 utest::v1::status_t test_setup(const size_t number_of_cases) {
00327     GREENTEA_SETUP(120, "udp_echo");
00328     return verbose_test_setup_handler(number_of_cases);
00329 }
00330 
00331 Case cases[] = {
00332     Case("UDP packet pressure parallel", test_udp_packet_pressure_parallel),
00333 };
00334 
00335 Specification specification(test_setup, cases);
00336 
00337 int main() {
00338     return !Harness::run(specification);
00339 }
00340 
00341