* Support the new {send,read}_data_no_buffer() protocol hooks to avoid

unnecessary data copies and waste of memory.
* Changed the storage backend to ring_buffer.


git-svn-id: file:///srv/svn/repos/haiku/haiku/trunk@25300 a95241bf-73f2-0310-859d-f6bbb57e9c96

src/add-ons/kernel/network/protocols/unix/UnixEndpoint.cpp

@@ -414,11 +414,12 @@ UnixEndpoint::Accept(net_socket **_acceptedSocket)
 }
 
 
-status_t
+ssize_t
-UnixEndpoint::Send(net_buffer *buffer)
+UnixEndpoint::Send(const iovec *vecs, size_t vecCount,
+	ancillary_data_container *ancillaryData)
 {
-	TRACE("[%ld] %p->UnixEndpoint::Send(%p)\n", find_thread(NULL), this,
+	TRACE("[%ld] %p->UnixEndpoint::Send(%p, %ld, %p)\n", find_thread(NULL),
-		buffer);
+		this, vecs, vecCount, ancillaryData);
 
 	bigtime_t timeout = absolute_timeout(socket->send.timeout);
 	if (gStackModule->is_restarted_syscall())
@@ -445,7 +446,7 @@ UnixEndpoint::Send(net_buffer *buffer)
 	locker.Unlock();
 	peerLocker.Unlock();
 
-	error = peerFifo->Write(buffer, timeout);
+	ssize_t result = peerFifo->Write(vecs, vecCount, ancillaryData, timeout);
 
 	// Notify select()ing readers, if we successfully wrote anything.
 	size_t readable = peerFifo->Readable();
@@ -471,17 +472,17 @@ UnixEndpoint::Send(net_buffer *buffer)
 	if (notifyWrite)
 		gSocketModule->notify(socket, B_SELECT_WRITE, writable);
 
-	switch (error) {
+	switch (result) {
 		case UNIX_FIFO_SHUTDOWN:
 			if (fPeerEndpoint == peerEndpoint
 					&& fState == UNIX_ENDPOINT_CONNECTED) {
 				// Orderly write shutdown on our side.
 				// Note: Linux and Solaris also send a SIGPIPE, but according
 				// the send() specification that shouldn't be done.
-				error = EPIPE;
+				result = EPIPE;
 			} else {
 				// The FD has been closed.
-				error = EBADF;
+				result = EBADF;
 			}
 			break;
 		case EPIPE:
@@ -493,19 +494,21 @@ UnixEndpoint::Send(net_buffer *buffer)
 		case B_TIMED_OUT:
 			// Translate non-blocking timeouts to the correct error code.
 			if (timeout == 0)
-				error = B_WOULD_BLOCK;
+				result = B_WOULD_BLOCK;
 			break;
 	}
 
-	RETURN_ERROR(error);
+	RETURN_ERROR(result);
 }
 
 
-status_t
+ssize_t
-UnixEndpoint::Receive(size_t numBytes, uint32 flags, net_buffer **_buffer)
+UnixEndpoint::Receive(const iovec *vecs, size_t vecCount,
+	ancillary_data_container **_ancillaryData, struct sockaddr *_address,
+	socklen_t *_addressLength)
 {
-	TRACE("[%ld] %p->UnixEndpoint::Receive(%ld, 0x%lx)\n", find_thread(NULL),
+	TRACE("[%ld] %p->UnixEndpoint::Receive(%p, %ld)\n", find_thread(NULL),
-		this, numBytes, flags);
+		this, vecs, vecCount);
 
 	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
 	if (gStackModule->is_restarted_syscall())
@@ -523,6 +526,14 @@ UnixEndpoint::Receive(size_t numBytes, uint32 flags, net_buffer **_buffer)
 	UnixEndpoint* peerEndpoint = fPeerEndpoint;
 	Reference<UnixEndpoint> peerReference(peerEndpoint);
 
+	// Copy the peer address upfront. This way, if we read something, we don't
+	// get into a potential race with Close().
+	if (_address != NULL) {
+		socklen_t addrLen = min_c(*_addressLength, socket->peer.ss_len);
+		memcpy(_address, &socket->peer, addrLen);
+		*_addressLength = addrLen;
+	}
+
 	// lock our FIFO
 	UnixFifo* fifo = fReceiveFifo;
 	Reference<UnixFifo> _(fifo);
@@ -531,17 +542,17 @@ UnixEndpoint::Receive(size_t numBytes, uint32 flags, net_buffer **_buffer)
 	// unlock endpoint
 	locker.Unlock();
 
-	status_t error = fifo->Read(numBytes, timeout, _buffer);
+	ssize_t result = fifo->Read(vecs, vecCount, _ancillaryData, timeout);
 
 	// Notify select()ing writers, if we successfully read anything.
 	size_t writable = fifo->Writable();
-	bool notifyWrite = (error == B_OK && writable > 0
+	bool notifyWrite = (result >= 0 && writable > 0
 		&& !fifo->IsWriteShutdown());
 
 	// Notify select()ing readers, if we failed to read anything and there's
 	// still something left to read.
 	size_t readable = fifo->Readable();
-	bool notifyRead = (error != B_OK && readable > 0
+	bool notifyRead = (result < 0 && readable > 0
 		&& !fifo->IsReadShutdown());
 
 	// re-lock our endpoint (unlock FIFO to respect locking order)
@@ -558,12 +569,12 @@ UnixEndpoint::Receive(size_t numBytes, uint32 flags, net_buffer **_buffer)
 	if (peerLocked && notifyWrite)
 		gSocketModule->notify(peerEndpoint->socket, B_SELECT_WRITE, writable);
 
-	switch (error) {
+	switch (result) {
 		case UNIX_FIFO_SHUTDOWN:
 			// Either our socket was closed or read shutdown.
 			if (fState == UNIX_ENDPOINT_CLOSED) {
 				// The FD has been closed.
-				error = EBADF;
+				result = EBADF;
 			} else {
 				// if (fReceiveFifo == fifo) {
 				// 		Orderly shutdown or the peer closed the connection.
@@ -571,18 +582,17 @@ UnixEndpoint::Receive(size_t numBytes, uint32 flags, net_buffer **_buffer)
 				//		Weird case: Peer closed connection and we are already
 				// 		reconnected (or listening).
 				// }
-				error = B_OK;
+				result = 0;
-				*_buffer = NULL;
 			}
 			break;
 		case B_TIMED_OUT:
 			// translate non-blocking timeouts to the correct error code
 			if (timeout == 0)
-				error = B_WOULD_BLOCK;
+				result = B_WOULD_BLOCK;
 			break;
 	}
 
-	RETURN_ERROR(error);
+	RETURN_ERROR(result);
 }
 
 
@@ -624,7 +634,7 @@ UnixEndpoint::Receivable()
 }
 
 
-void
+status_t
 UnixEndpoint::SetReceiveBufferSize(size_t size)
 {
 	TRACE("[%ld] %p->UnixEndpoint::SetReceiveBufferSize(%lu)\n",
@@ -632,11 +642,11 @@ UnixEndpoint::SetReceiveBufferSize(size_t size)
 
 	UnixEndpointLocker locker(this);
 
-	if (fState != UNIX_ENDPOINT_CONNECTED)
+	if (fReceiveFifo == NULL)
-		return;
+		return B_BAD_VALUE;
 
 	UnixFifoLocker fifoLocker(fReceiveFifo);
-	fReceiveFifo->SetBufferCapacity(size);
+	return fReceiveFifo->SetBufferCapacity(size);
 }
 
 

src/add-ons/kernel/network/protocols/unix/UnixEndpoint.h

@@ -66,13 +66,16 @@ public:
 	status_t Connect(const struct sockaddr *address);
 	status_t Accept(net_socket **_acceptedSocket);
 
-	status_t Send(net_buffer *buffer);
+	ssize_t Send(const iovec *vecs, size_t vecCount,
-	status_t Receive(size_t numBytes, uint32 flags, net_buffer **_buffer);
+		ancillary_data_container *ancillaryData);
+	ssize_t Receive(const iovec *vecs, size_t vecCount,
+		ancillary_data_container **_ancillaryData, struct sockaddr *_address,
+		socklen_t *_addressLength);
 
 	ssize_t Sendable();
 	ssize_t Receivable();
 
-	void SetReceiveBufferSize(size_t size);
+	status_t SetReceiveBufferSize(size_t size);
 
 	status_t Shutdown(int direction);
 

src/add-ons/kernel/network/protocols/unix/UnixFifo.cpp

@@ -5,8 +5,13 @@
 
 #include "UnixFifo.h"
 
+#include <new>
+
 #include <AutoDeleter.h>
 
+#include <net_stack.h>
+#include <util/ring_buffer.h>
+
 #include "unix.h"
 
 
@@ -15,314 +20,242 @@
 #include "UnixDebug.h"
 
 
-#if TRACE_BUFFER_QUEUE
+// #pragma mark - UnixRequest
-#	define TRACEBQ(x...)	ktrace_printf(x)
+
-#	define TRACEBQ_ONLY(x)	x
+
-#else
+UnixRequest::UnixRequest(const iovec* vecs, size_t count,
-#	define TRACEBQ(x...)	do {} while (false)
+		ancillary_data_container* ancillaryData)
-#	define TRACEBQ_ONLY(x)
+	:
-#endif
+	fVecs(vecs),
+	fVecCount(count),
+	fAncillaryData(ancillaryData),
+	fTotalSize(0),
+	fBytesTransferred(0),
+	fVecIndex(0),
+	fVecOffset(0)
+{
+	for (size_t i = 0; i < fVecCount; i++)
+		fTotalSize += fVecs[i].iov_len;
+}
+
+
+void
+UnixRequest::AddBytesTransferred(size_t size)
+{
+	fBytesTransferred += size;
+
+	// also adjust the current iovec index/offset
+	while (fVecIndex < fVecCount
+			&& fVecs[fVecIndex].iov_len - fVecOffset <= size) {
+		size -= fVecs[fVecIndex].iov_len - fVecOffset;
+		fVecIndex++;
+		fVecOffset = 0;
+	}
+
+	if (fVecIndex < fVecCount)
+		fVecOffset += size;
+}
+
+
+bool
+UnixRequest::GetCurrentChunk(void*& data, size_t& size)
+{
+	while (fVecIndex < fVecCount
+			&& fVecOffset >= fVecs[fVecIndex].iov_len) {
+		fVecIndex++;
+		fVecOffset = 0;
+	}
+	if (fVecIndex >= fVecCount)
+		return false;
+
+	data = (uint8*)fVecs[fVecIndex].iov_base + fVecOffset;
+	size = fVecs[fVecIndex].iov_len - fVecOffset;
+	return true;
+}
+
+
+void
+UnixRequest::SetAncillaryData(ancillary_data_container* data)
+{
+	fAncillaryData = data;
+}
+
+
+void
+UnixRequest::AddAncillaryData(ancillary_data_container* data)
+{
+	if (fAncillaryData != NULL) {
+		gStackModule->move_ancillary_data(data, fAncillaryData);
+		gStackModule->delete_ancillary_data_container(data);
+	} else
+		fAncillaryData = data;
+}
+
+
+// #pragma mark - UnixBufferQueue
 
 
 UnixBufferQueue::UnixBufferQueue(size_t capacity)
 	:
-	fSize(0),
+	fBuffer(NULL),
 	fCapacity(capacity)
-#if TRACE_BUFFER_QUEUE
-	, fWritten(0)
-	, fRead(0)
-#endif
 {
-	TRACEBQ_ONLY(
-		fParanoiaCheckBuffer = (uint8*)malloc(UNIX_FIFO_MAXIMAL_CAPACITY);
-		fParanoiaCheckBuffer2 = (uint8*)malloc(UNIX_FIFO_MAXIMAL_CAPACITY);
-	)
 }
 
 
 UnixBufferQueue::~UnixBufferQueue()
 {
-	while (net_buffer* buffer = fBuffers.RemoveHead())
+	while (AncillaryDataEntry* entry = fAncillaryData.RemoveHead()) {
-		gBufferModule->free(buffer);
+		gStackModule->delete_ancillary_data_container(entry->data);
+		delete entry;
+	}
 
-	TRACEBQ_ONLY(
+	delete_ring_buffer(fBuffer);
-		free(fParanoiaCheckBuffer);
-		free(fParanoiaCheckBuffer2);
-	)
 }
 
 
 status_t
-UnixBufferQueue::Read(size_t size, net_buffer** _buffer)
+UnixBufferQueue::Init()
 {
-	if (size > fSize)
+	fBuffer = create_ring_buffer(fCapacity);
-		size = fSize;
+	if (fBuffer == NULL)
-
+		return B_NO_MEMORY;
-	TRACEBQ("unix: UnixBufferQueue::Read(%lu): fSize: %lu, fRead: %lld, "
+	return B_OK;
-		"fWritten: %lld", size, fSize, fRead, fWritten);
-
-	TRACEBQ_ONLY(
-		MethodDeleter<UnixBufferQueue> _(this, &UnixBufferQueue::PostReadWrite);
-	)
-
-	if (size == 0)
-		return B_BAD_VALUE;
-
-TRACEBQ_ONLY(
-	if (fParanoiaCheckBuffer) {
-		size_t bufferSize = 0;
-		for (BufferList::Iterator it = fBuffers.GetIterator();
-			net_buffer* buffer = it.Next();) {
-			size_t toWrite = min_c(buffer->size, size - bufferSize);
-			if (toWrite == 0)
-				break;
-
-			gBufferModule->read(buffer, 0,
-				fParanoiaCheckBuffer + bufferSize, toWrite);
-			bufferSize += toWrite;
-		}
-	}
-	*_buffer = NULL;
-)
-
-	// If the first buffer has the right size or is smaller, we can just
-	// dequeue it.
-	net_buffer* buffer = fBuffers.Head();
-	if (buffer->size <= size) {
-		fBuffers.RemoveHead();
-		fSize -= buffer->size;
-		TRACEBQ_ONLY(fRead += buffer->size);
-		*_buffer = buffer;
-
-		if (buffer->size == size)
-{
-TRACEBQ("unix:   read full buffer %p (%lu)", buffer, buffer->size);
-TRACEBQ_ONLY(ParanoiaReadCheck(*_buffer));
-			return B_OK;
 }
 
-		// buffer is too small
 
- 		size_t bytesLeft = size - buffer->size;
+size_t
-TRACEBQ("unix:   read short buffer %p (%lu/%lu)", buffer, size, buffer->size);
+UnixBufferQueue::Readable() const
-
-		// Append from the following buffers, until we've read as much as we're
-		// supposed to.
-		while (bytesLeft > 0) {
-			net_buffer* nextBuffer = fBuffers.Head();
-			size_t toCopy = min_c(bytesLeft, nextBuffer->size);
-
-TRACEBQ("unix:   read next buffer %p (%lu/%lu)", nextBuffer, bytesLeft, nextBuffer->size);
-#if 0
-			if (gBufferModule->append_cloned(buffer, nextBuffer, 0, toCopy)
-					!= B_OK) {
-				// Too bad, but we've got some data, so we don't fail.
-TRACEBQ_ONLY(ParanoiaReadCheck(*_buffer));
-				return B_OK;
-			}
-#endif
-
-// TODO: Temporary work-around for the append_cloned() call above, which
-// doesn't seem to work right. Or maybe that's just in combination with the
-// remove_header() below.
 {
-	void* tmpBuffer = malloc(toCopy);
+	return ring_buffer_readable(fBuffer);
-	if (tmpBuffer == NULL)
-		return B_OK;
-	MemoryDeleter tmpBufferDeleter(tmpBuffer);
-
-	size_t offset = buffer->size;
-	if (gBufferModule->read(nextBuffer, 0, tmpBuffer, toCopy) != B_OK
-		|| gBufferModule->append_size(buffer, toCopy, NULL) != B_OK) {
-		return B_OK;
-	}
-
-	if (gBufferModule->write(buffer, offset, tmpBuffer, toCopy) != B_OK) {
-		gBufferModule->remove_trailer(buffer, toCopy);
-		return B_OK;
-	}
 }
 
-			// transfer the ancillary data
-			gBufferModule->transfer_ancillary_data(nextBuffer, buffer);
 
-			if (nextBuffer->size > toCopy) {
+size_t
-				// remove the part we've copied
+UnixBufferQueue::Writable() const
-//gBufferModule->read(nextBuffer, toCopy, fParanoiaCheckBuffer,
+{
-//nextBuffer->size - toCopy);
+	return ring_buffer_writable(fBuffer);
-				gBufferModule->remove_header(nextBuffer, toCopy);
+}
-//TRACEBQ_ONLY(ParanoiaReadCheck(nextBuffer));
+
-			} else {
+
-				// get rid of the buffer completely
+status_t
-				fBuffers.RemoveHead();
+UnixBufferQueue::Read(UnixRequest& request)
-				gBufferModule->free(nextBuffer);
+{
+	bool user = gStackModule->is_syscall();
+
+	size_t readable = Readable();
+	void* data;
+	size_t size;
+
+	while (readable > 0 && request.GetCurrentChunk(data, size)) {
+		if (size > readable)
+			size = readable;
+
+		ssize_t bytesRead;
+		if (user)
+			bytesRead = ring_buffer_user_read(fBuffer, (uint8*)data, size);
+		else
+			bytesRead = ring_buffer_read(fBuffer, (uint8*)data, size);
+
+		if (bytesRead < 0)
+			return bytesRead;
+		if (bytesRead == 0)
+			return B_ERROR;
+
+		// Adjust ancillary data entry offsets, respectively attach the ones
+		// that belong to the read data to the request.
+		if (AncillaryDataEntry* entry = fAncillaryData.Head()) {
+			size_t offsetDelta = bytesRead;
+			while (entry != NULL && offsetDelta > entry->offset) {
+				// entry data have been read -- add ancillary data to request
+				fAncillaryData.RemoveHead();
+				offsetDelta -= entry->offset;
+				request.AddAncillaryData(entry->data);
+				delete entry;
+
+				entry = fAncillaryData.Head();
 			}
 
-			bytesLeft -= toCopy;
+			if (entry != NULL)
-			fSize -= toCopy;
+				entry->offset -= offsetDelta;
-			TRACEBQ_ONLY(fRead += toCopy);
 		}
 
-TRACEBQ_ONLY(ParanoiaReadCheck(*_buffer));
+		request.AddBytesTransferred(bytesRead);
-		return B_OK;
+		readable -= bytesRead;
 	}
 
-	// buffer is too big
-
-	// Create a new buffer, and copy into it, as much as we need.
-	net_buffer* newBuffer = gBufferModule->create(256);
-	if (newBuffer == NULL)
-		return ENOBUFS;
-
-	status_t error = gBufferModule->append_cloned(newBuffer, buffer, 0, size);
-	if (error != B_OK) {
-		gBufferModule->free(newBuffer);
-		return error;
-	}
-
-	// transfer the ancillary data
-	gBufferModule->transfer_ancillary_data(buffer, newBuffer);
-
-	// remove the part we've copied
-TRACEBQ("unix:   read long buffer %p (%lu/%lu)", buffer, size, buffer->size);
-	gBufferModule->remove_header(buffer, size);
-
-	fSize -= size;
-	TRACEBQ_ONLY(fRead += size);
-	*_buffer = newBuffer;
-
-TRACEBQ_ONLY(ParanoiaReadCheck(*_buffer));
 	return B_OK;
 }
 
 
 status_t
-UnixBufferQueue::Write(net_buffer* buffer, size_t maxSize)
+UnixBufferQueue::Write(UnixRequest& request)
 {
-	TRACEBQ("unix: UnixBufferQueue::Write(%lu/%lu): fSize: %lu, fRead: %lld, "
+	bool user = gStackModule->is_syscall();
-		"fWritten: %lld", buffer->size, maxSize, fSize, fRead, fWritten);
 
-	TRACEBQ_ONLY(
+	size_t writable = Writable();
-		MethodDeleter<UnixBufferQueue> _(this, &UnixBufferQueue::PostReadWrite);
+	void* data;
-	)
+	size_t size;
 
-	maxSize = min_c(buffer->size, maxSize);
+	// If the request has ancillary data create an entry first.
-	if (maxSize > Writable())
+	AncillaryDataEntry* ancillaryEntry = NULL;
-		RETURN_ERROR(ENOBUFS);
+	ObjectDeleter<AncillaryDataEntry> ancillaryEntryDeleter;
+	if (writable > 0 && request.AncillaryData() != NULL) {
+		ancillaryEntry = new(std::nothrow) AncillaryDataEntry;
+		if (ancillaryEntry == NULL)
+			return B_NO_MEMORY;
 
-	// If we shall write the complete buffer, things are easy.
+		ancillaryEntryDeleter.SetTo(ancillaryEntry);
-	if (maxSize == buffer->size) {
+		ancillaryEntry->data = request.AncillaryData();
-		fBuffers.Add(buffer);
+		ancillaryEntry->offset = Readable();
-		fSize += buffer->size;
-		TRACEBQ_ONLY(fWritten += buffer->size);
 
-		return B_OK;
+		// The offsets are relative to the previous entry.
+		AncillaryDataList::Iterator it = fAncillaryData.GetIterator();
+		while (AncillaryDataEntry* entry = it.Next())
+			ancillaryEntry->offset -= entry->offset;
+			// TODO: This is inefficient when the list is long. Rather also
+			// store and maintain the absolute offset of the last queued entry.
 	}
 
-	// We shall write only a partial buffer. We need to create a new one and
+	// write as much as we can
-	// cut of the head of the old one.
+	while (writable > 0 && request.GetCurrentChunk(data, size)) {
-// TODO: This implementation obviously sucks, but we can't use the split method,
+		if (size > writable)
-// since it would split off the wrong buffer. The socket module requires us
+			size = writable;
-// to cut off the head of the given one.
 
-	// create a temporary buffer
+		ssize_t bytesWritten;
-	void* tmpBuffer = malloc(maxSize);
+		if (user)
-	if (tmpBuffer == NULL)
+			bytesWritten = ring_buffer_user_write(fBuffer, (uint8*)data, size);
-		return B_OK;
+		else
-	MemoryDeleter tmpBufferDeleter(tmpBuffer);
+			bytesWritten = ring_buffer_write(fBuffer, (uint8*)data, size);
 
-	// read the data to append into the temporary buffer
+		if (bytesWritten < 0)
-	status_t error = gBufferModule->read(buffer, 0, tmpBuffer, maxSize);
+			return bytesWritten;
-	if (error != B_OK)
+		if (bytesWritten == 0)
-		return error;
+			return B_ERROR;
 
-	// create the new buffer and append the data
+		if (ancillaryEntry != NULL) {
-	net_buffer* newBuffer = gBufferModule->create(256);
+			fAncillaryData.Add(ancillaryEntry);
-	if (newBuffer == NULL)
+			ancillaryEntryDeleter.Detach();
-		return ENOBUFS;
+			request.SetAncillaryData(NULL);
+			ancillaryEntry = NULL;
+		}
 
-	error = gBufferModule->append(newBuffer, tmpBuffer, maxSize);
+		request.AddBytesTransferred(bytesWritten);
-
+		writable -= bytesWritten;
-	// remove the header from the old buffer
-	if (error == B_OK)
-		error = gBufferModule->remove_header(buffer, maxSize);
-
-	if (error != B_OK) {
-		gBufferModule->free(newBuffer);
-		return error;
 	}
 
-	// transfer the ancillary data
-	gBufferModule->transfer_ancillary_data(buffer, newBuffer);
-
-	// Everything went fine. Append the new buffer.
-	fBuffers.Add(newBuffer);
-	fSize += newBuffer->size;
-	TRACEBQ_ONLY(fWritten += newBuffer->size);
-
 	return B_OK;
 }
 
 
-void
+status_t
 UnixBufferQueue::SetCapacity(size_t capacity)
 {
-	fCapacity = capacity;
+// TODO:...
+return B_ERROR;
 }
 
 
-#if TRACE_BUFFER_QUEUE
-
-void
-UnixBufferQueue::ParanoiaReadCheck(net_buffer* buffer)
-{
-	if (!buffer || !fParanoiaCheckBuffer || !fParanoiaCheckBuffer2)
-		return;
-
-	gBufferModule->read(buffer, 0, fParanoiaCheckBuffer2, buffer->size);
-
-	if (memcmp(fParanoiaCheckBuffer, fParanoiaCheckBuffer2, buffer->size)
-			!= 0) {
-		// find offset of first difference
-		size_t i = 0;
-		for (; i < buffer->size; i++) {
-			if (fParanoiaCheckBuffer[i] != fParanoiaCheckBuffer2[i])
-				break;
-		}
-
-		panic("unix: UnixBufferQueue::ParanoiaReadCheck(): incorrect read! "
-			"offset of first difference: %lu", i);
-	}
-}
-
-
-void
-UnixBufferQueue::PostReadWrite()
-{
-	TRACEBQ("unix: post read/write: fSize: %lu, fRead: %lld, fWritten: %lld",
-		fSize, fRead, fWritten);
-
-	if (fWritten - fRead != fSize) {
-		panic("UnixBufferQueue::PostReadWrite(): fSize: %lu, fRead: %lld, "
-			"fWritten: %lld", fSize, fRead, fWritten);
-	}
-
-	// check buffer size sum
-	size_t bufferSize = 0;
-	for (BufferList::Iterator it = fBuffers.GetIterator();
-		 net_buffer* buffer = it.Next();) {
-		bufferSize += buffer->size;
-	}
-
-	if (bufferSize != fSize) {
-		panic("UnixBufferQueue::PostReadWrite(): fSize: %lu, bufferSize: %lu",
-			fSize, bufferSize);
-	}
-}
-
-#endif	// TRACE_BUFFER_QUEUE
-
-
 // #pragma mark -
 
 
@@ -351,7 +284,7 @@ UnixFifo::~UnixFifo()
 status_t
 UnixFifo::Init()
 {
-	return B_OK;
+	return fBuffer.Init();
 }
 
 
@@ -371,24 +304,25 @@ UnixFifo::Shutdown(uint32 shutdown)
 }
 
 
-status_t
+ssize_t
-UnixFifo::Read(size_t numBytes, bigtime_t timeout, net_buffer** _buffer)
+UnixFifo::Read(const iovec* vecs, size_t vecCount,
+	ancillary_data_container** _ancillaryData, bigtime_t timeout)
 {
-	TRACE("[%ld] %p->UnixFifo::Read(%lu, %lld)\n", find_thread(NULL), this,
+	TRACE("[%ld] %p->UnixFifo::Read(%p, %ld, %lld)\n", find_thread(NULL),
-		numBytes, timeout);
+		this, vecs, vecCount, timeout);
 
 	if (IsReadShutdown())
 		return UNIX_FIFO_SHUTDOWN;
 
-	Request request(numBytes);
+	UnixRequest request(vecs, vecCount, NULL);
 	fReaders.Add(&request);
-	fReadRequested += request.size;
+	fReadRequested += request.TotalSize();
 
-	status_t error = _Read(request, numBytes, timeout, _buffer);
+	status_t error = _Read(request, timeout);
 
 	bool firstInQueue = fReaders.Head() == &request;
 	fReaders.Remove(&request);
-	fReadRequested -= request.size;
+	fReadRequested -= request.TotalSize();
 
 	if (firstInQueue && !fReaders.IsEmpty() && fBuffer.Readable() > 0
 			&& !IsReadShutdown()) {
@@ -397,21 +331,30 @@ UnixFifo::Read(size_t numBytes, bigtime_t timeout, net_buffer** _buffer)
 		fReadCondition.NotifyAll();
 	}
 
-	if (error == B_OK && *_buffer != NULL && (*_buffer)->size > 0
+	if (request.BytesTransferred() > 0 && !fWriters.IsEmpty()
-			&& !fWriters.IsEmpty() && !IsWriteShutdown()) {
+			&& !IsWriteShutdown()) {
 		// We read something and there are writers. Notify them
 		fWriteCondition.NotifyAll();
 	}
 
+	*_ancillaryData = request.AncillaryData();
+
+	if (request.BytesTransferred() > 0) {
+		if (request.BytesTransferred() > SSIZE_MAX)
+			RETURN_ERROR(SSIZE_MAX);
+		RETURN_ERROR((ssize_t)request.BytesTransferred());
+	}
+
 	RETURN_ERROR(error);
 }
 
 
-status_t
+ssize_t
-UnixFifo::Write(net_buffer* buffer, bigtime_t timeout)
+UnixFifo::Write(const iovec* vecs, size_t vecCount,
+	ancillary_data_container* ancillaryData, bigtime_t timeout)
 {
-	TRACE("[%ld] %p->UnixFifo::Write(%p (%lu), %lld)\n", find_thread(NULL),
+	TRACE("[%ld] %p->UnixFifo::Write(%p, %ld, %p, %lld)\n", find_thread(NULL),
-		this, buffer, buffer->size, timeout);
+		this, vecs, vecCount, ancillaryData, timeout);
 
 	if (IsWriteShutdown())
 		return UNIX_FIFO_SHUTDOWN;
@@ -419,16 +362,15 @@ UnixFifo::Write(net_buffer* buffer, bigtime_t timeout)
 	if (IsReadShutdown())
 		return EPIPE;
 
-	Request request(buffer->size);
+	UnixRequest request(vecs, vecCount, ancillaryData);
 	fWriters.Add(&request);
-	fWriteRequested += request.size;
+	fWriteRequested += request.TotalSize();
-	size_t bytesWritten = 0;
 
-	status_t error = _Write(request, buffer, timeout, bytesWritten);
+	status_t error = _Write(request, timeout);
 
 	bool firstInQueue = fWriters.Head() == &request;
 	fWriters.Remove(&request);
-	fWriteRequested -= request.size;
+	fWriteRequested -= request.TotalSize();
 
 	if (firstInQueue && !fWriters.IsEmpty() && fBuffer.Writable() > 0
 			&& !IsWriteShutdown()) {
@@ -437,12 +379,18 @@ UnixFifo::Write(net_buffer* buffer, bigtime_t timeout)
 		fWriteCondition.NotifyAll();
 	}
 
-	if (bytesWritten > 0 && request.size > 0 && !fReaders.IsEmpty()
+	if (request.BytesTransferred() > 0 && !fReaders.IsEmpty()
 			&& !IsReadShutdown()) {
 		// We've written something and there are readers. Notify them.
 		fReadCondition.NotifyAll();
 	}
 
+	if (request.BytesTransferred() > 0) {
+		if (request.BytesTransferred() > SSIZE_MAX)
+			RETURN_ERROR(SSIZE_MAX);
+		RETURN_ERROR((ssize_t)request.BytesTransferred());
+	}
+
 	RETURN_ERROR(error);
 }
 
@@ -463,7 +411,7 @@ UnixFifo::Writable() const
 }
 
 
-void
+status_t
 UnixFifo::SetBufferCapacity(size_t capacity)
 {
 	// check against allowed minimal/maximal value
@@ -474,20 +422,23 @@ UnixFifo::SetBufferCapacity(size_t capacity)
 
 	size_t oldCapacity = fBuffer.Capacity();
 	if (capacity == oldCapacity)
-		return;
+		return B_OK;
 
 	// set capacity
-	fBuffer.SetCapacity(capacity);
+	status_t error = fBuffer.SetCapacity(capacity);
+	if (error != B_OK)
+		return error;
 
 	// wake up waiting writers, if the capacity increased
 	if (!fWriters.IsEmpty() && !IsWriteShutdown())
 		fWriteCondition.NotifyAll();
+
+	return B_OK;
 }
 
 
 status_t
-UnixFifo::_Read(Request& request, size_t numBytes, bigtime_t timeout,
+UnixFifo::_Read(UnixRequest& request, bigtime_t timeout)
-	net_buffer** _buffer)
 {
 	// wait for the request to reach the front of the queue
 	if (fReaders.Head() != &request && timeout == 0)
@@ -509,10 +460,8 @@ UnixFifo::_Read(Request& request, size_t numBytes, bigtime_t timeout,
 		return UNIX_FIFO_SHUTDOWN;
 
 	if (fBuffer.Readable() == 0) {
-		if (IsWriteShutdown()) {
+		if (IsWriteShutdown())
-			*_buffer = NULL;
+			RETURN_ERROR(0);
-			RETURN_ERROR(B_OK);
-		}
 
 		if (timeout == 0)
 			RETURN_ERROR(B_WOULD_BLOCK);
@@ -536,21 +485,18 @@ UnixFifo::_Read(Request& request, size_t numBytes, bigtime_t timeout,
 	if (IsReadShutdown())
 		return UNIX_FIFO_SHUTDOWN;
 
-	if (fBuffer.Readable() == 0 && IsWriteShutdown()) {
+	if (fBuffer.Readable() == 0 && IsWriteShutdown())
-		*_buffer = NULL;
+		RETURN_ERROR(0);
-		RETURN_ERROR(B_OK);
-	}
 
-	RETURN_ERROR(fBuffer.Read(numBytes, _buffer));
+	RETURN_ERROR(fBuffer.Read(request));
 }
 
 
 status_t
-UnixFifo::_Write(Request& request, net_buffer* buffer, bigtime_t timeout,
+UnixFifo::_Write(UnixRequest& request, bigtime_t timeout)
-	size_t& bytesWritten)
 {
 	if (timeout == 0)
-		RETURN_ERROR(_WriteNonBlocking(request, buffer, bytesWritten));
+		RETURN_ERROR(_WriteNonBlocking(request));
 
 	// wait for the request to reach the front of the queue
 	while (fWriters.Head() != &request && !IsWriteShutdown()) {
@@ -571,13 +517,12 @@ UnixFifo::_Write(Request& request, net_buffer* buffer, bigtime_t timeout,
 	if (IsReadShutdown())
 		return EPIPE;
 
-	if (request.size == 0)
+	if (request.TotalSize() == 0)
-		return B_OK;
+		return 0;
 
 	status_t error = B_OK;
-	size_t bytesLeft = buffer->size;
 
-	while (error == B_OK && bytesLeft > 0) {
+	while (error == B_OK && request.BytesRemaining() > 0) {
 		// wait for any space to become available
 		while (error == B_OK && fBuffer.Writable() == 0 && !IsWriteShutdown()
 				&& !IsReadShutdown()) {
@@ -599,14 +544,11 @@ UnixFifo::_Write(Request& request, net_buffer* buffer, bigtime_t timeout,
 			return EPIPE;
 
 		// write as much as we can
-		size_t toWrite = min_c(fBuffer.Writable(), bytesLeft);
+		error = fBuffer.Write(request);
-		error = fBuffer.Write(buffer, toWrite);
 
 		if (error == B_OK) {
 // TODO: Whenever we've successfully written a part, we should reset the
 // timeout!
-			bytesWritten += toWrite;
-			bytesLeft -= toWrite;
 		}
 	}
 
@@ -615,35 +557,16 @@ UnixFifo::_Write(Request& request, net_buffer* buffer, bigtime_t timeout,
 
 
 status_t
-UnixFifo::_WriteNonBlocking(Request& request, net_buffer* buffer,
+UnixFifo::_WriteNonBlocking(UnixRequest& request)
-	size_t& bytesWritten)
 {
 	// We need to be first in queue and space should be available right now,
 	// otherwise we need to fail.
 	if (fWriters.Head() != &request || fBuffer.Writable() == 0)
 		RETURN_ERROR(B_WOULD_BLOCK);
 
-	if (request.size == 0)
+	if (request.TotalSize() == 0)
-		return B_OK;
+		return 0;
 
 	// Write as much as we can.
-	size_t toWrite = min_c(fBuffer.Writable(), buffer->size);
+	RETURN_ERROR(fBuffer.Write(request));
-	status_t error;
-
-	if (buffer->size <= fBuffer.Writable()) {
-		// enough space available
-		error = fBuffer.Write(buffer, toWrite);
-		if (error == B_OK)
-			bytesWritten = toWrite;
-	} else {
-		// not enough space available -- write what we can, but return
-		// B_WOULD_BLOCK nevertheless
-		error = fBuffer.Write(buffer,toWrite);
-		if (error == B_OK) {
-			bytesWritten = toWrite;
-			error = B_WOULD_BLOCK;
-		}
-	}
-
-	RETURN_ERROR(error);
 }

src/add-ons/kernel/network/protocols/unix/UnixFifo.h

@@ -18,14 +18,40 @@
 #define UNIX_FIFO_SHUTDOWN_READ		1
 #define UNIX_FIFO_SHUTDOWN_WRITE	2
 
-#define UNIX_FIFO_SHUTDOWN			1
+#define UNIX_FIFO_SHUTDOWN			(B_ERRORS_END + 1)
 	// error code returned by Read()/Write()
 
 #define UNIX_FIFO_MINIMAL_CAPACITY	1024
 #define UNIX_FIFO_MAXIMAL_CAPACITY	(128 * 1024)
 
 
-#define TRACE_BUFFER_QUEUE	0
+struct ring_buffer;
+
+class UnixRequest : public DoublyLinkedListLinkImpl<UnixRequest> {
+public:
+	UnixRequest(const iovec* vecs, size_t count,
+			ancillary_data_container* ancillaryData);
+
+	off_t TotalSize() const			{ return fTotalSize; }
+	off_t BytesTransferred() const	{ return fBytesTransferred; }
+	off_t BytesRemaining() const	{ return fTotalSize - fBytesTransferred; }
+
+	void AddBytesTransferred(size_t size);
+	bool GetCurrentChunk(void*& data, size_t& size);
+
+	ancillary_data_container* AncillaryData() const	 { return fAncillaryData; }
+	void SetAncillaryData(ancillary_data_container* data);
+	void AddAncillaryData(ancillary_data_container* data);
+
+private:
+	const iovec*				fVecs;
+	size_t						fVecCount;
+	ancillary_data_container*	fAncillaryData;
+	off_t						fTotalSize;
+	off_t						fBytesTransferred;
+	size_t						fVecIndex;
+	size_t						fVecOffset;
+};
 
 
 class UnixBufferQueue {
@@ -33,38 +59,31 @@ public:
 	UnixBufferQueue(size_t capacity);
 	~UnixBufferQueue();
 
-	size_t	Readable() const	{ return fSize; }
+	status_t Init();
-	size_t	Writable() const
-		{ return fCapacity >= fSize ? fCapacity - fSize : 0; }
 
-	status_t Read(size_t size, net_buffer** _buffer);
+	size_t	Readable() const;
-	status_t Write(net_buffer* buffer, size_t maxSize);
+	size_t	Writable() const;
 
-	size_t Capacity() const		{ return fCapacity; }
+	status_t Read(UnixRequest& request);
-	void SetCapacity(size_t capacity);
+	status_t Write(UnixRequest& request);
 
-#if TRACE_BUFFER_QUEUE
+	size_t Capacity() const				{ return fCapacity; }
-	void ParanoiaReadCheck(net_buffer* buffer);
+	status_t SetCapacity(size_t capacity);
-	void PostReadWrite();
-#endif
 
 private:
-	typedef DoublyLinkedList<net_buffer, DoublyLinkedListCLink<net_buffer> >
+	struct AncillaryDataEntry : DoublyLinkedListLinkImpl<AncillaryDataEntry> {
-		BufferList;
+		ancillary_data_container*	data;
+		size_t						offset;
+	};
 
-	BufferList	fBuffers;
+	typedef DoublyLinkedList<AncillaryDataEntry> AncillaryDataList;
-	size_t		fSize;
+
-	size_t		fCapacity;
+	ring_buffer*		fBuffer;
-#if TRACE_BUFFER_QUEUE
+	size_t				fCapacity;
-	off_t		fWritten;
+	AncillaryDataList	fAncillaryData;
-	off_t		fRead;
-	uint8*		fParanoiaCheckBuffer;
-	uint8*		fParanoiaCheckBuffer2;
-#endif
 };
 
 
-
 class UnixFifo : public Referenceable {
 public:
 	UnixFifo(size_t capacity);
@@ -94,42 +113,31 @@ public:
 		return (fShutdown & UNIX_FIFO_SHUTDOWN_WRITE);
 	}
 
-	status_t Read(size_t numBytes, bigtime_t timeout, net_buffer** _buffer);
+	ssize_t Read(const iovec* vecs, size_t vecCount,
-	status_t Write(net_buffer* buffer, bigtime_t timeout);
+		ancillary_data_container** _ancillaryData, bigtime_t timeout);
+	ssize_t Write(const iovec* vecs, size_t vecCount,
+		ancillary_data_container* ancillaryData, bigtime_t timeout);
 
 	size_t Readable() const;
 	size_t Writable() const;
 
-	void SetBufferCapacity(size_t capacity);
+	status_t SetBufferCapacity(size_t capacity);
 
 private:
-	struct Request : DoublyLinkedListLinkImpl<Request> {
+	typedef DoublyLinkedList<UnixRequest> RequestList;
-		Request(size_t size)
-			:
-			size(size)
-		{
-		}
-
-		size_t		size;
-	};
-
-	typedef DoublyLinkedList<Request> RequestList;
 
 private:
-	status_t _Read(Request& request, size_t numBytes, bigtime_t timeout,
+	status_t _Read(UnixRequest& request, bigtime_t timeout);
-		net_buffer** _buffer);
+	status_t _Write(UnixRequest& request, bigtime_t timeout);
-	status_t _Write(Request& request, net_buffer* buffer, bigtime_t timeout,
+	status_t _WriteNonBlocking(UnixRequest& request);
-		size_t& bytesWritten);
-	status_t _WriteNonBlocking(Request& request, net_buffer* buffer,
-		size_t& bytesWritten);
 
 private:
 	mutex				fLock;
 	UnixBufferQueue		fBuffer;
 	RequestList			fReaders;
 	RequestList			fWriters;
-	size_t				fReadRequested;
+	off_t				fReadRequested;
-	size_t				fWriteRequested;
+	off_t				fWriteRequested;
 	ConditionVariable	fReadCondition;
 	ConditionVariable	fWriteCondition;
 	uint32				fShutdown;

src/add-ons/kernel/network/protocols/unix/unix.cpp

@@ -148,7 +148,9 @@ unix_setsockopt(net_protocol *protocol, int level, int option,
 			if (length != sizeof(int))
 				return B_BAD_VALUE;
 
-			endpoint->SetReceiveBufferSize(*(int*)_value);
+			status_t error = endpoint->SetReceiveBufferSize(*(int*)_value);
+			if (error != B_OK)
+				return error;
 		} else if (option == SO_SNDBUF) {
 			// We don't have a receive buffer, so silently ignore this one.
 		}
@@ -198,7 +200,7 @@ unix_send_routed_data(net_protocol *_protocol, struct net_route *route,
 status_t
 unix_send_data(net_protocol *_protocol, net_buffer *buffer)
 {
-	return ((UnixEndpoint*)_protocol)->Send(buffer);
+	return B_ERROR;
 }
 
 
@@ -213,7 +215,7 @@ status_t
 unix_read_data(net_protocol *_protocol, size_t numBytes, uint32 flags,
 	net_buffer **_buffer)
 {
-	return ((UnixEndpoint*)_protocol)->Receive(numBytes, flags, _buffer);
+	return B_ERROR;
 }
 
 
@@ -380,6 +382,25 @@ unix_process_ancillary_data(net_protocol *self,
 }
 
 
+ssize_t
+unix_send_data_no_buffer(net_protocol *_protocol, const iovec *vecs,
+	size_t vecCount, ancillary_data_container *ancillaryData,
+	const struct sockaddr *address, socklen_t addressLength)
+{
+	return ((UnixEndpoint*)_protocol)->Send(vecs, vecCount, ancillaryData);
+}
+
+
+ssize_t
+unix_read_data_no_buffer(net_protocol *_protocol, const iovec *vecs,
+	size_t vecCount, ancillary_data_container **_ancillaryData,
+	struct sockaddr *_address, socklen_t *_addressLength)
+{
+	return ((UnixEndpoint*)_protocol)->Receive(vecs, vecCount, _ancillaryData,
+		_address, _addressLength);
+}
+
+
 // #pragma mark -
 
 
@@ -469,7 +490,9 @@ net_protocol_module_info gUnixModule = {
 	unix_error,
 	unix_error_reply,
 	unix_add_ancillary_data,
-	unix_process_ancillary_data
+	unix_process_ancillary_data,
+	unix_send_data_no_buffer,
+	unix_read_data_no_buffer
 };
 
 module_dependency module_dependencies[] = {