Backport ponylang/ponyc#1578 and ponylang/ponyc#1626 to net actors

Backport ponylang/ponyc#1578 and ponylang/ponyc#1626 to net actors because the help resolve edge cases around the networking code with direct impacts to the stability of the network tests.
WallarooLabs · Sep 29, 2017 · 1f1c4a3 · 1f1c4a3
1 parent 1c0c77b
commit 1f1c4a3
Show file tree

Hide file tree

Showing 7 changed files with 29 additions and 6 deletions.
diff --git a/lib/wallaroo/core/boundary/boundary.pony b/lib/wallaroo/core/boundary/boundary.pony
@@ -454,9 +454,11 @@ actor OutgoingBoundary is Consumer
             _event = event
             _connected = true
             _writeable = true
+            _readable = true
 
             _notify.connected(this)
             _on_connected()
+            _pending_reads()
 
             ifdef not windows then
               if _pending_writes() then
@@ -485,6 +487,7 @@ actor OutgoingBoundary is Consumer
 
             _connected = true
             _writeable = true
+            _readable = true
 
             // set replaying to true since we might need to replay to
             // downstream before resuming
@@ -495,6 +498,7 @@ actor OutgoingBoundary is Consumer
             _shutdown_peer = false
 
             _notify.connected(this)
+            _pending_reads()
 
             try
               let connect_msg = ChannelMsgEncoder.data_connect(_worker_name,

diff --git a/lib/wallaroo/core/data_channel/data_channel.pony b/lib/wallaroo/core/data_channel/data_channel.pony
@@ -145,7 +145,10 @@ actor DataChannel
     _max_size = max_size
 
     _notify.accepted(this)
+
+    _readable = true
     _queue_read()
+    _pending_reads()
 
   be identify_data_receiver(dr: DataReceiver, sender_step_id: StepId) =>
     """
@@ -349,9 +352,11 @@ actor DataChannel
             _event = event
             _connected = true
             _writeable = true
+            _readable = true
 
             _notify.connected(this)
             _queue_read()
+            _pending_reads()
 
             // Don't call _complete_writes, as Windows will see this as a
             // closed connection.

diff --git a/lib/wallaroo/core/data_channel/data_channel_listener.pony b/lib/wallaroo/core/data_channel/data_channel_listener.pony
@@ -202,13 +202,13 @@ actor DataChannelListener
     _closed = true
 
     if not _event.is_null() then
-      @pony_os_socket_close[None](_fd)
-      _fd = -1
-
       // When not on windows, the unsubscribe is done immediately.
       ifdef not windows then
         @pony_asio_event_unsubscribe(_event)
       end
 
+      @pony_os_socket_close[None](_fd)
+      _fd = -1
+
       _notify.closed(this)
     end
diff --git a/lib/wallaroo/core/metrics/reconnecting_metrics_sink.pony b/lib/wallaroo/core/metrics/reconnecting_metrics_sink.pony
@@ -345,9 +345,11 @@ actor ReconnectingMetricsSink
             _event = event
             _connected = true
             _writeable = true
+            _readable = true
 
             _notify.connected(this)
             _queue_read()
+            _pending_reads()
 
             ifdef not windows then
               if _pending_writes() then
@@ -373,12 +375,14 @@ actor ReconnectingMetricsSink
 
             _connected = true
             _writeable = true
+            _readable = true
 
             _closed = false
             _shutdown = false
             _shutdown_peer = false
 
             _notify.connected(this)
+            _pending_reads()
 
             ifdef not windows then
               if _pending_writes() then

diff --git a/lib/wallaroo/core/sink/tcp_sink/tcp_sink.pony b/lib/wallaroo/core/sink/tcp_sink/tcp_sink.pony
@@ -317,6 +317,7 @@ actor TCPSink is Consumer
             _event = event
             _connected = true
             _writeable = true
+            _readable = true
 
             match _initializer
             | let initializer: LocalTopologyInitializer =>
@@ -327,6 +328,7 @@ actor TCPSink is Consumer
             end
 
             _notify.connected(this)
+            _pending_reads()
 
             for msg in _initial_msgs.values() do
               _writev(msg, None)
@@ -357,12 +359,14 @@ actor TCPSink is Consumer
 
             _connected = true
             _writeable = true
+            _readable = true
 
             _closed = false
             _shutdown = false
             _shutdown_peer = false
 
             _notify.connected(this)
+            _pending_reads()
 
             ifdef not windows then
               if _pending_writes() then

diff --git a/lib/wallaroo/core/source/tcp_source/tcp_source.pony b/lib/wallaroo/core/source/tcp_source/tcp_source.pony
@@ -133,6 +133,9 @@ actor TCPSource is Producer
     //listening until we are done recovering
     _notify.accepted(this)
 
+    _readable = true
+    _pending_reads()
+
     for consumer in routes.values() do
       _routes(consumer) =
         _route_builder(this, consumer, _metrics_reporter)
@@ -274,8 +277,11 @@ actor TCPSource is Producer
             _fd = fd
             _event = event
             _connected = true
+            _readable = true
 
             _notify.connected(this)
+
+            _pending_reads()
           else
             // The connection failed, unsubscribe the event and close.
             @pony_asio_event_unsubscribe(event)

diff --git a/lib/wallaroo/core/source/tcp_source/tcp_source_listener.pony b/lib/wallaroo/core/source/tcp_source/tcp_source_listener.pony
@@ -233,11 +233,11 @@ actor TCPSourceListener is SourceListener
     _closed = true
 
     if not _event.is_null() then
-      @pony_os_socket_close[None](_fd)
-      _fd = -1
-
       // When not on windows, the unsubscribe is done immediately.
       ifdef not windows then
         @pony_asio_event_unsubscribe(_event)
       end
+
+      @pony_os_socket_close[None](_fd)
+      _fd = -1
     end