GEODE-10300: C++ native client: Allow locator responses greater than … (#970)

* GEODE-10300: Fix locator response size limit in C++ client

If a response message from the locator to the C++ native client
is longer than 3000 bytes the C++ native client will only
read the first 3000 bytes.

* GEODE-10300: Updated after review

* GEODE-10300: Updated after review

* GEODE-10300: Updated after another review

* GEODE-10300: Updated after some more reviews

* GEODE-10300: Some more changes after review.

* GEODE-10300: Small change after review

* GEODE-10300: Remove unneeded space

Author: albertogpz

cppcache/include/geode/DataInput.hpp

@@ -73,7 +73,7 @@ class APACHE_GEODE_EXPORT DataInput {
    */
   inline bool readBoolean() {
     _GEODE_CHECK_BUFFER_SIZE(1);
-    return *(m_buf++) == 1 ? true : false;
+    return *(buffer_++) == 1 ? true : false;
   }
 
   /**
@@ -89,8 +89,8 @@ class APACHE_GEODE_EXPORT DataInput {
   inline void readBytesOnly(uint8_t* buffer, size_t len) {
     if (len > 0) {
       _GEODE_CHECK_BUFFER_SIZE(len);
-      std::memcpy(buffer, m_buf, len);
-      m_buf += len;
+      std::memcpy(buffer, buffer_, len);
+      buffer_ += len;
     }
   }
 
@@ -107,8 +107,8 @@ class APACHE_GEODE_EXPORT DataInput {
   inline void readBytesOnly(int8_t* buffer, size_t len) {
     if (len > 0) {
       _GEODE_CHECK_BUFFER_SIZE(len);
-      std::memcpy(buffer, m_buf, len);
-      m_buf += len;
+      std::memcpy(buffer, buffer_, len);
+      buffer_ += len;
     }
   }
 
@@ -129,8 +129,8 @@ class APACHE_GEODE_EXPORT DataInput {
     if (length > 0) {
       _GEODE_CHECK_BUFFER_SIZE(length);
       _GEODE_NEW(buffer, uint8_t[length]);
-      std::memcpy(buffer, m_buf, length);
-      m_buf += length;
+      std::memcpy(buffer, buffer_, length);
+      buffer_ += length;
     }
     *bytes = buffer;
   }
@@ -152,8 +152,8 @@ class APACHE_GEODE_EXPORT DataInput {
     if (length > 0) {
       _GEODE_CHECK_BUFFER_SIZE(length);
       _GEODE_NEW(buffer, int8_t[length]);
-      std::memcpy(buffer, m_buf, length);
-      m_buf += length;
+      std::memcpy(buffer, buffer_, length);
+      buffer_ += length;
     }
     *bytes = buffer;
   }
@@ -173,10 +173,10 @@ class APACHE_GEODE_EXPORT DataInput {
    */
   inline int32_t readInt32() {
     _GEODE_CHECK_BUFFER_SIZE(4);
-    int32_t tmp = *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
+    int32_t tmp = *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
     return tmp;
   }
 
@@ -186,14 +186,14 @@ class APACHE_GEODE_EXPORT DataInput {
   inline int64_t readInt64() {
     _GEODE_CHECK_BUFFER_SIZE(8);
     int64_t tmp;
-    tmp = *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
-    tmp = (tmp << 8) | *(m_buf++);
+    tmp = *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
+    tmp = (tmp << 8) | *(buffer_++);
     return tmp;
   }
 
@@ -396,33 +396,33 @@ class APACHE_GEODE_EXPORT DataInput {
    * as readonly and modification of contents using this internal pointer
    * has undefined behavior.
    */
-  inline const uint8_t* currentBufferPosition() const { return m_buf; }
+  inline const uint8_t* currentBufferPosition() const { return buffer_; }
 
   /** get the number of bytes read in the buffer */
-  inline size_t getBytesRead() const { return m_buf - m_bufHead; }
+  inline size_t getBytesRead() const { return buffer_ - bufferHead_; }
 
   /** get the number of bytes remaining to be read in the buffer */
   inline size_t getBytesRemaining() const {
-    return (m_bufLength - getBytesRead());
+    return (bufferLength_ - getBytesRead());
   }
 
   /** advance the cursor by given offset */
-  inline void advanceCursor(size_t offset) { m_buf += offset; }
+  inline void advanceCursor(size_t offset) { buffer_ += offset; }
 
   /** rewind the cursor by given offset */
-  inline void rewindCursor(size_t offset) { m_buf -= offset; }
+  inline void rewindCursor(size_t offset) { buffer_ -= offset; }
 
   /** reset the cursor to the start of buffer */
-  inline void reset() { m_buf = m_bufHead; }
+  inline void reset() { buffer_ = bufferHead_; }
 
   inline void setBuffer() {
-    m_buf = currentBufferPosition();
-    m_bufLength = getBytesRemaining();
+    buffer_ = currentBufferPosition();
+    bufferLength_ = getBytesRemaining();
   }
 
-  inline void resetPdx(size_t offset) { m_buf = m_bufHead + offset; }
+  inline void resetPdx(size_t offset) { buffer_ = bufferHead_ + offset; }
 
-  inline size_t getPdxBytes() const { return m_bufLength; }
+  inline size_t getPdxBytes() const { return bufferLength_; }
 
   static uint8_t* getBufferCopy(const uint8_t* from, size_t length) {
     uint8_t* result;
@@ -432,7 +432,7 @@ class APACHE_GEODE_EXPORT DataInput {
     return result;
   }
 
-  inline void reset(size_t offset) { m_buf = m_bufHead + offset; }
+  inline void reset(size_t offset) { buffer_ = bufferHead_ + offset; }
 
   uint8_t* getBufferCopyFrom(const uint8_t* from, size_t length) {
     uint8_t* result;
@@ -452,19 +452,19 @@ class APACHE_GEODE_EXPORT DataInput {
   DataInput& operator=(DataInput&&) = default;
 
  protected:
+  const uint8_t* buffer_;
+  const uint8_t* bufferHead_;
+  size_t bufferLength_;
+  Pool* pool_;
+  const CacheImpl* cache_;
+
   /** constructor given a pre-allocated byte array with size */
   DataInput(const uint8_t* buffer, size_t len, const CacheImpl* cache,
             Pool* pool);
 
   virtual const SerializationRegistry& getSerializationRegistry() const;
 
  private:
-  const uint8_t* m_buf;
-  const uint8_t* m_bufHead;
-  size_t m_bufLength;
-  Pool* m_pool;
-  const CacheImpl* m_cache;
-
   std::shared_ptr<Serializable> readObjectInternal(int8_t typeId = -1);
 
   template <typename mType>
@@ -502,21 +502,21 @@ class APACHE_GEODE_EXPORT DataInput {
 
   inline char readPdxChar() { return static_cast<char>(readInt16()); }
 
-  inline void _checkBufferSize(size_t size, int32_t line) {
-    if ((m_bufLength - (m_buf - m_bufHead)) < size) {
+  virtual void _checkBufferSize(size_t size, int32_t line) {
+    if ((bufferLength_ - (buffer_ - bufferHead_)) < size) {
       throw OutOfRangeException(
           "DataInput: attempt to read beyond buffer at line " +
           std::to_string(line) + ": available buffer size " +
-          std::to_string(m_bufLength - (m_buf - m_bufHead)) +
+          std::to_string(bufferLength_ - (buffer_ - bufferHead_)) +
           ", attempted read of size " + std::to_string(size));
     }
   }
 
-  inline int8_t readNoCheck() { return *(m_buf++); }
+  inline int8_t readNoCheck() { return *(buffer_++); }
 
   inline int16_t readInt16NoCheck() {
-    int16_t tmp = *(m_buf++);
-    tmp = static_cast<int16_t>((tmp << 8) | *(m_buf++));
+    int16_t tmp = *(buffer_++);
+    tmp = static_cast<int16_t>((tmp << 8) | *(buffer_++));
     return tmp;
   }
 
@@ -605,7 +605,7 @@ class APACHE_GEODE_EXPORT DataInput {
     value.assign(reinterpret_cast<const wchar_t*>(tmp.data()), tmp.length());
   }
 
-  Pool* getPool() const { return m_pool; }
+  Pool* getPool() const { return pool_; }
 
   friend Cache;
   friend CacheImpl;

cppcache/src/Connector.hpp

@@ -116,6 +116,11 @@ class Connector {
    */
   virtual uint16_t getPort() = 0;
 
+  /**
+   * Returns the remote endpoint for this connection in the form host:port
+   */
+  virtual std::string getRemoteEndpoint() = 0;
+
   /**
    * Writes an array of a known size to the underlying output stream.
    *

cppcache/src/DataInput.cpp

@@ -30,21 +30,21 @@ namespace client {
 
 DataInput::DataInput(const uint8_t* buffer, size_t len, const CacheImpl* cache,
                      Pool* pool)
-    : m_buf(buffer),
-      m_bufHead(buffer),
-      m_bufLength(len),
-      m_pool(pool),
-      m_cache(cache) {}
+    : buffer_(buffer),
+      bufferHead_(buffer),
+      bufferLength_(len),
+      pool_(pool),
+      cache_(cache) {}
 
 std::shared_ptr<Serializable> DataInput::readObjectInternal(int8_t typeId) {
   return getSerializationRegistry().deserialize(*this, typeId);
 }
 
 const SerializationRegistry& DataInput::getSerializationRegistry() const {
-  return *m_cache->getSerializationRegistry();
+  return *cache_->getSerializationRegistry();
 }
 
-Cache* DataInput::getCache() const { return m_cache->getCache(); }
+Cache* DataInput::getCache() const { return cache_->getCache(); }
 
 template <class _Traits, class _Allocator>
 void DataInput::readJavaModifiedUtf8(
@@ -63,7 +63,7 @@ void DataInput::readJavaModifiedUtf8(
   uint16_t length = readInt16();
   _GEODE_CHECK_BUFFER_SIZE(length);
   value = internal::JavaModifiedUtf8::decode(
-      reinterpret_cast<const char*>(m_buf), length);
+      reinterpret_cast<const char*>(buffer_), length);
   advanceCursor(length);
 }
 template APACHE_GEODE_EXPLICIT_TEMPLATE_EXPORT void

cppcache/src/GetAllServersResponse.cpp

@@ -21,10 +21,10 @@ namespace geode {
 namespace client {
 
 void GetAllServersResponse::toData(DataOutput& output) const {
-  int32_t numServers = static_cast<int32_t>(m_servers.size());
-  output.writeInt(numServers);
-  for (int32_t i = 0; i < numServers; i++) {
-    output.writeObject(m_servers.at(i));
+  auto numServers = servers_.size();
+  output.writeInt(static_cast<int32_t>(numServers));
+  for (unsigned int i = 0; i < numServers; i++) {
+    output.writeObject(servers_.at(i));
   }
 }
 void GetAllServersResponse::fromData(DataInput& input) {
@@ -33,7 +33,7 @@ void GetAllServersResponse::fromData(DataInput& input) {
   for (int i = 0; i < numServers; i++) {
     std::shared_ptr<ServerLocation> sLoc = std::make_shared<ServerLocation>();
     sLoc->fromData(input);
-    m_servers.push_back(sLoc);
+    servers_.push_back(sLoc);
   }
 }
 

cppcache/src/GetAllServersResponse.hpp

@@ -35,23 +35,27 @@ namespace client {
 
 class GetAllServersResponse : public internal::DataSerializableFixedId_t<
                                   internal::DSFid::GetAllServersResponse> {
-  std::vector<std::shared_ptr<ServerLocation> > m_servers;
-
  public:
   static std::shared_ptr<Serializable> create() {
     return std::make_shared<GetAllServersResponse>();
   }
   GetAllServersResponse() : Serializable() {}
+  explicit GetAllServersResponse(
+      std::vector<std::shared_ptr<ServerLocation> > servers)
+      : Serializable(), servers_(servers) {}
   void toData(DataOutput& output) const override;
   void fromData(DataInput& input) override;
 
   size_t objectSize() const override {
-    return sizeof(GetAllServersResponse) + m_servers.capacity();
+    return sizeof(GetAllServersResponse) + servers_.capacity();
   }
   std::vector<std::shared_ptr<ServerLocation> > getServers() {
-    return m_servers;
+    return servers_;
   }
   ~GetAllServersResponse() override = default;
+
+ private:
+  std::vector<std::shared_ptr<ServerLocation> > servers_;
 };
 
 }  // namespace client

cppcache/src/StreamDataInput.cpp

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 "StreamDataInput.hpp"
+
+#include <geode/DataInput.hpp>
+
+#include "Utils.hpp"
+#include "util/Log.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+constexpr size_t kBufferSize = 3000;
+
+StreamDataInput::StreamDataInput(std::chrono::milliseconds timeout,
+                                 std::unique_ptr<Connector> connector,
+                                 const CacheImpl* cache, Pool* pool)
+    : DataInput(nullptr, 0, cache, pool),
+      connector_(std::move(connector)),
+      remainingTimeBeforeTimeout_(timeout) {}
+
+void StreamDataInput::readDataIfNotAvailable(size_t size) {
+  char buff[kBufferSize];
+  while (getBytesRemaining() < size) {
+    const auto start = std::chrono::system_clock::now();
+
+    const auto receivedLength = connector_->receive_nothrowiftimeout(
+        buff, kBufferSize, remainingTimeBeforeTimeout_);
+
+    const auto timeSpent = std::chrono::system_clock::now() - start;
+
+    remainingTimeBeforeTimeout_ -=
+        std::chrono::duration_cast<decltype(remainingTimeBeforeTimeout_)>(
+            timeSpent);
+
+    LOGDEBUG(
+        "received %d bytes from %s: %s, time spent: "
+        "%ld millisecs, time remaining before timeout: %ld millisecs",
+        receivedLength, connector_->getRemoteEndpoint().c_str(),
+        Utils::convertBytesToString(reinterpret_cast<uint8_t*>(buff),
+                                    receivedLength)
+            .c_str(),
+        std::chrono::duration_cast<std::chrono::milliseconds>(timeSpent)
+            .count(),
+        remainingTimeBeforeTimeout_.count());
+
+    if (remainingTimeBeforeTimeout_ <= std::chrono::milliseconds::zero()) {
+      throw(TimeoutException(std::string("Timeout when receiving from ")
+                                 .append(connector_->getRemoteEndpoint())));
+    }
+
+    auto newLength = bufferLength_ + receivedLength;
+    auto currentPosition = getBytesRead();
+    streamBuf_.resize(newLength);
+    memcpy(streamBuf_.data() + bufferLength_, buff, receivedLength);
+
+    bufferHead_ = streamBuf_.data();
+    buffer_ = bufferHead_ + currentPosition;
+    bufferLength_ = newLength;
+  }
+}
+
+}  // namespace client
+}  // namespace geode
+}  // namespace apache