* 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

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, [email protected]
+ */
+
+
+#include "BufferQueue.h"
+
+#include <KernelExport.h>
+
+
+#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;
+}

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, [email protected]
+ */
+#ifndef BUFFER_QUEUE_H
+#define BUFFER_QUEUE_H
+
+
+#include "tcp.h"
+
+#include <util/DoublyLinkedList.h>
+
+
+typedef DoublyLinkedList<struct net_buffer, DoublyLinkedListCLink<struct net_buffer> > 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

src/add-ons/kernel/network/protocols/tcp/Jamfile

@@ -15,6 +15,7 @@ UsePrivateHeaders kernel net ;
 KernelAddon tcp :
 	tcp.cpp
 	TCPConnection.cpp
+	BufferQueue.cpp
 ;
 
 # Installation

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);

src/add-ons/kernel/network/protocols/tcp/TCPConnection.h

@@ -11,9 +11,11 @@
 
 
 #include "tcp.h"
+#include "BufferQueue.h"
 
 #include <net_protocol.h>
 #include <net_stack.h>
+#include <util/DoublyLinkedList.h>
 
 #include <stddef.h>
 
@@ -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;
 };

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