D8076: worker: Manually buffer reads from pickle stream

[Yuya Nishihara]

> +if pycompat.ispy3:
> +
> +    class _blockingreader(object):
> +        def __init__(self, wrapped):
> +            self._wrapped = wrapped
> +
> +        def __getattr__(self, attr):
> +            return getattr(self._wrapped, attr)
> +
> +        # issue multiple reads until size is fulfilled
> +        def read(self, size=-1):
> +            if size < 0:
> +                return self._wrapped.readall()
> +
> +            buf = bytearray(size)
> +            view = memoryview(buf)
> +            pos = 0
> +
> +            while pos < size:
> +                ret = self._wrapped.readinto(view[pos:])
> +                if not ret:
> +                    break
> +                pos += ret
> +
> +            del view
> +            del buf[pos:]
> +            return buf

Might be better to optimize the common case `wrapped.read(size) == size`.
FWIW, if we don't mind issuing extra `read()` syscalls, maybe we can abuse
BufferedReader of `buffer_size=1`.

Another option is to rewrite the select loop to fully manage response buffer
by ourselves.

```
for key, events in selector.select():
    ...
    our_buffer.extend(key.fileobj.read())
    temp_io = BytesIO(our_buffer)
    while ...:
        try:
            pickle.load(temp_io)
        except ...
            ...
    del our_buffer[:temp_io.tell()]
```