network.cpp
// network.cpp - written and placed in the public domain by Wei Dai
#include "pch.h"
#include "network.h"
#if !defined(NO_OS_DEPENDENCE) && defined(SOCKETS_AVAILABLE)
#include "wait.h"
#define CRYPTOPP_TRACE_NETWORK 0
NAMESPACE_BEGIN(CryptoPP)
lword LimitedBandwidth::ComputeCurrentTransceiveLimit()
{
if (!m_maxBytesPerSecond)
return ULONG_MAX;
const double curTime = GetCurTimeAndCleanUp();
CRYPTOPP_UNUSED(curTime);
lword total = 0;
for (OpQueue::size_type i=0; i!=m_ops.size(); ++i)
total += m_ops[i].second;
return SaturatingSubtract(m_maxBytesPerSecond, total);
}
double LimitedBandwidth::TimeToNextTransceive()
{
if (!m_maxBytesPerSecond)
return 0;
if (!m_nextTransceiveTime)
ComputeNextTransceiveTime();
return SaturatingSubtract(m_nextTransceiveTime, m_timer.ElapsedTimeAsDouble());
}
void LimitedBandwidth::NoteTransceive(lword size)
{
if (m_maxBytesPerSecond)
{
double curTime = GetCurTimeAndCleanUp();
m_ops.push_back(std::make_pair(curTime, size));
m_nextTransceiveTime = 0;
}
}
void LimitedBandwidth::ComputeNextTransceiveTime()
{
double curTime = GetCurTimeAndCleanUp();
lword total = 0;
for (unsigned int i=0; i!=m_ops.size(); ++i)
total += m_ops[i].second;
m_nextTransceiveTime =
(total < m_maxBytesPerSecond) ? curTime : m_ops.front().first + 1000;
}
double LimitedBandwidth::GetCurTimeAndCleanUp()
{
if (!m_maxBytesPerSecond)
return 0;
double curTime = m_timer.ElapsedTimeAsDouble();
while (m_ops.size() && (m_ops.front().first + 1000 < curTime))
m_ops.pop_front();
return curTime;
}
void LimitedBandwidth::GetWaitObjects(WaitObjectContainer &container, const CallStack &callStack)
{
double nextTransceiveTime = TimeToNextTransceive();
if (nextTransceiveTime)
container.ScheduleEvent(nextTransceiveTime, CallStack("LimitedBandwidth::GetWaitObjects()", &callStack));
}
// *************************************************************
size_t NonblockingSource::GeneralPump2(
lword& byteCount, bool blockingOutput,
unsigned long maxTime, bool checkDelimiter, byte delimiter)
{
m_blockedBySpeedLimit = false;
if (!GetMaxBytesPerSecond())
{
size_t ret = DoPump(byteCount, blockingOutput, maxTime, checkDelimiter, delimiter);
m_doPumpBlocked = (ret != 0);
return ret;
}
bool forever = (maxTime == INFINITE_TIME);
unsigned long timeToGo = maxTime;
Timer timer(Timer::MILLISECONDS, forever);
lword maxSize = byteCount;
byteCount = 0;
timer.StartTimer();
while (true)
{
lword curMaxSize = UnsignedMin(ComputeCurrentTransceiveLimit(), maxSize - byteCount);
if (curMaxSize || m_doPumpBlocked)
{
if (!forever) timeToGo = SaturatingSubtract(maxTime, timer.ElapsedTime());
size_t ret = DoPump(curMaxSize, blockingOutput, timeToGo, checkDelimiter, delimiter);
m_doPumpBlocked = (ret != 0);
if (curMaxSize)
{
NoteTransceive(curMaxSize);
byteCount += curMaxSize;
}
if (ret)
return ret;
}
if (maxSize != ULONG_MAX && byteCount >= maxSize)
break;
if (!forever)
{
timeToGo = SaturatingSubtract(maxTime, timer.ElapsedTime());
if (!timeToGo)
break;
}
double waitTime = TimeToNextTransceive();
if (!forever && waitTime > timeToGo)
{
m_blockedBySpeedLimit = true;
break;
}
WaitObjectContainer container;
LimitedBandwidth::GetWaitObjects(container, CallStack("NonblockingSource::GeneralPump2() - speed limit", 0));
container.Wait((unsigned long)waitTime);
}
return 0;
}
size_t NonblockingSource::PumpMessages2(unsigned int &messageCount, bool blocking)
{
if (messageCount == 0)
return 0;
messageCount = 0;
lword byteCount;
do {
byteCount = LWORD_MAX;
RETURN_IF_NONZERO(Pump2(byteCount, blocking));
} while(byteCount == LWORD_MAX);
if (!m_messageEndSent && SourceExhausted())
{
RETURN_IF_NONZERO(AttachedTransformation()->Put2(NULL, 0, GetAutoSignalPropagation(), true));
m_messageEndSent = true;
messageCount = 1;
}
return 0;
}
lword NonblockingSink::TimedFlush(unsigned long maxTime, size_t targetSize)
{
m_blockedBySpeedLimit = false;
size_t curBufSize = GetCurrentBufferSize();
if (curBufSize <= targetSize && (targetSize || !EofPending()))
return 0;
if (!GetMaxBytesPerSecond())
return DoFlush(maxTime, targetSize);
bool forever = (maxTime == INFINITE_TIME);
unsigned long timeToGo = maxTime;
Timer timer(Timer::MILLISECONDS, forever);
lword totalFlushed = 0;
timer.StartTimer();
while (true)
{
size_t flushSize = UnsignedMin(curBufSize - targetSize, ComputeCurrentTransceiveLimit());
if (flushSize || EofPending())
{
if (!forever) timeToGo = SaturatingSubtract(maxTime, timer.ElapsedTime());
size_t ret = (size_t)DoFlush(timeToGo, curBufSize - flushSize);
if (ret)
{
NoteTransceive(ret);
curBufSize -= ret;
totalFlushed += ret;
}
}
if (curBufSize <= targetSize && (targetSize || !EofPending()))
break;
if (!forever)
{
timeToGo = SaturatingSubtract(maxTime, timer.ElapsedTime());
if (!timeToGo)
break;
}
double waitTime = TimeToNextTransceive();
if (!forever && waitTime > timeToGo)
{
m_blockedBySpeedLimit = true;
break;
}
WaitObjectContainer container;
LimitedBandwidth::GetWaitObjects(container, CallStack("NonblockingSink::TimedFlush() - speed limit", 0));
container.Wait((unsigned long)waitTime);
}
return totalFlushed;
}
bool NonblockingSink::IsolatedFlush(bool hardFlush, bool blocking)
{
TimedFlush(blocking ? INFINITE_TIME : 0);
return hardFlush && (!!GetCurrentBufferSize() || EofPending());
}
// *************************************************************
NetworkSource::NetworkSource(BufferedTransformation *attachment)
: NonblockingSource(attachment), m_buf(1024*16)
, m_putSize(0), m_dataBegin(0), m_dataEnd(0)
, m_waitingForResult(false), m_outputBlocked(false)
{
}
unsigned int NetworkSource::GetMaxWaitObjectCount() const
{
return LimitedBandwidth::GetMaxWaitObjectCount()
+ GetReceiver().GetMaxWaitObjectCount()
+ AttachedTransformation()->GetMaxWaitObjectCount();
}
void NetworkSource::GetWaitObjects(WaitObjectContainer &container, CallStack const& callStack)
{
if (BlockedBySpeedLimit())
LimitedBandwidth::GetWaitObjects(container, CallStack("NetworkSource::GetWaitObjects() - speed limit", &callStack));
else if (!m_outputBlocked)
{
if (m_dataBegin == m_dataEnd)
AccessReceiver().GetWaitObjects(container, CallStack("NetworkSource::GetWaitObjects() - no data", &callStack));
else
container.SetNoWait(CallStack("NetworkSource::GetWaitObjects() - have data", &callStack));
}
AttachedTransformation()->GetWaitObjects(container, CallStack("NetworkSource::GetWaitObjects() - attachment", &callStack));
}
size_t NetworkSource::DoPump(lword &byteCount, bool blockingOutput, unsigned long maxTime, bool checkDelimiter, byte delimiter)
{
NetworkReceiver &receiver = AccessReceiver();
lword maxSize = byteCount;
byteCount = 0;
bool forever = maxTime == INFINITE_TIME;
Timer timer(Timer::MILLISECONDS, forever);
BufferedTransformation *t = AttachedTransformation();
if (m_outputBlocked)
goto DoOutput;
while (true)
{
if (m_dataBegin == m_dataEnd)
{
if (receiver.EofReceived())
break;
if (m_waitingForResult)
{
if (receiver.MustWaitForResult() &&
!receiver.Wait(SaturatingSubtract(maxTime, timer.ElapsedTime()),
CallStack("NetworkSource::DoPump() - wait receive result", 0)))
break;
unsigned int recvResult = receiver.GetReceiveResult();
#if CRYPTOPP_TRACE_NETWORK
OutputDebugString((IntToString((unsigned int)this) + ": Received " + IntToString(recvResult) + " bytes\n").c_str());
#endif
m_dataEnd += recvResult;
m_waitingForResult = false;
if (!receiver.MustWaitToReceive() && !receiver.EofReceived() && m_dataEnd != m_buf.size())
goto ReceiveNoWait;
}
else
{
m_dataEnd = m_dataBegin = 0;
if (receiver.MustWaitToReceive())
{
if (!receiver.Wait(SaturatingSubtract(maxTime, timer.ElapsedTime()),
CallStack("NetworkSource::DoPump() - wait receive", 0)))
break;
receiver.Receive(m_buf+m_dataEnd, m_buf.size()-m_dataEnd);
m_waitingForResult = true;
}
else
{
ReceiveNoWait:
m_waitingForResult = true;
// call Receive repeatedly as long as data is immediately available,
// because some receivers tend to return data in small pieces
#if CRYPTOPP_TRACE_NETWORK
OutputDebugString((IntToString((unsigned int)this) + ": Receiving " + IntToString(m_buf.size()-m_dataEnd) + " bytes\n").c_str());
#endif
while (receiver.Receive(m_buf+m_dataEnd, m_buf.size()-m_dataEnd))
{
unsigned int recvResult = receiver.GetReceiveResult();
#if CRYPTOPP_TRACE_NETWORK
OutputDebugString((IntToString((unsigned int)this) + ": Received " + IntToString(recvResult) + " bytes\n").c_str());
#endif
m_dataEnd += recvResult;
if (receiver.EofReceived() || m_dataEnd > m_buf.size() /2)
{
m_waitingForResult = false;
break;
}
}
}
}
}
else
{
m_putSize = UnsignedMin(m_dataEnd - m_dataBegin, maxSize - byteCount);
if (checkDelimiter)
m_putSize = std::find(m_buf+m_dataBegin, m_buf+m_dataBegin+m_putSize, delimiter) - (m_buf+m_dataBegin);
DoOutput:
size_t result = t->PutModifiable2(m_buf+m_dataBegin, m_putSize, 0, forever || blockingOutput);
if (result)
{
if (t->Wait(SaturatingSubtract(maxTime, timer.ElapsedTime()),
CallStack("NetworkSource::DoPump() - wait attachment", 0)))
goto DoOutput;
else
{
m_outputBlocked = true;
return result;
}
}
m_outputBlocked = false;
byteCount += m_putSize;
m_dataBegin += m_putSize;
if (checkDelimiter && m_dataBegin < m_dataEnd && m_buf[m_dataBegin] == delimiter)
break;
if (maxSize != ULONG_MAX && byteCount == maxSize)
break;
// once time limit is reached, return even if there is more data waiting
// but make 0 a special case so caller can request a large amount of data to be
// pumped as long as it is immediately available
if (maxTime > 0 && timer.ElapsedTime() > maxTime)
break;
}
}
return 0;
}
// *************************************************************
NetworkSink::NetworkSink(unsigned int maxBufferSize, unsigned int autoFlushBound)
: m_maxBufferSize(maxBufferSize), m_autoFlushBound(autoFlushBound)
, m_needSendResult(false), m_wasBlocked(false), m_eofState(EOF_NONE)
, m_buffer(STDMIN(16U*1024U+256, maxBufferSize)), m_skipBytes(0)
, m_speedTimer(Timer::MILLISECONDS), m_byteCountSinceLastTimerReset(0)
, m_currentSpeed(0), m_maxObservedSpeed(0)
{
}
float NetworkSink::ComputeCurrentSpeed()
{
if (m_speedTimer.ElapsedTime() > 1000)
{
m_currentSpeed = m_byteCountSinceLastTimerReset * 1000 / m_speedTimer.ElapsedTime();
m_maxObservedSpeed = STDMAX(m_currentSpeed, m_maxObservedSpeed * 0.98f);
m_byteCountSinceLastTimerReset = 0;
m_speedTimer.StartTimer();
// OutputDebugString(("max speed: " + IntToString((int)m_maxObservedSpeed) + " current speed: " + IntToString((int)m_currentSpeed) + "\n").c_str());
}
return m_currentSpeed;
}
float NetworkSink::GetMaxObservedSpeed() const
{
lword m = GetMaxBytesPerSecond();
return m ? STDMIN(m_maxObservedSpeed, float(CRYPTOPP_VC6_INT64 m)) : m_maxObservedSpeed;
}
unsigned int NetworkSink::GetMaxWaitObjectCount() const
{
return LimitedBandwidth::GetMaxWaitObjectCount() + GetSender().GetMaxWaitObjectCount();
}
void NetworkSink::GetWaitObjects(WaitObjectContainer &container, CallStack const& callStack)
{
if (BlockedBySpeedLimit())
LimitedBandwidth::GetWaitObjects(container, CallStack("NetworkSink::GetWaitObjects() - speed limit", &callStack));
else if (m_wasBlocked)
AccessSender().GetWaitObjects(container, CallStack("NetworkSink::GetWaitObjects() - was blocked", &callStack));
else if (!m_buffer.IsEmpty())
AccessSender().GetWaitObjects(container, CallStack("NetworkSink::GetWaitObjects() - buffer not empty", &callStack));
else if (EofPending())
AccessSender().GetWaitObjects(container, CallStack("NetworkSink::GetWaitObjects() - EOF pending", &callStack));
}
size_t NetworkSink::Put2(const byte *inString, size_t length, int messageEnd, bool blocking)
{
if (m_eofState == EOF_DONE)
{
if (length || messageEnd)
throw Exception(Exception::OTHER_ERROR, "NetworkSink::Put2() being called after EOF had been sent");
return 0;
}
if (m_eofState > EOF_NONE)
goto EofSite;
{
if (m_skipBytes)
{
assert(length >= m_skipBytes);
inString += m_skipBytes;
length -= m_skipBytes;
}
m_buffer.Put(inString, length);
if (!blocking || m_buffer.CurrentSize() > m_autoFlushBound)
TimedFlush(0, 0);
size_t targetSize = messageEnd ? 0 : m_maxBufferSize;
if (blocking)
TimedFlush(INFINITE_TIME, targetSize);
if (m_buffer.CurrentSize() > targetSize)
{
assert(!blocking);
m_wasBlocked = true;
m_skipBytes += length;
size_t blockedBytes = UnsignedMin(length, m_buffer.CurrentSize() - targetSize);
return STDMAX<size_t>(blockedBytes, 1);
}
m_wasBlocked = false;
m_skipBytes = 0;
}
if (messageEnd)
{
m_eofState = EOF_PENDING_SEND;
EofSite:
TimedFlush(blocking ? INFINITE_TIME : 0, 0);
if (m_eofState != EOF_DONE)
return 1;
}
return 0;
}
lword NetworkSink::DoFlush(unsigned long maxTime, size_t targetSize)
{
NetworkSender &sender = AccessSender();
bool forever = maxTime == INFINITE_TIME;
Timer timer(Timer::MILLISECONDS, forever);
unsigned int totalFlushSize = 0;
while (true)
{
if (m_buffer.CurrentSize() <= targetSize)
break;
if (m_needSendResult)
{
if (sender.MustWaitForResult() &&
!sender.Wait(SaturatingSubtract(maxTime, timer.ElapsedTime()),
CallStack("NetworkSink::DoFlush() - wait send result", 0)))
break;
unsigned int sendResult = sender.GetSendResult();
#if CRYPTOPP_TRACE_NETWORK
OutputDebugString((IntToString((unsigned int)this) + ": Sent " + IntToString(sendResult) + " bytes\n").c_str());
#endif
m_buffer.Skip(sendResult);
totalFlushSize += sendResult;
m_needSendResult = false;
if (!m_buffer.AnyRetrievable())
break;
}
unsigned long timeOut = maxTime ? SaturatingSubtract(maxTime, timer.ElapsedTime()) : 0;
if (sender.MustWaitToSend() && !sender.Wait(timeOut, CallStack("NetworkSink::DoFlush() - wait send", 0)))
break;
size_t contiguousSize = 0;
const byte *block = m_buffer.Spy(contiguousSize);
#if CRYPTOPP_TRACE_NETWORK
OutputDebugString((IntToString((unsigned int)this) + ": Sending " + IntToString(contiguousSize) + " bytes\n").c_str());
#endif
sender.Send(block, contiguousSize);
m_needSendResult = true;
if (maxTime > 0 && timeOut == 0)
break; // once time limit is reached, return even if there is more data waiting
}
m_byteCountSinceLastTimerReset += totalFlushSize;
ComputeCurrentSpeed();
if (m_buffer.IsEmpty() && !m_needSendResult)
{
if (m_eofState == EOF_PENDING_SEND)
{
sender.SendEof();
m_eofState = sender.MustWaitForEof() ? EOF_PENDING_DELIVERY : EOF_DONE;
}
while (m_eofState == EOF_PENDING_DELIVERY)
{
unsigned long timeOut = maxTime ? SaturatingSubtract(maxTime, timer.ElapsedTime()) : 0;
if (!sender.Wait(timeOut, CallStack("NetworkSink::DoFlush() - wait EOF", 0)))
break;
if (sender.EofSent())
m_eofState = EOF_DONE;
}
}
return totalFlushSize;
}
NAMESPACE_END
#endif // #ifdef SOCKETS_AVAILABLE
Waitable::Wait
bool Wait(unsigned long milliseconds, CallStack const &callStack)
Wait on this object.
Definition: wait.cpp:469
NetworkReceiver::Receive
virtual bool Receive(byte *buf, size_t bufLen)=0
receive data from network source, returns whether result is immediately available
LimitedBandwidth::GetMaxWaitObjectCount
unsigned int GetMaxWaitObjectCount() const
Definition: network.h:37
NetworkSender
Network Sender.
Definition: network.h:151
INFINITE_TIME
const unsigned long INFINITE_TIME
Represents infinite time.
Definition: cryptlib.h:110
NonblockingSource::PumpMessages2
size_t PumpMessages2(unsigned int &messageCount, bool blocking=true)
Pump messages to attached transformation.
Definition: network.cpp:144
NonblockingSource::Pump2
size_t Pump2(lword &byteCount, bool blocking=true)
Pump data to attached transformation.
Definition: network.h:77
BufferedTransformation
Interface for buffered transformations.
Definition: cryptlib.h:1352
NetworkSink::Put2
size_t Put2(const byte *inString, size_t length, int messageEnd, bool blocking)
Input multiple bytes for processing.
Definition: network.cpp:422
BufferedTransformation::GetWaitObjects
void GetWaitObjects(WaitObjectContainer &container, CallStack const &callStack)
Retrieves waitable objects.
Definition: cryptlib.cpp:432
NonblockingSink::TimedFlush
lword TimedFlush(unsigned long maxTime, size_t targetSize=0)
flush to device for no more than maxTime milliseconds
Definition: network.cpp:166
NetworkSink::ComputeCurrentSpeed
float ComputeCurrentSpeed()
compute the current speed of this sink in bytes per second
Definition: network.cpp:386
UnsignedMin
const T1 UnsignedMin(const T1 &a, const T2 &b)
Safe comparison of values that could be neagtive and incorrectly promoted.
Definition: misc.h:503
InputRejecting< Filter >::Put2
size_t Put2(const byte *inString, size_t length, int messageEnd, bool blocking)
Input a byte array for processing.
Definition: simple.h:138
BufferedTransformation::PutModifiable2
virtual size_t PutModifiable2(byte *inString, size_t length, int messageEnd, bool blocking)
Input multiple bytes that may be modified by callee.
Definition: cryptlib.h:1466
BufferedTransformation::Put
size_t Put(byte inByte, bool blocking=true)
Input a byte for processing.
Definition: cryptlib.h:1378
Waitable::GetWaitObjects
virtual void GetWaitObjects(WaitObjectContainer &container, CallStack const &callStack)=0
Retrieves waitable objects.
BufferedTransformation::GetMaxWaitObjectCount
unsigned int GetMaxWaitObjectCount() const
Retrieves the maximum number of waitable objects.
Definition: cryptlib.cpp:426
CallStack
Definition: wait.h:124
Exception
Base class for all exceptions thrown by the library.
Definition: cryptlib.h:139
STDMIN
const T & STDMIN(const T &a, const T &b)
Replacement function for std::min.
Definition: misc.h:467
STDMAX
const T & STDMAX(const T &a, const T &b)
Replacement function for std::max.
Definition: misc.h:477
BufferedTransformation::Skip
virtual lword Skip(lword skipMax=LWORD_MAX)
Discard skipMax bytes from the output buffer.
Definition: cryptlib.cpp:557
NonblockingSource::GeneralPump2
size_t GeneralPump2(lword &byteCount, bool blockingOutput=true, unsigned long maxTime=INFINITE_TIME, bool checkDelimiter=false, byte delimiter='\n')
pump up to maxSize bytes using at most maxTime milliseconds
Definition: network.cpp:80
IntToString
std::string IntToString(T value, unsigned int base=10)
Converts a value to a string.
Definition: misc.h:530
NetworkReceiver
Network Receiver.
Definition: network.h:93
NetworkSource::GetWaitObjects
void GetWaitObjects(WaitObjectContainer &container, CallStack const &callStack)
Retrieves waitable objects.
Definition: network.cpp:246
NetworkSource::GetMaxWaitObjectCount
unsigned int GetMaxWaitObjectCount() const
Retrieves the maximum number of waitable objects.
Definition: network.cpp:239
NetworkSink::GetMaxWaitObjectCount
unsigned int GetMaxWaitObjectCount() const
Retrieves the maximum number of waitable objects.
Definition: network.cpp:405
NonblockingSource
a Source class that can pump from a device for a specified amount of time.
Definition: network.h:54
SecBlock::size
size_type size() const
Provides the count of elements in the SecBlock.
Definition: secblock.h:524
ByteQueue::AnyRetrievable
bool AnyRetrievable() const
Determines whether bytes are ready for retrieval.
Definition: queue.h:37
Filter::AttachedTransformation
BufferedTransformation * AttachedTransformation()
Retrieve attached transformation.
Definition: filters.cpp:36
CryptoPP
Crypto++ library namespace.
Timer
high resolution timer
Definition: hrtimer.h:57
SaturatingSubtract
T1 SaturatingSubtract(const T1 &a, const T2 &b)
Performs a saturating subtract clamped at 0.
Definition: misc.h:961
NetworkSink::GetMaxObservedSpeed
float GetMaxObservedSpeed() const
get the maximum observed speed of this sink in bytes per second
Definition: network.cpp:399
NetworkSink::GetWaitObjects
void GetWaitObjects(WaitObjectContainer &container, CallStack const &callStack)
Retrieves waitable objects.
Definition: network.cpp:410
Waitable::GetMaxWaitObjectCount
virtual unsigned int GetMaxWaitObjectCount() const =0
Maximum number of wait objects that this object can return.
WaitObjectContainer
container of wait objects
Definition: wait.h:169
NonblockingSink::IsolatedFlush
bool IsolatedFlush(bool hardFlush, bool blocking)
Flushes data buffered by this object, without signal propagation.
Definition: network.cpp:224
Exception::OTHER_ERROR
@ OTHER_ERROR
Some other error occurred not belonging to other categories.
Definition: cryptlib.h:158
Source::SourceExhausted
virtual bool SourceExhausted() const =0
Determines if the Source is exhausted.