Changeset 31:bbd688924703 in xplcommon for test

Timestamp:

01/02/13 13:58:25 (12 years ago)

Author:

István Váradi <ivaradi@…>

Branch:

default

Phase:

public

Message:

Implemented the data stream

File:

• test/testblkstream.cc (modified) (7 diffs)

test/testblkstream.cc

@@ -37 +37 @@
 #include <xplcommon/LocalClientSocket.h>
 #include <xplcommon/LocalConnector.h>
-#include <xplcommon/BlockingStream.h>
+#include <xplcommon/DataStream.h>
 #include <xplcommon/WaitableEvent.h>
 #include <xplcommon/PseudoRandom.h>
@@ -59 +59 @@
 using xplcommon::LocalConnector;
 using xplcommon::BufferedStream;
-using xplcommon::BlockingStream;
+using xplcommon::DataStream;
 using xplcommon::PseudoRandom;
 
@@ -69 +69 @@
 {
 private:
-    BlockingStream* dataStream;
+    static const uint8_t OP_ADD = 1;
+
+    static const uint8_t OP_SUB = 2;
+
+    static const uint8_t OP_MUL = 3;
+
+    static const uint8_t OP_DIV = 4;
+
+private:
+    static int32_t perform(int32_t value, uint8_t operation, int32_t x);
+
+    DataStream* dataStream;
 
 protected:
@@ -83 +94 @@
 //------------------------------------------------------------------------------
 
+inline int32_t TestThread::perform(int32_t value, uint8_t operation, int32_t x)
+{
+    switch (operation) {
+      case OP_ADD:
+        return value + x;
+      case OP_SUB:
+        return value - x;
+      case OP_MUL:
+        return value * x;
+      case OP_DIV:
+        return value / x;
+      default:
+        return value;
+    }
+}
+
+//------------------------------------------------------------------------------
+
 inline TestThread::TestThread() :
     dataStream(0)
@@ -99 +128 @@
 void TestThread::communicate(BufferedStream& stream, unsigned seed, bool isClient)
 {
-    static const size_t bufferSize = 1024;
-
-    static const size_t minToWrite = 16;
-    static const size_t maxToWrite = 6124;
-
-    static const size_t reportingInterval = 100000000;
+    static const size_t reportInterval = 100000;
 
     const char* prefix = isClient ? "Client" : "Server";
 
-    dataStream = new BlockingStream(stream);
+    dataStream = new DataStream(stream);
     PseudoRandom random(seed);
 
-
-    size_t numBytesRead = 0;
-    size_t nextReportRead = reportingInterval;
-    size_t numBytesWritten = 0;
-    size_t nextReportWritten = reportingInterval;
-
     bool first = true;
 
-    unsigned char* buffer = new unsigned char[bufferSize];
-
-    while(true) {
-        if (isClient || !first) {
-            unsigned dataSize = random.nextUnsigned(maxToWrite, minToWrite);
-            *reinterpret_cast<uint32_t*>(buffer) = dataSize;
-            while(dataSize>0) {
-                unsigned toWrite = min(static_cast<size_t>(dataSize),
-                                       bufferSize);
-                if (!dataStream->write(buffer, toWrite)) break;
-                dataSize -= toWrite;
-                numBytesWritten += toWrite;
-                if (numBytesWritten>=nextReportWritten) {
-                    printf("%s: written %lu bytes\n", prefix,
-                           static_cast<unsigned long>(numBytesWritten));
-                    nextReportWritten += reportingInterval;
+    size_t numRequests = 0;
+    size_t nextReportRequests = reportInterval;
+
+    int32_t value = 0;
+    if (isClient) {
+        value = random.next();
+        dataStream->writeS32(value);
+    } else {
+        value = dataStream->readS32();
+    }
+
+    int32_t newValue = value;
+
+    while(*dataStream) {
+        if (!first || isClient) {
+            unsigned numOperations = random.nextUnsigned(20, 1);
+            dataStream->writeU16(numOperations);
+            //printf("%s: numOperations=%u\n", prefix, numOperations);
+            for(unsigned i = 0; i<numOperations; ++i) {
+                double r = random.nextDouble();
+                uint8_t operation =
+                    (r<0.4) ? OP_ADD : ((r<0.8) ? OP_SUB :
+                                        ((r<0.98) ? OP_MUL : OP_DIV));
+                int32_t x = random.next();
+
+                dataStream->writeU8(operation);
+                dataStream->writeS32(x);
+
+                newValue = perform(newValue, operation, x);
+            }
+            if (!dataStream->flush()) break;
+
+            if (isClient) {
+                value = newValue;
+                ++numRequests;
+                if (numRequests>=nextReportRequests) {
+                    printf("%s: after %lu requests, value=%d\n",
+                           prefix, static_cast<unsigned long>(numRequests),
+                           value);
+                    nextReportRequests += reportInterval;
                 }
             }
-            if (!dataStream->flush()) break;
-        }
-
-        uint32_t dataSize = 0;
-        if (!dataStream->read(&dataSize, sizeof(dataSize))) break;
-        assert(dataSize>=minToWrite);
-        assert(dataSize<=maxToWrite);
-        dataSize -= sizeof(dataSize);
-        numBytesRead += 4;
-        while(dataSize>0) {
-            unsigned toRead = min(static_cast<size_t>(dataSize),
-                                  bufferSize);
-            if (!dataStream->read(buffer, toRead)) break;
-            dataSize -= toRead;
-            numBytesRead += toRead;
-            if (numBytesRead>=nextReportRead) {
-                printf("%s: read %lu bytes\n", prefix,
-                       static_cast<unsigned long>(numBytesRead));
-                nextReportRead += reportingInterval;
+        }
+
+        unsigned numOperations = dataStream->readU16();
+        for (unsigned i = 0; *dataStream && i<numOperations; ++i) {
+            uint8_t operation = dataStream->readU8();
+            int32_t x = dataStream->readS32();
+            newValue = perform(newValue, operation, x);
+        }
+
+        if (*dataStream) {
+            if (!isClient) {
+                value = newValue;
+                ++numRequests;
+                if (numRequests>=nextReportRequests) {
+                    printf("%s: after %lu requests, value=%d\n",
+                           prefix, static_cast<unsigned long>(numRequests),
+                           value);
+                    nextReportRequests += reportInterval;
+                }
             }
         }
 
+
         first = false;
     }
-
-    delete [] buffer;
 
     if (dataStream->failed()) {
@@ -171 +212 @@
     }
 
-    printf("%s: written %lu bytes and read %lu bytes\n", prefix,
-           static_cast<unsigned long>(numBytesWritten),
-           static_cast<unsigned long>(numBytesRead));
+    printf("%s: value=%d, %lu requests\n", prefix, value,
+           static_cast<unsigned long>(numRequests));
 }
 
@@ -275 +315 @@
 
     Thread::sleep(60000);
-    //Thread::sleep(5000);
-    printf("Signalling the server thread\n");
-    serverThread.interrupt();
-    //clientThread.interrupt();
-    printf("Signalled the server thread\n");
+    //Thread::sleep(1000);
+    printf("Signalling the client thread\n");
+    //serverThread.interrupt();
+    clientThread.interrupt();
+    printf("Signalled the client thread\n");
 
     printf("Waiting for the client thread\n");