Improve Cancellation to take advantage of InterruptedException

Previous patches wrapped InterruptedExceptions to be IOExceptions close sources and sinks, and did some tricks to propagate interrupted status. This pattern forces an couples an entire pipeline a single thread which is a scalability bottleneck. Also, by using interrupted exceptions, we can better provide the intent of an interruption in a type-safe way.