Merge pull request #6958 from vespa-engine/havardpe/optimistic-write-in-fnet

Havardpe/optimistic write in fnet

2 files changed, 36 insertions, 15 deletions

diff --git a/fnet/src/vespa/fnet/transport_thread.cpp b/fnet/src/vespa/fnet/transport_thread.cpp
index b0388bdc140..2c0d00b22f3 100644
--- a/fnet/src/vespa/fnet/transport_thread.cpp
+++ b/fnet/src/vespa/fnet/transport_thread.cpp
@@ -151,6 +151,32 @@ FNET_TransportThread::DiscardEvent(FNET_ControlPacket *cpacket,
 }
 
 
+void
+FNET_TransportThread::handle_add_cmd(FNET_IOComponent *ioc)
+{
+    if (ioc->handle_add_event()) {
+        AddComponent(ioc);
+        ioc->_flags._ioc_added = true;
+        ioc->attach_selector(_selector);
+    } else {
+        ioc->Close();
+        AddDeleteComponent(ioc);
+    }
+}
+
+
+void
+FNET_TransportThread::handle_close_cmd(FNET_IOComponent *ioc)
+{
+    if (ioc->_flags._ioc_added) {
+        RemoveComponent(ioc);
+        ioc->SubRef();
+    }
+    ioc->Close();
+    AddDeleteComponent(ioc);
+}
+
+
 extern "C" {
 
     static void pipehandler(int)
@@ -423,14 +449,7 @@ FNET_TransportThread::handle_wakeup()
 
         switch (packet->GetCommand()) {
         case FNET_ControlPacket::FNET_CMD_IOC_ADD:
-            if (context._value.IOC->handle_add_event()) {
-                AddComponent(context._value.IOC);
-                context._value.IOC->_flags._ioc_added = true;
-                context._value.IOC->attach_selector(_selector);
-            } else {
-                context._value.IOC->Close();
-                AddDeleteComponent(context._value.IOC);
-            }
+            handle_add_cmd(context._value.IOC);
             break;
         case FNET_ControlPacket::FNET_CMD_IOC_ENABLE_READ:
             context._value.IOC->EnableReadEvent(true);
@@ -442,19 +461,18 @@ FNET_TransportThread::handle_wakeup()
             break;
         case FNET_ControlPacket::FNET_CMD_IOC_ENABLE_WRITE:
             context._value.IOC->EnableWriteEvent(true);
-            context._value.IOC->SubRef();
+            if (context._value.IOC->HandleWriteEvent()) {
+                context._value.IOC->SubRef();
+            } else {
+                handle_close_cmd(context._value.IOC);
+            }
             break;
         case FNET_ControlPacket::FNET_CMD_IOC_DISABLE_WRITE:
             context._value.IOC->EnableWriteEvent(false);
             context._value.IOC->SubRef();
             break;
         case FNET_ControlPacket::FNET_CMD_IOC_CLOSE:
-            if (context._value.IOC->_flags._ioc_added) {
-                RemoveComponent(context._value.IOC);
-                context._value.IOC->SubRef();
-            }
-            context._value.IOC->Close();
-            AddDeleteComponent(context._value.IOC);
+            handle_close_cmd(context._value.IOC);
             break;
         }
     }
diff --git a/fnet/src/vespa/fnet/transport_thread.h b/fnet/src/vespa/fnet/transport_thread.h
index 1b8d1fa4eeb..408d20619d2 100644
--- a/fnet/src/vespa/fnet/transport_thread.h
+++ b/fnet/src/vespa/fnet/transport_thread.h
@@ -143,6 +143,9 @@ private:
     FNET_Config *GetConfig() { return &_config; }
 
 
+    void handle_add_cmd(FNET_IOComponent *ioc);
+    void handle_close_cmd(FNET_IOComponent *ioc);
+
 public:
     /**
      * Construct a transport object. To activate your newly created