 ///
 /// If `forwardingSink` is non-nil, each incremental piece of data will be sent
 /// to it as data is received.
-private func aggregateDataReadFromPipe(pipe: Pipe, forwardingSink: SinkOf<NSData>?) -> SignalProducer<NSData, ReactiveTaskError> {
-	let readProducer = pipe.transferReadsToProducer()
+private func aggregateDataReadFromPipe<S: SchedulerType>(pipe: Pipe, forwardingSink: SinkOf<NSData>?, scheduler: S) -> SignalProducer<NSData, ReactiveTaskError> {
+	let readProducer = pipe.transferReadsToProducer() |> observeOn(scheduler)
 
 	return SignalProducer { observer, disposable in
 		let buffer = MutableBox<dispatch_data_t?>(nil)

	/// The file descriptor for writing data.
	let writeFD: Int32

	/// Creates an NSFileHandle corresponding to the `readFD`. The file handle
	/// will not automatically close the descriptor.
	var readHandle: NSFileHandle {

			}

			dispatch_io_set_low_water(channel, 1)
			dispatch_io_read(channel, 0, Int.max, self.queue) { (done, data, error) in
				if let data = data {
					sendNext(observer, data)
				}
