diff --git a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp index 63f339aa7d..fd79142aa7 100644 --- a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp +++ b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp @@ -26,7 +26,8 @@ BufferQueue::BufferQueue(size_t maxBytes) fNumBytes(0), fContiguousBytes(0), fFirstSequence(0), - fLastSequence(0) + fLastSequence(0), + fPushPointer(0) { } @@ -352,3 +353,9 @@ BufferQueue::Available(tcp_sequence sequence) const return fContiguousBytes + fFirstSequence - sequence; } + +void +BufferQueue::SetPushPointer(tcp_sequence sequence) +{ + fPushPointer = sequence; +} diff --git a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h index 39c872eee6..de0e146ca1 100644 --- a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h +++ b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h @@ -33,6 +33,9 @@ class BufferQueue { size_t Available() const { return fContiguousBytes; } size_t Available(tcp_sequence sequence) const; + size_t PushedData() const { return fPushPointer > fFirstSequence ? fPushPointer - fFirstSequence : 0; } + void SetPushPointer(tcp_sequence sequence); + size_t Used() const { return fNumBytes; } size_t Free() const { return fMaxBytes - fNumBytes; } @@ -49,6 +52,7 @@ class BufferQueue { size_t fContiguousBytes; tcp_sequence fFirstSequence; tcp_sequence fLastSequence; + tcp_sequence fPushPointer; }; #endif // BUFFER_QUEUE_H diff --git a/src/add-ons/kernel/network/protocols/tcp/TCPEndpoint.cpp b/src/add-ons/kernel/network/protocols/tcp/TCPEndpoint.cpp index 8cd9db38c2..9c2c1c870c 100644 --- a/src/add-ons/kernel/network/protocols/tcp/TCPEndpoint.cpp +++ b/src/add-ons/kernel/network/protocols/tcp/TCPEndpoint.cpp @@ -496,21 +496,27 @@ TCPEndpoint::ReadData(size_t numBytes, uint32 flags, net_buffer** _buffer) // read data out of buffer // TODO: add support for urgent data (MSG_OOB) - // TODO: wait until enough bytes are available bigtime_t timeout = system_time() + socket->receive.timeout; - do { + while (fReceiveQueue.PushedData() == 0 + && fReceiveQueue.Available() < numBytes + && (fFlags & FLAG_NO_RECEIVE) == 0) { locker.Unlock(); status_t status = acquire_sem_etc(fReceiveLock, 1, B_ABSOLUTE_TIMEOUT | B_CAN_INTERRUPT, timeout); - if (status < B_OK) - return status; locker.Lock(); - } while (fReceiveQueue.Available() < socket->receive.low_water_mark - && (fFlags & FLAG_NO_RECEIVE) == 0); + + if (status < B_OK) { + // TODO: If we are timing out, should we push the + // available data? + if (status == B_TIMED_OUT && fReceiveQueue.Available() > 0) + break; + return status; + } + } TRACE("ReadData(): read %lu bytes, %lu are available", numBytes, fReceiveQueue.Available()); @@ -793,6 +799,8 @@ TCPEndpoint::Receive(tcp_segment_header &segment, net_buffer *buffer) fReceiveNext += buffer->size; TRACE("Receive(): receive next = %lu", (uint32)fReceiveNext); fReceiveQueue.Add(buffer, segment.sequence); + if (segment.flags & TCP_FLAG_PUSH) + fReceiveQueue.SetPushPointer(fReceiveNext); release_sem_etc(fReceiveLock, 1, B_DO_NOT_RESCHEDULE); // TODO: real conditional locking needed! @@ -1018,6 +1026,9 @@ TCPEndpoint::Receive(tcp_segment_header &segment, net_buffer *buffer) } else gBufferModule->free(buffer); + if (segment.flags & TCP_FLAG_PUSH) + fReceiveQueue.SetPushPointer(fReceiveNext); + return action; }