libreoffice-online/net/DelaySocket.cpp

262 lines
8.6 KiB
C++
Raw Normal View History

2017-03-23 12:14:51 -05:00
/* -*- 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/.
*/
#include "config.h"
#include "net/DelaySocket.hpp"
2017-04-22 12:37:38 -05:00
#define DELAY_LOG(X) std::cerr << X << "\n";
2017-03-23 12:14:51 -05:00
class Delayer;
// FIXME: TerminatingPoll ?
static SocketPoll DelayPoll("delay_poll");
/// Reads from fd, delays that and then writes to _dest.
class DelaySocket : public Socket {
int _delayMs;
enum State { ReadWrite, // normal socket
EofFlushWrites, // finish up writes and close
Closed };
State _state;
std::shared_ptr<DelaySocket> _dest; // our writing twin.
2017-03-23 12:14:51 -05:00
const size_t WindowSize = 64 * 1024;
/// queued up data - sent to us by our opposite twin.
struct WriteChunk {
2017-03-23 12:14:51 -05:00
std::chrono::steady_clock::time_point _sendTime;
std::vector<char> _data;
WriteChunk(int delayMs)
2017-03-23 12:14:51 -05:00
{
_sendTime = std::chrono::steady_clock::now() +
std::chrono::milliseconds(delayMs);
}
bool isError() { return _data.size() == 0; }
private:
WriteChunk();
2017-03-23 12:14:51 -05:00
};
std::vector<std::shared_ptr<WriteChunk>> _chunks;
2017-03-23 12:14:51 -05:00
public:
DelaySocket(int delayMs, int fd) :
Socket (fd), _delayMs(delayMs),
_state(ReadWrite)
2017-03-23 12:14:51 -05:00
{
// setSocketBufferSize(Socket::DefaultSendBufferSize);
}
void setDestination(const std::shared_ptr<DelaySocket> &dest)
2017-03-23 12:14:51 -05:00
{
_dest = dest;
}
void dumpState(std::ostream& os) override
{
os << "\tfd: " << getFD()
<< "\n\tqueue: " << _chunks.size() << "\n";
auto now = std::chrono::steady_clock::now();
for (auto &chunk : _chunks)
{
os << "\t\tin: " <<
std::chrono::duration_cast<std::chrono::milliseconds>(
chunk->_sendTime - now).count() << "ms - "
<< chunk->_data.size() << "bytes\n";
}
}
// FIXME - really need to propagate 'noDelay' etc.
// have a debug only lookup of delayed sockets for this case ?
int getPollEvents(std::chrono::steady_clock::time_point now,
int &timeoutMaxMs) override
{
if (_chunks.size() > 0)
2017-03-23 12:14:51 -05:00
{
int remainingMs = std::chrono::duration_cast<std::chrono::milliseconds>(
(*_chunks.begin())->_sendTime - now).count();
if (remainingMs < timeoutMaxMs)
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " reset timeout max to " << remainingMs
<< "ms from " << timeoutMaxMs << "ms\n");
2017-03-23 12:14:51 -05:00
timeoutMaxMs = std::min(timeoutMaxMs, remainingMs);
}
if (_chunks.size() > 0 &&
now > (*_chunks.begin())->_sendTime)
return POLLIN | POLLOUT;
else
return POLLIN;
2017-03-23 12:14:51 -05:00
}
void pushCloseChunk()
2017-03-23 12:14:51 -05:00
{
_chunks.push_back(std::make_shared<WriteChunk>(_delayMs));
2017-03-23 12:14:51 -05:00
}
void changeState(State newState)
{
switch (newState)
{
case ReadWrite:
assert (false);
break;
case EofFlushWrites:
assert (_state == ReadWrite);
assert (_dest);
_dest->pushCloseChunk();
_dest = nullptr;
break;
case Closed:
if (_dest && _state == ReadWrite)
_dest->pushCloseChunk();
_dest = nullptr;
shutdown();
break;
}
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " changed to state " << newState << "\n");
_state = newState;
}
2017-03-23 12:14:51 -05:00
HandleResult handlePoll(std::chrono::steady_clock::time_point now, int events) override
{
if (_state == ReadWrite && (events & POLLIN))
2017-03-23 12:14:51 -05:00
{
auto chunk = std::make_shared<WriteChunk>(_delayMs);
2017-03-23 12:14:51 -05:00
char buf[64 * 1024];
ssize_t len;
size_t toRead = sizeof(buf); //std::min(sizeof(buf), WindowSize - _chunksSize);
2017-03-23 12:14:51 -05:00
do {
len = ::read(getFD(), buf, toRead);
} while (len < 0 && errno == EINTR);
if (len == 0) // EOF.
changeState(EofFlushWrites);
else if (len >= 0)
2017-03-23 12:14:51 -05:00
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " read " << len
<< " to queue: " << _chunks.size() << "\n");
2017-03-23 12:14:51 -05:00
chunk->_data.insert(chunk->_data.end(), &buf[0], &buf[len]);
if (_dest)
_dest->_chunks.push_back(chunk);
else
assert("no destination for data" && false);
2017-03-23 12:14:51 -05:00
}
else if (errno != EAGAIN && errno != EWOULDBLOCK)
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " error : " << errno << " " << strerror(errno) << "\n");
changeState(Closed); // FIXME - propagate the error ?
2017-03-23 12:14:51 -05:00
}
}
if (_chunks.size() == 0)
{
if (_state == EofFlushWrites)
changeState(Closed);
}
else // Write if we have delayed enough.
2017-03-23 12:14:51 -05:00
{
std::shared_ptr<WriteChunk> chunk = *_chunks.begin();
2017-03-23 12:14:51 -05:00
if (std::chrono::duration_cast<std::chrono::milliseconds>(
now - chunk->_sendTime).count() >= 0)
{
if (chunk->_data.size() == 0)
{ // delayed error or close
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " handling delayed close\n");
changeState(Closed);
2017-03-23 12:14:51 -05:00
}
else
2017-03-23 12:14:51 -05:00
{
ssize_t len;
do {
len = ::write(getFD(), &chunk->_data[0], chunk->_data.size());
} while (len < 0 && errno == EINTR);
if (len < 0)
2017-03-23 12:14:51 -05:00
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " full - waiting for write\n");
}
else
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " failed onwards write "
<< len << "bytes of "
<< chunk->_data.size()
2017-04-22 12:37:38 -05:00
<< " queue: " << _chunks.size() << " error "
<< strerror(errno) << "\n");
changeState(Closed);
}
2017-03-23 12:14:51 -05:00
}
else
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " written onwards " << len << "bytes of "
<< chunk->_data.size()
2017-04-22 12:37:38 -05:00
<< " queue: " << _chunks.size() << "\n");
if (len > 0)
chunk->_data.erase(chunk->_data.begin(), chunk->_data.begin() + len);
2017-03-23 12:14:51 -05:00
if (chunk->_data.size() == 0)
_chunks.erase(_chunks.begin(), _chunks.begin() + 1);
}
2017-03-23 12:14:51 -05:00
}
}
}
if (events & (POLLERR | POLLHUP | POLLNVAL))
{
2017-04-22 12:37:38 -05:00
DELAY_LOG("#" << getFD() << " error events: " << events << "\n");
changeState(Closed);
2017-03-23 12:14:51 -05:00
}
if (_state == Closed)
return HandleResult::SOCKET_CLOSED;
else
return HandleResult::CONTINUE;
2017-03-23 12:14:51 -05:00
}
};
/// Delayer:
///
/// Some terminology:
/// physical socket (DelaySocket's own fd) - what we accepted.
/// internalFd - the internal side of the socket-pair
/// delayFd - what we hand on to our un-suspecting wrapped socket
/// which looks like an external socket - but delayed.
namespace Delay {
int create(int delayMs, int physicalFd)
{
int pair[2];
int rc = socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, pair);
assert (rc == 0);
int internalFd = pair[0];
int delayFd = pair[1];
auto physical = std::make_shared<DelaySocket>(delayMs, physicalFd);
auto internal = std::make_shared<DelaySocket>(delayMs, internalFd);
physical->setDestination(internal);
internal->setDestination(physical);
DelayPoll.startThread();
DelayPoll.insertNewSocket(physical);
DelayPoll.insertNewSocket(internal);
return delayFd;
}
void dumpState(std::ostream &os)
{
if (DelayPoll.isAlive())
{
os << "Delay poll:\n";
DelayPoll.dumpState(os);
}
}
}
/* vim:set shiftwidth=4 softtabstop=4 expandtab: */