Decouple stream and instruction

oneflow/api/python/functional/tensor_api.cpp

@@ -288,8 +288,10 @@ class LocalTensorSharedNumpyDataFunctor {
 
     // Init blob
     JUST(tensor_impl->InitEagerBlobObject(NewLocalDepObject(), /*pin_memory=*/false));
-    const auto& stream = GetDefaultStreamByDevice(device);
-    JUST(tensor_impl->eager_blob_object())->set_last_used_stream(stream);
+    const auto& stream = JUST(GetDefaultStreamByDevice(device));
+    const auto& eager_blob_object = JUST(tensor_impl->eager_blob_object());
+    JUST(eager_blob_object->init_producer_stream(stream));
+    eager_blob_object->set_last_used_stream(stream);
     std::shared_ptr<Tensor> out(new MirroredTensor(tensor_impl));
     return out;
   }

oneflow/core/boxing/slice_boxing_util.h

@@ -18,6 +18,7 @@ limitations under the License.
 
 #include "oneflow/core/framework/tensor.h"
 #include "oneflow/core/framework/placed_nd_sbp.h"
+#include "oneflow/core/job/parallel_desc.h"
 
 namespace oneflow {
 

oneflow/core/common/add_and_read_vector.h

@@ -0,0 +1,89 @@
+/*
+Copyright 2020 The OneFlow Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#ifndef ONEFLOW_CORE_COMMON_ADD_AND_READ_VECTOR_H_
+#define ONEFLOW_CORE_COMMON_ADD_AND_READ_VECTOR_H_
+
+#include <vector>
+#include <array>
+#include <mutex>
+#include <cmath>
+#include <glog/logging.h>
+
+namespace oneflow {
+
+// `at` is lock free
+template<typename T, int N = 20>
+class AddAndReadVector {
+ public:
+  AddAndReadVector() : size_(0) {}
+  ~AddAndReadVector() = default;
+
+  using value_type = T;
+  using size_type = size_t;
+
+  // not thread safe.
+  size_t size() const { return size_; }
+
+  // lock free.
+  const T& at(size_t index) const {
+    CHECK_GE(index, 0);
+    CHECK_LT(index, size_);
+    int gran = GetGranularity(index + 1);
+    return granularity2vector_[gran].data()[index];
+  }
+
+  // lock free.
+  T& at(size_t index) {
+    CHECK_GE(index, 0);
+    CHECK_LT(index, size_);
+    int gran = GetGranularity(index + 1);
+    return granularity2vector_[gran].data()[index];
+  }
+
+  void push_back(const T& elem) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    int granularity = GetGranularity(size_);
+    if (granularity2vector_[granularity].size() == (1 << granularity)) {
+      int next_granularity = granularity + 1;
+      CHECK_LT(next_granularity, N);
+      CHECK_EQ(next_granularity, GetGranularity(size_ + 1));
+      granularity2vector_[next_granularity].reserve(1 << next_granularity);
+      granularity2vector_[next_granularity] = granularity2vector_[granularity];
+      granularity = next_granularity;
+    } else if (granularity2vector_[granularity].size() > (1 << granularity)) {
+      LOG(FATAL) << "fatal bug in AddAndReadVector::push_back";
+    } else {
+      // do nothing.
+    }
+    auto* vec = &granularity2vector_[granularity];
+    vec->push_back(elem);
+    size_ = vec->size();
+  }
+
+ private:
+  static int GetGranularity(size_t size) {
+    if (size == 0) { return 0; }
+    return std::ceil(std::log2(size));
+  }
+
+  std::atomic<size_t> size_;
+  std::mutex mutex_;
+  std::array<std::vector<T>, N> granularity2vector_;
+};
+
+}  // namespace oneflow
+
+#endif  // ONEFLOW_CORE_COMMON_ADD_AND_READ_VECTOR_H_

oneflow/core/common/add_and_read_vector_test.cpp

@@ -0,0 +1,199 @@
+/*
+Copyright 2020 The OneFlow Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#include "gtest/gtest.h"
+#include "oneflow/core/common/add_and_read_vector.h"
+
+namespace oneflow {
+namespace test {
+
+TEST(AddAndReadVector, simple0) {
+  AddAndReadVector<int> vec;
+  ASSERT_EQ(vec.size(), 0);
+}
+
+TEST(AddAndReadVector, simple1) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+}
+
+TEST(AddAndReadVector, simple2) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+}
+
+TEST(AddAndReadVector, simple3) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+}
+
+TEST(AddAndReadVector, simple4) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+}
+
+TEST(AddAndReadVector, simple5) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+  vec.push_back(4);
+  ASSERT_EQ(vec.at(4), 4);
+  ASSERT_EQ(vec.size(), 5);
+}
+
+TEST(AddAndReadVector, simple6) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+  vec.push_back(4);
+  ASSERT_EQ(vec.at(4), 4);
+  ASSERT_EQ(vec.size(), 5);
+  vec.push_back(5);
+  ASSERT_EQ(vec.at(5), 5);
+  ASSERT_EQ(vec.size(), 6);
+}
+
+TEST(AddAndReadVector, simple7) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+  vec.push_back(4);
+  ASSERT_EQ(vec.at(4), 4);
+  ASSERT_EQ(vec.size(), 5);
+  vec.push_back(5);
+  ASSERT_EQ(vec.at(5), 5);
+  ASSERT_EQ(vec.size(), 6);
+  vec.push_back(6);
+  ASSERT_EQ(vec.at(6), 6);
+  ASSERT_EQ(vec.size(), 7);
+}
+
+TEST(AddAndReadVector, simple8) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+  vec.push_back(4);
+  ASSERT_EQ(vec.at(4), 4);
+  ASSERT_EQ(vec.size(), 5);
+  vec.push_back(5);
+  ASSERT_EQ(vec.at(5), 5);
+  ASSERT_EQ(vec.size(), 6);
+  vec.push_back(6);
+  ASSERT_EQ(vec.at(6), 6);
+  ASSERT_EQ(vec.size(), 7);
+  vec.push_back(7);
+  ASSERT_EQ(vec.at(7), 7);
+  ASSERT_EQ(vec.size(), 8);
+}
+
+TEST(AddAndReadVector, simple9) {
+  AddAndReadVector<int> vec;
+  vec.push_back(0);
+  ASSERT_EQ(vec.at(0), 0);
+  ASSERT_EQ(vec.size(), 1);
+  vec.push_back(1);
+  ASSERT_EQ(vec.at(1), 1);
+  ASSERT_EQ(vec.size(), 2);
+  vec.push_back(2);
+  ASSERT_EQ(vec.at(2), 2);
+  ASSERT_EQ(vec.size(), 3);
+  vec.push_back(3);
+  ASSERT_EQ(vec.at(3), 3);
+  ASSERT_EQ(vec.size(), 4);
+  vec.push_back(4);
+  ASSERT_EQ(vec.at(4), 4);
+  ASSERT_EQ(vec.size(), 5);
+  vec.push_back(5);
+  ASSERT_EQ(vec.at(5), 5);
+  ASSERT_EQ(vec.size(), 6);
+  vec.push_back(6);
+  ASSERT_EQ(vec.at(6), 6);
+  ASSERT_EQ(vec.size(), 7);
+  vec.push_back(7);
+  ASSERT_EQ(vec.at(7), 7);
+  ASSERT_EQ(vec.size(), 8);
+  vec.push_back(8);
+  ASSERT_EQ(vec.at(8), 8);
+  ASSERT_EQ(vec.size(), 9);
+}
+
+}  // namespace test
+}  // namespace oneflow

oneflow/core/common/container_util.h

@@ -49,15 +49,15 @@ template<typename VecT>
 Maybe<scalar_or_const_ref_t<typename VecT::value_type>> VectorAt(const VecT& vec,
                                                                  typename VecT::size_type index) {
   CHECK_LT_OR_RETURN(index, vec.size());
-  return vec[index];
+  return vec.at(index);
 }
 
 template<typename VecT>
 Maybe<typename VecT::value_type&> VectorAt(VecT& vec, typename VecT::size_type index) {
   static_assert(!std::is_same<typename VecT::value_type, bool>::value,
                 "VectorAt(vector<bool>&, size_t) is not supported.");
   CHECK_LT_OR_RETURN(index, vec.size());
-  return vec[index];
+  return vec.at(index);
 }
 
 template<>

oneflow/core/common/device_type.proto

@@ -5,5 +5,5 @@ enum DeviceType {
   kInvalidDevice = 0;
   kCPU = 1;
   kCUDA = 2;
-  kMockDevice = 3;
+  kMockDevice = 3; // pseudo device for test.
 }