[BUG] race condition in local rendezvous
@kevint324 This race condition can be avoided by using a mutex or semaphore to protect the shared Rendezvous key. Proper synchronization technique should be used. Thank you! @sushreebarsa  
It's not easy to do it with mutex during runtime. You'll have to create-then-grab a mutex for each parsed key at SendOp::Compute and release corresponding mutex when RecvOp::ComputeAsync is done.

Also it doens't smell right that SEND/RECV should clean the mess that eager runtime created.

Eager context should create tensors with unique `tensor_name` so that SEND/RECV should not worry about parsed key confliction but it will affect lots of things.

~~Another "fix" is to create unique parsed key at the ctor of SendOp and RecvOp so that no send/recv pairs has the same parsed key globally.~~

```
--- a/tensorflow/core/kernels/sendrecv_ops.cc
+++ b/tensorflow/core/kernels/sendrecv_ops.cc
@@ -57,6 +57,26 @@ static FrameAndIter GetFrameAndIter(OpKernelContext* ctx,
   }
 }
 
+static absl::flat_hash_map<std::string, int> key_prefix_map;
+static absl::Mutex key_prefix_map_mutex;
+
+static std::string UniqueKeyPrefix(const std::string& key_prefix) {
+  absl::MutexLock lock(&key_prefix_map_mutex);
+
+  auto it = key_prefix_map.find(key_prefix);
+  if (it != key_prefix_map.end()) {
+    it->second++;
+    // For each SEND/RECV pair we need to use same key prefix.
+    if (it->second == 1) {
+      return key_prefix;
+    }
+    return absl::StrCat(key_prefix, "_", it->second / 2);
+  } else {
+    key_prefix_map[key_prefix] = 0;
+    return key_prefix;
+  }
+}
+
 SendOp::SendOp(OpKernelConstruction* ctx) : OpKernel(ctx) {
   string send_device;
   OP_REQUIRES_OK(ctx, ctx->GetAttr("send_device", &send_device));
@@ -70,6 +90,7 @@ SendOp::SendOp(OpKernelConstruction* ctx) : OpKernel(ctx) {
   OP_REQUIRES_OK(ctx, ctx->GetAttr("tensor_name", &tensor_name));
   key_prefix_ = GetRendezvousKeyPrefix(send_device, recv_device,
                                        send_device_incarnation, tensor_name);
+  key_prefix_ = UniqueKeyPrefix(key_prefix_);
   // The vast majority of Send nodes are outside any loop context, so
   // proactively cache the rendezvous key for the top-level.
   GetRendezvousKey(key_prefix_, {0, 0}, &parsed_key_.buf_);
@@ -151,6 +172,7 @@ RecvOp::RecvOp(OpKernelConstruction* ctx) : AsyncOpKernel(ctx) {
   OP_REQUIRES_OK(ctx, ctx->GetAttr("tensor_name", &tensor_name));
   key_prefix_ = GetRendezvousKeyPrefix(send_device, recv_device,
                                        send_device_incarnation, tensor_name);
+  key_prefix_ = UniqueKeyPrefix(key_prefix_);
   // The vast majority of Recv nodes are outside any loop context, so
   // proactively cache the rendezvous key for the top-level.
   GetRendezvousKey(key_prefix_, {0, 0}, &parsed_key_.buf_);
@@ -185,6 +207,8 @@ Rendezvous::DoneCallback make_recv_callback(OpKernelContext* ctx,
       // The runtime checks whether the tensor received here is
       // the same type.
       if (!is_dead) {
+        VLOG(1) << "make_recv_callback tensor ptr " << &val
+                << " shape " << val.shape().DebugString();
         ctx->set_output(0, val);
       }
     }
```

~~Still it makes send recv ops looks a little bit weird since it's clean someone else's mess but it has the minimal damage.If this looks good to you I'll file a PR.~~

Sorry, above patch won't work. Send/Recv nodes are instantiated out of order and it won't be easy to match correct send-recv pair with a global map. I think the right path is still try to fix the origin.

Thanks another possible fix is to manipulate the generation of `tensor name`

```diff
diff --git a/tensorflow/core/common_runtime/eager/execute.cc b/tensorflow/core/common_runtime/eager/execute.cc
index 28aa93a7bbd..328468ada31 100644
--- a/tensorflow/core/common_runtime/eager/execute.cc
+++ b/tensorflow/core/common_runtime/eager/execute.cc
@@ -890,10 +890,27 @@ Status WrapInCallOp(EagerOperation* op, EagerOperation** wrapped_op) {
     NodeDef* ndef = fdef.add_node_def();
     ndef->set_op(op->Name());
     ndef->set_name(op->Name());  // This could be anything.
+
+    static std::atomic<int64_t> unique_input_id = 0;
+    auto mutable_signature = fdef.mutable_signature();
+    for (size_t i = 0; i < mutable_signature->input_arg_size(); i++) {
+      const std::string& original_name =
+        mutable_signature->mutable_input_arg(i)->name();
+      int64_t uid = unique_input_id.fetch_add(1);
+      mutable_signature->mutable_input_arg(i)->set_name(
+        absl::StrCat(original_name, "_", uid));
+    }
+
     const auto& signature = fdef.signature();
     for (size_t i = 0; i < signature.input_arg_size(); i++) {
       ndef->add_input(absl::StrCat(fdef.signature().input_arg(i).name(), ":0"));
     }
+
```

 @SuryanarayanaY @sushreebarsa 
I filled a PR to avoid the key conflict.

https://github.com/tensorflow/tensorflow/pull/62568

 Thanks @SuryanarayanaY @rohan100jain @gbaned 

https://colab.research.google.com/drive/1CUiyuG2Aob-Fj8xllrmbx2w3cNBXFYqU?usp=sharing

Here is a minimum case derived from the [DLRM from NV DLE](https://github.com/NVIDIA/DeepLearningExamples/tree/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM) 

Click run all and you can reproduce the error in a few seconds.

 We are currently using TF 2.13.0 and I can confirm a segfault due to a race condition in local rendezvous.
When inspecting the `core` file with gdb, this is the obtained stack trace:
```
(gdb) bt full
#0  0x00007f9441158cf0 in tensorflow::LocalRendezvous::~LocalRendezvous() () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#1  0x00007f94414dff89 in tensorflow::RefCountedIntraProcessRendezvous::~RefCountedIntraProcessRendezvous() () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#2  0x00007f94414e00d2 in tensorflow::RefCountedIntraProcessRendezvous::~RefCountedIntraProcessRendezvous() () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#3  0x00007f9441467808 in std::_Function_handler<void (absl::lts_20230125::Status const&), tensorflow::ProcessFunctionLibraryRuntime::ApplyCleanUpToDoneCallback(std::vector<std::unique_ptr<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem, std::default_delete<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem> >, std::allocator<std::unique_ptr<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem, std::default_delete<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem> > > >*, std::function<void (absl::lts_20230125::Status const&)>, tensorflow::FunctionLibraryRuntime::Options const&, tsl::core::RefCountPtr<tensorflow::Rendezvous>) const::{lambda(absl::lts_20230125::Status const&)#1}>::_M_invoke(std::_Any_data const&, absl::lts_20230125::Status const&) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#4  0x00007f94414600a3 in std::_Function_handler<void (absl::lts_20230125::Status const&), tensorflow::ProcessFunctionLibraryRuntime::Run(tensorflow::FunctionLibraryRuntime::Options const&, unsigned long, absl::lts_20230125::Span<tensorflow::Tensor const>, std::vector<tensorflow::Tensor, std::allocator<tensorflow::Tensor> >*, std::function<void (absl::lts_20230125::Status const&)>) const::{lambda(absl::lts_20230125::Status const&)#1}>::_M_invoke(std::_Any_data const&, absl::lts_20230125::Status const&) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#5  0x00007f944145f7bc in tensorflow::ReffedStatusCallback::~ReffedStatusCallback() () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#6  0x00007f9441467f34 in tensorflow::ProcessFunctionLibraryRuntime::RunMultiDeviceAsync(tensorflow::FunctionLibraryRuntime::Options const&, unsigned long, std::vector<std::variant<tensorflow::Tensor, tensorflow::TensorShape>, std::allocator<std::variant<tensorflow::Tensor, tensorflow::TensorShape> > >*, std::vector<std::unique_ptr<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem, std::default_delete<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem> >, std::allocator<std::unique_ptr<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem, std::default_delete<tensorflow::ProcessFunctionLibraryRuntime::CleanUpItem> > > >*, std::function<void (absl::lts_20230125::Status const&)>, std::function<absl::lts_20230125::Status (tensorflow::ProcessFunctionLibraryRuntime::ComponentFunctionData const&, tensorflow::ProcessFunctionLibraryRuntime::InternalArgs*)>) const::{lambda(absl::lts_20230125::Status const&)#2}::operator()(absl::lts_20230125::Status const&) const () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#7  0x00007f9441461570 in std::_Function_handler<void (absl::lts_20230125::Status const&), tensorflow::(anonymous namespace)::TensorsToFunctionRetsDoneCallback(std::vector<std::variant<tensorflow::Tensor, tensorflow::TensorShape>, std::allocator<std::variant<tensorflow::Tensor, tensorflow::TensorShape> > >*, std::vector<tensorflow::Tensor, std::allocator<tensorflow::Tensor> >*, std::function<void (absl::lts_20230125::Status const&)>)::{lambda(absl::lts_20230125::Status const&)#1}>::_M_invoke(std::_Any_data const&, absl::lts_20230125::Status const&) () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#8  0x00007f944144db50 in std::_Function_handler<void (absl::lts_20230125::Status const&), tensorflow::FunctionLibraryRuntimeImpl::Run(tensorflow::FunctionLibraryRuntime::Options const&, unsigned long, absl::lts_20230125::Span<tensorflow::Tensor const>, std::vector<tensorflow::Tensor, std::allocator<tensorflow::Tensor> >*, std::function<void (absl::lts_20230125::Status const&)>)::{lambda(absl::lts_20230125::Status const&)#3}>::_M_invoke(std::_Any_data const&, absl::lts_20230125::Status const&) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#9  0x00007f9441512455 in std::_Function_handler<void (), tensorflow::(anonymous namespace)::ExecutorState<tensorflow::PropagatorState>::Finish()::{lambda()#4}>::_M_invoke(std::_Any_data const&) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#10 0x00007f9441afe121 in std::_Function_handler<void (), std::_Bind<tensorflow::data::RunnerWithMaxParallelism(std::function<void (std::function<void ()>)>, int)::{lambda(std::function<void (std::function<void ()>)> const&, std::function<void ()>)#1}::operator()(std::function<void (std::function<void ()>)> const&, std::function<void ()>) const::{lambda(std::function<void ()> const&)#1} (std::function<void ()>)> >::_M_invoke(std::_Any_data const&) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#11 0x00007f94406eb645 in Eigen::ThreadPoolTempl<tsl::thread::EigenEnvironment>::WorkerLoop(int) () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#12 0x00007f94406ed2b8 in void absl::lts_20230125::internal_any_invocable::RemoteInvoker<false, void, tsl::thread::EigenEnvironment::CreateThread(std::function<void ()>)::{lambda()#1}&>(absl::lts_20230125::internal_any_invocable::TypeErasedState*) ()
   from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#13 0x00007f9441344fcf in tsl::(anonymous namespace)::PThread::ThreadFn(void*) () from /usr/local/lib/python3.10/dist-packages/tensorflow/python/platform/../../libtensorflow_framework.so.2
No symbol table info available.
#14 0x00007f94ec8d4ac3 in start_thread (arg=<optimized out>) at ./nptl/pthread_create.c:442
        ret = <optimized out>
        pd = <optimized out>
        out = <optimized out>
        unwind_buf = {cancel_jmp_buf = {{jmp_buf = {140732530118736, 6006152095181854869, 140224407516736, 0, 140277600569296, 140732530119088, -6031673960749836139, -6020192743340746603}, mask_was_saved = 0}}, priv = {pad = {0x0, 0x0, 0x0, 0x0}, data = {prev = 0x0, 
              cleanup = 0x0, canceltype = 0}}}
        not_first_call = <optimized out>
#15 0x00007f94ec965bf4 in epoll_wait (epfd=-699346607, events=0x7f88940327e0, maxevents=-938782704, timeout=-938782712) at ../sysdeps/unix/sysv/linux/epoll_wait.c:30
        __arg4 = -938782712
        __arg2 = 140224575514592
        _a3 = -938782704
        _a1 = -699346607
        resultvar = <optimized out>
        __arg3 = -938782704
        __arg1 = -699346607
        _a4 = 0
        _a2 = 140224575514592
        sc_ret = <optimized out>
        sc_ret = <optimized out>
#16 0x0000000000000000 in ?? ()
No symbol table info available.
```

### Issue type

Bug

### Have you reproduced the bug with TensorFlow Nightly?

No

### TensorFlow version

2.10, 2.12, doesn't really matter

### Custom code

Yes

### OS platform and distribution

ubuntu 20.04

### Python version

3.8, 3.9

### Bazel version

5.3.0

### GCC/compiler version

9.4

### Current behavior?

The story is kinda bit long and it took us months to debug this issue. I'll try to keep it short.

### Background and what the problem is
We run the [DLRM from NV DLE](https://github.com/NVIDIA/DeepLearningExamples/tree/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM) in our CI daily test. 

Very rare, like once every few weeks, the daily test report such error.

`
W tensorflow/core/framework/op_kernel.cc:1874] OP_REQUIRES failed at strided_slice_op.cc:112 : INVALID_ARGUMENT: Expected begin, end, and strides to be 1D equal size tensors, but got shapes [65536], [1], and [1] instead.
`

The[ strided slice op validator](https://github.com/tensorflow/tensorflow/blob/3a029b19c9c156cd68cab671b5ce95bde839f15e/tensorflow/core/util/strided_slice_op.cc#L214) isn't happy about the input parameters.

The problem is like a ghost. It appears once every 1 or 2 weeks which makes it very hard to debug. But one thing is for sure is that it's not a random dram bit flip caused by cosmic rays because the error systoms is stable: one of (begin, end, stride) tensor's shape is incorrectly set to [65535] instead of [1]

### Evidence

I'll skip the lengthy debug process and jump to the last step.

![image](https://github.com/tensorflow/tensorflow/assets/8800468/d73ddf6d-7cc9-4a10-90f8-1e4d3885c91b)
![image](https://github.com/tensorflow/tensorflow/assets/8800468/0b99c252-59b0-490d-8321-924adb70be72)


We observed a malfunctioned SEND/RECV pair.

The item tensor in SENDOP has a shape of [1] but during the local rendezvous process the recevied tensor got a shape of [65536]. This tensor later flows to strided slice op and triggers the grumpy validator.

![image](https://github.com/tensorflow/tensorflow/assets/8800468/7626ef2f-0de4-4eb0-8c25-3cfd9618b359)

We grep the key hash (red highlighted 1632...9087) in the debug log.  There are two threads: 66787 and 63997. Most of the time the rendezvous runs in a SEND-RECV pattern within the same thread. 

But in the green rectangle you can see two threads interleaved and a SEND-SEND-RECV-RECV pattern is observed. Soon after this the bomb exploded.

### Analysis

It seems the root cause is a bug in the rendezvous mechanism. Two unrelated operations generated the same communication Rendezvous key. 

Assuming there are 2 threads, when the CPU load is not heavy, in most cases the scheduling order is

- THREAD A: SEND(KEY, VALUE_A)

- THREAD A: RECV(KEY)

- THREAD B: SEND(KEY, VALUE_B)

- THREAD B: RECV(KEY)

In this way, everyone will be fine even if the keys are the same. It is equivalent to time-division multiplexing of the same KEY.

When the CPU load becomes heavy (which is the case in our daily test scenario), thread scheduling becomes unpredictable. It is possible that such a pattern may occur

- THREAD A: SEND(KEY, VALUE_A)

- THREAD B: SEND(KEY, VALUE_B)

- THREAD A: RECV(KEY)

- THREAD B: RECV(KEY)

This will cause THREAD A to incorrectly receive the data sent by THREAD B.


### Related python code

1. Thread A is running the normal [training loop.](https://github.com/NVIDIA/DeepLearningExamples/blob/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM/main.py#L281)
2. Thread B is created for the RawBinaryDataset class. The pre-processing creates an [asynchronous thread pool.](https://github.com/NVIDIA/DeepLearningExamples/blob/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM/split_binary_dataset.py#L138)
3. Thread A is running[ x = x[self.begin_idx:self.end_idx]](https://github.com/NVIDIA/DeepLearningExamples/blob/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM/model.py#L58). Translated into StridedSlice operation by eager runtime. begin_idx, end_idx and the implicit stride are all constant tensors on CPU, and StridedSlice is an device operator. So these three tensors need to be sent to the device side.
4. Thread B is running[ tensor = tf.expand_dims(tensor, axis=1). ](https://github.com/NVIDIA/DeepLearningExamples/blob/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM/split_binary_dataset.py#L204). Execute expand_dims. Similar to thread A, the expand_dims operator runs on the device side while the input is on CPU, and the tensor needs to be sent to the device. The shape of this tensor is [65536].
5. Due to the defect of Rendezous::CreateKey, the keys generated by these two operators in the eager runtime are exactly the same.
6. Multi-threading + HASH KEY collision + SEND-SEND-RECV-RECV, all three together, BOOOM.

### FIX

![image](https://github.com/tensorflow/tensorflow/assets/8800468/f50ed76a-9cef-4bf7-87b2-e4a298236dd9)

The key string is 
/job:localhost/replica:0/task:0/device:CPU:0;ea74dbce35f0ab7e;/job:localhost/replica:0/task:0/device:MLU:0;edge_2_input;0:0

The rendezvous key consists 5 parts.
- src_device: /job:localhost/replica:0/task:0/device:CPU:0;
- src_incarnation: ea74dbce35f0ab7e;
- dst_device: /job:localhost/replica:0/task:0/device:MLU:0;
- name: edge_2_input;
- frame_inter: 0:0

In this case,  4 out of 5 (src_device, src_incarnation, dst_device_frame_iter) are naturely indistinguishable.

Unless we add new field in the key, the only field we can play with is `name`.

The creation of `edge_2_input` involes another few tons of code.

![image](https://github.com/tensorflow/tensorflow/assets/8800468/b2d85cd7-1e47-492b-be50-1e3d63f13d55)

A quick fix for my current problem is simply add `dst_name` in the `tensor_name_attr` during graph partition. The names will become `edge_2_input_strided_slice` and `edge_2_input_expand_dims` thus my problem is solved.

```diff
diff --git a/tensorflow/core/graph/graph_partition.cc b/tensorflow/core/graph/graph_partition.cc
index a4f09383c63..57a4e919526 100644
--- a/tensorflow/core/graph/graph_partition.cc
+++ b/tensorflow/core/graph/graph_partition.cc
@@ -1147,7 +1147,8 @@ Status Partition(const PartitionOptions& opts, Graph* g,
         tensor_name_attr = opts.get_tensor_name_attr(edge);
       } else {
         tensor_name_attr =
-            strings::StrCat("edge_", edge->id(), "_", edge->src()->name());
+            strings::StrCat("edge_", edge->id(), "_", edge->src()->name(),
+            "_", edge->dst()->name());
       }
```

But a more approriate and generic fix might be to have a unique src node name. The source name will be like input_xxxxxx and input_yyyyyy. But this sounds like a fundamental change and I'm not sure if this would break too many things. And I'm not sure about the right place to make the change, like manipulating the input name a little bit [here](https://github.com/tensorflow/tensorflow/blob/master/tensorflow/core/common_runtime/eager/execute.cc#L988).

I'd like to hear your opinion on this problem and I'd like to file a PR if you could point me the right place to apply the fix.

Cheers
Hengwen

### Standalone code to reproduce the issue

A minimum case derived from the [DLRM from NV DLE](https://github.com/NVIDIA/DeepLearningExamples/tree/3f82b7f982a1d9a9c5392bd110a5b93dfa7e80eb/Tensorflow2/Recommendation/DLRM) 

Click run all and you can reproduce the error in a few seconds.

https://colab.research.google.com/drive/1CUiyuG2Aob-Fj8xllrmbx2w3cNBXFYqU?usp=sharing




