Changeset 31:bbd688924703 in xplcommon

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

Files:

• src/xplcommon/BlockingStream.cc (modified) (6 diffs)
• src/xplcommon/BlockingStream.h (modified) (7 diffs)
• src/xplcommon/DataStream.cc (added)
• src/xplcommon/DataStream.h (added)
• src/xplcommon/Makefile.am (modified) (2 diffs)
• test/testblkstream.cc (modified) (7 diffs)

src/xplcommon/BlockingStream.cc

@@ -46 +46 @@
 //------------------------------------------------------------------------------
 
-bool BlockingStream::isInterrupted()
+inline bool BlockingStream::checkInterrupted()
 {
-    if (failed()) return false;
-
     if (!interrupted) {
         interrupted = event.check();
@@ -83 +81 @@
 //------------------------------------------------------------------------------
 
+bool BlockingStream::skip(size_t length)
+{
+    ReadingBuffer& readingBuffer = stream.getReadingBuffer();
+    while (length>0) {
+        size_t toSkip = min(length, readingBuffer.getLength() - readingOffset);
+        readingOffset += toSkip;
+        length -= toSkip;
+
+        if (length!=0) {
+            if (!fill()) return false;
+        }
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
 bool BlockingStream::write(const void* src, size_t length)
 {
@@ -105 +121 @@
 {
     WritingBuffer& writingBuffer = stream.getWritingBuffer();
-    while (!failed()) {
-        if (isInterrupted()) return false;
+    while (*this) {
+        if (checkInterrupted()) return false;
 
         if (writingBuffer.write()) {
@@ -112 +128 @@
         } else if (writingBuffer.failed()) {
             setErrorCode(writingBuffer.getErrorCode());
-            return false;
         } else {
             Waiter* waiter = stream.getWaiter();
@@ -132 +147 @@
     readingBuffer.reset();
     readingOffset = 0;
-    while (true) {
-        if (isInterrupted()) return false;
+    while (*this) {
+        if (checkInterrupted()) return false;
 
         if (readingBuffer.read()) {
-            return !readingBuffer.isEmpty();
+            eof = readingBuffer.isEmpty();
+            return !eof;
         } else if (readingBuffer.failed()) {
             setErrorCode(readingBuffer.getErrorCode());
-            return false;
         } else {
             Waiter* waiter = stream.getWaiter();
@@ -149 +164 @@
         }
     }
+    return false;
 }
 

src/xplcommon/BlockingStream.h

@@ -35 +35 @@
 
 #include "BufferedStream.h"
+
 #include "Waiter.h"
 #include "WaitableEvent.h"
@@ -76 +77 @@
     size_t readingOffset;
 
+    /**
+     * Indicate if the end-of-file has been reached while reading.
+     */
+    bool eof;
+
 public:
     /**
@@ -83 +89 @@
      */
     BlockingStream(BufferedStream& stream);
+
+    /**
+     * Determine if the stream has neither failed nor been
+     * interrupted.
+     */
+    operator bool() const;
 
     /**
@@ -102 +114 @@
      */
     bool read(void* dest, size_t length);
+
+    /**
+     * Skip the given number of bytes.
+     */
+    bool skip(size_t length);
 
     /**
@@ -125 +142 @@
      */
     bool fill();
+
+    /**
+     * Check for the stream being interrupted.
+     *
+     * @return if the stream is interrupted, false otherwise
+     */
+    bool checkInterrupted();
 };
 
@@ -135 +159 @@
     event(stream.getWaiter()),
     interrupted(false),
-    readingOffset(0)
+    readingOffset(0),
+    eof(false)
 {
+}
+
+//------------------------------------------------------------------------------
+
+inline BlockingStream::operator bool() const
+{
+    return !failed() && !interrupted && !eof;
 }
 
@@ -144 +176 @@
 {
     event.fire();
+}
+
+//------------------------------------------------------------------------------
+
+inline bool BlockingStream::isInterrupted()
+{
+    return interrupted;
 }
 

src/xplcommon/Makefile.am

@@ -11 +11 @@
 libxplcommon_la_SOURCES= \
         PseudoRandom.cc         \
-        BlockingStream.cc
+        BlockingStream.cc       \
+        DataStream.cc
 
 if TARGET_API_POSIX
@@ -39 +40 @@
         LocalConnector.h        \
         LocalClientSocket.h     \
-        BlockingStream.h
+        BlockingStream.h        \
+        DataStream.h

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");