[Embedding] Change init value of version in EV from 0 to -1.

Signed-off-by: lixy9474 <lxy268263@alibaba-inc.com>

Author: lixy9474

tensorflow/core/framework/embedding/value_ptr.h

@@ -114,6 +114,7 @@ struct NormalHeader {
     memset(this, 0, sizeof(NormalHeader));
     meta.SetLayoutType(LayoutType::NORMAL);
     meta.SetHeaderSize(sizeof(NormalHeader) / sizeof(int64));
+    SetGlobalStep(-1);
   }
 
   inline int64 GetGlobalStep() {
@@ -157,6 +158,7 @@ struct FixedLengthHeader {
 
   FixedLengthHeader() {
     memset(this, 0, sizeof(FixedLengthHeader));
+    SetGlobalStep(-1);
   }
 
    inline int64 GetGlobalStep() {

tensorflow/core/kernels/embedding_variable_ops_test.cc

@@ -1782,6 +1782,54 @@ TEST(KVInterfaceTest, TestDirectIoFile) {
   TestReadEmbFile();
 }
 
+
+void InsertKey(EmbeddingVar<int64, float>* variable, int value_size) {
+  float *val = (float *)malloc((value_size+1)*sizeof(float));
+  for (int64 i = 0; i < 100000000; i++) {
+    variable->LookupOrCreate(20, val, nullptr);
+  }
+  LOG(INFO)<<"Finish Insert";
+}
+
+void RemoveKey(EmbeddingVar<int64, float>* variable) {
+  for (int64 i = 0; i < 10; i++) {
+    sleep(1);
+    variable->storage_manager()->Remove(20);
+  }
+  LOG(INFO)<<"Remove thread finish";
+}
+
+TEST(EmbeddingVariableTest, TestLookupRemoveConcurrency) {
+  int value_size = 10;
+  Tensor value(DT_FLOAT, TensorShape({value_size}));
+  test::FillValues<float>(&value, std::vector<float>(value_size, 10.0));
+  auto emb_config = EmbeddingConfig(
+      /*emb_index = */0, /*primary_emb_index = */0,
+      /*block_num = */1, /*slot_num = */0,
+      /*name = */"", /*steps_to_live = */0,
+      /*filter_freq = */2, /*max_freq = */999999,
+      /*l2_weight_threshold = */-1.0, /*layout = */"normal",
+      /*max_element_size = */0, /*false_positive_probability = */-1.0,
+      /*counter_type = */DT_UINT64);
+  auto storage_manager = new embedding::StorageManager<int64, float>(
+      "EmbeddingVar", embedding::StorageConfig());
+  auto var = new EmbeddingVar<int64, float>("EmbeddingVar",
+      storage_manager,
+      emb_config,
+      cpu_allocator());
+
+   var->Init(value, 1);
+   int thread_num = 5;
+   std::vector<std::thread> insert_threads(thread_num);
+   for (size_t i = 0 ; i < thread_num - 1; i++) {
+     insert_threads[i] = std::thread(InsertKey, var, value_size);
+   }
+   insert_threads[thread_num - 1] = std::thread(RemoveKey, var);
+   for (auto &t : insert_threads) {
+     t.join();
+   }
+ }
+
 } // namespace
 } // namespace embedding
 } // namespace tensorflow

tensorflow/python/ops/embedding_variable_ops_test.py

@@ -239,7 +239,7 @@ def testEmbeddingVariableForExport(self):
                             [1., 1., 1.],
                             [1., 1., 1.],
                             [1., 1., 1.]], fetches[1])
-        self.assertAllEqual([0, 0, 0, 0, 0, 0], fetches[2])
+        self.assertAllEqual([-1, -1, -1, -1, -1, -1], fetches[2])
         self.assertAllEqual([1, 1, 1, 1, 1, 1], fetches[3])
 
   def testEmbeddingVariableForGetShape(self):
@@ -2081,7 +2081,7 @@ def testEmbeddingVariableForGetFrequencyAndVersion(self):
       s, f, v = sess.run([shape, frequency, version])
       self.assertAllEqual(np.array([5,3]), s)
       self.assertAllEqual(np.array([3,1,2,0,2,0,1]), f)
-      self.assertAllEqual(np.array([2,0,1,0,2,0,2]), v)
+      self.assertAllEqual(np.array([2,0,1,-1,2,-1,2]), v)
 
   def testEmbeddingVariableForInference(self):
     print("testEmbeddingVariableForInference")

third_party/sparsehash_c11.patch

@@ -1,4 +1,4 @@
-From dcd5910ddff22b2896dc5221eab766a6c6d2307d Mon Sep 17 00:00:00 2001
+From 2d87035732d9ae347515b8497ca7a75d3b36f8cf Mon Sep 17 00:00:00 2001
 From: Tongxuan Liu <tongxuan.ltx@alibaba-inc.com>
 Date: Mon, 13 Mar 2023 08:58:56 +0800
 Subject: [PATCH] Avoid fetching nullptr when use feature filter.
@@ -8,14 +8,14 @@ Subject: [PATCH] Avoid fetching nullptr when use feature filter.
  sparsehash/dense_hash_map_lockless            |  447 ++++
  sparsehash/dense_hash_set_lockless            |  381 +++
  sparsehash/internal/densehashtable.h          |   16 +-
- sparsehash/internal/densehashtable_lockless.h | 2033 +++++++++++++++++
+ sparsehash/internal/densehashtable_lockless.h | 2035 +++++++++++++++++
  sparsehash/internal/hashtable-common.h        |    4 +
  sparsehash/internal/sparsehashtable.h         |   18 +-
  sparsehash/traits                             |   10 +-
  tests/bench_lockless.cc                       | 1466 ++++++++++++
  tests/dense_hash_map_unittests.cc             |  137 +-
  tests/rwlock.h                                |  224 ++
- 11 files changed, 4726 insertions(+), 23 deletions(-)
+ 11 files changed, 4728 insertions(+), 23 deletions(-)
  create mode 100644 sparsehash/dense_hash_map_lockless
  create mode 100644 sparsehash/dense_hash_set_lockless
  create mode 100644 sparsehash/internal/densehashtable_lockless.h
@@ -958,10 +958,10 @@ index e254126..3bc3c16 100644
      for (; dist > 0; --dist, ++f) {
 diff --git a/sparsehash/internal/densehashtable_lockless.h b/sparsehash/internal/densehashtable_lockless.h
 new file mode 100644
-index 0000000..64f677f
+index 0000000..2f8a80b
 --- /dev/null
 +++ b/sparsehash/internal/densehashtable_lockless.h
-@@ -0,0 +1,2033 @@
+@@ -0,0 +1,2035 @@
 +// Copyright (c) 2005, Google Inc.
 +// All rights reserved.
 +//
@@ -2204,9 +2204,11 @@ index 0000000..64f677f
 +      }else if(test_deleted(bucknum, tmp_pointer)) {
 +        if(insert_pos == ILLEGAL_BUCKET) insert_pos = bucknum;
 +      }else if (equals(key, get_key(tmp_pointer->table_[bucknum]))) {
-+        std::pair<K, T> tmp(tmp_pointer->table_[bucknum].first,
-+                            tmp_pointer->table_[bucknum].second);
-+        if(tmp.first == key){
++        //Force to read from volatile memory
++        volatile K* key_ptr = const_cast<K*>(&(tmp_pointer->table_[bucknum].first));
++        volatile T* value_ptr = const_cast<T*>(&(tmp_pointer->table_[bucknum].second));
++        std::pair<K, T> tmp(*key_ptr, *value_ptr);
++        if(*key_ptr == key){
 +          return tmp;
 +        }else{
 +          return std::pair<K,T>(tmp_pointer->key_info_.empty_key, empty_value);