The idea is to avoid creating task in majority of the cases, e.g.:

In the stream __init__:

   self._nop = self.loop.create_future()
   self._nop.set_result(None)

and then in _fast_drain:

   if not self._paused:
        return self._nop

But I do almost the same a few lines above.
There is no writer._paused attribute but writer._protocol._paused one (which has undefined state if the protocol is closing).
That's why I have slightly more comlex logic that you mentioned in a comment but the functionality is the same.

Ah, I see, never mind. Could you please add a comment before return self._complete_fut that it's a "fast path"?

Done