Rewrite nile_Funnel to work with large input streams.
Rewrite nile_Funnel to work with large input streams.

Before we'd deadlock on large input streams because work_until_below
was waiting for the stage after Funnel to unclog (input.n to drop),
but that stage was sometimes waiting for Funnel to remove it
(because it was swapped), producing deadlock.

Now the work_until_below waits for the Funnel to change state
away from BLOCKED_ON_CONSUMER.

damelang/nile,damelang/nile,damelang/nile,damelang/nile,damelang/nile,damelang/nile