Changeset 19:54c3f68d2d46 in xplcommon

Timestamp:

12/29/12 18:47:52 (11 years ago)

Author:

István Váradi <ivaradi@…>

Branch:

default

Phase:

public

Message:

Fixed problems with non-blocking mode and extended the socket test program to actually communicate

Files:

• src/xplcommon/posix/ServerSocket.cc (modified) (1 diff)
• src/xplcommon/posix/Socket.h (modified) (2 diffs)
• test/testlocsock.cc (modified) (8 diffs)

src/xplcommon/posix/ServerSocket.cc

@@ -48 +48 @@
 {
     if (fd<0) setErrorCodeFromErrno();
+    else if (waiter!=0) setNonBlocking(fd);
 }
 

src/xplcommon/posix/Socket.h

@@ -89 +89 @@
 {
     if (fd<0) setErrorCodeFromErrno();
+    else if (waiter!=0) setNonBlocking(fd);
 }
 
@@ -97 +98 @@
     BufferedStream(waiter, fd, readingCapacity, writingCapacity)
 {
+    if (waiter!=0) setNonBlocking(fd);
 }
 

test/testlocsock.cc

@@ -39 +39 @@
 #include <xplcommon/ReadingBuffer.h>
 #include <xplcommon/WritingBuffer.h>
-
+#include <xplcommon/WaitableEvent.h>
+
+#include <cstdlib>
 #include <cstdio>
+#include <cassert>
+
+#include <signal.h>
+
+#include <sys/time.h>
 
 //------------------------------------------------------------------------------
@@ -51 +58 @@
 using xplcommon::LocalClientSocket;
 using xplcommon::Connector;
+using xplcommon::ReadingBuffer;
+using xplcommon::WritingBuffer;
+using xplcommon::WaitableEvent;
+
+using std::min;
+
+//------------------------------------------------------------------------------
+
+void communicate(ReadingBuffer& readingBuffer, WritingBuffer& writingBuffer,
+                 Waiter& waiter, unsigned seed, const char* prefix,
+                 WaitableEvent* quitEvent = 0)
+{
+    static const size_t minToWrite = 2076;
+    static const size_t maxToWrite = 1345670;
+    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 int sleepRemainder = RAND_MAX - sleepThreshold;
+    static const unsigned minSleep = 10;
+    static const unsigned maxSleep = 250;
+    static const unsigned sleepRange = maxSleep - minSleep;
+
+    size_t numBytesRead = 0;
+    size_t nextReportRead = reportingInterval;
+    size_t numWriteBlocked = 0;
+    size_t numBytesWritten = 0;
+    size_t nextReportWritten = reportingInterval;
+    size_t numReadBlocked = 0;
+    size_t numMillisSlept = 0;
+
+    bool toQuit = false;
+
+    while(!toQuit) {
+        int r = rand_r(&seed);
+        size_t toWrite = minToWrite +
+            static_cast<size_t>( static_cast<double>(r) *
+                                 rangeToWrite / RAND_MAX );
+
+        while(toWrite>0) {
+            writingBuffer.reset();
+            size_t length = min(writingBuffer.getCapacity(), toWrite);
+            writingBuffer.addLength(length);
+
+            if (quitEvent!=0 && quitEvent->check()) {
+                toQuit = true;
+                break;
+            }
+
+            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;
+                }
+            } 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 (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;
+                }
+                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;
+            }
+
+            r = rand_r(&seed);
+            if (r>=sleepThreshold) {
+                unsigned toSleep = minSleep +
+                    static_cast<unsigned>(static_cast<double>(r - sleepThreshold) *
+                                          sleepRange / sleepRemainder);
+                assert(toSleep>=minSleep);
+                assert(toSleep<maxSleep);
+                Thread::sleep(toSleep);
+                numMillisSlept += toSleep;
+            }
+
+            if (length!=0) {
+                waiter.wait();
+                if (waiter.failed()) {
+                    printf("%s: waiting failed with error: %lu\n",
+                           prefix,
+                           static_cast<unsigned long>(waiter.getErrorCode()));
+                    toQuit = true;
+                    break;
+                }
+            }
+        }
+    }
+
+    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);
+}
 
 //------------------------------------------------------------------------------
@@ -56 +182 @@
 class ServerThread : public Thread
 {
+private:
+    Waiter waiter;
+
+    WaitableEvent event;
+
+public:
+    ServerThread();
+
+    void signal();
+
     virtual void run();
 };
@@ -61 +197 @@
 //------------------------------------------------------------------------------
 
+ServerThread::ServerThread() :
+    event(&waiter)
+{
+}
+
+//------------------------------------------------------------------------------
+
+void ServerThread::signal()
+{
+    event.fire();
+}
+
+//------------------------------------------------------------------------------
+
 void ServerThread::run()
 {
-    printf("ServerThread::run\n");
-
-    Waiter waiter;
+    unsigned seed = static_cast<unsigned>(time(0)*1.2);
+    printf("ServerThread::run: seed=%u\n", seed);
 
     LocalServerSocket serverSocket("test", &waiter);
     LocalAcceptor& acceptor = serverSocket.getAcceptor();
 
-    while(true) {
-        while(!acceptor.accept()) {
-            if (acceptor.failed()) {
-                printf("ServerThread::run: acceptor failed...\n");
-                return;
-            }
-            printf("ServerThread::run: waiting...\n");
-            waiter.wait();
+    while(!acceptor.accept()) {
+        if (acceptor.failed()) {
+            printf("ServerThread::run: acceptor failed...\n");
+            return;
         }
-
-        printf("ServerThread::run: waiting done, received connection\n");
-        LocalSocket* socket = acceptor.getSocket();
-        delete socket;
-        break;
+        printf("ServerThread::run: waiting...\n");
+        waiter.wait();
     }
+
+    printf("ServerThread::run: waiting done, received connection\n");
+    LocalSocket* socket = acceptor.getSocket();
+
+    communicate(socket->getReadingBuffer(), socket->getWritingBuffer(),
+                waiter, seed, "ServerThread", &event);
+
+
+    delete socket;
 }
 
@@ -100 +251 @@
 void ClientThread::run()
 {
+    unsigned seed = static_cast<unsigned>(time(0)*1.8);
+    printf("ClientThread::run: seed=%u\n", seed);
+
     printf("ClientThread::run: sleeping\n");
-    sleep(2000);
+    sleep(500);
     printf("ClientThread::run: connecting\n");
 
@@ -120 +274 @@
 
     printf("ClientThread::run: connected\n");
+
+    communicate(socket.getReadingBuffer(), socket.getWritingBuffer(),
+                waiter, seed, "ClientThread");
 }
 
@@ -127 +284 @@
 int main()
 {
+    signal(SIGPIPE, SIG_IGN);
+
     ServerThread serverThread;
     ClientThread clientThread;
@@ -133 +292 @@
     serverThread.start();
     clientThread.start();
+
+    Thread::sleep(60000);
+    serverThread.signal();
+    printf("Signalled the server thread\n");
 
     printf("Waiting for the client thread\n");