Reduce impact of media plugin message parsing when many media surfaces present with binary llsd

Author: RyeMutt

indra/llplugin/llplugininstance.cpp

@@ -102,7 +102,7 @@ void LLPluginInstance::sendMessage(const std::string &message)
     if(mPluginSendMessageFunction)
     {
         LL_DEBUGS("Plugin") << "sending message to plugin: \"" << message << "\"" << LL_ENDL;
-        mPluginSendMessageFunction(message.c_str(), &mPluginUserData);
+        mPluginSendMessageFunction(message.data(), message.size(), &mPluginUserData);
     }
     else
     {
@@ -119,28 +119,31 @@ void LLPluginInstance::idle(void)
 }
 
 // static
-void LLPluginInstance::staticReceiveMessage(const char *message_string, void **user_data)
+void LLPluginInstance::staticReceiveMessage(const char *message_string, size_t message_size, void **user_data)
 {
     // TODO: validate that the user_data argument is still a valid LLPluginInstance pointer
     // we could also use a key that's looked up in a map (instead of a direct pointer) for safety, but that's probably overkill
     LLPluginInstance *self = (LLPluginInstance*)*user_data;
-    self->receiveMessage(message_string);
+    self->receiveMessage(message_string, message_size);
 }
 
 /**
  * Plugin receives message from plugin loader shell.
  *
- * @param[in] message_string Message
+ * @param[in] message_string Message bytes (binary LLSD, may contain embedded NULs)
+ * @param[in] message_size Length of message_string in bytes
  */
-void LLPluginInstance::receiveMessage(const char *message_string)
+void LLPluginInstance::receiveMessage(const char *message_string, size_t message_size)
 {
+    // Reconstruct the message with its exact length: the payload is binary and
+    // may contain embedded NULs, so it can't be treated as a C string.
+    std::string message(message_string, message_size);
     if(mOwner)
     {
-        LL_DEBUGS("Plugin") << "processing incoming message: \"" << message_string << "\"" << LL_ENDL;
-        mOwner->receivePluginMessage(message_string);
+        mOwner->receivePluginMessage(message);
     }
     else
     {
-        LL_WARNS("Plugin") << "dropping incoming message: \"" << message_string << "\"" << LL_ENDL;
+        LL_WARNS("Plugin") << "dropping incoming message: \"" << message << "\"" << LL_ENDL;
     }
 }

indra/llplugin/llplugininstance.h

@@ -67,10 +67,11 @@ class LLPluginInstance
 
     /** The signature of the function for sending a message from plugin to plugin loader shell.
     *
-     * @param[in] message_string Null-terminated C string
+     * @param[in] message_string Message bytes (may contain embedded NULs - it is binary LLSD)
+     * @param[in] message_size Length of message_string in bytes
     * @param[in] user_data The opaque reference that the callee supplied during setup.
     */
-    typedef void (*sendMessageFunction) (const char *message_string, void **user_data);
+    typedef void (*sendMessageFunction) (const char *message_string, size_t message_size, void **user_data);
 
     /** The signature of the plugin init function. TODO:DOC check direction (pluging loader shell to plugin?)
     *
@@ -87,8 +88,8 @@ class LLPluginInstance
     static void setStaticInitFunction(pluginInitFunction func);
 
 private:
-    static void staticReceiveMessage(const char *message_string, void **user_data);
-    void receiveMessage(const char *message_string);
+    static void staticReceiveMessage(const char *message_string, size_t message_size, void **user_data);
+    void receiveMessage(const char *message_string, size_t message_size);
 
     void *mPluginUserData;
     sendMessageFunction mPluginSendMessageFunction;

indra/llplugin/llpluginmessage.cpp

@@ -358,8 +358,8 @@ std::string LLPluginMessage::generate(void) const
     std::ostringstream result;
 
     // Pretty XML may be slightly easier to deal with while debugging...
-//  LLSDSerialize::toXML(mMessage, result);
-    LLSDSerialize::toPrettyXML(mMessage, result);
+//  LLSDSerialize::toPrettyXML(mMessage, result);
+    LLSDSerialize::toBinary(mMessage, result);
 
     return result.str();
 }
@@ -374,9 +374,10 @@ int LLPluginMessage::parse(const std::string &message)
     // clear any previous state
     clear();
 
-    std::istringstream input(message);
+    boost::iostreams::stream<boost::iostreams::array_source> input(message.data(), message.size());
 
-    S32 parse_result = LLSDSerialize::fromXML(mMessage, input);
+    //S32 parse_result = LLSDSerialize::fromXML(mMessage, input);
+    S32 parse_result = LLSDSerialize::fromBinary(mMessage, input, LLSDSerialize::SIZE_UNLIMITED);
 
     return (int)parse_result;
 }

indra/llplugin/llpluginmessagepipe.cpp

@@ -33,7 +33,11 @@
 
 #include "llapr.h"
 
-static const char MESSAGE_DELIMITER = '\0';
+// Each message on the wire is framed with a 4-byte big-endian length prefix
+// followed by exactly that many payload bytes. A single delimiter byte can't
+// be used because the payload is now binary LLSD, which contains embedded NULs
+// (and every other byte value) in its length/size fields.
+static constexpr size_t MESSAGE_HEADER_SIZE = 4;
 
 LLPluginMessagePipeOwner::LLPluginMessagePipeOwner() :
     mMessagePipe(NULL),
@@ -121,8 +125,16 @@ bool LLPluginMessagePipe::addMessage(const std::string &message)
         mOutputStartIndex = 0;
     }
 
+    // Frame the message with a 4-byte big-endian length prefix.
+    U32 msg_size = static_cast<U32>(message.size());
+    const char header[MESSAGE_HEADER_SIZE] = {
+        static_cast<char>((msg_size >> 24) & 0xFF),
+        static_cast<char>((msg_size >> 16) & 0xFF),
+        static_cast<char>((msg_size >> 8) & 0xFF),
+        static_cast<char>(msg_size & 0xFF)
+    };
+    mOutput.append(header, MESSAGE_HEADER_SIZE);
     mOutput += message;
-    mOutput += MESSAGE_DELIMITER;   // message separator
 
     return true;
 }
@@ -177,9 +189,10 @@ bool LLPluginMessagePipe::pumpOutput()
 
         LLMutexLock lock(&mOutputMutex);
 
-        const char * output_data = &(mOutput.data()[mOutputStartIndex]);
-        if(*output_data != '\0')
+        if(mOutput.size() > mOutputStartIndex)
         {
+            const char * output_data = &(mOutput.data()[mOutputStartIndex]);
+
             // write any outgoing messages
             in_size = (apr_size_t) (mOutput.size() - mOutputStartIndex);
             out_size = in_size;
@@ -370,27 +383,39 @@ bool LLPluginMessagePipe::pumpInput(F64 timeout)
 
 void LLPluginMessagePipe::processInput(void)
 {
-    // Look for input delimiter(s) in the input buffer.
-    size_t delim;
     mInputMutex.lock();
-    while((delim = mInput.find(MESSAGE_DELIMITER)) != std::string::npos)
+    // Each message is a 4-byte big-endian length followed by that many payload
+    // bytes. Dispatch complete messages, leaving any trailing partial message
+    // in the buffer to be completed by a later read.
+    while(mInput.size() >= MESSAGE_HEADER_SIZE)
     {
-        // Let the owner process this message
-        if (mOwner)
+        const U8* buf = reinterpret_cast<const U8*>(mInput.data());
+        size_t msg_size =
+            (static_cast<size_t>(buf[0]) << 24) |
+            (static_cast<size_t>(buf[1]) << 16) |
+            (static_cast<size_t>(buf[2]) << 8) |
+             static_cast<size_t>(buf[3]);
+
+        if(mInput.size() < MESSAGE_HEADER_SIZE + msg_size)
         {
-            // Pull the message out of the input buffer before calling receiveMessageRaw.
-            // It's now possible for this function to get called recursively (in the case where the plugin makes a blocking request)
-            // and this guarantees that the messages will get dequeued correctly.
-            std::string message(mInput, 0, delim);
-            mInput.erase(0, delim + 1);
-            mInputMutex.unlock();
-            mOwner->receiveMessageRaw(message);
-            mInputMutex.lock();
+            // The full message body hasn't arrived yet; wait for more.
+            break;
         }
-        else
+
+        if(!mOwner)
         {
             LL_WARNS("Plugin") << "!mOwner" << LL_ENDL;
+            break;
         }
+
+        // Pull the message out of the input buffer before calling receiveMessageRaw.
+        // It's now possible for this function to get called recursively (in the case where the plugin makes a blocking request)
+        // and this guarantees that the messages will get dequeued correctly.
+        std::string message(mInput, MESSAGE_HEADER_SIZE, msg_size);
+        mInput.erase(0, MESSAGE_HEADER_SIZE + msg_size);
+        mInputMutex.unlock();
+        mOwner->receiveMessageRaw(message);
+        mInputMutex.lock();
     }
     mInputMutex.unlock();
 }

indra/media_plugins/base/media_plugin_base.cpp

@@ -104,13 +104,14 @@ void MediaPluginBase::setStatus(EStatus status)
  * @param[in] user_data Message data
  *
  */
-void MediaPluginBase::staticReceiveMessage(const char *message_string, void **user_data)
+void MediaPluginBase::staticReceiveMessage(const char *message_string, size_t message_size, void **user_data)
 {
     MediaPluginBase *self = (MediaPluginBase*)*user_data;
 
     if(self != NULL)
     {
-        self->receiveMessage(message_string);
+        // Reconstruct with the exact length: binary LLSD may contain NULs.
+        self->receiveMessage(std::string(message_string, message_size));
 
         // If the plugin has processed the delete message, delete it.
         if(self->mDeleteMe)
@@ -130,7 +131,7 @@ void MediaPluginBase::staticReceiveMessage(const char *message_string, void **us
 void MediaPluginBase::sendMessage(const LLPluginMessage &message)
 {
     std::string output = message.generate();
-    mHostSendFunction(output.c_str(), &mHostUserData);
+    mHostSendFunction(output.data(), output.size(), &mHostUserData);
 }
 
 /**

indra/media_plugins/base/media_plugin_base.h

@@ -82,9 +82,9 @@ class MediaPluginBase
     virtual ~MediaPluginBase() {}
 
    /** Handle received message from plugin loader shell. */
-    virtual void receiveMessage(const char *message_string) = 0;
+    virtual void receiveMessage(const std::string &message_string) = 0;
 
-    static void staticReceiveMessage(const char *message_string, void **user_data);
+    static void staticReceiveMessage(const char *message_string, size_t message_size, void **user_data);
 
 protected:
 

indra/media_plugins/cef/media_plugin_cef.cpp

@@ -234,7 +234,7 @@ class MediaPluginCEF :
     ~MediaPluginCEF();
 
     /*virtual*/
-    void receiveMessage(const char* message_string);
+    void receiveMessage(const std::string& message_string);
 
 private:
     bool init();
@@ -1078,7 +1078,7 @@ void MediaPluginCEF::authResponse(LLPluginMessage &message)
 
 ////////////////////////////////////////////////////////////////////////////////
 //
-void MediaPluginCEF::receiveMessage(const char* message_string)
+void MediaPluginCEF::receiveMessage(const std::string& message_string)
 {
     //  std::cerr << "MediaPluginCEF::receiveMessage: received message: \"" << message_string << "\"" << std::endl;
     LLPluginMessage message_in;

indra/media_plugins/example/media_plugin_example.cpp

@@ -45,7 +45,7 @@ class mediaPluginExample :
     mediaPluginExample(LLPluginInstance::sendMessageFunction host_send_func, void *host_user_data);
     ~mediaPluginExample();
 
-    /*virtual*/ void receiveMessage(const char* message_string);
+    /*virtual*/ void receiveMessage(const std::string& message_string);
 
 private:
     bool init();
@@ -89,7 +89,7 @@ mediaPluginExample::~mediaPluginExample()
 
 ////////////////////////////////////////////////////////////////////////////////
 //
-void mediaPluginExample::receiveMessage(const char* message_string)
+void mediaPluginExample::receiveMessage(const std::string& message_string)
 {
     //  std::cerr << "MediaPluginWebKit::receiveMessage: received message: \"" << message_string << "\"" << std::endl;
     LLPluginMessage message_in;

indra/media_plugins/gstreamer10/media_plugin_gstreamer10.cpp

@@ -66,7 +66,7 @@ class MediaPluginGStreamer10 : public MediaPluginBase
     MediaPluginGStreamer10(LLPluginInstance::sendMessageFunction host_send_func, void *host_user_data);
     ~MediaPluginGStreamer10();
 
-    /* virtual */ void receiveMessage(const char *message_string);
+    /* virtual */ void receiveMessage(const std::string &message_string);
 
     static bool startup();
     static bool closedown();
@@ -743,7 +743,7 @@ std::string MediaPluginGStreamer10::getVersion()
     return plugin_version;
 }
 
-void MediaPluginGStreamer10::receiveMessage(const char *message_string)
+void MediaPluginGStreamer10::receiveMessage(const std::string &message_string)
 {
     LLPluginMessage message_in;
 

indra/media_plugins/libvlc/media_plugin_libvlc.cpp

@@ -56,7 +56,7 @@ class MediaPluginLibVLC :
     MediaPluginLibVLC(LLPluginInstance::sendMessageFunction host_send_func, void *host_user_data);
     ~MediaPluginLibVLC();
 
-    /*virtual*/ void receiveMessage(const char* message_string);
+    /*virtual*/ void receiveMessage(const std::string& message_string);
 
 private:
     bool init();
@@ -464,7 +464,7 @@ void MediaPluginLibVLC::setVolume(const F64 volume)
 
 ////////////////////////////////////////////////////////////////////////////////
 //
-void MediaPluginLibVLC::receiveMessage(const char* message_string)
+void MediaPluginLibVLC::receiveMessage(const std::string& message_string)
 {
     LLPluginMessage message_in;