Fixed race condition in flushBatchRequests(): connections that are not yet

    validated are now ignored when flushing.

6 files changed, 69 insertions, 16 deletions

diff --git a/cpp/src/Ice/Connection.cpp b/cpp/src/Ice/Connection.cpp
index 9fcfae99fa7..d23fa6ce8f7 100644
--- a/cpp/src/Ice/Connection.cpp
+++ b/cpp/src/Ice/Connection.cpp
@@ -340,6 +340,13 @@ IceInternal::Connection::validate()
     setState(StateHolding);
 }
 
+bool
+IceInternal::Connection::isValidated() const
+{
+    IceUtil::Monitor<IceUtil::RecMutex>::Lock sync(*this);
+    return _state > StateNotValidated;
+}
+
 void
 IceInternal::Connection::incProxyCount()
 {
@@ -376,7 +383,8 @@ IceInternal::Connection::sendRequest(Outgoing* out, bool oneway)
     {
 	_exception->ice_throw();
     }
-    assert(_state > StateNotValidated && _state < StateClosing);
+    assert(_state > StateNotValidated);
+    assert(_state < StateClosing);
 
     Int requestId;
 
@@ -486,7 +494,8 @@ IceInternal::Connection::sendAsyncRequest(const OutgoingAsyncPtr& out)
     {
 	_exception->ice_throw();
     }
-    assert(_state > StateNotValidated && _state < StateClosing);
+    assert(_state > StateNotValidated);
+    assert(_state < StateClosing);
 
     Int requestId;
 
@@ -590,7 +599,8 @@ IceInternal::Connection::prepareBatchRequest(BasicStream* os)
 	unlock();
 	_exception->ice_throw();
     }
-    assert(_state > StateNotValidated && _state < StateClosing);
+    assert(_state > StateNotValidated);
+    assert(_state < StateClosing);
 
     if(_batchStream.b.empty())
     {
@@ -622,7 +632,8 @@ IceInternal::Connection::finishBatchRequest(BasicStream* os)
 	unlock();
 	_exception->ice_throw();
     }
-    assert(_state > StateNotValidated && _state < StateClosing);
+    assert(_state > StateNotValidated);
+    assert(_state < StateClosing);
 
     _batchStream.swap(*os); // Get the batch stream back.
     ++_batchRequestNum; // Increment the number of requests in the batch.
@@ -645,7 +656,8 @@ IceInternal::Connection::flushBatchRequest()
     {
 	_exception->ice_throw();
     }
-    assert(_state > StateNotValidated && _state < StateClosing);
+    assert(_state > StateNotValidated);
+    assert(_state < StateClosing);
     
     if(!_batchStream.b.empty())
     {
diff --git a/cpp/src/Ice/Connection.h b/cpp/src/Ice/Connection.h
index bf3061c3fea..107d92a3f9a 100644
--- a/cpp/src/Ice/Connection.h
+++ b/cpp/src/Ice/Connection.h
@@ -64,6 +64,7 @@ public:
     void monitor();
 
     void validate();
+    bool isValidated() const;
 
     void incProxyCount();
     void decProxyCount();
diff --git a/cpp/src/Ice/ConnectionFactory.cpp b/cpp/src/Ice/ConnectionFactory.cpp
index ae99f562611..1d6227ecd81 100644
--- a/cpp/src/Ice/ConnectionFactory.cpp
+++ b/cpp/src/Ice/ConnectionFactory.cpp
@@ -510,11 +510,26 @@ IceInternal::IncomingConnectionFactory::connections() const
     return result;
 }
 
+namespace IceInternal {
+
+struct FlushIfValidated
+{
+    void operator() (ConnectionPtr p)
+    {
+	if(p->isValidated())
+	{
+	    p->flushBatchRequest();
+	}
+    }
+};
+
+}
+
 void
 IceInternal::IncomingConnectionFactory::flushBatchRequests()
 {
-    list<ConnectionPtr> c = connections(); // connections() is synchronized, so need to synchronize here.
-    for_each(c.begin(), c.end(), Ice::voidMemFun(&Connection::flushBatchRequest));
+    list<ConnectionPtr> c = connections(); // connections() is synchronized, so no need to synchronize here.
+    for_each(c.begin(), c.end(), FlushIfValidated());
 }
 
 bool
diff --git a/cpp/src/Ice/Instance.cpp b/cpp/src/Ice/Instance.cpp
index 52ba8ea430d..770e7aef0cf 100644
--- a/cpp/src/Ice/Instance.cpp
+++ b/cpp/src/Ice/Instance.cpp
@@ -363,19 +363,16 @@ IceInternal::Instance::messageSizeMax() const
 void
 IceInternal::Instance::flushBatchRequests()
 {
-    OutgoingConnectionFactoryPtr factory;
-    std::map<std::string, ::Ice::ObjectAdapterIPtr> adapters;
+    OutgoingConnectionFactoryPtr connectionFactory;
+    ObjectAdapterFactoryPtr adapterFactory;
     {
 	IceUtil::RecMutex::Lock sync(*this);
 
-	factory = _outgoingConnectionFactory;
-	adapters = _objectAdapterFactory->_adapters;
-    }
-    factory->flushBatchRequests();
-    for(std::map<std::string, ::Ice::ObjectAdapterIPtr>::const_iterator p = adapters.begin(); p != adapters.end(); ++p)
-    {
-	p->second->flushBatchRequests();
+	connectionFactory = _outgoingConnectionFactory;
+	adapterFactory = _objectAdapterFactory;
     }
+    connectionFactory->flushBatchRequests();
+    adapterFactory->flushBatchRequests();
 }
 
 IceInternal::Instance::Instance(const CommunicatorPtr& communicator, int& argc, char* argv[],
diff --git a/cpp/src/Ice/ObjectAdapterFactory.cpp b/cpp/src/Ice/ObjectAdapterFactory.cpp
index c87ce94bf31..569317fb33f 100644
--- a/cpp/src/Ice/ObjectAdapterFactory.cpp
+++ b/cpp/src/Ice/ObjectAdapterFactory.cpp
@@ -143,6 +143,33 @@ IceInternal::ObjectAdapterFactory::findObjectAdapter(const ObjectPrx& proxy)
     return 0;
 }
 
+namespace IceInternal {
+
+struct FlushAdapter
+{
+    void operator() (ObjectAdapterIPtr p)
+    {
+	p->flushBatchRequests();
+    }
+};
+
+}
+
+void
+IceInternal::ObjectAdapterFactory::flushBatchRequests() const
+{
+    list<ObjectAdapterIPtr> a;
+    {
+	IceUtil::Monitor<IceUtil::Mutex>::Lock sync(*this);
+
+	for(map<string, ObjectAdapterIPtr>::const_iterator p = _adapters.begin(); p != _adapters.end(); ++p)
+	{
+	    a.push_back(p->second);
+	}
+    }
+    for_each(a.begin(), a.end(), FlushAdapter());
+}
+
 IceInternal::ObjectAdapterFactory::ObjectAdapterFactory(const InstancePtr& instance,
 							const CommunicatorPtr& communicator) :
     _instance(instance),
diff --git a/cpp/src/Ice/ObjectAdapterFactory.h b/cpp/src/Ice/ObjectAdapterFactory.h
index 448d2b26615..4e885140a56 100644
--- a/cpp/src/Ice/ObjectAdapterFactory.h
+++ b/cpp/src/Ice/ObjectAdapterFactory.h
@@ -31,6 +31,7 @@ public:
 
     ::Ice::ObjectAdapterPtr createObjectAdapter(const std::string&);
     ::Ice::ObjectAdapterPtr findObjectAdapter(const ::Ice::ObjectPrx&);
+    void flushBatchRequests() const;
 
 private: