Changeset 21:eb59943050c9 in xplcommon for test/testlocsock.cc

Timestamp:

12/31/12 14:17:32 (11 years ago)

Author:

István Váradi <ivaradi@…>

Branch:

default

Phase:

public

Message:

Added the implementation of the local sockets for Win32 and it seems to work

File:

• test/testlocsock.cc (modified) (15 diffs)

test/testlocsock.cc

@@ -36 +36 @@
 #include <xplcommon/LocalSocket.h>
 #include <xplcommon/LocalClientSocket.h>
-#include <xplcommon/Connector.h>
+#include <xplcommon/LocalConnector.h>
 #include <xplcommon/ReadingBuffer.h>
 #include <xplcommon/WritingBuffer.h>
@@ -57 +57 @@
 using xplcommon::LocalSocket;
 using xplcommon::LocalClientSocket;
-using xplcommon::Connector;
+using xplcommon::LocalConnector;
 using xplcommon::ReadingBuffer;
 using xplcommon::WritingBuffer;
@@ -66 +66 @@
 //------------------------------------------------------------------------------
 
-void communicate(ReadingBuffer& readingBuffer, WritingBuffer& writingBuffer,
+void communicate(ReadingBuffer* readingBuffer, WritingBuffer* writingBuffer,
                  Waiter& waiter, unsigned seed, const char* prefix,
                  WaitableEvent* quitEvent = 0)
@@ -74 +74 @@
     static const size_t rangeToWrite = maxToWrite - minToWrite;
 
-    static const size_t reportingInterval = 100000;
-
-    static const int sleepThreshold = static_cast<int>(RAND_MAX * 0.95);
+    static const size_t reportingInterval = 10000;
+
+    static const int sleepThreshold = static_cast<int>(RAND_MAX * 0.99);
     static const int sleepRemainder = RAND_MAX - sleepThreshold;
     static const unsigned minSleep = 10;
@@ -93 +93 @@
 
     while(!toQuit) {
+#if TARGET_API_POSIX
         int r = rand_r(&seed);
+#endif
+#if TARGET_API_WIN32
+        int r = rand();
+#endif
         size_t toWrite = minToWrite +
             static_cast<size_t>( static_cast<double>(r) *
                                  rangeToWrite / RAND_MAX );
 
+        size_t lastWriteLength = 0;
+        if (writingBuffer!=0) {
+            lastWriteLength = min(writingBuffer->getCapacity(), toWrite);
+            writingBuffer->reset();
+            writingBuffer->addLength(lastWriteLength);
+        }
+
         while(toWrite>0) {
-            writingBuffer.reset();
-            size_t length = min(writingBuffer.getCapacity(), toWrite);
-            writingBuffer.addLength(length);
-
             if (quitEvent!=0 && quitEvent->check()) {
                 toQuit = true;
@@ -108 +116 @@
             }
 
-            if (writingBuffer.write()) {
-                numBytesWritten += length;
-                toWrite -= length;
-                length = 0;
-
-                if (numBytesWritten>=nextReportWritten) {
-                    printf("%s: written %zu bytes with %zu blocking and %zu ms sleeping\n",
-                           prefix, numBytesWritten, numWriteBlocked,
-                           numMillisSlept);
-                    nextReportWritten += reportingInterval;
+            bool hasPending = false;
+
+            if (writingBuffer!=0) {
+                size_t loopCount = 0;
+                while(toWrite>0) {
+                    ++loopCount;
+                    if (writingBuffer->write()) {
+                        numBytesWritten += lastWriteLength;
+                        toWrite -= lastWriteLength;
+
+                        if (numBytesWritten>=nextReportWritten) {
+                            printf("%s: written %lu bytes with %lu blocking and %lu ms sleeping\n",
+                                   prefix,
+                                   static_cast<unsigned long>(numBytesWritten),
+                                   static_cast<unsigned long>(numWriteBlocked),
+                                   static_cast<unsigned long>(numMillisSlept));
+                            nextReportWritten += reportingInterval;
+                        }
+
+                        if (toWrite>0) {
+                            if (quitEvent!=0 && quitEvent->check()) {
+                                toQuit = true;
+                                break;
+                            } else {
+                                lastWriteLength = min(writingBuffer->getCapacity(), toWrite);
+                                writingBuffer->addLength(lastWriteLength);
+                                if (loopCount>10) {
+                                    break;
+                                }
+                            }
+                        }
+                    } else if (writingBuffer->failed()) {
+                        printf("%s: writing failed with error: %lu\n",
+                               prefix,
+                               static_cast<unsigned long>(writingBuffer->getErrorCode()));
+                        toQuit = true;
+                        break;
+                    } else {
+                        ++numWriteBlocked;
+                        hasPending = true;
+                        break;
+                    }
                 }
-            } else if (writingBuffer.failed()) {
-                printf("%s: writing failed with error: %lu\n",
-                       prefix,
-                       static_cast<unsigned long>(writingBuffer.getErrorCode()));
-                toQuit = true;
-                break;
-            } else {
-                ++numWriteBlocked;
+                if (toQuit) break;
             }
 
-            if (readingBuffer.read()) {
-                numBytesRead += readingBuffer.getLength();
-                if (numBytesRead>=nextReportRead) {
-                    printf("%s: read %zu bytes with %zu blocking and %zu ms sleeping\n",
-                           prefix, numBytesRead, numReadBlocked, numMillisSlept);
-                    nextReportRead += reportingInterval;
+            if (readingBuffer!=0) {
+                size_t loopCount = 0;
+                while(true) {
+                    ++loopCount;
+                    if (readingBuffer->read()) {
+                        size_t l = readingBuffer->getLength();
+                        if (l==0) {
+                            printf("%s: reading encountered end of file\n", prefix);
+                            toQuit = true;
+                            break;
+                        }
+                        numBytesRead += l;
+                        if (numBytesRead>=nextReportRead) {
+                            printf("%s: read %lu bytes with %lu blocking and %lu ms sleeping\n",
+                                   prefix,
+                                   static_cast<unsigned long>(numBytesRead),
+                                   static_cast<unsigned long>(numReadBlocked),
+                                   static_cast<unsigned long>(numMillisSlept));
+                            nextReportRead += reportingInterval;
+                        }
+                        readingBuffer->reset();
+                        if (quitEvent!=0 && quitEvent->check()) {
+                            toQuit = true;
+                            break;
+                        } else if (loopCount>10) {
+                            break;
+                        }
+                    } else if (readingBuffer->failed()) {
+                        printf("%s: reading failed with error: %lu\n",
+                               prefix,
+                               static_cast<unsigned long>(readingBuffer->getErrorCode()));
+                        toQuit = true;
+                        break;
+                    } else {
+                        ++numReadBlocked;
+                        hasPending = true;
+                        break;
+                    }
                 }
-                readingBuffer.reset();
-            } else if (readingBuffer.failed()) {
-                printf("%s: reading failed with error: %lu\n",
-                       prefix,
-                       static_cast<unsigned long>(readingBuffer.getErrorCode()));
-                toQuit = true;
-                break;
-            } else {
-                ++numReadBlocked;
+                if (toQuit) break;
             }
 
+#if TARGET_API_POSIX
             r = rand_r(&seed);
+#endif
+#if TARGET_API_WIN32
+            r = rand();
+#endif
             if (r>=sleepThreshold) {
                 unsigned toSleep = minSleep +
@@ -158 +220 @@
             }
 
-            if (length!=0) {
+            if (toWrite>0 && hasPending) {
                 waiter.wait();
                 if (waiter.failed()) {
@@ -171 +233 @@
     }
 
-    printf("%s: written %zu bytes with %zu blocking\n",
-           prefix, numBytesWritten, numWriteBlocked);
-    printf("%s: read %zu bytes with %zu blocking\n",
-           prefix, numBytesRead, numReadBlocked);
-    printf("%s: slept %zu ms\n", prefix, numMillisSlept);
+    printf("%s: written %lu bytes with %lu blocking\n",
+           prefix,
+           static_cast<unsigned long>(numBytesWritten),
+           static_cast<unsigned long>(numWriteBlocked));
+    printf("%s: read %lu bytes with %lu blocking\n",
+           prefix,
+           static_cast<unsigned long>(numBytesRead),
+           static_cast<unsigned long>(numReadBlocked));
+    printf("%s: slept %lu ms\n", prefix,
+           static_cast<unsigned long>(numMillisSlept));
 }
 
@@ -219 +286 @@
     LocalAcceptor& acceptor = serverSocket.getAcceptor();
 
+    bool eventFired = false;
     while(!acceptor.accept()) {
+        eventFired = event.check();
+        if (eventFired) break;
         if (acceptor.failed()) {
-            printf("ServerThread::run: acceptor failed...\n");
+            printf("ServerThread::run: acceptor failed...: %lu\n",
+                   (unsigned long)acceptor.getErrorCode());
             return;
         }
@@ -228 +299 @@
     }
 
+    if (eventFired) {
+        printf("ServerThread::run: waiting done, quitting\n");
+        return;
+    }
+
     printf("ServerThread::run: waiting done, received connection\n");
     LocalSocket* socket = acceptor.getSocket();
 
-    communicate(socket->getReadingBuffer(), socket->getWritingBuffer(),
+    communicate(&socket->getReadingBuffer(), &socket->getWritingBuffer(),
                 waiter, seed, "ServerThread", &event);
 
@@ -243 +319 @@
 class ClientThread : public Thread
 {
+private:
+    Waiter waiter;
+
+    WaitableEvent event;
+
 public:
+    ClientThread();
+
+    void signal();
+
     virtual void run();
 };
+
+//------------------------------------------------------------------------------
+
+ClientThread::ClientThread() :
+    event(&waiter)
+{
+}
+
+//------------------------------------------------------------------------------
+
+void ClientThread::signal()
+{
+    event.fire();
+}
 
 //------------------------------------------------------------------------------
@@ -258 +357 @@
     printf("ClientThread::run: connecting\n");
 
-    Waiter waiter;
-
     LocalClientSocket socket("test", &waiter);
 
-    Connector& connector = socket.getConnector();
+    LocalConnector& connector = socket.getConnector();
 
     while(!connector.connect()) {
@@ -275 +372 @@
     printf("ClientThread::run: connected\n");
 
-    communicate(socket.getReadingBuffer(), socket.getWritingBuffer(),
-                waiter, seed, "ClientThread");
+    communicate(&socket.getReadingBuffer(), &socket.getWritingBuffer(),
+                waiter, seed, "ClientThread", &event);
 }
 
@@ -284 +381 @@
 int main()
 {
+#if TARGET_API_POSIX
     signal(SIGPIPE, SIG_IGN);
+#endif
 
     ServerThread serverThread;
@@ -294 +393 @@
 
     Thread::sleep(60000);
-    serverThread.signal();
+    //Thread::sleep(5000);
+    printf("Signalling the server thread\n");
+    //serverThread.signal();
+    clientThread.signal();
     printf("Signalled the server thread\n");