Fixes false-postive when testing if pointer is a "VALID" shared memor…

…y piece. (#1078)


Signed-off-by: Tao He <[email protected]>

python/client.cc

@@ -625,7 +625,8 @@ void bind_client(py::module& mod) {
           "target"_a)
       .def("is_shared_memory",
            [](Client* self, const uintptr_t target) -> bool {
-             return self->IsSharedMemory(target);
+             ObjectID object_id = InvalidObjectID();
+             return self->IsSharedMemory(target, object_id);
            })
       .def("find_shared_memory",
            [](Client* self, const uintptr_t target) -> py::object {

python/vineyard/core/resolver.py

@@ -55,11 +55,11 @@ def run(self, obj, **kw):
                         value = resolver(obj, resolver=self)
                     else:
                         value = resolver(obj)
-                except Exception:
+                except Exception as e:
                     raise RuntimeError(  # pylint: disable=raise-missing-from
                         'Unable to construct the object using resolver: '
                         'typename is %s, resolver is %s' % (obj.meta.typename, resolver)
-                    )
+                    ) from e
             if value is None:
                 # if the obj has been resolved by pybind types, and there's no proper
                 # resolver, it shouldn't be an error

python/vineyard/data/tensor.py

@@ -68,22 +68,23 @@ def numpy_ndarray_builder(client, value, **kw):
 
 def numpy_ndarray_resolver(obj):
     meta = obj.meta
-    value_name = meta['value_type_']
-    if value_name == 'object':
+    value_type_name = meta['value_type_']
+    if value_type_name == 'object':
         view = memoryview(obj.member('buffer_'))
         return pickle.loads(view, fix_imports=True)
 
-    value_type = normalize_dtype(value_name, meta.get('value_type_meta_', None))
+    value_type = normalize_dtype(value_type_name, meta.get('value_type_meta_', None))
     shape = from_json(meta['shape_'])
     if 'order_' in meta:
         order = from_json(meta['order_'])
     else:
         order = 'C'
     if np.prod(shape) == 0:
         return np.zeros(shape, dtype=value_type)
-    c_array = np.frombuffer(
-        memoryview(obj.member('buffer_')), dtype=value_type
-    ).reshape(shape)
+    mem = memoryview(obj.member('buffer_'))[
+        0 : int(np.prod(shape)) * np.dtype(value_type).itemsize
+    ]
+    c_array = np.frombuffer(mem, dtype=value_type).reshape(shape)
     # TODO: revise the memory copy of asfortranarray
     array = c_array if order == 'C' else np.asfortranarray(c_array)
     return array.view(ndarray)

src/client/client.cc

@@ -494,19 +494,27 @@ std::vector<std::shared_ptr<Object>> Client::ListObjects(
 }
 
 bool Client::IsSharedMemory(const void* target) const {
-  return shm_->Exists(target);
+  ObjectID object_id = InvalidObjectID();
+  return IsSharedMemory(target, object_id);
 }
 
 bool Client::IsSharedMemory(const uintptr_t target) const {
-  return shm_->Exists(target);
+  ObjectID object_id = InvalidObjectID();
+  return IsSharedMemory(target, object_id);
 }
 
 bool Client::IsSharedMemory(const void* target, ObjectID& object_id) const {
-  return shm_->Exists(target, object_id);
+  return IsSharedMemory(reinterpret_cast<uintptr_t>(target), object_id);
 }
 
 bool Client::IsSharedMemory(const uintptr_t target, ObjectID& object_id) const {
-  return shm_->Exists(target, object_id);
+  if (shm_->Exists(target, object_id)) {
+    // verify that the blob is not deleted on the server side
+    json tree;
+    Client* mutable_this = const_cast<Client*>(this);
+    return mutable_this->GetData(object_id, tree, false, false).ok();
+  }
+  return false;
 }
 
 Status Client::AllocatedSize(const ObjectID id, size_t& size) {
@@ -1346,8 +1354,9 @@ Status SharedMemoryManager::Mmap(int fd, ObjectID id, int64_t map_size,
                                  uint8_t* pointer, bool readonly, bool realign,
                                  uint8_t** ptr) {
   RETURN_ON_ERROR(this->Mmap(fd, map_size, pointer, readonly, realign, ptr));
-  segments_.emplace(reinterpret_cast<uintptr_t>(*ptr) + data_offset,
-                    std::make_pair(data_size, id));
+  // override deleted blobs
+  segments_[reinterpret_cast<uintptr_t>(*ptr) + data_offset] =
+      std::make_pair(data_size, id);
   return Status::OK();
 }
 
@@ -1366,12 +1375,12 @@ void SharedMemoryManager::PreMmap(int fd, std::vector<int>& fds,
 }
 
 bool SharedMemoryManager::Exists(const uintptr_t target) {
-  ObjectID id;
+  ObjectID id = InvalidObjectID();
   return Exists(target, id);
 }
 
 bool SharedMemoryManager::Exists(const void* target) {
-  ObjectID id;
+  ObjectID id = InvalidObjectID();
   return Exists(target, id);
 }
 
@@ -1386,7 +1395,8 @@ bool SharedMemoryManager::Exists(const uintptr_t target, ObjectID& object_id) {
   std::clog << "[trace] pointer that been queried: "
             << reinterpret_cast<void*>(target) << std::endl;
   for (auto const& item : segments_) {
-    std::clog << "[trace] [" << reinterpret_cast<void*>(item.first) << ", "
+    std::clog << "[trace] " << ObjectIDToString(item.second.second) << ": ["
+              << reinterpret_cast<void*>(item.first) << ", "
               << reinterpret_cast<void*>(item.first + item.second.first) << ")"
               << std::endl;
   }
@@ -1420,7 +1430,11 @@ ObjectID SharedMemoryManager::resolveObjectID(const uintptr_t target,
                                               const uintptr_t key,
                                               const uintptr_t data_size,
                                               const ObjectID object_id) {
-  if (key <= target && target < key + data_size) {
+  // With a more strict constraint: the target pointer must be starts froms the
+  // given blob (key), as blob slicing is not supported yet.
+  //
+  // if (key <= target && target < key + data_size) {
+  if (key == target) {
 #if defined(WITH_VERBOSE)
     std::clog << "[trace] resuing blob " << ObjectIDToString(object_id)
               << " for pointer " << reinterpret_cast<void*>(target)

src/client/client.h

@@ -107,9 +107,11 @@ class SharedMemoryManager {
   // compute the set of fds that needs to `recv` from the server
   void PreMmap(int fd, std::vector<int>& fds, std::set<int>& dedup);
 
-  bool Exists(const uintptr_t target);
+  bool Exists(const uintptr_t target)
+      __attribute__((deprecated("Use Exists(target, object_id) instead.")));
 
-  bool Exists(const void* target);
+  bool Exists(const void* target)
+      __attribute__((deprecated("Use Exists(target, object_id) instead.")));
 
   bool Exists(const uintptr_t target, ObjectID& object_id);
 
@@ -638,7 +640,8 @@ class Client final : public BasicIPCClient,
    * Return true if the address (client-side address) comes from the vineyard
    * server.
    */
-  bool IsSharedMemory(const void* target) const;
+  bool IsSharedMemory(const void* target) const __attribute__((
+      deprecated("Use IsSharedMemory(target, object_id) instead.")));
 
   /**
    * Check if the given address belongs to the shared memory region.
@@ -648,7 +651,8 @@ class Client final : public BasicIPCClient,
    * Return true if the address (client-side address) comes from the vineyard
    * server.
    */
-  bool IsSharedMemory(const uintptr_t target) const;
+  bool IsSharedMemory(const uintptr_t target) const __attribute__((
+      deprecated("Use IsSharedMemory(target, object_id) instead.")));
 
   /**
    * Check if the given address belongs to the shared memory region.

src/client/client_base.cc

@@ -41,8 +41,9 @@ Status ClientBase::GetData(const ObjectID id, json& tree,
   RETURN_ON_ERROR(doWrite(message_out));
   json message_in;
   RETURN_ON_ERROR(doRead(message_in));
-  RETURN_ON_ERROR(ReadGetDataReply(message_in, tree));
-  return Status::OK();
+  auto status = ReadGetDataReply(message_in, tree);
+  return Status::Wrap(
+      status, "failed to get metadata for '" + ObjectIDToString(id) + "'");
 }
 
 Status ClientBase::GetData(const std::vector<ObjectID>& ids,

src/common/util/status.h

@@ -321,6 +321,14 @@ class VINEYARD_MUST_USE_TYPE Status {
   /// Return a success status
   inline static Status OK() { return Status(); }
 
+  /// Wrap a status with customized extra message
+  inline static Status Wrap(const Status& s, const std::string& message) {
+    if (s.ok()) {
+      return s;
+    }
+    return Status(s.code(), message + ": " + s.message());
+  }
+
   /// Return an error status for invalid data (for example a string that
   /// fails parsing).
   static Status Invalid() { return Status(StatusCode::kInvalid, ""); }