- Added support for non-blocking AMI/batch requests, connection

  creation.
- Added support for AMI oneway requests.
- Changed collocation optimization to not perform any DNS lookups.

1 files changed, 203 insertions, 190 deletions

diff --git a/cpp/src/Ice/TraceUtil.cpp b/cpp/src/Ice/TraceUtil.cpp
index cb371f4e35a..b8c4d1f10b1 100644
--- a/cpp/src/Ice/TraceUtil.cpp
+++ b/cpp/src/Ice/TraceUtil.cpp
@@ -44,6 +44,26 @@ printIdentityFacetOperation(ostream& s, BasicStream& stream)
     s << "\noperation = " << operation;
 }
 
+static string
+getMessageTypeAsString(Byte type)
+{
+    switch(type)
+    {
+        case requestMsg:
+            return "request";
+        case requestBatchMsg:
+            return "batch request";
+        case replyMsg:
+            return "reply";
+        case closeConnectionMsg:
+            return "close connection";
+        case validateConnectionMsg:
+            return "validate connection";
+        default:
+            return "unknown";
+    }
+}
+
 static void
 printRequestHeader(ostream& s, BasicStream& stream)
 {
@@ -95,7 +115,7 @@ printRequestHeader(ostream& s, BasicStream& stream)
     }
 }
 
-static void
+static Byte
 printHeader(ostream& s, BasicStream& stream)
 {
     Byte magicNumber;
@@ -120,105 +140,238 @@ printHeader(ostream& s, BasicStream& stream)
 
     Byte type;
     stream.read(type);
-    s << "\nmessage type = "  << static_cast<int>(type) << ' ';
+    s << "\nmessage type = "  << static_cast<int>(type) << " (" << getMessageTypeAsString(type) << ')';
 
-    switch(type)
+    Byte compress;
+    stream.read(compress);
+    s << "\ncompression status = "  << static_cast<int>(compress) << ' ';
+
+    switch(compress)
     {
-        case requestMsg:
+        case 0:
         {
-            s << "(request)";
+            s << "(not compressed; do not compress response, if any)";
             break;
         }
 
-        case requestBatchMsg:
+        case 1:
         {
-            s << "(batch request)";
+            s << "(not compressed; compress response, if any)";
             break;
         }
 
-        case replyMsg:
+        case 2:
         {
-            s << "(reply)";
+            s << "(compressed; compress response, if any)";
             break;
         }
 
-        case closeConnectionMsg:
+        default:
         {
-            s << "(close connection)";
+            s << "(unknown)";
             break;
         }
+    }
 
-        case validateConnectionMsg:
+    Int size;
+    stream.read(size);
+    s << "\nmessage size = " << size;
+
+    return type;
+}
+
+static void
+printRequest(ostream& s, BasicStream& stream)
+{
+    Int requestId;
+    stream.read(requestId);
+    s << "\nrequest id = " << requestId;
+    if(requestId == 0)
+    {
+        s << " (oneway)";
+    }
+
+    printRequestHeader(s, stream);
+}
+
+static void
+printBatchRequest(ostream& s, BasicStream& stream)
+{
+    int batchRequestNum;
+    stream.read(batchRequestNum);
+    s << "\nnumber of requests = " << batchRequestNum;
+
+    for(int i = 0; i < batchRequestNum; ++i)
+    {
+        s << "\nrequest #" << i << ':';
+        printRequestHeader(s, stream);
+        stream.skipEncaps();
+    }
+}
+
+static void
+printReply(ostream& s, BasicStream& stream)
+{
+    Int requestId;
+    stream.read(requestId);
+    s << "\nrequest id = " << requestId;
+
+    Byte replyStatus;
+    stream.read(replyStatus);
+    s << "\nreply status = " << static_cast<int>(replyStatus) << ' ';
+    switch(replyStatus)
+    {
+    case replyOK:
+    {
+        s << "(ok)";
+        break;
+    }
+
+    case replyUserException:
+    {
+        s << "(user exception)";
+        break;
+    }
+
+    case replyObjectNotExist:
+    case replyFacetNotExist:
+    case replyOperationNotExist:
+    {
+        switch(replyStatus)
+        {
+        case replyObjectNotExist:
+        {
+            s << "(object not exist)";
+            break;
+        }
+
+        case replyFacetNotExist:
+        {
+            s << "(facet not exist)";
+            break;
+        }
+
+        case replyOperationNotExist:
         {
-            s << "(validate connection)";
+            s << "(operation not exist)";
             break;
         }
 
         default:
         {
-            s << "(unknown)";
+            assert(false);
             break;
         }
-    }
+        }
 
-    Byte compress;
-    stream.read(compress);
-    s << "\ncompression status = "  << static_cast<int>(compress) << ' ';
+        printIdentityFacetOperation(s, stream);
+        break;
+    }
 
-    switch(compress)
+    case replyUnknownException:
+    case replyUnknownLocalException:
+    case replyUnknownUserException:
     {
-        case 0:
+        switch(replyStatus)
         {
-            s << "(not compressed; do not compress response, if any)";
+        case replyUnknownException:
+        {
+            s << "(unknown exception)";
             break;
         }
 
-        case 1:
+        case replyUnknownLocalException:
         {
-            s << "(not compressed; compress response, if any)";
+            s << "(unknown local exception)";
             break;
         }
 
-        case 2:
+        case replyUnknownUserException:
         {
-            s << "(compressed; compress response, if any)";
+            s << "(unknown user exception)";
             break;
         }
 
         default:
         {
-            s << "(unknown)";
+            assert(false);
             break;
         }
+        }
+                
+        string unknown;
+        stream.read(unknown, false);
+        s << "\nunknown = " << unknown;
+        break;
     }
 
-    Int size;
-    stream.read(size);
-    s << "\nmessage size = " << size;
+    default:
+    {
+        s << "(unknown)";
+        break;
+    }
+    }
 }
 
-void
-IceInternal::traceHeader(const char* heading, const BasicStream& str, const LoggerPtr& logger,
-                         const TraceLevelsPtr& tl)
+static Byte
+printMessage(ostream& s, BasicStream& stream)
 {
-    if(tl->protocol >= 1)
+    Byte type = printHeader(s, stream);
+
+    switch(type)
     {
-        BasicStream& stream = const_cast<BasicStream&>(str);
-        BasicStream::Container::iterator p = stream.i;
-        stream.i = stream.b.begin();
+    case closeConnectionMsg:
+    case validateConnectionMsg:
+    {
+        // We're done.
+        break;
+    }
+        
+    case requestMsg:
+    {
+        printRequest(s, stream);
+        break;
+    }
+        
+    case requestBatchMsg:
+    {
+        printBatchRequest(s, stream);
+        break;
+    }
+        
+    case replyMsg:
+    {
+        printReply(s, stream);
+        break;
+    }
+        
+    default:
+    {
+        break;
+    }    
+    }
 
-        ostringstream s;
-        s << heading;
-        printHeader(s, stream);
+    return type;
+}
 
-        logger->trace(tl->protocolCat, s.str());
-        stream.i = p;
+static IceUtil::StaticMutex slicingMutex = ICE_STATIC_MUTEX_INITIALIZER;
+
+void
+IceInternal::traceSlicing(const char* kind, const string& typeId, const char* slicingCat, const LoggerPtr& logger)
+{
+    IceUtil::StaticMutex::Lock lock(slicingMutex);
+    static set<string> slicingIds;
+    if(slicingIds.insert(typeId).second)
+    {
+        string s("unknown ");
+        s += kind;
+        s += " type `" + typeId + "'";
+        logger->trace(slicingCat, s);
     }
 }
 
 void
-IceInternal::traceRequest(const char* heading, const BasicStream& str, const LoggerPtr& logger,
-                          const TraceLevelsPtr& tl)
+IceInternal::traceSend(const BasicStream& str, const LoggerPtr& logger, const TraceLevelsPtr& tl)
 {
     if(tl->protocol >= 1)
     {
@@ -227,27 +380,15 @@ IceInternal::traceRequest(const char* heading, const BasicStream& str, const Log
         stream.i = stream.b.begin();
 
         ostringstream s;
-        s << heading;
-        printHeader(s, stream);
+        Byte type = printMessage(s, stream);
 
-        Int requestId;
-        stream.read(requestId);
-        s << "\nrequest id = " << requestId;
-        if(requestId == 0)
-        {
-            s << " (oneway)";
-        }
-
-        printRequestHeader(s, stream);
-
-        logger->trace(tl->protocolCat, s.str());
+        logger->trace(tl->protocolCat, "sending " + getMessageTypeAsString(type) + " " + s.str());
         stream.i = p;
     }
 }
 
 void
-IceInternal::traceBatchRequest(const char* heading, const BasicStream& str, const LoggerPtr& logger,
-                               const TraceLevelsPtr& tl)
+IceInternal::traceRecv(const BasicStream& str, const LoggerPtr& logger, const TraceLevelsPtr& tl)
 {
     if(tl->protocol >= 1)
     {
@@ -256,28 +397,15 @@ IceInternal::traceBatchRequest(const char* heading, const BasicStream& str, cons
         stream.i = stream.b.begin();
 
         ostringstream s;
-        s << heading;
-        printHeader(s, stream);
+        Byte type = printMessage(s, stream);
 
-        int batchRequestNum;
-        stream.read(batchRequestNum);
-        s << "\nnumber of requests = " << batchRequestNum;
-
-        for(int i = 0; i < batchRequestNum; ++i)
-        {
-            s << "\nrequest #" << i << ':';
-            printRequestHeader(s, stream);
-            stream.skipEncaps();
-        }
-
-        logger->trace(tl->protocolCat, s.str());
+        logger->trace(tl->protocolCat, "received " + getMessageTypeAsString(type) + " " + s.str());
         stream.i = p;
     }
 }
 
 void
-IceInternal::traceReply(const char* heading, const BasicStream& str, const LoggerPtr& logger,
-                        const TraceLevelsPtr& tl)
+IceInternal::trace(const char* heading, const BasicStream& str, const LoggerPtr& logger, const TraceLevelsPtr& tl)
 {
     if(tl->protocol >= 1)
     {
@@ -287,125 +415,10 @@ IceInternal::traceReply(const char* heading, const BasicStream& str, const Logge
 
         ostringstream s;
         s << heading;
-        printHeader(s, stream);
-
-        Int requestId;
-        stream.read(requestId);
-        s << "\nrequest id = " << requestId;
-
-        Byte replyStatus;
-        stream.read(replyStatus);
-        s << "\nreply status = " << static_cast<int>(replyStatus) << ' ';
-        switch(replyStatus)
-        {
-            case replyOK:
-            {
-                s << "(ok)";
-                break;
-            }
-
-            case replyUserException:
-            {
-                s << "(user exception)";
-                break;
-            }
-
-            case replyObjectNotExist:
-            case replyFacetNotExist:
-            case replyOperationNotExist:
-            {
-                switch(replyStatus)
-                {
-                    case replyObjectNotExist:
-                    {
-                        s << "(object not exist)";
-                        break;
-                    }
-
-                    case replyFacetNotExist:
-                    {
-                        s << "(facet not exist)";
-                        break;
-                    }
-
-                    case replyOperationNotExist:
-                    {
-                        s << "(operation not exist)";
-                        break;
-                    }
-
-                    default:
-                    {
-                        assert(false);
-                        break;
-                    }
-                }
-
-                printIdentityFacetOperation(s, stream);
-                break;
-            }
-
-            case replyUnknownException:
-            case replyUnknownLocalException:
-            case replyUnknownUserException:
-            {
-                switch(replyStatus)
-                {
-                    case replyUnknownException:
-                    {
-                        s << "(unknown exception)";
-                        break;
-                    }
-
-                    case replyUnknownLocalException:
-                    {
-                        s << "(unknown local exception)";
-                        break;
-                    }
-
-                    case replyUnknownUserException:
-                    {
-                        s << "(unknown user exception)";
-                        break;
-                    }
-
-                    default:
-                    {
-                        assert(false);
-                        break;
-                    }
-                }
-                
-                string unknown;
-                stream.read(unknown, false);
-                s << "\nunknown = " << unknown;
-                break;
-            }
-
-            default:
-            {
-                s << "(unknown)";
-                break;
-            }
-        }
+        printMessage(s, stream);
 
         logger->trace(tl->protocolCat, s.str());
         stream.i = p;
     }
 }
 
-static IceUtil::StaticMutex slicingMutex = ICE_STATIC_MUTEX_INITIALIZER;
-
-void
-IceInternal::traceSlicing(const char* kind, const string& typeId, const char* slicingCat, const LoggerPtr& logger)
-{
-    IceUtil::StaticMutex::Lock lock(slicingMutex);
-    static set<string> slicingIds;
-    if(slicingIds.insert(typeId).second)
-    {
-        string s("unknown ");
-        s += kind;
-        s += " type `" + typeId + "'";
-        logger->trace(slicingCat, s);
-    }
-}