From 5fcf044823d5325bd1b6de7c82ef170aaa86439b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Axel=20D=C3=B6rfler?= Date: Sat, 25 Nov 2006 16:13:30 +0000 Subject: [PATCH] * Added a helper class tcp_sequence which hides the semantics of comparing sequences (this was completely broken in the code before). * Wrote a buffer queue class to replace the previous algorithm - instead of merging all buffers together, they're kept in a list, so that the most work will be done in the application's thread and only very little when the data is received; maybe we should add an append_move() function to net_buffer, and use that instead, though, to keep the number of fragments small. * The advertised receive window is now bound to 65535, the receive window shift is correctly computed, but not yet used. * The new buffer queue is now also responsible for the send buffer. * TCPConnection::ListenReceive() used the wrong address to retrieve the target route. * Fixed TCPConnection::ReadData() to also return data when the connection is already closed, and to wait if the connection is not yet established (in SO_NONBLOCK mode); it still doesn't wait until data is available, though... git-svn-id: file:///srv/svn/repos/haiku/haiku/trunk@19372 a95241bf-73f2-0310-859d-f6bbb57e9c96 --- .../network/protocols/tcp/BufferQueue.cpp | 316 ++++++++++++++++++ .../network/protocols/tcp/BufferQueue.h | 50 +++ .../kernel/network/protocols/tcp/Jamfile | 1 + .../network/protocols/tcp/TCPConnection.cpp | 269 ++++++--------- .../network/protocols/tcp/TCPConnection.h | 33 +- .../kernel/network/protocols/tcp/tcp.h | 20 ++ 6 files changed, 512 insertions(+), 177 deletions(-) create mode 100644 src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp create mode 100644 src/add-ons/kernel/network/protocols/tcp/BufferQueue.h diff --git a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp new file mode 100644 index 0000000000..41cded4043 --- /dev/null +++ b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.cpp @@ -0,0 +1,316 @@ +/* + * Copyright 2006, Haiku, Inc. All Rights Reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Axel Dörfler, axeld@pinc-software.de + */ + + +#include "BufferQueue.h" + +#include + + +#define TRACE_BUFFER_QUEUE +#ifdef TRACE_BUFFER_QUEUE +# define TRACE(x) dprintf x +#else +# define TRACE(x) +#endif + + +BufferQueue::BufferQueue(size_t maxBytes) + : + fMaxBytes(maxBytes), + fNumBytes(0), + fContiguousBytes(0), + fFirstSequence(0), + fLastSequence(0) +{ +} + + +BufferQueue::~BufferQueue() +{ + // free up any buffers left in the queue + + net_buffer *buffer; + while ((buffer = fList.RemoveHead()) != NULL) { + gBufferModule->free(buffer); + } +} + + +void +BufferQueue::SetMaxBytes(size_t maxBytes) +{ + fMaxBytes = maxBytes; +} + + +void +BufferQueue::SetInitialSequence(tcp_sequence sequence) +{ + fFirstSequence = fLastSequence = sequence; +} + + + +void +BufferQueue::Add(net_buffer *buffer) +{ + Add(buffer, fLastSequence); +} + + +void +BufferQueue::Add(net_buffer *buffer, tcp_sequence sequence) +{ + buffer->sequence = sequence; + if (fList.IsEmpty() || fFirstSequence > sequence) + fFirstSequence = sequence; + + if (fList.IsEmpty() || sequence >= fLastSequence) { + // we usually just add the buffer to the + fList.Add(buffer); + + if (sequence == fLastSequence && fLastSequence - fFirstSequence == fNumBytes) { + // there is no hole in the buffer, we can make the whole buffer available + fContiguousBytes += buffer->size; + } + + fLastSequence = sequence + buffer->size; + fNumBytes += buffer->size; + return; + } + + if (fLastSequence < sequence + buffer->size) + fLastSequence = sequence + buffer->size; + + // find for the place where to insert the buffer into the queue + + SegmentList::ReverseIterator iterator = fList.GetReverseIterator(); + net_buffer *previous = NULL; + net_buffer *next = NULL; + while ((previous = iterator.Next()) != NULL) { + if (sequence >= previous->sequence) { + // The new fragment can be inserted after this one + break; + } + + next = previous; + } + + // check if we have duplicate data, and remove it if that is the case + if (previous != NULL) { + if (sequence == previous->sequence) { + // we already have at least part of this data - ignore new data whenever + // it makes sense (because some TCP implementations send bogus data when + // probing the window) + if (previous->size >= buffer->size) { + gBufferModule->free(buffer); + buffer = NULL; + } else { + fList.Remove(previous); + gBufferModule->free(previous); + } + } else if (tcp_sequence(previous->sequence + previous->size) > sequence) + gBufferModule->remove_header(buffer, previous->sequence + previous->size - sequence); + } + + if (buffer != NULL && next != NULL + && tcp_sequence(sequence + buffer->size) > next->sequence) { + // we already have at least part of this data + if (tcp_sequence(next->sequence + next->size) < sequence + buffer->size) { + gBufferModule->free(next); + next = (net_buffer *)next->link.next; + } else + gBufferModule->remove_trailer(buffer, next->sequence - (sequence + buffer->size)); + } + + if (buffer == NULL) + return; + + fList.Insert(next, buffer); + + // we might need to update the number of bytes available + + if (fLastSequence - fFirstSequence == fNumBytes) + fContiguousBytes = fNumBytes; + else if (fFirstSequence + fContiguousBytes == sequence) { + // the complicated case: the new segment may have connected almost all + // buffers in the queue (but not all, or the above would be true) + + do { + fContiguousBytes += buffer->size; + + buffer = (struct net_buffer *)buffer->link.next; + } while (buffer != NULL && fFirstSequence + fContiguousBytes == buffer->sequence); + } +} + + +/*! + Removes all data in the queue up to the \a sequence number as specified. + + NOTE: + If there are missing segments in the buffers to be removed, + fContiguousBytes is not maintained correctly! +*/ +status_t +BufferQueue::RemoveUntil(tcp_sequence sequence) +{ + SegmentList::Iterator iterator = fList.GetIterator(); + net_buffer *buffer = NULL; + while ((buffer = iterator.Next()) != NULL) { + if (sequence <= buffer->sequence) { + fFirstSequence = buffer->sequence; + break; + } + + if (sequence >= buffer->sequence + buffer->size) { + // remove this buffer completely + iterator.Remove(); + fNumBytes -= buffer->size; + + fContiguousBytes -= buffer->size; + gBufferModule->free(buffer); + } else { + // remove the header as far as needed + size_t size = sequence - buffer->sequence; + gBufferModule->remove_header(buffer, size); + + fNumBytes -= size; + fContiguousBytes -= size; + } + } + + return B_OK; +} + + +/*! + Clones the requested data in the buffer queue into the provided \a buffer. +*/ +status_t +BufferQueue::Get(net_buffer *buffer, tcp_sequence sequence, size_t bytes) +{ + if (bytes == 0) + return B_OK; + + if (sequence >= fLastSequence) { + // we don't have the requested data + return B_BAD_VALUE; + } + if (tcp_sequence(sequence + bytes) > fLastSequence) + bytes = fLastSequence - sequence; + + size_t bytesLeft = bytes; + + // find first buffer matching the sequence + + SegmentList::Iterator iterator = fList.GetIterator(); + net_buffer *source = NULL; + while ((source = iterator.Next()) != NULL) { + if (tcp_sequence(sequence + bytes) <= source->sequence) + break; + } + + if (source == NULL) + panic("we should have had that data..."); + + // clone the data + + uint32 offset = source->sequence - sequence; + + while (source != NULL && bytesLeft > 0) { + size_t size = min_c(buffer->size - offset, bytesLeft); + status_t status = gBufferModule->append_cloned(buffer, source, offset, size); + if (status < B_OK) + return status; + + bytesLeft -= size; + offset = 0; + source = iterator.Next(); + } + + return B_OK; +} + + +/*! + Creates a new buffer containing \a bytes bytes from the start of the + buffer queue. If \a remove is \c true, the data is removed from the + queue, if not, the data is cloned from the queue. +*/ +status_t +BufferQueue::Get(size_t bytes, bool remove, net_buffer **_buffer) +{ + if (Available() < bytes || bytes == 0) + return B_BAD_VALUE; + + net_buffer *buffer = fList.First(); + size_t bytesLeft = bytes; + ASSERT(buffer != NULL); + + if (!remove || buffer->size > bytes) { + // we need a new buffer + buffer = gBufferModule->create(256); + if (buffer == NULL) + return B_NO_MEMORY; + } else { + // we can reuse this buffer + bytesLeft -= buffer->size; + fList.Remove(buffer); + + if (fList.First() != NULL) + fFirstSequence = fList.First()->sequence; + } + + // clone/copy the remaining data + + SegmentList::Iterator iterator = fList.GetIterator(); + net_buffer *source = NULL; + status_t status = B_OK; + while (bytesLeft > 0 && (source = iterator.Next()) != NULL) { + size_t size = min_c(source->size, bytesLeft); + status_t status = gBufferModule->append_cloned(buffer, source, 0, size); + if (status < B_OK) + break; + + bytesLeft -= size; + + if (!remove) + continue; + + // remove either the whole buffer or only the part we cloned + + if (size == source->size) { + iterator.Remove(); + gBufferModule->free(source); + } else + gBufferModule->remove_header(source, size); + } + + if (status == B_OK) { + *_buffer = buffer; + if (remove) { + fNumBytes -= bytes; + fContiguousBytes -= bytes; + } + } else + gBufferModule->free(buffer); + + return status; +} + + +size_t +BufferQueue::Available(tcp_sequence sequence) const +{ + if (sequence > (uint32)fFirstSequence + fContiguousBytes) + return 0; + + return fContiguousBytes + fFirstSequence - sequence; +} diff --git a/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h new file mode 100644 index 0000000000..79dfd8a1fc --- /dev/null +++ b/src/add-ons/kernel/network/protocols/tcp/BufferQueue.h @@ -0,0 +1,50 @@ +/* + * Copyright 2006, Haiku, Inc. All Rights Reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Axel Dörfler, axeld@pinc-software.de + */ +#ifndef BUFFER_QUEUE_H +#define BUFFER_QUEUE_H + + +#include "tcp.h" + +#include + + +typedef DoublyLinkedList > SegmentList; + +class BufferQueue { + public: + BufferQueue(size_t maxBytes); + ~BufferQueue(); + + void SetMaxBytes(size_t maxBytes); + void SetInitialSequence(tcp_sequence sequence); + + void Add(net_buffer *buffer); + void Add(net_buffer *buffer, tcp_sequence sequence); + status_t RemoveUntil(tcp_sequence sequence); + status_t Get(net_buffer *buffer, tcp_sequence sequence, size_t bytes); + status_t Get(size_t bytes, bool remove, net_buffer **_buffer); + + size_t Available() const { return fContiguousBytes; } + size_t Available(tcp_sequence sequence) const; + + size_t Used() const { return fNumBytes; } + size_t Free() const { return fMaxBytes - fNumBytes; } + + tcp_sequence LastSequence() const { return fLastSequence; } + + private: + SegmentList fList; + size_t fMaxBytes; + size_t fNumBytes; + size_t fContiguousBytes; + tcp_sequence fFirstSequence; + tcp_sequence fLastSequence; +}; + +#endif // BUFFER_QUEUE_H diff --git a/src/add-ons/kernel/network/protocols/tcp/Jamfile b/src/add-ons/kernel/network/protocols/tcp/Jamfile index 9e41ebf221..7fa0797498 100644 --- a/src/add-ons/kernel/network/protocols/tcp/Jamfile +++ b/src/add-ons/kernel/network/protocols/tcp/Jamfile @@ -15,6 +15,7 @@ UsePrivateHeaders kernel net ; KernelAddon tcp : tcp.cpp TCPConnection.cpp + BufferQueue.cpp ; # Installation diff --git a/src/add-ons/kernel/network/protocols/tcp/TCPConnection.cpp b/src/add-ons/kernel/network/protocols/tcp/TCPConnection.cpp index f9ddbbcb95..e259423080 100644 --- a/src/add-ons/kernel/network/protocols/tcp/TCPConnection.cpp +++ b/src/add-ons/kernel/network/protocols/tcp/TCPConnection.cpp @@ -86,25 +86,26 @@ tcp_segment::~tcp_segment() TCPConnection::TCPConnection(net_socket *socket) : + fSendWindowShift(0), + fReceiveWindowShift(0), fLastAcknowledged(0), //system_time()), fSendNext(fLastAcknowledged), fSendWindow(0), - fSendBuffer(NULL), + fSendQueue(socket->send.buffer_size), fRoute(NULL), fReceiveNext(0), - fReceiveWindow(32768), - fAvgRTT(TCP_INITIAL_RTT), - fReceiveBuffer(NULL), + fReceiveWindow(socket->receive.buffer_size), + fReceiveQueue(socket->receive.buffer_size), + fRoundTripTime(TCP_INITIAL_RTT), fState(CLOSED), fError(B_OK) { gStackModule->init_timer(&fTimer, _TimeWait, this); - list_init(&fReorderQueue); - list_init(&fWaitQueue); - benaphore_init(&fReceiveLock, "tcp receive"); - benaphore_init(&fSendLock, "tcp send"); - fAcceptSemaphore = create_sem(0, "tcp accept"); + //benaphore_init(&fReceiveLock, "tcp receive"); + //benaphore_init(&fSendLock, "tcp send"); + fSendLock = create_sem(0, "tcp send"); + fReceiveLock = create_sem(0, "tcp receive"); } @@ -112,21 +113,23 @@ TCPConnection::~TCPConnection() { gStackModule->set_timer(&fTimer, -1); - benaphore_destroy(&fReceiveLock); - benaphore_destroy(&fSendLock); - delete_sem(fAcceptSemaphore); + //benaphore_destroy(&fReceiveLock); + //benaphore_destroy(&fSendLock); + //delete_sem(fAcceptSemaphore); + delete_sem(fReceiveLock); + delete_sem(fSendLock); } status_t TCPConnection::InitCheck() const { - if (fReceiveLock.sem < B_OK) - return fReceiveLock.sem; - if (fSendLock.sem < B_OK) - return fSendLock.sem; - if (fAcceptSemaphore < B_OK) - return fAcceptSemaphore; + if (fReceiveLock < B_OK) + return fReceiveLock; + if (fSendLock < B_OK) + return fSendLock; + //if (fAcceptSemaphore < B_OK) + // return fAcceptSemaphore; return B_OK; } @@ -172,7 +175,7 @@ TCPConnection::Open() status_t TCPConnection::Close() { - BenaphoreLocker lock(&fSendLock); + //BenaphoreLocker lock(&fSendLock); TRACE(("TCP:%p.Close()\n", this)); if (fState == SYNCHRONIZE_SENT || fState == LISTEN) { fState = CLOSED; @@ -224,7 +227,9 @@ TCPConnection::Connect(const struct sockaddr *address) if (address->sa_family != AF_INET) return EAFNOSUPPORT; - BenaphoreLocker lock(&fSendLock); + //BenaphoreLocker lock(&fSendLock); + + TRACE((" TCP: Connect(): in state %d\n", fState)); // Can only call connect() from CLOSED or LISTEN states // otherwise connection is considered already connected @@ -234,8 +239,6 @@ TCPConnection::Connect(const struct sockaddr *address) } else if (fState != CLOSED) return EISCONN; - TRACE((" TCP: Connect(): in state %d\n", fState)); - // get a net_route if there isn't one // TODO: get a net_route_info instead! if (fRoute == NULL) { @@ -245,10 +248,18 @@ TCPConnection::Connect(const struct sockaddr *address) return ENETUNREACH; } + remove_connection(this); + // we need to temporarily remove us from the connection list, as we're + // changing our addresses + // need to associate this connection with a real address, not INADDR_ANY if (gAddressModule->is_empty_address((sockaddr *)&socket->address, false)) { TRACE((" TCP: Connect(): Local Address is INADDR_ANY\n")); - gAddressModule->set_to((sockaddr *)&socket->address, (sockaddr *)fRoute->interface->address); + uint16 port = gAddressModule->get_port((sockaddr *)&socket->address); + gAddressModule->set_to((sockaddr *)&socket->address, + (sockaddr *)fRoute->interface->address); + gAddressModule->set_port((sockaddr *)&socket->address, port); + // need to reset the port after overwriting the address } gAddressModule->set_to((sockaddr *)&socket->peer, address); @@ -260,10 +271,21 @@ TCPConnection::Connect(const struct sockaddr *address) return status; } + fMaxReceiveSize = next->module->get_mtu(next, (sockaddr *)address) + - sizeof(tcp_header); + + // Compute the window shift we advertise to our peer - if it doesn't support + // this option, this will be reset to 0 (when its SYN is received) + fReceiveWindowShift = 0; + while (fReceiveWindowShift < TCP_MAX_WINDOW_SHIFT + && (0xffffUL << fReceiveWindowShift) < socket->receive.buffer_size) { + fReceiveWindowShift++; + } +dprintf("************************* size = %ld, shift = %d\n", socket->receive.buffer_size, fReceiveWindowShift); + TRACE((" TCP: Connect(): starting 3-way handshake...\n")); fState = SYNCHRONIZE_SENT; - fMaxReceiveSize = fRoute->mtu - 40; // send SYN status = _SendQueuedData(TCP_FLAG_SYNCHRONIZE, false); @@ -272,9 +294,18 @@ TCPConnection::Connect(const struct sockaddr *address) return status; } - // TODO: wait until 3-way handshake is complete - TRACE((" TCP: Connect(): Connection complete\n")); - return B_OK; + // wait until 3-way handshake is complete (if needed) + + bigtime_t timeout = min_c(socket->send.timeout, TCP_CONNECTION_TIMEOUT); + if (timeout == 0) { + // we're a non-blocking socket + return EINPROGRESS; + } + + status = acquire_sem_etc(fSendLock, 1, B_RELATIVE_TIMEOUT | B_CAN_INTERRUPT, timeout); + + TRACE((" TCP: Connect(): Connection complete: %s\n", strerror(status))); + return status; } @@ -287,7 +318,8 @@ TCPConnection::Accept(struct net_socket **_acceptedSocket) // TODO: test for non-blocking I/O status_t status; do { - status = acquire_sem(fAcceptSemaphore); + status = acquire_sem_etc(fReceiveLock, 1, B_RELATIVE_TIMEOUT, + socket->receive.timeout); if (status < B_OK) return status; @@ -307,7 +339,7 @@ TCPConnection::Bind(sockaddr *address) if (address->sa_family != AF_INET) return EAFNOSUPPORT; - BenaphoreLocker lock(&fSendLock); + //BenaphoreLocker lock(&fSendLock); // TODO: there is no lock yet for these things... if (fState != CLOSED) @@ -339,7 +371,7 @@ TCPConnection::Unbind(struct sockaddr *address) { TRACE(("TCP:%p.Unbind()\n", this )); - BenaphoreLocker lock(&fSendLock); + //BenaphoreLocker lock(&fSendLock); // TODO: there is no lock yet for these things... status_t status = remove_connection(this); @@ -356,7 +388,7 @@ status_t TCPConnection::Listen(int count) { TRACE(("TCP:%p.Listen()\n", this)); - BenaphoreLocker lock(&fSendLock); + //BenaphoreLocker lock(&fSendLock); if (fState != CLOSED) return B_BAD_VALUE; @@ -382,13 +414,7 @@ TCPConnection::SendData(net_buffer *buffer) { TRACE(("TCP:%p.SendData()\n", this)); - BenaphoreLocker lock(&fSendLock); - if (fSendBuffer != NULL) { - status_t status = gBufferModule->merge(fSendBuffer, buffer, true); - if (status != B_OK) - return status; - } else - fSendBuffer = buffer; + fSendQueue.Add(buffer); return _SendQueuedData(TCP_FLAG_ACKNOWLEDGE, false); } @@ -398,11 +424,8 @@ size_t TCPConnection::SendAvailable() { TRACE(("TCP:%p.SendAvailable()\n", this)); - BenaphoreLocker lock(&fSendLock); - if (fSendBuffer != NULL) - return TCP_MAX_SEND_BUF - fSendBuffer->size; - return TCP_MAX_SEND_BUF; + return fSendQueue.Free(); } @@ -411,27 +434,24 @@ TCPConnection::ReadData(size_t numBytes, uint32 flags, net_buffer** _buffer) { TRACE(("TCP:%p.ReadData()\n", this)); - BenaphoreLocker lock(&fReceiveLock); + //BenaphoreLocker lock(&fReceiveLock); - // must be in a synchronous state - if (fState != ESTABLISHED || fState != FINISH_SENT || fState != FINISH_ACKNOWLEDGED) { - // TODO: is this correct semantics? -dprintf(" TCP state = %d\n", fState); - return B_ERROR; + if (fState == SYNCHRONIZE_SENT || fState == SYNCHRONIZE_RECEIVED) { + // we need to wait until the connection becomes established + if (flags & MSG_DONTWAIT) + return B_WOULD_BLOCK; + + status_t status = acquire_sem_etc(fSendLock, 1, + B_RELATIVE_TIMEOUT | B_CAN_INTERRUPT, socket->receive.timeout); + if (status < B_OK) + return status; } -dprintf(" TCP error = %ld\n", fError); - if (fError != B_OK) - return fError; + // read data out of buffer + // TODO: add support for urgent data (MSG_OOB) + // TODO: wait until enough bytes are available - if (fReceiveBuffer->size < numBytes) - numBytes = fReceiveBuffer->size; - - *_buffer = gBufferModule->split(fReceiveBuffer, numBytes); - if (*_buffer == NULL) - return B_NO_MEMORY; - - return B_OK; + return fReceiveQueue.Get(numBytes, (flags & MSG_PEEK) == 0, _buffer); } @@ -439,88 +459,8 @@ size_t TCPConnection::ReadAvailable() { TRACE(("TCP:%p.ReadAvailable()\n", this)); - BenaphoreLocker lock(&fReceiveLock); - if (fReceiveBuffer != NULL) - return fReceiveBuffer->size; - - return 0; -} - - -/*! - You must hold the connection's receive lock when calling this method -*/ -status_t -TCPConnection::_EnqueueReceivedData(net_buffer *buffer, uint32 sequence) -{ - TRACE(("TCP:%p.EnqueueReceivedData(%p, %lu)\n", this, buffer, sequence)); - status_t status; - - if (sequence == fReceiveNext) { - // first check if the received buffer meets up with the first - // segment in the ReorderQueue - tcp_segment *next; - while ((next = (tcp_segment *)list_get_first_item(&fReorderQueue)) != NULL) { - if (sequence + buffer->size >= next->sequence) { - if (sequence + buffer->size > next->sequence) { - status = gBufferModule->trim(buffer, sequence - next->sequence); - if (status != B_OK) - return status; - } - status = gBufferModule->merge(buffer, next->buffer, true); - if (status != B_OK) - return status; - list_remove_item(&fReorderQueue, next); - delete next; - } else - break; - } - - fReceiveNext += buffer->size; - - if (fReceiveBuffer != NULL) { - status = gBufferModule->merge(fReceiveBuffer, buffer, true); - if (status < B_OK) { - fReceiveNext -= buffer->size; - return status; - } - } else - fReceiveBuffer = buffer; - } else { - // add this buffer into the ReorderQueue in the correct place - // creating a new tcp_segment if necessary - tcp_segment *next = NULL; - do { - next = (tcp_segment *)list_get_next_item(&fReorderQueue, next); - if (next != NULL && next->sequence < sequence) - continue; - if (next != NULL && sequence + buffer->size >= next->sequence) { - // merge the new buffer with the next buffer - if (sequence + buffer->size > next->sequence) { - status = gBufferModule->trim(buffer, sequence - next->sequence); - if (status != B_OK) - return status; - } - status = gBufferModule->merge(buffer, next->buffer, true); - if (status != B_OK) - return status; - - next->buffer = buffer; - next->sequence = sequence; - break; - } - tcp_segment *segment = new(std::nothrow) tcp_segment(buffer, sequence, -1); - if (segment == NULL) - return B_NO_MEMORY; - - if (next == NULL) - list_add_item(&fReorderQueue, segment); - else - list_insert_item_before(&fReorderQueue, next, segment); - } while (next != NULL); - } - - return B_OK; + //BenaphoreLocker lock(&fReceiveLock); + return fReceiveQueue.Available(); } @@ -575,7 +515,7 @@ TCPConnection::ListenReceive(tcp_segment_header& segment, net_buffer *buffer) // TODO: proper error handling! connection->fRoute = gDatalinkModule->get_route(gDomain, - (sockaddr *)&newSocket->address); + (sockaddr *)&newSocket->peer); if (connection->fRoute == NULL) return DROP; @@ -592,10 +532,10 @@ TCPConnection::ListenReceive(tcp_segment_header& segment, net_buffer *buffer) if (segment.max_segment_size > 0) connection->fMaxSegmentSize = segment.max_segment_size; - benaphore_lock(&connection->fSendLock); + //benaphore_lock(&connection->fSendLock); status_t status = connection->_SendQueuedData( TCP_FLAG_SYNCHRONIZE | TCP_FLAG_ACKNOWLEDGE, false); - benaphore_unlock(&connection->fSendLock); + //benaphore_unlock(&connection->fSendLock); if (status < B_OK) return DROP; @@ -626,6 +566,8 @@ TCPConnection::SynchronizeSentReceive(tcp_segment_header& segment, net_buffer *b if (segment.flags & TCP_FLAG_ACKNOWLEDGE) { // the connection has been established fState = ESTABLISHED; + release_sem_etc(fSendLock, 1, B_DO_NOT_RESCHEDULE); + // TODO: this is not enough - we need to use B_RELEASE_ALL } else { // simultaneous open fState = SYNCHRONIZE_RECEIVED; @@ -721,11 +663,9 @@ TCPConnection::Receive(tcp_segment_header& segment, net_buffer *buffer) // TODO: This isn't the most efficient way to do it, and will need to be changed // to deal with Silly Window Syndrome - if (buffer->size > 0) { - status = _EnqueueReceivedData(buffer, segment.sequence); - if (status != B_OK) - return DROP; - } else + if (buffer->size > 0) + fReceiveQueue.Add(buffer, segment.sequence); + else gBufferModule->free(buffer); if (fState != CLOSING && fState != WAIT_FOR_FINISH_ACKNOWLEDGE) @@ -765,26 +705,21 @@ TCPConnection::_SendQueuedData(uint16 flags, bool empty) if (fRoute == NULL) return B_ERROR; - net_buffer *buffer; uint32 effectiveWindow = min_c(next->module->get_mtu(next, (sockaddr *)&socket->address), fSendWindow); + uint32 available = fSendQueue.Available(fSendNext); - if (empty || effectiveWindow == 0 || fSendBuffer == NULL || fSendBuffer->size == 0) { - if (flags == 0) { - // there is just nothing left to do - return B_OK; - } + if (effectiveWindow > available) + effectiveWindow = available; - buffer = gBufferModule->create(256); - if (buffer == NULL) - return ENOBUFS; - } else { - if (effectiveWindow == fSendBuffer->size) { - buffer = fSendBuffer; - fSendBuffer = NULL; - } else - buffer = gBufferModule->split(fSendBuffer, effectiveWindow); - } + if (effectiveWindow == 0 && flags == 0) + return B_OK; + + // TODO: determine if we should send anything at all! + + net_buffer *buffer = gBufferModule->create(256); + if (buffer == NULL) + return B_NO_MEMORY; gAddressModule->set_to((sockaddr *)&buffer->source, (sockaddr *)&socket->address); gAddressModule->set_to((sockaddr *)&buffer->destination, (sockaddr *)&socket->peer); @@ -799,12 +734,14 @@ TCPConnection::_SendQueuedData(uint16 flags, bool empty) segment.flags = (uint8)flags; segment.sequence = fSendNext; segment.acknowledge = fReceiveNext; - segment.advertised_window = fReceiveWindow; + segment.advertised_window = min_c(65535, fReceiveWindow); + // TODO: support shift option! segment.urgent_offset = 0; if ((flags & TCP_FLAG_SYNCHRONIZE) != 0) { // add connection establishment options segment.max_segment_size = fMaxReceiveSize; + //segment.window_shift = fReceiveWindowShift; } status_t status = add_tcp_header(segment, buffer); diff --git a/src/add-ons/kernel/network/protocols/tcp/TCPConnection.h b/src/add-ons/kernel/network/protocols/tcp/TCPConnection.h index c11b2b9f3c..08079373c6 100644 --- a/src/add-ons/kernel/network/protocols/tcp/TCPConnection.h +++ b/src/add-ons/kernel/network/protocols/tcp/TCPConnection.h @@ -11,9 +11,11 @@ #include "tcp.h" +#include "BufferQueue.h" #include #include +#include #include @@ -59,22 +61,23 @@ class TCPConnection : public net_protocol { bool _IsSequenceValid(uint32 sequence, uint32 length); status_t _SendQueuedData(uint16 flags, bool empty); - status_t _EnqueueReceivedData(net_buffer *buffer, uint32 sequenceNumber); static void _TimeWait(struct net_timer *timer, void *data); TCPConnection *fHashNext; - benaphore fSendLock; - benaphore fReceiveLock; - benaphore fLock; - sem_id fAcceptSemaphore; + //benaphore fLock; + sem_id fReceiveLock; + sem_id fSendLock; + + uint8 fSendWindowShift; + uint8 fReceiveWindowShift; uint32 fLastAcknowledged; uint32 fSendNext; uint32 fSendWindow; uint32 fMaxSegmentSize; - net_buffer *fSendBuffer; + BufferQueue fSendQueue; net_route *fRoute; // TODO: don't use a net_route, but a net_route_info!!! @@ -82,16 +85,24 @@ class TCPConnection : public net_protocol { uint32 fReceiveNext; uint32 fReceiveWindow; uint32 fMaxReceiveSize; - bigtime_t fAvgRTT; - net_buffer *fReceiveBuffer; + BufferQueue fReceiveQueue; + + // round trip time and retransmit timeout computation + int32 fRoundTripTime; + int32 fRetransmitTimeoutBase; + bigtime_t fRetransmitTimeout; + int32 fRoundTripDeviation; + bigtime_t fTrackingTimeStamp; + uint32 fTrackingSequence; + bool fTracking; + + uint32 fCongestionWindow; + uint32 fSlowStartThreshold; tcp_state fState; status_t fError; vint32 fDelayedAcknowledge; - struct list fReorderQueue; - struct list fWaitQueue; - // timer net_timer fTimer; }; diff --git a/src/add-ons/kernel/network/protocols/tcp/tcp.h b/src/add-ons/kernel/network/protocols/tcp/tcp.h index ba702df98d..d8561d92ae 100644 --- a/src/add-ons/kernel/network/protocols/tcp/tcp.h +++ b/src/add-ons/kernel/network/protocols/tcp/tcp.h @@ -69,6 +69,22 @@ struct tcp_header { uint16 UrgentOffset() const { return ntohs(urgent_offset); } } _PACKED; +class tcp_sequence { + public: + tcp_sequence(uint32 sequence) : number(sequence) {} + + operator uint32() const { return number; } + void operator=(uint32 sequence) { number = sequence; } + bool operator>(uint32 sequence) const { return (int32)(number - sequence) > 0; } + bool operator>=(uint32 sequence) const { return (int32)(number - sequence) >= 0; } + bool operator<(uint32 sequence) const { return (int32)(number - sequence) < 0; } + bool operator<=(uint32 sequence) const { return (int32)(number - sequence) <= 0; } + uint32 operator+=(uint32 sequence) { return number += sequence; } + + private: + uint32 number; +}; + // TCP flag constants #define TCP_FLAG_FINISH 0x01 #define TCP_FLAG_SYNCHRONIZE 0x02 @@ -79,6 +95,8 @@ struct tcp_header { #define TCP_FLAG_ECN 0x40 // Explicit Congestion Notification echo #define TCP_FLAG_CWR 0x80 // Congestion Window Reduced +#define TCP_CONNECTION_TIMEOUT 75000000 // 75 secs + struct tcp_option { uint8 kind; uint8 length; @@ -98,6 +116,8 @@ enum tcp_option_kind { TCP_OPTION_TIMESTAMP = 8, }; +#define TCP_MAX_WINDOW_SHIFT 14 + struct tcp_segment_header { tcp_segment_header() : window_shift(0), max_segment_size(0) {} // constructor zeros options