Client uses less memory allocations

client/client_connection.cpp

@@ -20,6 +20,7 @@
 #include "common/aixlog.hpp"
 #include "common/snap_exception.hpp"
 #include "common/str_compat.hpp"
+#include "message/factory.hpp"
 #include "message/hello.hpp"
 #include <iostream>
 #include <mutex>
@@ -32,6 +33,8 @@ ClientConnection::ClientConnection(MessageReceiver* receiver, const std::string&
     : socket_(io_context_), active_(false), messageReceiver_(receiver), reqId_(1), host_(host), port_(port), readerThread_(nullptr),
       sumTimeout_(chronos::msec(0))
 {
+    base_msg_size_ = base_message_.getSize();
+    buffer_.resize(base_msg_size_);
 }
 
 
@@ -130,9 +133,10 @@ bool ClientConnection::send(const msg::BaseMessage* message)
 }
 
 
-unique_ptr<msg::SerializedMessage> ClientConnection::sendRequest(const msg::BaseMessage* message, const chronos::msec& timeout)
+
+unique_ptr<msg::BaseMessage> ClientConnection::sendRequest(const msg::BaseMessage* message, const chronos::msec& timeout)
 {
-    unique_ptr<msg::SerializedMessage> response(nullptr);
+    unique_ptr<msg::BaseMessage> response(nullptr);
     if (++reqId_ >= 10000)
         reqId_ = 1;
     message->id = reqId_;
@@ -168,32 +172,26 @@ unique_ptr<msg::SerializedMessage> ClientConnection::sendRequest(const msg::Base
 
 void ClientConnection::getNextMessage()
 {
-    msg::BaseMessage baseMessage;
+    socketRead(&buffer_[0], base_msg_size_);
-    size_t baseMsgSize = baseMessage.getSize();
+    base_message_.deserialize(buffer_.data());
-    vector<char> buffer(baseMsgSize);
-    socketRead(&buffer[0], baseMsgSize);
-    baseMessage.deserialize(&buffer[0]);
     //	LOG(DEBUG) << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " <<
     // baseMessage.refersTo << "\n";
-    if (baseMessage.size > buffer.size())
+    if (base_message_.size > buffer_.size())
-        buffer.resize(baseMessage.size);
+        buffer_.resize(base_message_.size);
     //	{
     //		std::lock_guard<std::mutex> socketLock(socketMutex_);
-    socketRead(&buffer[0], baseMessage.size);
+    socketRead(buffer_.data(), base_message_.size);
     tv t;
-    baseMessage.received = t;
+    base_message_.received = t;
     //	}
 
     { // scope for lock
         std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
         for (auto req : pendingRequests_)
         {
-            if (req->id() == baseMessage.refersTo)
+            if (req->id() == base_message_.refersTo)
             {
-                auto response = make_unique<msg::SerializedMessage>();
+                auto response = msg::factory::createMessage(base_message_, buffer_.data());
-                response->message = baseMessage;
-                response->buffer = (char*)malloc(baseMessage.size);
-                memcpy(response->buffer, &buffer[0], baseMessage.size);
                 req->setValue(std::move(response));
                 return;
             }
@@ -201,7 +199,7 @@ void ClientConnection::getNextMessage()
     }
 
     if (messageReceiver_ != nullptr)
-        messageReceiver_->onMessageReceived(this, baseMessage, &buffer[0]);
+        messageReceiver_->onMessageReceived(this, base_message_, buffer_.data());
 }
 
 

client/client_connection.hpp

@@ -47,7 +47,7 @@ public:
     };
 
     template <typename Rep, typename Period>
-    std::unique_ptr<msg::SerializedMessage> waitForResponse(const std::chrono::duration<Rep, Period>& timeout)
+    std::unique_ptr<msg::BaseMessage> waitForResponse(const std::chrono::duration<Rep, Period>& timeout)
     {
         try
         {
@@ -60,7 +60,7 @@ public:
         return nullptr;
     }
 
-    void setValue(std::unique_ptr<msg::SerializedMessage> value)
+    void setValue(std::unique_ptr<msg::BaseMessage> value)
     {
         promise_.set_value(std::move(value));
     }
@@ -73,8 +73,8 @@ public:
 private:
     uint16_t id_;
 
-    std::promise<std::unique_ptr<msg::SerializedMessage>> promise_;
+    std::promise<std::unique_ptr<msg::BaseMessage>> promise_;
-    std::future<std::unique_ptr<msg::SerializedMessage>> future_;
+    std::future<std::unique_ptr<msg::BaseMessage>> future_;
 };
 
 
@@ -109,18 +109,24 @@ public:
     virtual bool send(const msg::BaseMessage* message);
 
     /// Send request to the server and wait for answer
-    virtual std::unique_ptr<msg::SerializedMessage> sendRequest(const msg::BaseMessage* message, const chronos::msec& timeout = chronos::msec(1000));
+    virtual std::unique_ptr<msg::BaseMessage> sendRequest(const msg::BaseMessage* message, const chronos::msec& timeout = chronos::msec(1000));
 
     /// Send request to the server and wait for answer of type T
     template <typename T>
     std::unique_ptr<T> sendReq(const msg::BaseMessage* message, const chronos::msec& timeout = chronos::msec(1000))
     {
-        std::unique_ptr<msg::SerializedMessage> reply = sendRequest(message, timeout);
+        std::unique_ptr<msg::BaseMessage> response = sendRequest(message, timeout);
-        if (!reply)
+        if (!response)
             return nullptr;
-        std::unique_ptr<T> msg(new T);
+
-        msg->deserialize(reply->message, reply->buffer);
+        T* tmp = dynamic_cast<T*>(response.get());
-        return msg;
+        std::unique_ptr<T> result;
+        if (tmp != nullptr)
+        {
+            response.release();
+            result.reset(tmp);
+        }
+        return result;
     }
 
     std::string getMacAddress();
@@ -136,6 +142,10 @@ protected:
     void socketRead(void* to, size_t bytes);
     void getNextMessage();
 
+    msg::BaseMessage base_message_;
+    std::vector<char> buffer_;
+    size_t base_msg_size_;
+
     boost::asio::io_context io_context_;
     mutable std::mutex socketMutex_;
     tcp::socket socket_;

client/controller.cpp

@@ -178,6 +178,16 @@ void Controller::start(const PcmDevice& pcmDevice, const std::string& host, size
 }
 
 
+void Controller::run(const PcmDevice& pcmDevice, const std::string& host, size_t port, int latency)
+{
+    pcmDevice_ = pcmDevice;
+    latency_ = latency;
+    clientConnection_.reset(new ClientConnection(this, host, port));
+    worker();
+    // controllerThread_ = thread(&Controller::worker, this);
+}
+
+
 void Controller::stop()
 {
     LOG(DEBUG) << "Stopping Controller" << endl;

client/controller.hpp

@@ -50,6 +50,7 @@ class Controller : public MessageReceiver
 public:
     Controller(const std::string& clientId, size_t instance, std::shared_ptr<MetadataAdapter> meta);
     void start(const PcmDevice& pcmDevice, const std::string& host, size_t port, int latency);
+    void run(const PcmDevice& pcmDevice, const std::string& host, size_t port, int latency);
     void stop();
 
     /// Implementation of MessageReceiver.

client/snapclient.cpp

@@ -215,10 +215,16 @@ int main(int argc, char** argv)
 #endif
 
         bool active = true;
+        std::shared_ptr<Controller> controller;
         auto signal_handler = install_signal_handler({SIGHUP, SIGTERM, SIGINT},
-                                                     [&active](int signal, const std::string& strsignal) {
+                                                     [&active, &controller](int signal, const std::string& strsignal) {
                                                          SLOG(INFO) << "Received signal " << signal << ": " << strsignal << "\n";
                                                          active = false;
+                                                         if (controller)
+                                                         {
+                                                             LOG(INFO) << "Stopping controller\n";
+                                                             controller->stop();
+                                                         }
                                                      });
         if (host.empty())
         {
@@ -258,11 +264,11 @@ int main(int argc, char** argv)
             if (metaStderr)
                 meta.reset(new MetaStderrAdapter);
 
-            std::unique_ptr<Controller> controller(new Controller(hostIdValue->value(), instance, meta));
+            controller = make_shared<Controller>(hostIdValue->value(), instance, meta);
             LOG(INFO) << "Latency: " << latency << "\n";
-            controller->start(pcmDevice, host, port, latency);
+            controller->run(pcmDevice, host, port, latency);
-            signal_handler.wait();
+            // signal_handler.wait();
-            controller->stop();
+            // controller->stop();
         }
     }
     catch (const std::exception& e)

common/message/factory.hpp

@@ -0,0 +1,77 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef MESSAGE_FACTORY_HPP
+#define MESSAGE_FACTORY_HPP
+
+#include "codec_header.hpp"
+#include "hello.hpp"
+#include "server_settings.hpp"
+#include "stream_tags.hpp"
+#include "time.hpp"
+#include "wire_chunk.hpp"
+
+#include "common/str_compat.hpp"
+#include "common/utils.hpp"
+#include "json_message.hpp"
+#include <string>
+
+
+namespace msg
+{
+namespace factory
+{
+
+template <typename T>
+static std::unique_ptr<T> createMessage(const BaseMessage& base_message, char* buffer)
+{
+    std::unique_ptr<T> result = std::make_unique<T>();
+    if (!result)
+        return nullptr;
+    result->deserialize(base_message, buffer);
+    return result;
+}
+
+
+static std::unique_ptr<BaseMessage> createMessage(const BaseMessage& base_message, char* buffer)
+{
+    std::unique_ptr<BaseMessage> result;
+    switch (base_message.type)
+    {
+        case kCodecHeader:
+            return createMessage<CodecHeader>(base_message, buffer);
+        case kHello:
+            return createMessage<Hello>(base_message, buffer);
+        case kServerSettings:
+            return createMessage<ServerSettings>(base_message, buffer);
+        case kStreamTags:
+            return createMessage<StreamTags>(base_message, buffer);
+        case kTime:
+            return createMessage<Time>(base_message, buffer);
+        case kWireChunk:
+            return createMessage<WireChunk>(base_message, buffer);
+        default:
+            return nullptr;
+    }
+}
+
+
+} // namespace factory
+} // namespace msg
+
+#endif

common/message/message.hpp

@@ -291,17 +291,6 @@ protected:
     virtual void doserialize(std::ostream& /*stream*/) const {};
 };
 
-
+} // namespace msg
-struct SerializedMessage
-{
-    ~SerializedMessage()
-    {
-        free(buffer);
-    }
-
-    BaseMessage message;
-    char* buffer;
-};
-}
 
 #endif