Concurrent producer/consumer processing for co