Add socket source #227

martindurant · 2019-03-13T15:13:26Z

Could be used, for example, with python logging-to-socket

martindurant · 2019-03-13T15:36:14Z

I would also like to make a HTTP server source, i.e., one event for each incoming GET, say, containing the URL parameters and data. How, though, do you tell a tornado Application or Server to start on a given event loop, or otherwise get it to run in the right place?

CJ-Wright · 2019-03-13T15:39:40Z

I'm not certain, I'd like this as well, since I've been looking at making a ZMQ source.

martindurant · 2019-03-13T16:00:03Z

Of course, this fails on kafka, as all the PRs seem to now. If #226 is to be merged, should we simply drop the unbatched kafka sources, because they cause us so much trouble? or, @skmatti , might your reuse of confluent objects help those too?

martindurant · 2019-03-13T18:10:50Z

@CJ-Wright OK, solved the tornado server too.

martindurant · 2019-03-13T18:21:01Z

@mrocklin , can you review, and we should come to an agreement on how to move ahead regarding the stalling kafka tests.

skmatti · 2019-03-13T19:05:51Z

I think unbatched kafka source failures could be fixed by this commit from #216.

mrocklin

Some comments/questions

mrocklin · 2019-03-13T15:39:41Z

streamz/tests/test_sources.py

+        time.sleep(0.02)
+        assert l == []
+        sock.send(b'\n')
+        time.sleep(0.02)


I've found that fixed time sleeps like this can be brittle, especially on travis-ci, where a stray GC call may take more than 20ms (or a full second if you're unlucky).

Typically rather than sleeps I do something like

start = time.time() while not l: time.sleep(0.01 ) assert time.time() < start + 2

Or, even better, you could find some way to trigger when things show up.

mrocklin · 2019-03-13T15:40:10Z

streamz/tests/test_sources.py

+        sock2.close()
+        sock.close()
+
+    assert l == [b'data\n', b'data\n', b'data2\n']


It might be nice also to have an async test, if you're comfortable with the async API

mrocklin · 2019-03-16T02:55:19Z

streamz/sources.py

+            self.sock.bind(("", self.port))
+            self.sock.listen(128)
+            self.loop.add_handler(self.sock.fileno(), self.connection_ready,
+                                  self.loop.READ)


I'm somewhat surprised to see the use of raw sockets here. I'm curious, was there a reason to go this route rather than create a Tornado IOStream here directly? I see that you use them later on.

I don't know a ton about managing nonblocking sockets manually, I'm probably just missing something.

This code is adapted from the tornado IOLoop documentation: as far as I understand it, you do not have any IOStream until you accept an incoming connection on the open socket.

Ah, maybe the right abstraction here then is a Tornado TCPServer?

I didn't remove the socket server verion yet, so that thhe two could be compared

martindurant · 2019-03-17T20:41:06Z

@mrocklin , I added a tornado.TCPServer version of the same. I could just remove the direct socket version, but I am not convinced that either is better than the other.

codecov-io · 2019-03-20T18:05:30Z

Codecov Report

Merging #227 into master will increase coverage by 0.14%.
The diff coverage is 96.92%.

@@            Coverage Diff             @@
##           master     #227      +/-   ##
==========================================
+ Coverage   93.38%   93.53%   +0.14%     
==========================================
  Files          13       13              
  Lines        1481     1546      +65     
==========================================
+ Hits         1383     1446      +63     
- Misses         98      100       +2

Impacted Files	Coverage Δ
streamz/sources.py	`95.31% <96.92%> (+0.54%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 090b9ba...f453a96. Read the comment docs.

mrocklin · 2019-03-20T18:32:16Z

I could just remove the direct socket version, but I am not convinced that either is better than the other

Yeah, I don't know either. I don't know non-blocking sockets that well. I tend to just let Tornado/asyncio handle it.

martindurant · 2019-03-20T19:03:03Z

I tend to just let Tornado/asyncio handle it.

I strongly suspect that the two forms are actually identical, so it might just be a matter of choice - or leave both.

mrocklin · 2019-03-20T19:21:37Z

I strongly suspect that the two forms are actually identical, so it might just be a matter of choice - or leave both.

I recommend that we stick with TCPServer if that's ok.

Any comments on the async test and timing questions above?

martindurant · 2019-03-20T19:36:46Z

Any comments on the async test and timing questions above?

Sorry, missed that. I added an async test, and changed sleeps to (a)wait_for.

mrocklin

Thanks for working on this. A few minor comments/requests. Hopefully nothing major though.

mrocklin · 2019-03-21T01:14:49Z

streamz/sources.py

+        """Shutdown HTTP server"""
+        if not self.stopped:
+            self.server.stop()
+            self.server = None


You remove the server here, but not in the TCP solution. Is there a reason?

mrocklin · 2019-03-21T01:15:21Z

streamz/sources.py

+        application = Application([
+            (self.path, Handler),
+        ])
+        self.server = HTTPServer(application)


I can imagine wanting to pass in keyword arguments to the constructor and have them percolate to here (same with TCPServer) . I'm not sure though.

mrocklin · 2019-03-21T01:16:20Z

streamz/tests/test_sources.py

+    finally:
+        s.stop()
+
+    wait_for(lambda: out == [b'data\n', b'data\n', b'data2\n'], 2, period=0.01)


Do we need to clean up the sockets here? Does this leak file descriptors?

mrocklin · 2019-03-21T01:18:48Z

streamz/tests/test_sources.py

+    s = Source.from_http_server(port)
+    out = s.sink_to_list()
+    s.start()
+    time.sleep(0.02)  # allow loop to run


Is this necessary? If so, why this time?

In general, I'll probably challenge any sleep like this. If it's necessary to make things run then it's likely too short in all cases (GC on travis can take arbitrarily long times) and so results in intermittent failures. I spend a non-trivial time tracking these things down in the dask/distributed codebase, so I've grown fairly allergic to them.

- Removed sleeps - Closed sockets at test end - Allowed for kwargs to pass to tornado servers (not tested - don't know what parameters may be reasonable here) - remove tcp server instance on stop; not necessary, but reasonable

mrocklin · 2019-03-21T15:30:20Z

streamz/sources.py

+    def __init__(self, port, path='/.*', start=False, server_kwargs=None):
        self.port = port
        self.path = path
+        self.server_kwargs = server_kwargs or {}


In one case you use tcp_kwargs and in the other server_kwargs. Maybe use server_kwargs in both places for consistency of the API?

mrocklin · 2019-03-21T20:24:04Z

Thanks for this @martindurant . Should be fun to play with !

Add docket source

4272cf7

Add HTTP server source

4a5ea0f

martindurant mentioned this pull request Mar 13, 2019

to_kafka stream #228

Closed

mrocklin reviewed Mar 16, 2019

View reviewed changes

Same, but with TCPServer

72cfcac

I didn't remove the socket server verion yet, so that thhe two could be compared

Martin Durant added 5 commits March 19, 2019 12:27

Import for older tornado

e28ce59

Merge branch 'master' into sockets

6d33f49

flake

0421353

add small sleeps to run loop

f3436d7

flake again

7b2f3c0

fix coverage

dc3d801

remove from_socket; add async test

b76aa14

mrocklin reviewed Mar 21, 2019

View reviewed changes

Responses to feedback

e116639

- Removed sleeps - Closed sockets at test end - Allowed for kwargs to pass to tornado servers (not tested - don't know what parameters may be reasonable here) - remove tcp server instance on stop; not necessary, but reasonable

mrocklin reviewed Mar 21, 2019

View reviewed changes

Martin Durant added 2 commits March 21, 2019 11:34

tcp_kwargs to server_kwargs

8b798ac

Move wait to becfore shutdown

f453a96

mrocklin merged commit 0b93c94 into python-streamz:master Mar 21, 2019

martindurant deleted the sockets branch March 21, 2019 23:20

martindurant mentioned this pull request Mar 26, 2019

ENH: Add streaming HTTP/TCP #200

Closed

Add socket source #227

Add socket source #227

Uh oh!

Conversation

martindurant commented Mar 13, 2019

Uh oh!

martindurant commented Mar 13, 2019

Uh oh!

CJ-Wright commented Mar 13, 2019

Uh oh!

martindurant commented Mar 13, 2019

Uh oh!

martindurant commented Mar 13, 2019

Uh oh!

martindurant commented Mar 13, 2019

Uh oh!

skmatti commented Mar 13, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mrocklin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

martindurant commented Mar 17, 2019

Uh oh!

codecov-io commented Mar 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

mrocklin commented Mar 20, 2019

Uh oh!

martindurant commented Mar 20, 2019

Uh oh!

mrocklin commented Mar 20, 2019

Uh oh!

martindurant commented Mar 20, 2019

Uh oh!

mrocklin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrocklin commented Mar 21, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

skmatti commented Mar 13, 2019 •

edited

Loading

codecov-io commented Mar 20, 2019 •

edited

Loading