implementation using Parallel::DataPipe