In qmfengine, if a method call or method response requires a buffer larger than the

static buffer used for communication, allocate a large-enough buffer temporarily from the
heap.

A test is included to verify large-buffer behavior.


git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1058709 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 163 insertions, 13 deletions

diff --git a/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb b/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb
index 1c4a0a2105..972d5977b8 100755
--- a/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb
+++ b/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb
@@ -354,6 +354,43 @@ class ConsoleTest < ConsoleTestBase
     end
   end
 
+  def test_H_map_list_method_call_big
+    parent = @qmfc.object(:class => "parent")
+    assert(parent, "Number of 'parent' objects")
+
+    big_string = ""
+    segment = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+    for idx in 1...1500
+      big_string = big_string + segment
+    end
+
+    inMap = {'aLong' => 9999999999,
+             'aInt'  => 54321,
+             'aSigned' => -666,
+             'aString' => big_string,
+             'another' => big_string,
+             'aFloat' => 3.1415,
+             'aList' => ['x', -1, 'y', 2],
+             'abool' => false}
+
+    inList = ['aString', 1, -1, 2.7182, {'aMap'=> -8}, true]
+
+    result = parent.test_map_list(inMap, inList)
+    assert_equal(result.status, 0)
+    assert_equal(result.text, "OK")
+
+    # verify returned values
+    assert_equal(inMap.length, result.args['outMap'].length)
+    result.args['outMap'].each do |k,v|
+      assert_equal(inMap[k], v)
+    end
+
+    assert_equal(inList.length, result.args['outList'].length)
+    for idx in 0...inList.length
+      assert_equal(inList[idx], result.args['outList'][idx])
+    end
+  end
+
 end
 
 app = ConsoleTest.new
diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp
index 067f53471f..1f08dded94 100644
--- a/qpid/cpp/src/qmf/engine/Agent.cpp
+++ b/qpid/cpp/src/qmf/engine/Agent.cpp
@@ -356,8 +356,7 @@ void AgentImpl::heartbeat()
     QPID_LOG(trace, "SENT HeartbeatIndication");
 }
 
-void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text,
-                                     const Value& argMap)
+void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& argMap)
 {
     Mutex::ScopedLock _lock(lock);
     map<uint32_t, AgentQueryContext::Ptr>::iterator iter = contextMap.find(sequence);
@@ -366,7 +365,32 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text,
     AgentQueryContext::Ptr context = iter->second;
     contextMap.erase(iter);
 
-    Buffer buffer(outputBuffer, MA_BUFFER_SIZE);
+    char* buf(outputBuffer);
+    uint32_t bufLen(114 + strlen(text)); // header(8) + status(4) + mstring(2 + size) + margin(100)
+    bool allocated(false);
+
+    if (status == 0) {
+        for (vector<const SchemaArgument*>::const_iterator aIter = context->schemaMethod->impl->arguments.begin();
+             aIter != context->schemaMethod->impl->arguments.end(); aIter++) {
+            const SchemaArgument* schemaArg = *aIter;
+            if (schemaArg->getDirection() == DIR_OUT || schemaArg->getDirection() == DIR_IN_OUT) {
+                if (argMap.keyInMap(schemaArg->getName())) {
+                    const Value* val = argMap.byKey(schemaArg->getName());
+                    bufLen += val->impl->encodedSize();
+                } else {
+                    Value val(schemaArg->getType());
+                    bufLen += val.impl->encodedSize();
+                }
+            }
+        }
+    }
+
+    if (bufLen > MA_BUFFER_SIZE) {
+        buf = (char*) malloc(bufLen);
+        allocated = true;
+    }
+
+    Buffer buffer(buf, bufLen);
     Protocol::encodeHeader(buffer, Protocol::OP_METHOD_RESPONSE, context->sequence);
     buffer.putLong(status);
     buffer.putMediumString(text);
@@ -386,6 +410,8 @@ void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text,
         }
     }
     sendBufferLH(buffer, context->exchange, context->key);
+    if (allocated)
+        free(buf);
     QPID_LOG(trace, "SENT MethodResponse seq=" << context->sequence << " status=" << status << " text=" << text);
 }
 
diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp
index 69acc2dd37..5fc71979fd 100644
--- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp
+++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp
@@ -269,6 +269,31 @@ string BrokerProxyImpl::encodeMethodArguments(const SchemaMethod* schema, const
     return string();
 }
 
+string BrokerProxyImpl::encodedSizeMethodArguments(const SchemaMethod* schema, const Value* argmap, uint32_t& size)
+{
+    int argCount = schema->getArgumentCount();
+
+    if (argmap == 0 || !argmap->isMap())
+        return string("Arguments must be in a map value");
+
+    for (int aIdx = 0; aIdx < argCount; aIdx++) {
+        const SchemaArgument* arg(schema->getArgument(aIdx));
+        if (arg->getDirection() == DIR_IN || arg->getDirection() == DIR_IN_OUT) {
+            if (argmap->keyInMap(arg->getName())) {
+                const Value* argVal(argmap->byKey(arg->getName()));
+                if (argVal->getType() != arg->getType())
+                    return string("Argument is the wrong type: ") + arg->getName();
+                size += argVal->impl->encodedSize();
+            } else {
+                Value defaultValue(arg->getType());
+                size += defaultValue.impl->encodedSize();
+            }
+        }
+    }
+
+    return string();
+}
+
 void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls,
                                         const string& methodName, const Value* args, void* userContext)
 {
@@ -280,7 +305,23 @@ void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass*
             Mutex::ScopedLock _lock(lock);
             SequenceContext::Ptr methodContext(new MethodContext(*this, userContext, method));
             stringstream key;
-            Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
+            char* buf(outputBuffer);
+            uint32_t bufLen(1024);
+            bool allocated(false);
+
+            string argErrorString = encodedSizeMethodArguments(method, args, bufLen);
+            if (!argErrorString.empty()) {
+                MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString));
+                eventQueue.push_back(eventMethodResponse(userContext, argError));
+                return;
+            }
+
+            if (bufLen > MA_BUFFER_SIZE) {
+                buf = (char*) malloc(bufLen);
+                allocated = true;
+            }
+
+            Buffer outBuffer(buf, bufLen);
             uint32_t sequence(seqMgr.reserve(methodContext));
 
             Protocol::encodeHeader(outBuffer, Protocol::OP_METHOD_REQUEST, sequence);
@@ -288,15 +329,14 @@ void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass*
             cls->getClassKey()->impl->encode(outBuffer);
             outBuffer.putShortString(methodName);
 
-            string argErrorString = encodeMethodArguments(method, args, outBuffer);
-            if (argErrorString.empty()) {
-                key << "agent.1." << oid->impl->getAgentBank();
-                sendBufferLH(outBuffer, QMF_EXCHANGE, key.str());
-                QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str());
-            } else {
-                MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString));
-                eventQueue.push_back(eventMethodResponse(userContext, argError));
-            }
+            encodeMethodArguments(method, args, outBuffer);
+            key << "agent.1." << oid->impl->getAgentBank();
+            sendBufferLH(outBuffer, QMF_EXCHANGE, key.str());
+            QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str());
+
+            if (allocated)
+                free(buf);
+
             return;
         }
     }
diff --git a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h
index 494da5e239..0542b67dbb 100644
--- a/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h
+++ b/qpid/cpp/src/qmf/engine/BrokerProxyImpl.h
@@ -142,6 +142,7 @@ namespace engine {
         void sendQuery(const Query& query, void* context, const AgentProxy* agent);
         bool sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent);
         std::string encodeMethodArguments(const SchemaMethod* schema, const Value* args, qpid::framing::Buffer& buffer);
+        std::string encodedSizeMethodArguments(const SchemaMethod* schema, const Value* args, uint32_t& size);
         void sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, const std::string& method, const Value* args, void* context);
 
         void addBinding(const std::string& exchange, const std::string& key);
diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.cpp b/qpid/cpp/src/qmf/engine/ValueImpl.cpp
index 409bf64c35..f9ebbf5028 100644
--- a/qpid/cpp/src/qmf/engine/ValueImpl.cpp
+++ b/qpid/cpp/src/qmf/engine/ValueImpl.cpp
@@ -394,6 +394,51 @@ void ValueImpl::encode(Buffer& buf) const
     }
 }
 
+uint32_t ValueImpl::encodedSize() const
+{
+    FieldTable ft;
+    List fl;
+
+    switch (typecode) {
+    case TYPE_UINT8     :
+    case TYPE_BOOL      :
+    case TYPE_INT8      : return 1;
+ 
+    case TYPE_UINT16    :
+    case TYPE_INT16     : return 2;
+
+    case TYPE_UINT32    :
+    case TYPE_INT32     :
+    case TYPE_FLOAT     : return 4;
+
+    case TYPE_UINT64    :
+    case TYPE_INT64     :
+    case TYPE_DOUBLE    :
+    case TYPE_ABSTIME   :
+    case TYPE_DELTATIME : return 8;
+
+    case TYPE_UUID      : 
+    case TYPE_REF       : return 16;
+
+    case TYPE_SSTR      : return 1 + stringVal.size();
+    case TYPE_LSTR      : return 2 + stringVal.size();
+    case TYPE_MAP:
+        mapToFieldTable(ft);
+        return ft.encodedSize();
+
+    case TYPE_LIST:
+        listToFramingList(fl);
+        return fl.encodedSize();
+
+    case TYPE_ARRAY:
+    case TYPE_OBJECT:
+    default:
+        break;
+    }
+
+    return 0;
+}
+
 bool ValueImpl::keyInMap(const char* key) const
 {
     return typecode == TYPE_MAP && mapVal.count(key) > 0;
diff --git a/qpid/cpp/src/qmf/engine/ValueImpl.h b/qpid/cpp/src/qmf/engine/ValueImpl.h
index 3b535834fd..8de8c5329f 100644
--- a/qpid/cpp/src/qmf/engine/ValueImpl.h
+++ b/qpid/cpp/src/qmf/engine/ValueImpl.h
@@ -79,6 +79,7 @@ namespace engine {
         ~ValueImpl();
 
         void encode(qpid::framing::Buffer& b) const;
+        uint32_t encodedSize() const;
 
         Typecode getType() const { return typecode; }
         bool isNull() const { return !valid; }