fixed some TLS errors when combining protocols SSL/TLS and VST

arangodb · jsteemann · Sep 19, 2018 · Sep 19, 2018 · Sep 19, 2018 · Sep 19, 2018
commit 2549f2e6bd14df9e735b3f40f5b8f8ac22255ca6
diff --git a/arangod/Scheduler/SocketTask.cpp b/arangod/Scheduler/SocketTask.cpp
@@ -119,7 +119,7 @@ bool SocketTask::start() {
 
   if (_closeRequested.load(std::memory_order_acquire)) {
     LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
-        << "cannot start, close alread in progress";
+        << "cannot start, close already in progress";
     return false;
   }
 
@@ -191,6 +191,7 @@ void SocketTask::closeStream() {
   if (_abandoned.load(std::memory_order_acquire)) {
     return;
   }
+
   // strand::dispatch may execute this immediately if this
   // is called on a thread inside the same strand
   auto self = shared_from_this();
@@ -204,7 +205,7 @@ void SocketTask::closeStream() {
 void SocketTask::closeStreamNoLock() {
   TRI_ASSERT(_peer != nullptr);
   TRI_ASSERT(_peer->runningInThisThread());
-
+  
   bool mustCloseSend = !_closedSend.load(std::memory_order_acquire);
   bool mustCloseReceive = !_closedReceive.load(std::memory_order_acquire);
 
@@ -291,6 +292,7 @@ bool SocketTask::trySyncRead() {
 
   asio_ns::error_code err;
   TRI_ASSERT(_peer != nullptr);
+
   if (0 == _peer->available(err)) {
     return false;
   }
@@ -315,17 +317,15 @@ bool SocketTask::trySyncRead() {
 
   _readBuffer.increaseLength(bytesRead);
 
-  if (err) {
-    if (err == asio_ns::error::would_block) {
-      return false;
-    } else {
-      LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "trySyncRead failed with: "
-                                              << err.message();
-      return false;
-    }
+  if (!err) {
+    return true;
   }
 
-  return true;
+  if (err != asio_ns::error::would_block && err != asio_ns::error::try_again) {
+    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "trySyncRead failed with: " << err.message();
+  }
+
+  return false;
 }
 
 // caller must hold the _lock
@@ -411,7 +411,7 @@ void SocketTask::asyncReadSome() {
     closeStreamNoLock();
     return;
   }
-
+  
   // try to read more bytes
   if (_abandoned.load(std::memory_order_acquire)) {
     return;
@@ -460,54 +460,72 @@ void SocketTask::asyncWriteSome() {
   if (_writeBuffer.empty()) {
     return;
   }
+
+  TRI_ASSERT(_writeBuffer._buffer != nullptr);
   size_t total = _writeBuffer._buffer->length();
   size_t written = 0;
 
   TRI_ASSERT(!_abandoned);
-  TRI_ASSERT(_peer != nullptr);
 
   asio_ns::error_code err;
-  err.clear();
-  while (true) {
-    RequestStatistics::SET_WRITE_START(_writeBuffer._statistics);
-    written = _peer->writeSome(_writeBuffer._buffer, err);
+
+  if (!_peer->isEncrypted()) {
+    // try some direct writes only for non-SSL mode
+    // in SSL mode it will fall apart when mixing direct writes and async
+    // writes later
+    while (true) {
+      TRI_ASSERT(_writeBuffer._buffer != nullptr);
+
+      // we can directly skip sending empty buffers
+      if (_writeBuffer._buffer->length() > 0) {
+        RequestStatistics::SET_WRITE_START(_writeBuffer._statistics);
+        written = _peer->writeSome(_writeBuffer._buffer, err);
+
+        RequestStatistics::ADD_SENT_BYTES(_writeBuffer._statistics, written);
+
+        if (err || written != total) {
+          // unable to write everything at once, might be a lot of data
+          // above code does not update the buffer positon
+          break;
+        }
 
-    if (err) {
-      break;
-    }
+        TRI_ASSERT(written > 0);
+      }
 
-    RequestStatistics::ADD_SENT_BYTES(_writeBuffer._statistics, written);
+      if (!completedWriteBuffer()) {
+        return;
+      }
 
-    if (written != total) {
-      // unable to write everything at once, might be a lot of data
-      // above code does not update the buffer positon
-      break;
+      // try to send next buffer
+      TRI_ASSERT(_writeBuffer._buffer != nullptr);
+      total = _writeBuffer._buffer->length();
     }
 
-    if (!completedWriteBuffer()) {
+    // write could have blocked which is the only acceptable error
+    if (err && err != asio_ns::error::would_block && err != asio_ns::error::try_again) {
+      LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "sync write on stream failed with: "
+        << err.message();
+      closeStreamNoLock();
       return;
     }
+  } // !_peer->isEncrypted 
 
-    // try to send next buffer
-    total = _writeBuffer._buffer->length();
-    written = 0;
-  }
-
-  // write could have blocked which is the only acceptable error
-  if (err && err != ::asio_ns::error::would_block) {
-    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "sync write on failed with: "
-                                            << err.message();
-    closeStreamNoLock();
-    return;
-  }
+  // we will be getting here in the following cases
+  // - encrypted mode (SSL)
+  // - we send only parts of the write buffer, but have more to send
+  // - we got the error would_block/try_again when sending data
+  // in this case we dispatch an async write
 
   if (_abandoned.load(std::memory_order_acquire)) {
     return;
   }
-
+
+  TRI_ASSERT(_writeBuffer._buffer != nullptr);
+
   // so the code could have blocked at this point or not all data
   // was written in one go, begin writing at offset (written)
   auto self = shared_from_this();
+
   _peer->asyncWrite(
       asio_ns::buffer(_writeBuffer._buffer->begin() + written, total - written),
       [self, this](const asio_ns::error_code& ec, std::size_t transferred) {
@@ -516,7 +534,8 @@ void SocketTask::asyncWriteSome() {
 
         if (_abandoned.load(std::memory_order_acquire)) {
           return;
-        } else if (ec) {
+        } 
+        if (ec) {
           LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "write on failed with: "
                                                   << ec.message();
           closeStream();
@@ -527,11 +546,9 @@ void SocketTask::asyncWriteSome() {
                                           transferred);
 
         if (completedWriteBuffer()) {
-          _peer->post([self, this] {
-            if (!_abandoned.load(std::memory_order_acquire)) {
-              asyncWriteSome();
-            }
-          });
+          if (!_abandoned.load(std::memory_order_acquire)) {
+            asyncWriteSome();
+          }
         }
       });
 }