loolwsd/LOOLSession.cpp | 20 ++++--
loolwsd/LOOLSession.hpp | 9 ++-
loolwsd/LOOLWSD.cpp | 109 +++++++++++++++++++++++++++++---------
loolwsd/LoadTest.cpp | 3 -
loolwsd/loolwsd-systemplate-setup | 2
loolwsd/protocol.txt | 8 ++
loolwsd/tsqueue.h | 65 ++++++++++++++++++++++
7 files changed, 180 insertions(+), 36 deletions(-)
New commits:
commit 9e98a3898b0825c9d57b25df9905700a904c96c3
Author: Tor Lillqvist <t...@collabora.com>
Date: Fri Jun 5 16:12:06 2015 +0300
Add a "canceltiles" message to the protocol and handle it
Implementing this Was harder than I first expected. The basic idea is as
follows: The master process puts each message arriving from a client that
isn't "canceltiles" into a (client-specific) queue. A separate thread that
pulls messages from the queue at its own pace and handles them as
before. Incoming "canceltiles" messages are handled specially, though: The
queue is emptied of "tile" messages.
The above sounds simple but there are several details that were a bit tricky
to get right.
diff --git a/loolwsd/LOOLSession.cpp b/loolwsd/LOOLSession.cpp
index 5b30949..159955b 100644
--- a/loolwsd/LOOLSession.cpp
+++ b/loolwsd/LOOLSession.cpp
@@ -87,11 +87,21 @@ LOOLSession::~LOOLSession()
void LOOLSession::sendTextFrame(const std::string& text)
{
+ std::unique_lock<std::mutex> lock(_mutex);
+
_ws->sendFrame(text.data(), text.size());
}
void LOOLSession::sendBinaryFrame(const char *buffer, int length)
{
+ std::unique_lock<std::mutex> lock(_mutex);
+
+ if (length > 1000)
+ {
+ std::string nextmessage = "nextmessage: size=" +
std::to_string(length);
+ _ws->sendFrame(nextmessage.data(), nextmessage.size());
+ }
+
_ws->sendFrame(buffer, length, WebSocket::FRAME_BINARY);
}
@@ -529,8 +539,6 @@ void MasterProcessSession::sendTile(const char *buffer, int
length, StringTokeni
cachedTile->read(output.data() + pos, size);
cachedTile->close();
- sendTextFrame("nextmessage: size=" + std::to_string(output.size()));
-
sendBinaryFrame(output.data(), output.size());
return;
@@ -628,7 +636,7 @@ void MasterProcessSession::forwardToPeer(const char
*buffer, int length)
auto peer = _peer.lock();
if (!peer)
return;
- peer->_ws->sendFrame(buffer, length, WebSocket::FRAME_BINARY);
+ peer->sendBinaryFrame(buffer, length);
}
ChildProcessSession::ChildProcessSession(std::shared_ptr<WebSocket> ws,
LibreOfficeKit *loKit) :
@@ -893,8 +901,6 @@ void ChildProcessSession::sendTile(const char *buffer, int
length, StringTokeniz
delete[] pixmap;
- sendTextFrame("nextmessage: size=" + std::to_string(output.size()));
-
sendBinaryFrame(output.data(), output.size());
}
diff --git a/loolwsd/LOOLSession.hpp b/loolwsd/LOOLSession.hpp
index 93fee3c..2fd7d0b 100644
--- a/loolwsd/LOOLSession.hpp
+++ b/loolwsd/LOOLSession.hpp
@@ -69,6 +69,9 @@ protected:
// In the master, the actual URL. In the child, the copy inside the chroot
jail.
std::string _docURL;
+
+private:
+ std::mutex _mutex;
};
template<typename charT, typename traits>
diff --git a/loolwsd/LOOLWSD.cpp b/loolwsd/LOOLWSD.cpp
index 18c21b3..cdced5f 100644
--- a/loolwsd/LOOLWSD.cpp
+++ b/loolwsd/LOOLWSD.cpp
@@ -86,6 +86,7 @@ DEALINGS IN THE SOFTWARE.
#include "LOOLProtocol.hpp"
#include "LOOLSession.hpp"
#include "LOOLWSD.hpp"
+#include "tsqueue.h"
#include "Util.hpp"
using namespace LOOLProtocol;
@@ -120,6 +121,36 @@ using Poco::Util::Option;
using Poco::Util::OptionSet;
using Poco::Util::ServerApplication;
+class FromClientQueueHandler: public Runnable
+{
+public:
+ FromClientQueueHandler(tsqueue<std::string>& queue):
+ _queue(queue)
+ {
+ }
+
+ void setSession(std::shared_ptr<MasterProcessSession> session)
+ {
+ _session = session;
+ }
+
+ void run() override
+ {
+ while (true)
+ {
+ std::string input = _queue.get();
+ if (input == "eof")
+ break;
+ if (!_session->handleInput(input.c_str(), input.size()))
+ break;
+ }
+ }
+
+private:
+ std::shared_ptr<MasterProcessSession> _session;
+ tsqueue<std::string>& _queue;
+};
+
class WebSocketRequestHandler: public HTTPRequestHandler
/// Handle a WebSocket connection.
{
@@ -139,26 +170,37 @@ public:
}
Application& app = Application::instance();
+
+ tsqueue<std::string> queue;
+ Thread queueHandlerThread;
+ FromClientQueueHandler handler(queue);
+
try
{
try
{
std::shared_ptr<WebSocket> ws(new WebSocket(request,
response));
- std::shared_ptr<MasterProcessSession> session;
+ LOOLSession::Kind kind;
if (request.getURI() == LOOLWSD::CHILD_URI &&
request.serverAddress().port() == LOOLWSD::MASTER_PORT_NUMBER)
- {
- session.reset(new MasterProcessSession(ws,
LOOLSession::Kind::ToPrisoner));
- }
+ kind = LOOLSession::Kind::ToPrisoner;
else
+ kind = LOOLSession::Kind::ToClient;
+
+ std::shared_ptr<MasterProcessSession> session(new
MasterProcessSession(ws, kind));
+
+ // For ToClient sessions, we store incoming messages in a
queue and have a separate
+ // thread that handles them. This is so that we can empty the
queue when we get a
+ // "canceltiles" message.
+ if (kind == LOOLSession::Kind::ToClient)
{
- session.reset(new MasterProcessSession(ws,
LOOLSession::Kind::ToClient));
+ handler.setSession(session);
+ queueHandlerThread.start(handler);
}
- // Loop, receiving WebSocket messages either from the
- // client, or from the child process (to be forwarded to
- // the client).
+ // Loop, receiving WebSocket messages either from the client,
or from the child
+ // process (to be forwarded to the client).
int flags;
int n;
ws->setReceiveTimeout(0);
@@ -167,25 +209,43 @@ public:
char buffer[100000];
n = ws->receiveFrame(buffer, sizeof(buffer), flags);
+ std::string firstLine = getFirstLine(buffer, n);
+ StringTokenizer tokens(firstLine, " ",
StringTokenizer::TOK_IGNORE_EMPTY | StringTokenizer::TOK_TRIM);
+
if (n > 0 && (flags & WebSocket::FRAME_OP_BITMASK) !=
WebSocket::FRAME_OP_CLOSE)
{
- if (!session->handleInput(buffer, n))
- n = 0;
- }
- if (n > 0 && (flags & WebSocket::FRAME_OP_BITMASK) !=
WebSocket::FRAME_OP_CLOSE)
- {
- std::string firstLine = getFirstLine(buffer, n);
- StringTokenizer tokens(firstLine, " ",
StringTokenizer::TOK_IGNORE_EMPTY | StringTokenizer::TOK_TRIM);
-
- int size;
- if (tokens.count() == 2 && tokens[0] == "nextmessage:"
&& getTokenInteger(tokens[1], "size", size) && size > 0)
+ if (kind == LOOLSession::Kind::ToClient &&
firstLine.size() == static_cast<std::string::size_type>(n))
{
- char largeBuffer[size];
-
- n = ws->receiveFrame(largeBuffer, size, flags);
- if (n > 0 && (flags & WebSocket::FRAME_OP_BITMASK)
!= WebSocket::FRAME_OP_CLOSE)
+ // Check if it is a "canceltiles" and in that case
remove outstanding
+ // "tile" messages from the queue.
+ if (tokens.count() == 1 && tokens[0] ==
"canceltiles")
+ {
+ queue.remove_if([](std::string& x){ return
x.find("tile ") == 0;});
+ }
+ else
+ {
+ queue.put(firstLine);
+ }
+ }
+ else
+ {
+ // Check if it is a "nextmessage:" and in that
case read the large
+ // follow-up message separately, and handle that
only.
+ int size;
+ if (tokens.count() == 2 && tokens[0] ==
"nextmessage:" && getTokenInteger(tokens[1], "size", size) && size > 0)
+ {
+ char largeBuffer[size];
+
+ n = ws->receiveFrame(largeBuffer, size, flags);
+ if (n > 0 && (flags &
WebSocket::FRAME_OP_BITMASK) != WebSocket::FRAME_OP_CLOSE)
+ {
+ if (!session->handleInput(largeBuffer, n))
+ n = 0;
+ }
+ }
+ else
{
- if (!session->handleInput(largeBuffer, n))
+ if (!session->handleInput(buffer, n))
n = 0;
}
}
@@ -215,6 +275,9 @@ public:
{
app.logger().error(Util::logPrefix() + "IOException: " +
exc.message());
}
+ queue.clear();
+ queue.put("eof");
+ queueHandlerThread.join();
}
};
diff --git a/loolwsd/LoadTest.cpp b/loolwsd/LoadTest.cpp
index 7b7ca38..c0c2eed 100644
--- a/loolwsd/LoadTest.cpp
+++ b/loolwsd/LoadTest.cpp
@@ -241,8 +241,9 @@ private:
break;
}
y = ((y + 1) % ((output._height-1)/DOCTILESIZE + 1));
- Thread::sleep(200);
+ Thread::sleep(50);
}
+ sendTextFrame(ws, "canceltiles");
Thread::sleep(10000);
diff --git a/loolwsd/protocol.txt b/loolwsd/protocol.txt
index 82d3b90..655dcba 100644
--- a/loolwsd/protocol.txt
+++ b/loolwsd/protocol.txt
@@ -11,6 +11,12 @@ tiles proactively (guessing what the client might need). Etc.
client -> server
================
+canceltiles
+
+ All outstanding tile messages from the client to the server are
+ dropped and will not be handled. There is no guarantee of exacely
+ which tile: messages might still be sent back to the client.
+
invalidatetiles part=<partNumber> tileposx=<xpos> tileposy=<ypos>
tilewidth=<tileWidth> tileheight=<tileHeight>
All parameters are numbers. Makes the server remove any cached
@@ -88,7 +94,7 @@ nextmessage: size=<byteSize>
message). Can be ignored by clients using an API that can read
arbitrarily large buffers from a WebSocket (like JavaScript), but
must be handled by clients that cannot (like those using Poco
- 1.6.0).
+ 1.6.0, like the "loadtest" program in the loolwsd sources).
status: type=<typeName> parts=<numberOfParts> current=<currentPartNumber>
width=<width> height=<height>
diff --git a/loolwsd/tsqueue.h b/loolwsd/tsqueue.h
new file mode 100644
index 0000000..9511efb
--- /dev/null
+++ b/loolwsd/tsqueue.h
@@ -0,0 +1,65 @@
+/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4;
fill-column: 100 -*- */
+/*
+ * This file is part of the LibreOffice project.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+#ifndef INCLUDED_TSQUEUE_H
+#define INCLUDED_TSQUEUE_H
+
+#include "config.h"
+
+#include <condition_variable>
+#include <mutex>
+#include <deque>
+
+// Thread-safe queue
+
+template <class T>
+class tsqueue
+{
+public:
+ void put(const T& value)
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _queue.push_back(value);
+ lock.unlock();
+ _cv.notify_one();
+ }
+
+ T get()
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _cv.wait(lock, [this] { return _queue.size() > 0; });
+ T result = _queue.front();
+ _queue.pop_front();
+ return result;
+ }
+
+ void clear()
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ while (_queue.size())
+ _queue.pop_front();
+ }
+
+ template<class UnaryPredicate>
+ void remove_if(UnaryPredicate p)
+ {
+ std::unique_lock<std::mutex> lock(_mutex);
+ _queue.erase(std::remove_if(_queue.begin(), _queue.end(), p),
+ _queue.end());
+ }
+
+private:
+ std::mutex _mutex;
+ std::condition_variable _cv;
+ std::deque<T> _queue;
+};
+
+#endif
+
+/* vim:set shiftwidth=4 softtabstop=4 expandtab: */
commit f849c5f31e43517804294b129e0c56bb65906f3c
Author: Tor Lillqvist <t...@collabora.com>
Date: Mon Jun 8 16:58:34 2015 +0300
Clarify usage message
diff --git a/loolwsd/loolwsd-systemplate-setup
b/loolwsd/loolwsd-systemplate-setup
index 8af0517..7e310fb 100755
--- a/loolwsd/loolwsd-systemplate-setup
+++ b/loolwsd/loolwsd-systemplate-setup
@@ -1,6 +1,6 @@
#!/bin/bash
-test $# -eq 2 || { echo "Usage: $0 <chroot template directory> <LO
installation directory>"; exit 1; }
+test $# -eq 2 || { echo "Usage: $0 <chroot template directory for system libs
to create> <LO installation directory>"; exit 1; }
# No provision for spaces or other weird characters in pathnames. So sue me.
commit b5856cb61cd2d1aa185cabf8054da5ef03d4a55c
Author: Tor Lillqvist <t...@collabora.com>
Date: Mon Jun 8 16:35:52 2015 +0300
The buffer parameter to handleInput() can be const
diff --git a/loolwsd/LOOLSession.cpp b/loolwsd/LOOLSession.cpp
index 9843832..5b30949 100644
--- a/loolwsd/LOOLSession.cpp
+++ b/loolwsd/LOOLSession.cpp
@@ -123,7 +123,7 @@ MasterProcessSession::~MasterProcessSession()
}
}
-bool MasterProcessSession::handleInput(char *buffer, int length)
+bool MasterProcessSession::handleInput(const char *buffer, int length)
{
Application::instance().logger().information(Util::logPrefix() + "Input: "
+ getAbbreviatedMessage(buffer, length));
@@ -647,7 +647,7 @@ ChildProcessSession::~ChildProcessSession()
Util::shutdownWebSocket(*_ws);
}
-bool ChildProcessSession::handleInput(char *buffer, int length)
+bool ChildProcessSession::handleInput(const char *buffer, int length)
{
Application& app = Application::instance();
diff --git a/loolwsd/LOOLSession.hpp b/loolwsd/LOOLSession.hpp
index c73456c..93fee3c 100644
--- a/loolwsd/LOOLSession.hpp
+++ b/loolwsd/LOOLSession.hpp
@@ -53,7 +53,7 @@ protected:
const Kind _kind;
- virtual bool handleInput(char *buffer, int length) = 0;
+ virtual bool handleInput(const char *buffer, int length) = 0;
void sendBinaryFrame(const char *buffer, int length);
@@ -94,7 +94,7 @@ public:
MasterProcessSession(std::shared_ptr<Poco::Net::WebSocket> ws, Kind kind);
virtual ~MasterProcessSession();
- virtual bool handleInput(char *buffer, int length) override;
+ virtual bool handleInput(const char *buffer, int length) override;
bool haveSeparateProcess();
@@ -152,7 +152,7 @@ public:
ChildProcessSession(std::shared_ptr<Poco::Net::WebSocket> ws,
LibreOfficeKit *loKit);
virtual ~ChildProcessSession();
- virtual bool handleInput(char *buffer, int length) override;
+ virtual bool handleInput(const char *buffer, int length) override;
virtual bool getStatus(const char *buffer, int length);
_______________________________________________
Libreoffice-commits mailing list
libreoffice-comm...@lists.freedesktop.org
http://lists.freedesktop.org/mailman/listinfo/libreoffice-commits
