message: ring_reducer.cc errata:
  1. Block in the current (blockable) thread when pre-copying input to
  output rather than continuing in the callback which cannot block.
  2. Clear RingField array on exit to more promptly release Refs on
  output tensor buffer.
  3. Properly set the forward_from_array parameter in SubContext.
diff: @@ -157,21 +157,27 @@ void RingReducer::Run(StatusCallback done) {
  // we're not computing in-place on the input tensor.
  if ((input_ != output_) &&
  (DMAHelper::base(input_) != DMAHelper::base(output_))) {
  + // We are running in a blockable thread and the callback can't block so
  + // just wait here on the copy.
  + Notification note;
  CollectiveRemoteAccessLocal::MemCpyAsync(
  ctx_->input_device_context(0), ctx_->op_device_context(), device_,
  device_, ctx_->input_alloc_attr(0), ctx_->output_alloc_attr(0), input_,
  - output_, [this](const Status& s) {
  - if (!s.ok()) {
  - done_(s);
  - } else {
  - ContinueAfterInputCopy();
  - }
  + output_, [this, &note, &status](const Status& s) {
  + status.Update(s);
  + note.Notify();
  });
  - } else {
  - ContinueAfterInputCopy();
  + note.WaitForNotification();
  + if (!status.ok()) {
  + done_(status);
  + return;
  + }
  }
  + ContinueAfterInputCopy();
  }
  +// Note that this function is blocking and must not run in any thread
  +// which cannot be blocked.
  void RingReducer::ContinueAfterInputCopy() {
  AllocatorAttributes attr = ctx_->output_alloc_attr(0);
  ca_.reset(MakeCollectiveAdapter(output_, group_size_ * num_subdivs_,
  @@ -235,6 +241,7 @@ void RingReducer::Finish(bool ok) {
  mutex_lock l(status_mu_);
  s = status_;
  }
  + rfv_.clear(); // Give up Refs on output tensor.
  done_(s);
  }
  @@ -252,6 +259,7 @@ RingReducer::SubContext::SubContext(OpKernelContext* ctx,
  sub_params_.input_device_contexts = &sub_input_dc_;
  sub_params_.eigen_gpu_device = nullptr;
  sub_params_.ensure_eigen_gpu_device();
  + sub_params_.forward_from_array = &forward_from_;
  sub_ctx_ = new OpKernelContext(&sub_params_, 1);
  }
  