Merge remote-tracking branch 'origin/3.6'

5 files changed, 118 insertions, 101 deletions

diff --git a/cpp/src/Glacier2/Blobject.cpp b/cpp/src/Glacier2/Blobject.cpp
index a9c2ce55f80..1f58da34ffb 100644
--- a/cpp/src/Glacier2/Blobject.cpp
+++ b/cpp/src/Glacier2/Blobject.cpp
@@ -48,7 +48,7 @@ Glacier2::Blobject::Blobject(const InstancePtr& instance, const ConnectionPtr& r
                         _instance->properties()->getPropertyAsInt(clientTraceOverride)),
     _context(context)
 {
-    RequestQueueThreadPtr t = _reverseConnection ? _instance->serverRequestQueueThread() : 
+    RequestQueueThreadPtr t = _reverseConnection ? _instance->serverRequestQueueThread() :
                                                    _instance->clientRequestQueueThread();
     if(t)
     {
@@ -86,17 +86,14 @@ Glacier2::Blobject::invokeResponse(bool ok, const pair<const Byte*, const Byte*>
 }
 
 void
-Glacier2::Blobject::invokeSent(bool sent, const AMD_Object_ice_invokePtr& amdCB)
+Glacier2::Blobject::invokeSent(bool, const AMD_Object_ice_invokePtr& amdCB)
 {
-    if(sent)
-    {
 #if (defined(_MSC_VER) && (_MSC_VER >= 1600))
-        amdCB->ice_response(true, pair<const Byte*, const Byte*>(static_cast<const Byte*>(nullptr), 
-                                                                 static_cast<const Byte*>(nullptr)));
+    amdCB->ice_response(true, pair<const Byte*, const Byte*>(static_cast<const Byte*>(nullptr),
+                                                             static_cast<const Byte*>(nullptr)));
 #else
-        amdCB->ice_response(true, pair<const Byte*, const Byte*>(0, 0));
+    amdCB->ice_response(true, pair<const Byte*, const Byte*>(0, 0));
 #endif
-    }
 }
 
 void
@@ -124,7 +121,7 @@ Glacier2::Blobject::invokeException(const Exception& ex, const AMD_Object_ice_in
 }
 
 void
-Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amdCB, 
+Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amdCB,
                            const std::pair<const Byte*, const Byte*>& inParams, const Current& current)
 {
     //
@@ -171,7 +168,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
                     proxy = proxy->ice_twoway();
                     break;
                 }
-                
+
                 case 'o':
                 {
                     if(_alwaysBatch && _requestQueue)
@@ -184,7 +181,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
                     }
                     break;
                 }
-                
+
                 case 'd':
                 {
                     if(_alwaysBatch && _requestQueue)
@@ -197,7 +194,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
                     }
                     break;
                 }
-                
+
                 case 'O':
                 {
                     if(_requestQueue)
@@ -210,7 +207,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
                     }
                     break;
                 }
-                
+
                 case 'D':
                 {
                     if(_requestQueue)
@@ -223,19 +220,19 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
                     }
                     break;
                 }
-                
+
                 case 's':
                 {
                     proxy = proxy->ice_secure(true);
                     break;
                 }
-                
+
                 case 'z':
                 {
                     proxy = proxy->ice_compress(true);
                     break;
                 }
-                
+
                 default:
                 {
                     Warning out(_instance->logger());
@@ -245,7 +242,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
             }
         }
     }
-    
+
     if(_requestTraceLevel >= 1)
     {
         Trace out(_instance->logger(), "Glacier2");
@@ -284,7 +281,7 @@ Glacier2::Blobject::invoke(ObjectPrx& proxy, const AMD_Object_ice_invokePtr& amd
     }
 
     if(_requestQueue)
-    {    
+    {
         //
         // If we are in buffered mode, we create a new request and add
         // it to the request queue. If the request is twoway, we use
diff --git a/cpp/src/Glacier2/RequestQueue.cpp b/cpp/src/Glacier2/RequestQueue.cpp
index 78eff63bce0..5acc66dbfa9 100644
--- a/cpp/src/Glacier2/RequestQueue.cpp
+++ b/cpp/src/Glacier2/RequestQueue.cpp
@@ -10,7 +10,6 @@
 #include <Glacier2/RequestQueue.h>
 #include <Glacier2/Instance.h>
 #include <Glacier2/SessionRouterI.h>
-#include <set>
 
 using namespace std;
 using namespace Ice;
@@ -33,6 +32,19 @@ Glacier2::Request::Request(const ObjectPrx& proxy, const std::pair<const Byte*,
     }
 }
 
+void
+Glacier2::Request::addBatchProxy(set<Ice::ObjectPrx>& batchProxies)
+{
+    set<Ice::ObjectPrx>::const_iterator p = batchProxies.find(_proxy);
+    if(p == batchProxies.end())
+    {
+        batchProxies.insert(_proxy);
+    }
+    else if(p->get() != _proxy.get())
+    {
+        const_cast<Ice::ObjectPrx&>(_proxy) = *p;
+    }
+}
 
 Ice::AsyncResultPtr
 Glacier2::Request::invoke(const Callback_Object_ice_invokePtr& cb)
@@ -52,7 +64,7 @@ Glacier2::Request::invoke(const Callback_Object_ice_invokePtr& cb)
     {
         ByteSeq outParams;
         if(_forwardContext)
-        { 
+        {
             if(_sslContext.size() > 0)
             {
                 Ice::Context ctx = _current.ctx;
@@ -81,7 +93,7 @@ Glacier2::Request::invoke(const Callback_Object_ice_invokePtr& cb)
     {
         Ice::AsyncResultPtr result;
         if(_forwardContext)
-        { 
+        {
             if(_sslContext.size() > 0)
             {
                 Ice::Context ctx = _current.ctx;
@@ -131,7 +143,7 @@ Glacier2::Request::override(const RequestPtr& other) const
 
     //
     // Don't override if the override isn't the same.
-    // 
+    //
     if(_override != other->_override)
     {
         return false;
@@ -163,13 +175,13 @@ Glacier2::Request::exception(const Ice::Exception& ex)
     }
 }
 
-void 
+void
 Glacier2::Request::queued()
 {
     if(!_proxy->ice_isTwoway())
     {
 #if (defined(_MSC_VER) && (_MSC_VER >= 1600))
-        _amdCB->ice_response(true, pair<const Byte*, const Byte*>(static_cast<const Byte*>(nullptr), 
+        _amdCB->ice_response(true, pair<const Byte*, const Byte*>(static_cast<const Byte*>(nullptr),
                                                                   static_cast<const Byte*>(nullptr)));
 #else
         _amdCB->ice_response(true, pair<const Byte*, const Byte*>(0, 0));
@@ -177,8 +189,8 @@ Glacier2::Request::queued()
     }
 }
 
-Glacier2::RequestQueue::RequestQueue(const RequestQueueThreadPtr& requestQueueThread, 
-                                     const InstancePtr& instance, 
+Glacier2::RequestQueue::RequestQueue(const RequestQueueThreadPtr& requestQueueThread,
+                                     const InstancePtr& instance,
                                      const Ice::ConnectionPtr& connection) :
     _requestQueueThread(requestQueueThread),
     _instance(instance),
@@ -219,7 +231,17 @@ Glacier2::RequestQueue::addRequest(const RequestPtr& request)
             }
         }
     }
-    
+
+    if(!_connection)
+    {
+        //
+        // If it's a batch request, we make sure we use a unique batch proxy object for the queued
+        // batch requests. We want all the requests for the same batch proxy to be queued on the
+        // same proxy object.
+        //
+        request->addBatchProxy(_batchProxies);
+    }
+
     //
     // No override, we add the new request.
     //
@@ -237,7 +259,7 @@ Glacier2::RequestQueue::addRequest(const RequestPtr& request)
 }
 
 void
-Glacier2::RequestQueue::flushRequests(set<Ice::ObjectPrx>& batchProxies)
+Glacier2::RequestQueue::flushRequests()
 {
     IceUtil::Mutex::Lock lock(*this);
     if(_connection)
@@ -250,7 +272,34 @@ Glacier2::RequestQueue::flushRequests(set<Ice::ObjectPrx>& batchProxies)
     }
     else
     {
-        flush(batchProxies);
+        for(deque<RequestPtr>::const_iterator p = _requests.begin(); p != _requests.end(); ++p)
+        {
+            try
+            {
+                if(_observer)
+                {
+                    _observer->forwarded(!_connection);
+                }
+                assert(_callback);
+                (*p)->invoke(_callback);
+            }
+            catch(const Ice::LocalException&)
+            {
+                // Ignore, this can occur for batch requests.
+            }
+        }
+        _requests.clear();
+
+        for(set<Ice::ObjectPrx>::const_iterator q = _batchProxies.begin(); q != _batchProxies.end(); ++q)
+        {
+            (*q)->begin_ice_flushBatchRequests();
+        }
+        _batchProxies.clear();
+    }
+
+    if(_destroyed && _requests.empty())
+    {
+        destroyInternal();
     }
 }
 
@@ -279,20 +328,6 @@ Glacier2::RequestQueue::updateObserver(const Glacier2::Instrumentation::SessionO
 }
 
 void
-Glacier2::RequestQueue::destroyInternal()
-{
-    //
-    // Must be called with the mutex locked.
-    //
-
-    //
-    // Remove cyclic references.
-    //
-    const_cast<Ice::Callback_Object_ice_invokePtr&>(_callback) = 0;
-    const_cast<Ice::Callback_Connection_flushBatchRequestsPtr&>(_flushCallback) = 0;
-}
-
-void
 Glacier2::RequestQueue::flush()
 {
     assert(_connection);
@@ -314,7 +349,7 @@ Glacier2::RequestQueue::flush()
             if(!result)
             {
                 flushBatchRequests = true;
-            } 
+            }
             else if(!result->sentSynchronously() && !result->isCompleted())
             {
                 _pendingSend = true;
@@ -346,44 +381,20 @@ Glacier2::RequestQueue::flush()
             _pendingSendRequest = 0;
         }
     }
-
-    if(_destroyed && _requests.empty())
-    {
-        destroyInternal();
-    }
 }
 
 void
-Glacier2::RequestQueue::flush(set<Ice::ObjectPrx>& batchProxies)
+Glacier2::RequestQueue::destroyInternal()
 {
-    assert(!_connection);
-
-    for(deque<RequestPtr>::const_iterator p = _requests.begin(); p != _requests.end(); ++p)
-    {
-        try
-        {
-            if(_observer)
-            {
-                _observer->forwarded(!_connection);
-            }
-            assert(_callback);
-            Ice::AsyncResultPtr result = (*p)->invoke(_callback);
-            if(!result)
-            {
-                batchProxies.insert((*p)->getProxy());
-            } 
-        }
-        catch(const Ice::LocalException&)
-        {
-            // Ignore, this can occur for batch requests.
-        }
-    }
-    _requests.clear();
+    //
+    // Must be called with the mutex locked.
+    //
 
-    if(_destroyed)
-    {
-        destroyInternal();
-    }
+    //
+    // Remove cyclic references.
+    //
+    const_cast<Ice::Callback_Object_ice_invokePtr&>(_callback) = 0;
+    const_cast<Ice::Callback_Connection_flushBatchRequestsPtr&>(_flushCallback) = 0;
 }
 
 void
@@ -427,7 +438,7 @@ Glacier2::RequestQueue::exception(const Ice::Exception& ex, const RequestPtr& re
     }
 }
 
-void 
+void
 Glacier2::RequestQueue::sent(bool sentSynchronously, const RequestPtr& request)
 {
     if(_connection && !sentSynchronously)
@@ -454,12 +465,12 @@ Glacier2::RequestQueueThread::~RequestQueueThread()
     assert(_queues.empty());
 }
 
-void 
+void
 Glacier2::RequestQueueThread::destroy()
 {
     {
         IceUtil::Monitor<IceUtil::Mutex>::Lock lock(*this);
-        
+
         assert(!_destroy);
         _destroy = true;
         _sleep = false;
@@ -534,7 +545,7 @@ Glacier2::RequestQueueThread::run()
             }
 
             //
-            // If the queue is being destroyed and there's no requests or responses 
+            // If the queue is being destroyed and there's no requests or responses
             // to send, we're done.
             //
             if(_destroy && _queues.empty())
@@ -543,7 +554,7 @@ Glacier2::RequestQueueThread::run()
             }
 
             assert(!_queues.empty() && !_sleep);
-            
+
             queues.swap(_queues);
 
             if(_sleepTime > IceUtil::Time())
@@ -552,16 +563,10 @@ Glacier2::RequestQueueThread::run()
                 _sleepDuration = _sleepTime;
             }
         }
-        
-        set<Ice::ObjectPrx> flushProxySet;
-        for(vector<RequestQueuePtr>::const_iterator p = queues.begin(); p != queues.end(); ++p)
-        {
-            (*p)->flushRequests(flushProxySet);
-        }
 
-        for(set<Ice::ObjectPrx>::const_iterator q = flushProxySet.begin(); q != flushProxySet.end(); ++q)
+        for(vector<RequestQueuePtr>::const_iterator p = queues.begin(); p != queues.end(); ++p)
         {
-            (*q)->begin_ice_flushBatchRequests();
+            (*p)->flushRequests();
         }
     }
 }
diff --git a/cpp/src/Glacier2/RequestQueue.h b/cpp/src/Glacier2/RequestQueue.h
index d918584e1d5..94ca48b089f 100644
--- a/cpp/src/Glacier2/RequestQueue.h
+++ b/cpp/src/Glacier2/RequestQueue.h
@@ -36,10 +36,10 @@ public:
 
     Request(const Ice::ObjectPrx&, const std::pair<const Ice::Byte*, const Ice::Byte*>&, const Ice::Current&, bool,
             const Ice::Context&, const Ice::AMD_Object_ice_invokePtr&);
-    
+
     Ice::AsyncResultPtr invoke(const Ice::Callback_Object_ice_invokePtr& callback);
     bool override(const RequestPtr&) const;
-    const Ice::ObjectPrx& getProxy() const { return _proxy; }
+    void addBatchProxy(std::set<Ice::ObjectPrx>&);
     bool hasOverride() const { return !_override.empty(); }
 
 private:
@@ -65,7 +65,7 @@ public:
     RequestQueue(const RequestQueueThreadPtr&, const InstancePtr&, const Ice::ConnectionPtr&);
 
     bool addRequest(const RequestPtr&);
-    void flushRequests(std::set<Ice::ObjectPrx>&);
+    void flushRequests();
 
     void destroy();
 
@@ -76,12 +76,11 @@ private:
     void destroyInternal();
 
     void flush();
-    void flush(std::set<Ice::ObjectPrx>&);
 
     void response(bool, const std::pair<const Ice::Byte*, const Ice::Byte*>&, const RequestPtr&);
     void exception(const Ice::Exception&, const RequestPtr&);
     void sent(bool, const RequestPtr&);
-    
+
     const RequestQueueThreadPtr _requestQueueThread;
     const InstancePtr _instance;
     const Ice::ConnectionPtr _connection;
@@ -89,6 +88,7 @@ private:
     const Ice::Callback_Connection_flushBatchRequestsPtr _flushCallback;
 
     std::deque<RequestPtr> _requests;
+    std::set<Ice::ObjectPrx> _batchProxies;
     bool _pendingSend;
     RequestPtr _pendingSendRequest;
     bool _destroyed;
diff --git a/cpp/src/IceSSL/OpenSSLTransceiverI.cpp b/cpp/src/IceSSL/OpenSSLTransceiverI.cpp
index 99248a74a2d..2178c292ea0 100644
--- a/cpp/src/IceSSL/OpenSSLTransceiverI.cpp
+++ b/cpp/src/IceSSL/OpenSSLTransceiverI.cpp
@@ -466,6 +466,8 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf)
         return IceInternal::SocketOperationNone;
     }
 
+    _stream->ready(IceInternal::SocketOperationRead, false);
+
     //
     // It's impossible for packetSize to be more than an Int.
     //
@@ -555,6 +557,11 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf)
         }
     }
 
+    //
+    // Check if there's still buffered data to read, set the read ready status.
+    //
+    _stream->ready(IceInternal::SocketOperationRead, SSL_pending(_ssl) > 0);
+
     return IceInternal::SocketOperationNone;
 }
 
diff --git a/cpp/src/IceSSL/SecureTransportTransceiverI.cpp b/cpp/src/IceSSL/SecureTransportTransceiverI.cpp
index dc9d421b957..b043ef19713 100644
--- a/cpp/src/IceSSL/SecureTransportTransceiverI.cpp
+++ b/cpp/src/IceSSL/SecureTransportTransceiverI.cpp
@@ -407,17 +407,13 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf)
         return _stream->read(buf);
     }
 
-    //
-    // Note: we assume that SecureTransport doesn't read more SSL records
-    // than necessary to fill the requested data and that the sender sends
-    // Ice messages in individual SSL records.
-    //
-
     if(buf.i == buf.b.end())
     {
         return  IceInternal::SocketOperationNone;
     }
 
+    _stream->ready(IceInternal::SocketOperationRead, false);
+
     size_t packetSize = std::min(static_cast<size_t>(buf.b.end() - buf.i), _maxRecvPacketSize);
     while(buf.i != buf.b.end())
     {
@@ -464,6 +460,18 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf)
             packetSize = buf.b.end() - buf.i;
         }
     }
+
+    //
+    // Check if there's still buffered data to read. In this case, set the read ready status.
+    //
+    size_t buffered = 0;
+    OSStatus err = SSLGetBufferedReadSize(_ssl, &buffered);
+    if(err)
+    {
+        errno = err;
+        throw SocketException(__FILE__, __LINE__, IceInternal::getSocketErrno());
+    }
+    _stream->ready(IceInternal::SocketOperationRead, buffered > 0);
     return IceInternal::SocketOperationNone;
 }