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