Fix slowdown in throttle due to improper use of stepLeg

[mpilquist]

See this Scastie, provided on Discord: https://scastie.scala-lang.org/mn98/oNhMT2wZR0O42S6lN0lUMQ/8

https://discord.com/channels/632277896739946517/632310980449402880/950382134110081084

When the tick resolution argument of throttle is 100 millis, the output occurs at the correct speed. When the tick resolution is changed to 10 millis, the outputs get progressively slower.

[diesalbla]

@mpilquist It may help to add in the commit message, amended, a note about what was causing the slowdown in the old code that is not causing it in the new one. Was it an allocation and retention of not strictly necessary objects? Was this allocation growing over time running the throttle?

Do we have any way of measuring or testing for slow-downs, just as we have for memory leaks? For instance, if the cause of the slowdown was a growing number of objects being allocated and retained , would running a throttle in a small heap highlight it?

[mpilquist]

I'm not sure actually. The problem was caused by converting the StepLeg to a stream on each step, which the docs explicitly say not to do:

fs2/core/shared/src/main/scala/fs2/Stream.scala

Lines 4837 to 4843 in 3f40b8c

	/** Converts this leg back to regular stream. Scope is updated to the scope associated with this leg.
	* Note that when this is invoked, no more interleaving legs are allowed, and this must be very last
	* leg remaining.
	*
	* Note that resulting stream won't contain the `head` of this leg.
	*/
	def stream: Stream[F, O] =

I didn't work through exactly what was accumulating though.

We don't currently have a way to measure for this type of slowdown.

[diesalbla]

We don't currently have a way to measure for this type of slowdown.

Is there a snippet that we may try running with ammonite, so we may manually observe that slowdown? The PR description does not link to any actual issue for it.

[mpilquist]

Ha, whoops! It was reported on discord. I updated the description of this PR.

[mn98]

Here's my original snippet that I shared with Michael and I first discussed on discord: https://scastie.scala-lang.org/mn98/oNhMT2wZR0O42S6lN0lUMQ/11

[diesalbla]

I have tried locally the given snippet. On the main branch, it only emits elements up to 14. With the new branch, it emits up to 20. So this is good.