QPID-3883: Using application headers in messages causes a very large slowdown

Encode Variants directly without translating to FieldTables

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

1 files changed, 224 insertions, 81 deletions

diff --git a/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp b/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
index b976a5d09b..da2c84d6d5 100644
--- a/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
+++ b/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
@@ -52,9 +52,7 @@ template <class T, class U, class F> void convert(const T& from, U& to, F f)
 }
 
 Variant::Map::value_type toVariantMapEntry(const FieldTable::value_type& in);
-FieldTable::value_type toFieldTableEntry(const Variant::Map::value_type& in);
 Variant toVariant(boost::shared_ptr<FieldValue> in);
-boost::shared_ptr<FieldValue> toFieldValue(const Variant& in);
 
 template <class T, class U, class F> void translate(boost::shared_ptr<FieldValue> in, U& u, F f) 
 {
@@ -70,20 +68,6 @@ template <class T, class U, class F> T* toFieldValueCollection(const U& u, F f)
     return new T(t);
 }
 
-FieldTableValue* toFieldTableValue(const Variant::Map& map) 
-{
-    FieldTable ft;
-    convert(map, ft, &toFieldTableEntry);
-    return new FieldTableValue(ft);
-}
-
-ListValue* toListValue(const Variant::List& list) 
-{
-    List l;
-    convert(list, l, &toFieldValue);
-    return new ListValue(l);
-}
-
 void setEncodingFor(Variant& out, uint8_t code)
 {
     switch(code){
@@ -151,7 +135,7 @@ Variant toVariant(boost::shared_ptr<FieldValue> in)
 
       case 0xf0: break;//void, which is the default value for Variant
       case 0xf1: out.setEncoding(amqp0_10_bit); break;//treat 'bit' as void, which is the default value for Variant
-        
+
         //Variable Width types:
         //strings:
       case 0x80: 
@@ -217,89 +201,229 @@ boost::shared_ptr<FieldValue> convertString(const std::string& value, const std:
     }
 }
 
-boost::shared_ptr<FieldValue> toFieldValue(const Variant& in)
+Variant::Map::value_type toVariantMapEntry(const FieldTable::value_type& in)
 {
-    boost::shared_ptr<FieldValue> out;
-    switch (in.getType()) {
-      case VAR_VOID: out = boost::shared_ptr<FieldValue>(new VoidValue()); break;
-      case VAR_BOOL: out = boost::shared_ptr<FieldValue>(new BoolValue(in.asBool())); break;
-      case VAR_UINT8: out = boost::shared_ptr<FieldValue>(new Unsigned8Value(in.asUint8())); break;
-      case VAR_UINT16: out = boost::shared_ptr<FieldValue>(new Unsigned16Value(in.asUint16())); break;
-      case VAR_UINT32: out = boost::shared_ptr<FieldValue>(new Unsigned32Value(in.asUint32())); break;
-      case VAR_UINT64: out = boost::shared_ptr<FieldValue>(new Unsigned64Value(in.asUint64())); break;
-      case VAR_INT8: out = boost::shared_ptr<FieldValue>(new Integer8Value(in.asInt8())); break;
-      case VAR_INT16: out = boost::shared_ptr<FieldValue>(new Integer16Value(in.asInt16())); break;
-      case VAR_INT32: out = boost::shared_ptr<FieldValue>(new Integer32Value(in.asInt32())); break;
-      case VAR_INT64: out = boost::shared_ptr<FieldValue>(new Integer64Value(in.asInt64())); break;
-      case VAR_FLOAT: out = boost::shared_ptr<FieldValue>(new FloatValue(in.asFloat())); break;
-      case VAR_DOUBLE: out = boost::shared_ptr<FieldValue>(new DoubleValue(in.asDouble())); break;
-      case VAR_STRING: out = convertString(in.asString(), in.getEncoding()); break;
-      case VAR_UUID: out = boost::shared_ptr<FieldValue>(new UuidValue(in.asUuid().data())); break;
-      case VAR_MAP: 
-        out = boost::shared_ptr<FieldValue>(toFieldTableValue(in.asMap()));
-        break;
-      case VAR_LIST: 
-        out = boost::shared_ptr<FieldValue>(toListValue(in.asList()));
+    return Variant::Map::value_type(in.first, toVariant(in.second));
+}
+
+struct DecodeBuffer
+{
+    Buffer buffer;
+
+    DecodeBuffer(const std::string& s) : buffer(const_cast<char*>(s.data()), s.size()) {}
+
+    template <class T> void decode(T& t) { t.decode(buffer); }
+
+};
+
+template <class T, class U, class F> void _decode(const std::string& data, U& value, F f)
+{
+    T t;
+    DecodeBuffer buffer(data);
+    buffer.decode(t);
+    convert(t, value, f);
+}
+
+uint32_t encodedSize(const Variant::Map& values);
+uint32_t encodedSize(const Variant::List& values);
+uint32_t encodedSize(const std::string& value);
+
+uint32_t encodedSize(const Variant& value)
+{
+    switch (value.getType()) {
+      case VAR_VOID:
+        return 0;
+      case VAR_BOOL:
+      case VAR_UINT8:
+      case VAR_INT8:
+        return 1;
+      case VAR_UINT16:
+      case VAR_INT16:
+        return 2;
         break;
+      case VAR_UINT32:
+      case VAR_INT32:
+      case VAR_FLOAT:
+        return 4;
+      case VAR_UINT64:
+      case VAR_INT64:
+      case VAR_DOUBLE:
+        return 8;
+      case VAR_UUID:
+        return 16;
+      case VAR_MAP:
+        return encodedSize(value.asMap());
+      case VAR_LIST:
+        return encodedSize(value.asList());
+      case VAR_STRING:
+        return encodedSize(value.getString());
+      default:
+        throw Exception("Couldn't encode Variant: Illegal type code");
     }
-    return out;
 }
 
-Variant::Map::value_type toVariantMapEntry(const FieldTable::value_type& in)
+uint32_t encodedSize(const Variant::Map& values)
 {
-    return Variant::Map::value_type(in.first, toVariant(in.second));
+    uint32_t size = 4/*size field*/ + 4/*count field*/;
+    for(Variant::Map::const_iterator i = values.begin(); i != values.end(); ++i) {
+        size += 1/*size of key*/ + (i->first).size() + 1/*typecode*/ + encodedSize(i->second);
+    }
+    return size;
 }
 
-FieldTable::value_type toFieldTableEntry(const Variant::Map::value_type& in)
+uint32_t encodedSize(const Variant::List& values)
 {
-    return FieldTable::value_type(in.first, toFieldValue(in.second));
+    uint32_t size = 4/*size field*/ + 4/*count field*/;
+    for(Variant::List::const_iterator i = values.begin(); i != values.end(); ++i) {
+        size += 1/*typecode*/ + encodedSize(*i);
+    }
+    return size;
 }
 
-struct EncodeBuffer
+uint32_t encodedSize(const std::string& value)
 {
-    char* data;
-    Buffer buffer;
+    uint32_t size = value.size();
+    if (size > std::numeric_limits<uint16_t>::max()) {
+        return size + 4; /*Long size*/
+    } else {
+        return size + 2; /*Short size*/
+    }
+}
 
-    EncodeBuffer(size_t size) : data(new char[size]), buffer(data, size) {}
-    ~EncodeBuffer() { delete[] data; }
+void encode(const std::string& value, const std::string& encoding, qpid::framing::Buffer& buffer)
+{
+    uint32_t size = value.size();
+    if (size > std::numeric_limits<uint16_t>::max()) {
+        if (encoding == utf8 || encoding == utf16 || encoding == iso885915) {
+            throw Exception(QPID_MSG("Could not encode " << encoding << " character string - too long (" << size << " bytes)"));
+        } else {
+            buffer.putOctet(0xa0);
+            buffer.putLong(size);
+            buffer.putRawData(value);
+        }
+    } else {
+        if (encoding == utf8) {
+            buffer.putOctet(0x95);
+        } else if (encoding == utf16) {
+            buffer.putOctet(0x96);
+        } else if (encoding == iso885915) {
+            buffer.putOctet(0x94);
+        } else {
+            buffer.putOctet(0x90);
+        }
+        buffer.putShort(size);
+        buffer.putRawData(value);
+    }
+}
 
-    template <class T> void encode(T& t) { t.encode(buffer); }
+void encode(const Variant::Map& map, uint32_t len, qpid::framing::Buffer& buffer);
+void encode(const Variant::List& list, uint32_t len, qpid::framing::Buffer& buffer);
 
-    void getData(std::string& s) { 
-        s.assign(data, buffer.getSize()); 
+void encode(const Variant& value, qpid::framing::Buffer& buffer)
+{
+    switch (value.getType()) {
+      case VAR_VOID:
+        buffer.putOctet(0xf0);
+        break;
+      case VAR_BOOL:
+        buffer.putOctet(0x08);
+        buffer.putOctet(value.asBool());
+        break;
+      case VAR_INT8:
+        buffer.putOctet(0x01);
+        buffer.putInt8(value.asInt8());
+        break;
+      case VAR_UINT8:
+        buffer.putOctet(0x02);
+        buffer.putOctet(value.asUint8());
+        break;
+      case VAR_INT16:
+        buffer.putOctet(0x11);
+        buffer.putInt16(value.asInt16());
+        break;
+      case VAR_UINT16:
+        buffer.putOctet(0x12);
+        buffer.putShort(value.asUint16());
+        break;
+      case VAR_INT32:
+        buffer.putOctet(0x21);
+        buffer.putInt32(value.asInt32());
+        break;
+      case VAR_UINT32:
+        buffer.putOctet(0x22);
+        buffer.putLong(value.asUint32());
+        break;
+      case VAR_FLOAT:
+        buffer.putOctet(0x23);
+        buffer.putFloat(value.asFloat());
+        break;
+      case VAR_INT64:
+        buffer.putOctet(0x31);
+        buffer.putInt64(value.asInt64());
+        break;
+      case VAR_UINT64:
+        buffer.putOctet(0x32);
+        buffer.putLongLong(value.asUint64());
+        break;
+      case VAR_DOUBLE:
+        buffer.putOctet(0x33);
+        buffer.putDouble(value.asDouble());
+        break;
+      case VAR_UUID:
+        buffer.putOctet(0x48);
+        buffer.putBin128(value.asUuid().data());
+        break;
+      case VAR_MAP:
+        buffer.putOctet(0xa8);
+        encode(value.asMap(), encodedSize(value.asMap()), buffer);
+        break;
+      case VAR_LIST:
+        buffer.putOctet(0xa9);
+        encode(value.asList(), encodedSize(value.asList()), buffer);
+        break;
+      case VAR_STRING:
+        encode(value.getString(), value.getEncoding(), buffer);
+        break;
     }
-};
+}
 
-struct DecodeBuffer
+void encode(const Variant::Map& map, uint32_t len, qpid::framing::Buffer& buffer)
 {
-    Buffer buffer;
-
-    DecodeBuffer(const std::string& s) : buffer(const_cast<char*>(s.data()), s.size()) {}
-
-    template <class T> void decode(T& t) { t.decode(buffer); }
-    
-};
+    uint32_t s = buffer.getPosition();
+    buffer.putLong(len - 4);//exclusive of the size field itself
+    buffer.putLong(map.size());
+    for (Variant::Map::const_iterator i = map.begin(); i != map.end(); ++i) {
+        buffer.putShortString(i->first);
+    	encode(i->second, buffer);
+    }
+    assert(s + len == buffer.getPosition());
+}
 
-template <class T, class U, class F> void _encode(const U& value, std::string& data, F f)
+void encode(const Variant::List& list, uint32_t len, qpid::framing::Buffer& buffer)
 {
-    T t;
-    convert(value, t, f);
-    EncodeBuffer buffer(t.encodedSize());
-    buffer.encode(t);
-    buffer.getData(data);
+    uint32_t s = buffer.getPosition();
+    buffer.putLong(len - 4);//exclusive of the size field itself
+    buffer.putLong(list.size());
+    for (Variant::List::const_iterator i = list.begin(); i != list.end(); ++i) {
+    	encode(*i, buffer);
+    }
+    assert(s + len == buffer.getPosition());
 }
 
-template <class T, class U, class F> void _decode(const std::string& data, U& value, F f)
+void decode(qpid::framing::Buffer&, Variant::Map&)
 {
-    T t;
-    DecodeBuffer buffer(data);
-    buffer.decode(t);
-    convert(t, value, f);
 }
 
+
 void MapCodec::encode(const Variant::Map& value, std::string& data)
 {
-    _encode<FieldTable>(value, data, &toFieldTableEntry);
+    uint32_t len = qpid::amqp_0_10::encodedSize(value);
+    std::vector<char> space(len);
+    qpid::framing::Buffer buff(&space[0], len);
+
+    qpid::amqp_0_10::encode(value, len, buff);
+    assert( len == buff.getPosition() );
+    data.assign(&space[0], len);
 }
 
 void MapCodec::decode(const std::string& data, Variant::Map& value)
@@ -309,14 +433,18 @@ void MapCodec::decode(const std::string& data, Variant::Map& value)
 
 size_t MapCodec::encodedSize(const Variant::Map& value)
 {
-    std::string encoded;
-    encode(value, encoded);
-    return encoded.size();
+    return qpid::amqp_0_10::encodedSize(value);
 }
 
 void ListCodec::encode(const Variant::List& value, std::string& data)
 {
-    _encode<List>(value, data, &toFieldValue);
+    uint32_t len = qpid::amqp_0_10::encodedSize(value);
+    std::vector<char> space(len);
+    qpid::framing::Buffer buff(&space[0], len);
+
+    qpid::amqp_0_10::encode(value, len, buff);
+    assert( len == buff.getPosition() );
+    data.assign(&space[0], len);
 }
 
 void ListCodec::decode(const std::string& data, Variant::List& value)
@@ -326,14 +454,29 @@ void ListCodec::decode(const std::string& data, Variant::List& value)
 
 size_t ListCodec::encodedSize(const Variant::List& value)
 {
-    std::string encoded;
-    encode(value, encoded);
-    return encoded.size();
+    return qpid::amqp_0_10::encodedSize(value);
 }
 
 void translate(const Variant::Map& from, FieldTable& to)
 {
-    convert(from, to, &toFieldTableEntry);
+    // Create buffer of correct size to encode Variant::Map
+    uint32_t len = encodedSize(from);
+    std::vector<char> space(len);
+    qpid::framing::Buffer buff(&space[0], len);
+
+    // Encode Variant::Map into buffer directly -
+    // We pass the already calculated length in to avoid
+    // recalculating it.
+    encode(from, len, buff);
+
+    // Give buffer to FieldTable
+    // Could speed this up a bit by avoiding copying
+    // the buffer we just created into the FieldTable
+    assert( len == buff.getPosition() );
+    buff.reset();
+    to.decode(buff);
+
+    //convert(from, to, &toFieldTableEntry);
 }
 
 void translate(const FieldTable& from, Variant::Map& to)