Rework BatchedSend logic #661

pitrou · 2016-11-14T11:54:04Z

Does away with the timeout and looking up a private attribute on IOStream.
Refs PR #653.

Does away with the timeout and looking up a private attribute on IOStream. Refs PR dask#653.

mrocklin · 2016-11-14T12:50:33Z

distributed/batched.py

+            except Exception:
+                logger.exception("Error in batched write")
+                break
+            self.next_deadline = self.loop.time() + self.interval


We might want to base the next deadline on when we started the last send rather than when we finsihed it.

Well, I don't know. What are the intended semantics?

This class generally tries to solve the problem of streams on which we want to send thousands of small messages per second, such as might occur in the following situation:

for x in range(10000): future = client.submit(inc, x) futures.append(future)

(or on the worker to scheduler side, as the worker reports status updates)

We found that these situations were significantly faster if we never sent two messages within a few milliseconds of each other, preferring instead to batch them. If it has been more than a few milliseconds since the last payload was dispatched and the last payload has finished then I think we should be able to send again.

Then the yield write(...) wasn't really useful in the previous version? There's no need to wait on the write future if we want to base the deadline on the start of the write operation.

Related question: what is with gen.with_timeout(timedelta(seconds=0.01),...) in d.core.write?

That is related to what was going on here and is also a possible source of error. Some explanation here: #653

Oh, right. We can remove it.

mrocklin · 2016-11-14T12:54:41Z

distributed/batched.py

+                if self.next_deadline is not None:
+                    delay = self.next_deadline - self.loop.time()
+                    if delay > 0:
+                        yield gen.sleep(delay)


Why this added delay?

It mirrors the yield self.last_send that was here previously. Perhaps I'm misunderstanding the intent :-)

mrocklin · 2016-11-14T12:55:12Z

distributed/tests/test_batched.py

@@ -37,7 +37,7 @@ def handle_stream(self, stream, address):
                self.count += 1
                yield write(stream, msg)
            except StreamClosedError as e:
-                pass
+                return


I would expect this to be a syntax error in Python 2

No, only return with an explicit value is forbidden. Bare return allows exiting the coroutine.

mrocklin · 2016-11-14T12:55:47Z

This looks pretty nice to me

pitrou · 2016-11-14T13:57:08Z

By the way, BatchedStream doesn't seem used anymore, perhaps we should remove it?

mrocklin · 2016-11-14T13:59:07Z

Removing it sounds fine to me. Generally I am happy to yield to your judgment on anything related to this issue. I suspect that you have a lot more experience here than I do.

mrocklin · 2016-11-14T14:33:03Z

Testing failures here are unrelated. Addressing them in #662 .

mrocklin · 2016-11-14T15:21:56Z

distributed/core.py

-                break
-            except gen.TimeoutError:
-                pass
+    yield stream.write(frames[-1])


I think that this could fail if we write to the same stream in another coroutine. Ideally we shouldn't do this. Normally the rpc class creates new streams as necessary to handle concurrent communications to the same destination. All cases that I can find when a coroutine writes directly to a stream it creates and owns that stream exclusively.

Still though, we were running into problems in the wild that suggested that this might be an issue.

I'm not sure how that's different from the old code, though? It would also wait on futures[-1] and only catch timeout errors.

We would raise a timeout error if the future didn't complete quickly and then fall back to checking if the stream's write_buffer was empty.

I'm a bit surprised that this would make a difference. What were the symptoms of the problems?

Somewhere some coroutine is stuck waiting on yield write(...). This first occurred on yield self.last_write within BatchedSend, and resulted in messages waiting in the worker's message buffer.

You're right. That's because IOStream.write can forget previous futures. So how about we don't wait for the write at all? We could simply yield gen.moment so that write() remains a coroutine...

I think we would want to yield on the write if we were going to apply backpressure. We're not doing this yet at other stages though so yes, I suspect that that would work fine. We're moving the data pile-up from the BatchedSend buffer to the Tornado write buffer, which is probably appropriate anyway.

Turns out we must wait for the write() to be issued before closing the stream. This is gonna be a bit hairy...

More or less hairy than polling on the _write_buffer?

I think the solution is to flush the stream explicitly before closing. Let me try it out.

…ched_send

@mrocklin

@mrocklin

mrocklin · 2016-11-14T17:31:29Z

distributed/core.py

+    This is recommended before closing the stream.
+    """
+    if stream.writing():
+        yield stream.write(b'')


How do we know that this will complete?

The API expects that write() isn't called before flush() completes.

Are we confident that this expectation is fulfilled?

I've updated the docstring to better inform the reader. But perhaps we only want to expose close() so that we don't do any further mistakes. What do you think?

I'm mostly concerned about us as users. Only very advanced dask/distributed users should use read/write/close directly.

However, given that we've had problems reported it's possible that we aren't handling everything well internally.

I have no problem with read/write/close as an API generally.

mrocklin · 2016-11-14T17:31:51Z

distributed/core.py

+    """
+    if not stream.closed():
+        try:
+            flush(stream)


Should we yield on this?

Yes, you're right, my bad.

mrocklin · 2016-11-15T12:02:50Z

It would be good to develop some tests to stress communication in a few ways. However, I'm not entirely sure how this needs to be stressed. One thing that I've found to be useful in the past is to change 300000 to 1000000 in distributed/tests/test_batched.py::test_sending_traffic_jam.

mrocklin · 2016-11-16T19:33:42Z

Thoughts on ignoring the RuntimeError around the h5py test?

mrocklin · 2016-11-16T19:51:27Z

This all seems fine to me. +1

pitrou · 2016-11-16T20:46:06Z

Thoughts on ignoring the RuntimeError around the h5py test?

I would hope h5py merges the pull request that would fix the issue.

mrocklin · 2016-11-16T20:47:11Z

I do not expect h5py to merge or release quickly.

Rework BatchedSend logic

d8897a2

Does away with the timeout and looking up a private attribute on IOStream. Refs PR dask#653.

mrocklin mentioned this pull request Nov 14, 2016

Use SerializableLock in hdf5 test #662

Merged

mrocklin reviewed Nov 14, 2016

View reviewed changes

Address review comments

1f931c0

pitrou force-pushed the batched_send branch from 0cbed81 to 1f931c0 Compare November 14, 2016 13:52

Remove BatchedStream

e746b32

mrocklin reviewed Nov 14, 2016

View reviewed changes

pitrou added 3 commits November 14, 2016 17:14

Remove fragile wait on IOStream.write()

6b6c492

Merge branch 'master' of https://github.com/dask/distributed into bat…

1a2e190

…ched_send

Flush streams before closing to avoid losing messages.

ac1fbb1

@mrocklin

mrocklin reviewed Nov 14, 2016

View reviewed changes

pitrou added 3 commits November 14, 2016 18:51

Yield on the flush() call

05b7c92

Avoid exposing close()

eb6afe8

Fix name clash with close()

b0e92f0

Add a large traffic jam test

a22650d

pitrou merged commit 3f1cc73 into dask:master Nov 16, 2016

gjoseph92 mentioned this pull request Nov 4, 2021

Do not drop BatchedSend payload if worker reconnects #5457

Closed

1 task

Rework BatchedSend logic #661

Rework BatchedSend logic #661

Conversation

pitrou commented Nov 14, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pitrou Nov 14, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mrocklin commented Nov 14, 2016

pitrou commented Nov 14, 2016

mrocklin commented Nov 14, 2016

mrocklin commented Nov 14, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mrocklin commented Nov 15, 2016

mrocklin commented Nov 16, 2016

mrocklin commented Nov 16, 2016

pitrou commented Nov 16, 2016

mrocklin commented Nov 16, 2016

pitrou Nov 14, 2016 •

edited