Implement timeout and key routing for Stream.partition#376
Implement timeout and key routing for Stream.partition#376martindurant merged 6 commits intopython-streamz:masterfrom
Conversation
|
Looks like flake8 check failed, but not on the files that I changed. Should I fix these? |
|
The callback flushing mechanism is flawed, data accumulates in pending callbacks, and this breaks backpressure. Will have to implement some kind of locking mechanism to prevent this. |
|
Nope, it isn't, I was just building my test pipeline in an incorrect way. Decribed in #378 |
Please do fix - it may be due to a new version of flake8 |
| self._retain_refs(metadata) | ||
| self._buffer.append(x) | ||
| key = self._get_key(x) | ||
| buffer = self._buffer[key] |
There was a problem hiding this comment.
Do we need to create new variables buffer and metadata_buffer here?
There was a problem hiding this comment.
cc: @jsmaupin for review because this PR is changing how metadata is being handled in partition.
There was a problem hiding this comment.
Do we need to create new variables buffer and metadata_buffer here?
We really don't have to, but since they're referenced multiple times, I figured it'd me more readable than self._metadata_buffer[key].
There was a problem hiding this comment.
Thanks for the cc. The metadata changes look okay to me.
| assert L == [(0, 2), (1, 3), (4, 6), (5, 7)] | ||
|
|
||
|
|
||
| def test_partition_size_one(): |
There was a problem hiding this comment.
What is the goal of this test?
There was a problem hiding this comment.
Here:
I didn't have self.n > 1 condition at first and self._callbacks[key].cancel() raised KeyError when n=1. So it's kind of an edge case test. If you think it's unnecessary, I'll remove it.
There was a problem hiding this comment.
I'm thinking if we should add a check in partition (and update docstring) forcing n>1. If n=1, the user would be better off avoiding the use of partition altogether. What do you think @martindurant?
There was a problem hiding this comment.
For the sake of test and debug, I think it's OK to allow a redundant partition with n=1.
|
Feel free to push an empty commit here to restart the CI |
Codecov Report
@@ Coverage Diff @@
## master #376 +/- ##
==========================================
+ Coverage 95.77% 95.96% +0.19%
==========================================
Files 16 16
Lines 2508 2527 +19
==========================================
+ Hits 2402 2425 +23
+ Misses 106 102 -4
Continue to review full report at Codecov.
|
4 participants
Resolves #375