From 91d37082ab008e3e3d1cae21ac285a01eb57fbd8 Mon Sep 17 00:00:00 2001 From: Simon Fels Date: Fri, 18 Nov 2016 12:53:27 +0100 Subject: [PATCH] First attempt on getting rid of the socket server --- external/android-emugl/CMakeLists.txt | 1 + .../host/include/libOpenglRender/IOStream.h | 4 +- .../host/libs/GLESv1_dec/CMakeLists.txt | 2 +- .../host/libs/GLESv2_dec/CMakeLists.txt | 2 +- .../libs/renderControl_dec/CMakeLists.txt | 2 +- src/anbox/graphics/emugl/ColorBuffer.cpp | 5 + src/anbox/graphics/emugl/FrameBuffer.cpp | 8 + src/anbox/graphics/emugl/LayerManager.cpp | 4 +- src/anbox/graphics/emugl/RenderApi.cpp | 55 ---- src/anbox/graphics/emugl/RenderControl.cpp | 5 + src/anbox/graphics/emugl/RenderThread.cpp | 51 ++-- src/anbox/graphics/gl_renderer_server.cpp | 3 + .../graphics/opengles_message_processor.cpp | 238 +++++++++++++----- .../graphics/opengles_message_processor.h | 1 + src/anbox/network/base_socket_messenger.cpp | 11 + src/anbox/network/base_socket_messenger.h | 1 + src/anbox/network/message_sender.h | 2 + 17 files changed, 243 insertions(+), 152 deletions(-) diff --git a/external/android-emugl/CMakeLists.txt b/external/android-emugl/CMakeLists.txt index 29edadb..b328b17 100644 --- a/external/android-emugl/CMakeLists.txt +++ b/external/android-emugl/CMakeLists.txt @@ -3,6 +3,7 @@ set(CMAKE_C_FLAGS "-Wall") include_directories( + ${CMAKE_SOURCE_DIR}/src ${CMAKE_SOURCE_DIR}/external/android-emugl/shared ${CMAKE_SOURCE_DIR}/external/android-emugl/host/include ${CMAKE_SOURCE_DIR}/external/android-emugl/shared/OpenglCodecCommon diff --git a/external/android-emugl/host/include/libOpenglRender/IOStream.h b/external/android-emugl/host/include/libOpenglRender/IOStream.h index 672f5a7..9bb671e 100644 --- a/external/android-emugl/host/include/libOpenglRender/IOStream.h +++ b/external/android-emugl/host/include/libOpenglRender/IOStream.h @@ -21,6 +21,8 @@ #include "ErrorLog.h" +#include "anbox/logger.h" + class IOStream { public: @@ -70,7 +72,7 @@ public: } int flush() { - + DEBUG("buf %p free %d buf size %d", m_buf, m_free, m_bufsize); if (!m_buf || m_free == m_bufsize) return 0; int stat = commitBuffer(m_bufsize - m_free); diff --git a/external/android-emugl/host/libs/GLESv1_dec/CMakeLists.txt b/external/android-emugl/host/libs/GLESv1_dec/CMakeLists.txt index e8d83e2..3377489 100644 --- a/external/android-emugl/host/libs/GLESv1_dec/CMakeLists.txt +++ b/external/android-emugl/host/libs/GLESv1_dec/CMakeLists.txt @@ -14,7 +14,7 @@ add_custom_command( set(SOURCES GLESv1Decoder.cpp) -if ("${cmake_build_type_lower}" STREQUAL "trace") +if ("${cmake_build_type_lower}" STREQUAL "debug") set(OPENGL_DEBUG "-DOPENGL_DEBUG_PRINTOUT -DCHECK_GL_ERROR") set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${OPENGL_DEBUG}") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${OPENGL_DEBUG}") diff --git a/external/android-emugl/host/libs/GLESv2_dec/CMakeLists.txt b/external/android-emugl/host/libs/GLESv2_dec/CMakeLists.txt index 140d061..34a700e 100644 --- a/external/android-emugl/host/libs/GLESv2_dec/CMakeLists.txt +++ b/external/android-emugl/host/libs/GLESv2_dec/CMakeLists.txt @@ -14,7 +14,7 @@ add_custom_command( set(SOURCES GLESv2Decoder.cpp) -if ("${cmake_build_type_lower}" STREQUAL "trace") +if ("${cmake_build_type_lower}" STREQUAL "debug") set(OPENGL_DEBUG "-DOPENGL_DEBUG_PRINTOUT -DCHECK_GL_ERROR") set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${OPENGL_DEBUG}") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${OPENGL_DEBUG}") diff --git a/external/android-emugl/host/libs/renderControl_dec/CMakeLists.txt b/external/android-emugl/host/libs/renderControl_dec/CMakeLists.txt index cb67c5a..5bc3558 100644 --- a/external/android-emugl/host/libs/renderControl_dec/CMakeLists.txt +++ b/external/android-emugl/host/libs/renderControl_dec/CMakeLists.txt @@ -10,7 +10,7 @@ add_custom_command( WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} DEPENDS emugen) -if ("${cmake_build_type_lower}" STREQUAL "trace") +if ("${cmake_build_type_lower}" STREQUAL "debug") set(OPENGL_DEBUG "-DOPENGL_DEBUG_PRINTOUT -DCHECK_GL_ERROR") set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${OPENGL_DEBUG}") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${OPENGL_DEBUG}") diff --git a/src/anbox/graphics/emugl/ColorBuffer.cpp b/src/anbox/graphics/emugl/ColorBuffer.cpp index 270ccac..8687654 100644 --- a/src/anbox/graphics/emugl/ColorBuffer.cpp +++ b/src/anbox/graphics/emugl/ColorBuffer.cpp @@ -255,12 +255,17 @@ void ColorBuffer::readPixels(int x, GLenum p_format, GLenum p_type, void* pixels) { + DEBUG(""); + ScopedHelperContext context(m_helper); if (!context.isOk()) { return; } + DEBUG("Before bind"); + if (bindFbo(&m_fbo, m_tex)) { + DEBUG("Off to GL to read pixels"); s_gles2.glReadPixels(x, y, width, height, p_format, p_type, pixels); unbindFbo(); } diff --git a/src/anbox/graphics/emugl/FrameBuffer.cpp b/src/anbox/graphics/emugl/FrameBuffer.cpp index 64b8323..68ed50a 100644 --- a/src/anbox/graphics/emugl/FrameBuffer.cpp +++ b/src/anbox/graphics/emugl/FrameBuffer.cpp @@ -545,7 +545,9 @@ HandleType FrameBuffer::genHandle() HandleType FrameBuffer::createColorBuffer(int p_width, int p_height, GLenum p_internalFormat) { + DEBUG("width %d height %d", p_width, p_height); emugl::Mutex::AutoLock mutex(m_lock); + DEBUG("Got lock"); HandleType ret = 0; ColorBufferPtr cb(ColorBuffer::create( @@ -748,10 +750,16 @@ void FrameBuffer::readColorBuffer(HandleType p_colorbuffer, int x, int y, int width, int height, GLenum format, GLenum type, void *pixels) { + DEBUG("handle %d x %d y %d width %d height %d", + p_colorbuffer, x, y, width, height); + emugl::Mutex::AutoLock mutex(m_lock); + DEBUG("Got lock"); + ColorBufferMap::iterator c( m_colorbuffers.find(p_colorbuffer) ); if (c == m_colorbuffers.end()) { + DEBUG("Didn't found color buffer"); // bad colorbuffer handle return; } diff --git a/src/anbox/graphics/emugl/LayerManager.cpp b/src/anbox/graphics/emugl/LayerManager.cpp index ab2e74c..c7bb97c 100644 --- a/src/anbox/graphics/emugl/LayerManager.cpp +++ b/src/anbox/graphics/emugl/LayerManager.cpp @@ -73,8 +73,8 @@ LayerManager::~LayerManager() { } void LayerManager::post_layer(const LayerInfo &layer) { - if (is_layer_blacklisted(layer.name)) - return; + // if (is_layer_blacklisted(layer.name)) + // return; FrameBufferWindow *window = nullptr; for (auto &l : layers_) { diff --git a/src/anbox/graphics/emugl/RenderApi.cpp b/src/anbox/graphics/emugl/RenderApi.cpp index d0b40e9..c0f742e 100644 --- a/src/anbox/graphics/emugl/RenderApi.cpp +++ b/src/anbox/graphics/emugl/RenderApi.cpp @@ -79,61 +79,6 @@ RENDER_APICALL int RENDER_APIENTRY initOpenGLRenderer( set_emugl_crash_reporter(crashfunc); set_emugl_logger(logfuncs.coarse); set_emugl_cxt_logger(logfuncs.fine); - // - // Fail if renderer is already initialized - // - if (s_renderThread) { - return false; - } - - // kUseThread is used to determine whether the RenderWindow should use - // a separate thread to manage its subwindow GL/GLES context. - // For now, this feature is disabled entirely for the following - // reasons: - // - // - It must be disabled on Windows at all times, otherwise the main window becomes - // unresponsive after a few seconds of user interaction (e.g. trying to - // move it over the desktop). Probably due to the subtle issues around - // input on this platform (input-queue is global, message-queue is - // per-thread). Also, this messes considerably the display of the - // main window when running the executable under Wine. - // - // - On Linux/XGL and OSX/Cocoa, this used to be necessary to avoid corruption - // issues with the GL state of the main window when using the SDL UI. - // After the switch to Qt, this is no longer necessary and may actually cause - // undesired interactions between the UI thread and the RenderWindow thread: - // for example, in a multi-monitor setup the context might be recreated when - // dragging the window between monitors, triggering a Qt-specific callback - // in the context of RenderWindow thread, which will become blocked on the UI - // thread, which may in turn be blocked on something else. - bool kUseThread = false; - - // - // initialize the renderer and listen to connections - // on a thread in the current process. - // - s_renderWindow = new RenderWindow(native_display, kUseThread); - if (!s_renderWindow) { - ERR("Could not create rendering window class"); - GL_LOG("Could not create rendering window class"); - return false; - } - if (!s_renderWindow->isValid()) { - ERR("Could not initialize emulated framebuffer\n"); - delete s_renderWindow; - s_renderWindow = NULL; - return false; - } - - s_renderThread = RenderServer::create(addr, addrLen); - if (!s_renderThread) { - return false; - } - strncpy(s_renderAddr, addr, sizeof(s_renderAddr)); - - s_renderThread->start(); - - GL_LOG("OpenGL renderer initialized successfully"); return true; } diff --git a/src/anbox/graphics/emugl/RenderControl.cpp b/src/anbox/graphics/emugl/RenderControl.cpp index 942e7a4..a147870 100644 --- a/src/anbox/graphics/emugl/RenderControl.cpp +++ b/src/anbox/graphics/emugl/RenderControl.cpp @@ -25,6 +25,8 @@ #include "OpenGLESDispatch/EGLDispatch.h" +#include "anbox/logger.h" + #include #include @@ -236,6 +238,7 @@ static void rcDestroyWindowSurface(uint32_t windowSurface) static uint32_t rcCreateColorBuffer(uint32_t width, uint32_t height, GLenum internalFormat) { + DEBUG(""); FrameBuffer *fb = FrameBuffer::getFB(); if (!fb) { return 0; @@ -351,6 +354,8 @@ static void rcReadColorBuffer(uint32_t colorBuffer, GLint width, GLint height, GLenum format, GLenum type, void* pixels) { + DEBUG(""); + FrameBuffer *fb = FrameBuffer::getFB(); if (!fb) { return; diff --git a/src/anbox/graphics/emugl/RenderThread.cpp b/src/anbox/graphics/emugl/RenderThread.cpp index db9c6a9..be61412 100644 --- a/src/anbox/graphics/emugl/RenderThread.cpp +++ b/src/anbox/graphics/emugl/RenderThread.cpp @@ -26,15 +26,19 @@ #include "OpenGLESDispatch/GLESv1Dispatch.h" #include "../../../shared/OpenglCodecCommon/ChecksumCalculatorThreadInfo.h" +#include "anbox/logger.h" + #define STREAM_BUFFER_SIZE 4*1024*1024 RenderThread::RenderThread(IOStream *stream, emugl::Mutex *lock) : emugl::Thread(), m_lock(lock), - m_stream(stream) {} + m_stream(stream) { + DEBUG(""); +} RenderThread::~RenderThread() { - delete m_stream; + DEBUG(""); } // static @@ -43,78 +47,75 @@ RenderThread* RenderThread::create(IOStream *stream, emugl::Mutex *lock) { } void RenderThread::forceStop() { + DEBUG(""); m_stream->forceStop(); } intptr_t RenderThread::main() { RenderThreadInfo tInfo; + // Not used below but will store a reference of itself in TLS so that + // it can be accessed down the stack in the same thread when decoding + // any of the commands. ChecksumCalculatorThreadInfo tChecksumInfo; - // - // initialize decoders - // tInfo.m_glDec.initGL(gles1_dispatch_get_proc_func, NULL); tInfo.m_gl2Dec.initGL(gles2_dispatch_get_proc_func, NULL); initRenderControlContext(&tInfo.m_rcDec); ReadBuffer readBuf(STREAM_BUFFER_SIZE); - while (1) { + DEBUG("Started"); + while (1) { int stat = readBuf.getData(m_stream); if (stat <= 0) { + DEBUG("Connection closed"); break; } + DEBUG("Got %d bytes for decoding", readBuf.validData()); + bool progress; do { progress = false; - m_lock->lock(); - // - // try to process some of the command buffer using the GLESv1 decoder - // + DEBUG("Locked"); + size_t last = tInfo.m_glDec.decode(readBuf.buf(), readBuf.validData(), m_stream); if (last > 0) { + DEBUG("Ran GL commands"); progress = true; readBuf.consume(last); } - // - // try to process some of the command buffer using the GLESv2 decoder - // last = tInfo.m_gl2Dec.decode(readBuf.buf(), readBuf.validData(), m_stream); if (last > 0) { + DEBUG("Ran GL2 commands"); progress = true; readBuf.consume(last); } - // - // try to process some of the command buffer using the - // renderControl decoder - // last = tInfo.m_rcDec.decode(readBuf.buf(), readBuf.validData(), m_stream); if (last > 0) { + DEBUG("Ran RC commands"); readBuf.consume(last); progress = true; } m_lock->unlock(); - - } while( progress ); + DEBUG("Unlocked"); + } while (progress); } - // + DEBUG("Shutting down"); + // Release references to the current thread's context/surfaces if any - // FrameBuffer::getFB()->bindContext(0, 0, 0); - if (tInfo.currContext || tInfo.currDrawSurf || tInfo.currReadSurf) { - fprintf(stderr, "ERROR: RenderThread exiting with current context/surfaces\n"); - } + if (tInfo.currContext || tInfo.currDrawSurf || tInfo.currReadSurf) + ERROR("Exiting with current context/surfaces"); FrameBuffer::getFB()->drainWindowSurface(); - FrameBuffer::getFB()->drainRenderContext(); return 0; diff --git a/src/anbox/graphics/gl_renderer_server.cpp b/src/anbox/graphics/gl_renderer_server.cpp index 00d8ec1..dcd8e78 100644 --- a/src/anbox/graphics/gl_renderer_server.cpp +++ b/src/anbox/graphics/gl_renderer_server.cpp @@ -71,6 +71,9 @@ void GLRendererServer::start() { log_funcs.coarse = logger_write; log_funcs.fine = logger_write; + // HACK: This will do nothing but set our log functions + initOpenGLRenderer(nullptr, nullptr, 0, log_funcs, logger_write); + FrameBuffer::initialize(window_creator_->native_display()); } } // namespace graphics diff --git a/src/anbox/graphics/opengles_message_processor.cpp b/src/anbox/graphics/opengles_message_processor.cpp index 3452eff..c067917 100644 --- a/src/anbox/graphics/opengles_message_processor.cpp +++ b/src/anbox/graphics/opengles_message_processor.cpp @@ -26,122 +26,227 @@ #include #include +#include namespace { -class DirectIOStream : public IOStream { +constexpr const size_t default_buffer_size{384}; +constexpr const size_t max_send_buffer_size{1024}; + +class DelayedIOStream : public IOStream { public: - explicit DirectIOStream(const std::shared_ptr &messenger, - const size_t &buffer_size = 10000) : + typedef std::vector Buffer; + + explicit DelayedIOStream(const std::shared_ptr &messenger, + size_t buffer_size = default_buffer_size) : IOStream(buffer_size), messenger_(messenger) { + // writer_thread_(std::bind(&DelayedIOStream::worker_thread, this)) { } - virtual ~DirectIOStream() { - if (send_buffer_ != nullptr) { - free(send_buffer_); - send_buffer_ = nullptr; - } + virtual ~DelayedIOStream() { + DEBUG(""); + forceStop(); + DEBUG("Shutting down"); } void* allocBuffer(size_t min_size) override { - size_t size = (send_buffer_size_ < min_size ? min_size : send_buffer_size_); - if (!send_buffer_) - send_buffer_ = (unsigned char *) malloc(size); - else if (send_buffer_size_ < size) { - unsigned char *p = (unsigned char *)realloc(send_buffer_, size); - if (p != NULL) { - send_buffer_ = p; - send_buffer_size_ = size; - } else { - free(send_buffer_); - send_buffer_ = NULL; - send_buffer_size_ = 0; - } - } - return send_buffer_; + DEBUG("min size %d", min_size); + if (buffer_.size() < min_size) + buffer_.resize(min_size); + return buffer_.data(); } int commitBuffer(size_t size) override { - messenger_->send(reinterpret_cast(send_buffer_), size); - return size; + DEBUG("size %d", size); + std::unique_lock l(read_mutex_); + +#if 0 + if (buffer_.capacity() <= 2 * size) { + buffer_.resize(size); + out_queue_.push(std::move(buffer_)); + } else { + out_queue_.push(Buffer(buffer_.data(), buffer_.data() + size)); + } + DEBUG("Submitted data into output queue (%d bytes)", size); + can_write_.notify_all(); +#else + ssize_t bytes_left = size; + while (bytes_left > 0) { + const ssize_t written = messenger_->send_raw(buffer_.data() + (size - bytes_left), bytes_left); + if (written < 0 ) { + if (errno != EINTR) { + ERROR("Failed to write data: %s", std::strerror(errno)); + break; + } + WARNING("Socket busy, trying again"); + } else + bytes_left -= written; + } + + DEBUG("Sent data to remote (%d bytes)", buffer_.size()); +#endif + return static_cast(size); } - const unsigned char* readFully(void*, size_t) override { - ERROR("Not implemented"); - return nullptr; + const unsigned char* readFully(void *buffer, size_t length) override { + size_t size = length; + auto data = read(buffer, &size); + if (size < length) + return nullptr; + return data; } - const unsigned char* read(void *data, size_t *size) override { - if (!wait_for_data() || buffer_.size() == 0) { - *size = 0; + const unsigned char* read(void *buffer, size_t *length) override { + std::unique_lock l(read_mutex_); + + if (stopped_) { + DEBUG("Aborting"); return nullptr; } - auto bytes_to_read = *size; - if (bytes_to_read > buffer_.size()) - bytes_to_read = buffer_.size(); + if (current_read_buffer_left_ == 0 && in_queue_.empty()) { + DEBUG("Waiting for data to be available"); + can_read_.wait(l); - ::memcpy(data, buffer_.data(), bytes_to_read); - buffer_.erase(buffer_.begin(), buffer_.begin() + bytes_to_read); + if (stopped_) { + DEBUG("Aborting"); + return nullptr; + } + } - *size = bytes_to_read; + DEBUG("Trying to read %d bytes", *length); + size_t read = 0; + auto buf = static_cast(buffer); + const auto buffer_end = buf + *length; + while (buf != buffer_end) { + if (current_read_buffer_left_ == 0) { + // If we don't have anymore buffers we need to stop reading here + if (in_queue_.empty()) + break; - return static_cast(data); + current_read_buffer_ = in_queue_.front(); + in_queue_.pop(); + current_read_buffer_left_ = current_read_buffer_.size(); + } + + const size_t current_size = std::min(buffer_end - buf, + current_read_buffer_left_); + ::memcpy(buffer, current_read_buffer_.data() + + (current_read_buffer_.size() - current_read_buffer_left_), + current_size); + + read += current_size; + buf += current_size; + current_read_buffer_left_ -= current_size; + + DEBUG("Size %d, left to read %d", current_size, current_read_buffer_left_); + } + + if (read == 0) + return nullptr; + + *length = read; + + DEBUG("Read %d bytes (buffers left %d)", read, in_queue_.size()); + + return buf; } - int writeFully(const void*, size_t) override { + int writeFully(const void *buffer, size_t length) override { + (void) buffer; + (void) length; ERROR("Not implemented"); - return 0; + return -1; } void forceStop() override { - std::unique_lock l(mutex_); - buffer_.clear(); + DEBUG(""); + stopped_ = true; + can_read_.notify_all(); + can_write_.notify_all(); } - void submitData(const std::vector &data) { - std::unique_lock l(mutex_); - for (const auto &byte : data) - buffer_.push_back(byte); - // buffer_.insert(buffer_.end(), data.begin(), data.end()); - lock_.notify_one(); + void post_data(const Buffer &buffer) { + DEBUG("Got data and waiting for lock"); + std::unique_lock l(read_mutex_); + DEBUG("Received %d bytes", buffer.size()); + in_queue_.push(std::move(buffer)); + can_read_.notify_all(); } private: - bool wait_for_data() { - std::unique_lock l(mutex_); + void worker_thread() { + DEBUG("Running send thread"); + while (true) { + std::unique_lock l(write_mutex_); + while (out_queue_.empty() && !stopped_) { + can_write_.wait(l, [&]() { return !out_queue_.empty() || stopped_; }); + DEBUG("Woke up (queue size %d)", out_queue_.size()); + } - if (!l.owns_lock()) - return false; + if (stopped_) + break; - lock_.wait(l, [&]() { return !buffer_.empty(); }); - return true; + DEBUG("Going to send out %d bytes", out_queue_.front().size()); + + auto buffer = out_queue_.front(); + out_queue_.pop(); + + ssize_t bytes_left = buffer.size(); + while (bytes_left > 0) { + const ssize_t written = messenger_->send_raw(buffer.data() + (buffer.size() - bytes_left), bytes_left); + if (written < 0 ) { + if (errno != EINTR) { + ERROR("Failed to write data: %s", std::strerror(errno)); + break; + } + WARNING("Socket busy, trying again"); + } else + bytes_left -= written; + } + DEBUG("Sent %d bytes to client (queue size %d)", buffer.size(), out_queue_.size()); + } + + DEBUG("Shutting down"); } std::shared_ptr messenger_; - std::mutex mutex_; - std::condition_variable lock_; - std::vector buffer_; - unsigned char *send_buffer_ = nullptr; - size_t send_buffer_size_ = 0; + Buffer buffer_; + std::thread writer_thread_; + std::queue out_queue_; + Buffer current_write_buffer_; + size_t current_write_buffer_left_ = 0; + std::queue in_queue_; + Buffer current_read_buffer_; + size_t current_read_buffer_left_ = 0; + std::mutex write_mutex_; + std::mutex read_mutex_; + std::condition_variable can_write_; + std::condition_variable can_read_; + bool stopped_ = false; }; } namespace anbox { namespace graphics { emugl::Mutex OpenGlesMessageProcessor::global_lock{}; - +static int next_id = 0; OpenGlesMessageProcessor::OpenGlesMessageProcessor(const std::shared_ptr &messenger) : messenger_(messenger), - stream_(std::make_shared(messenger_)), - renderer_(RenderThread::create(stream_.get(), &global_lock)) { + id_(next_id++), + stream_(std::make_shared(messenger_)) { // We have to read the client flags first before we can continue // processing the actual commands - std::array buffer; - messenger_->receive_msg(boost::asio::buffer(buffer)); + unsigned int client_flags = 0; + auto err = messenger_->receive_msg(boost::asio::buffer(&client_flags, sizeof(unsigned int))); + if (err) + ERROR("%s", err.message()); + renderer_.reset(RenderThread::create(stream_.get(), &global_lock)); renderer_->start(); + + DEBUG("Started new OpenGL ES message processor"); } OpenGlesMessageProcessor::~OpenGlesMessageProcessor() { @@ -151,8 +256,9 @@ OpenGlesMessageProcessor::~OpenGlesMessageProcessor() { } bool OpenGlesMessageProcessor::process_data(const std::vector &data) { - auto stream = std::static_pointer_cast(stream_); - stream->submitData(data); + DEBUG("[%d] Got %d bytes", id_, data.size()); + auto stream = std::static_pointer_cast(stream_); + stream->post_data(DelayedIOStream::Buffer(data.data(), data.data() + data.size())); return true; } } // namespace graphics diff --git a/src/anbox/graphics/opengles_message_processor.h b/src/anbox/graphics/opengles_message_processor.h index 374af97..97381d1 100644 --- a/src/anbox/graphics/opengles_message_processor.h +++ b/src/anbox/graphics/opengles_message_processor.h @@ -45,6 +45,7 @@ private: static emugl::Mutex global_lock; std::shared_ptr messenger_; + int id_; std::shared_ptr stream_; std::shared_ptr renderer_; }; diff --git a/src/anbox/network/base_socket_messenger.cpp b/src/anbox/network/base_socket_messenger.cpp index 1cce830..fa75cae 100644 --- a/src/anbox/network/base_socket_messenger.cpp +++ b/src/anbox/network/base_socket_messenger.cpp @@ -74,6 +74,16 @@ Credentials BaseSocketMessenger::creds() const { return {cr.pid, cr.uid, cr.gid}; } +template +ssize_t BaseSocketMessenger::send_raw(char const* data, size_t length) +{ + VariableLengthArray whole_message{length}; + std::copy(data, data + length, whole_message.data()); + + std::unique_lock lg(message_lock); + return ::send(socket_fd, data, length, 0); +} + template void BaseSocketMessenger::send(char const* data, size_t length) { @@ -88,6 +98,7 @@ void BaseSocketMessenger::send(char const* data, size_t length) boost::asio::transfer_all()); } catch (const boost::system::system_error &err) { + DEBUG("Got error: %s", err.what()); if (err.code() == boost::asio::error::try_again) continue; } diff --git a/src/anbox/network/base_socket_messenger.h b/src/anbox/network/base_socket_messenger.h index c28f037..243374a 100644 --- a/src/anbox/network/base_socket_messenger.h +++ b/src/anbox/network/base_socket_messenger.h @@ -37,6 +37,7 @@ public: unsigned short local_port() const override; void send(char const* data, size_t length) override; + ssize_t send_raw(char const* data, size_t length) override; void async_receive_msg(AnboxReadHandler const& handle, boost::asio::mutable_buffers_1 const &buffer) override; boost::system::error_code receive_msg(boost::asio::mutable_buffers_1 const& buffer) override; size_t available_bytes() override; diff --git a/src/anbox/network/message_sender.h b/src/anbox/network/message_sender.h index 750715d..ab4a10d 100644 --- a/src/anbox/network/message_sender.h +++ b/src/anbox/network/message_sender.h @@ -20,6 +20,7 @@ #define ANBOX_NETWORK_MESSAGE_SENDER_H_ #include +#include namespace anbox { namespace network { @@ -27,6 +28,7 @@ class MessageSender { public: virtual void send(char const* data, size_t length) = 0; + virtual ssize_t send_raw(char const* data, size_t length) = 0; protected: MessageSender() = default;