- further streamline I/O blocking

src/libfibre/EventScope.h

@@ -43,8 +43,8 @@ class EventScope {
     SyncSem RD;
     SyncSem WR;
 #if TESTING_LAZY_FD_REGISTRATION
-    FibreMutex regLock;
-    size_t status;
+    FibreMutex lock;
+    size_t     status;
     SyncFD() : status(0) {}
 #endif
   } *fdSyncVector;
@@ -142,7 +142,7 @@ public:
     RASSERT0(fd >= 0 && fd < fdCount);
     SyncFD& fdsync = fdSyncVector[fd];
     if ((fdsync.status & target) == target) return; // outside of lock: faster, but double regs possible...
-    ScopedLock<FibreMutex> sl(fdsync.regLock);
+    ScopedLock<FibreMutex> sl(fdsync.lock);
     bool change = fdsync.status;                    // already registered for polling?
     fdsync.status |= target;
 #endif
@@ -167,7 +167,7 @@ public:
     RASSERT0(fdsync.RD.empty());
     RASSERT0(fdsync.WR.empty());
 #if TESTING_LAZY_FD_REGISTRATION
-    ScopedLock<FibreMutex> sl(fdsync.regLock);
+    ScopedLock<FibreMutex> sl(fdsync.lock);
     fdsync.status = 0;
 #endif
     if (RemoveFromPollSet) {                        // only called from lfConnect
@@ -259,14 +259,14 @@ public:
 #if TESTING_LAZY_FD_REGISTRATION
     registerFD<Input,!Input,false,false>(fd);
 #endif
-    Fibre::yield();
     SyncSem& sem = Input ? fdSyncVector[fd].RD : fdSyncVector[fd].WR;
+    sem.P_yield();
     for (;;) {
-      sem.P();
       ret = iofunc(fd, a...);
-      if (ret >= 0 || !TestEAGAIN<Input>()) break;
+      if (ret >= 0) { sem.V(); break; }
+      if (!TestEAGAIN<Input>()) break;
+      sem.P();
     }
-    sem.V();
     return ret;
   }
 };

src/runtime/BlockingSync.h

@@ -224,7 +224,7 @@ public:
     while (!empty()) Pause();     // wait for timed out events to disappear
   }
   template<typename Lock>
-  bool block(Lock& lock, bool wait) {
+  bool block(Lock& lock, bool wait = true) {
     if (wait) {
       BlockingInfo<Lock> bi(lock);
       DBG::outl(DBG::Level::Blocking, "Stack ", FmtHex(Context::CurrStack()), " blocking on ", FmtHex(&queue));
@@ -250,9 +250,6 @@ public:
     return false;
   }
 
-  template<typename Lock>
-  bool block(Lock& lock) { return block(lock, true); }
-
   template<bool Enqueue = true>
   StackContext* unblock() {       // not concurrency-safe; better hold lock
     for (StackContext* s = queue.front(); s != queue.edge(); s = BlockedStackList::next(*s)) {
@@ -276,12 +273,13 @@ protected:
   ssize_t counter;
   BQ bq;
 
-  template<typename... Args>
+  template<bool Yield = false, typename... Args>
   bool internalP(const Args&... args) {
     // need baton passing: counter unchanged, if blocking fails (timeout)
     if (counter < 1) return bq.block(lock, args...);
     counter -= 1;
     lock.release();
+    if (Yield) StackContext::yield();
     return true;
   }
 
@@ -307,7 +305,12 @@ public:
   bool P_unlock(Lock2& l) {
     lock.acquire();
     l.release();
-    return internalP(true);
+    return internalP();
+  }
+
+  bool P_yield() {
+    lock.acquire();
+    return internalP<true>();
   }
 
   void P_fake(size_t c = 1) {
@@ -561,7 +564,7 @@ public:
       lock.release();
       return true;
     } else {
-      bq.block(lock, true);
+      bq.block(lock);
       return false;
     }
   }
@@ -574,7 +577,7 @@ class Condition {
 public:
   bool empty() { return bq.empty(); }
   void reset(Lock& lock) { bq.reset(lock); }
-  bool wait(Lock& lock) { return bq.block(lock, true); }
+  bool wait(Lock& lock) { return bq.block(lock); }
   bool wait(Lock& lock, const Time& timeout) { return bq.block(lock, timeout); }
   template<bool Broadcast = false>
   void signal() { while (bq.unblock() && Broadcast); }