unix: Implement datagram sockets

Implement `SOCK_DGRAM` sockets for `AF_UNIX` family.

Change-Id: If3d6f408a7d881635ccf04b080391905fdc94b13
Reviewed-on: https://review.haiku-os.org/c/haiku/+/6617
Reviewed-by: Jérôme Duval <[email protected]>
Tested-by: Commit checker robot <[email protected]>

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

@@ -9,4 +9,7 @@ KernelAddon unix :
 	UnixAddress.cpp
 	UnixEndpoint.cpp
 	UnixFifo.cpp
+
+	UnixDatagramEndpoint.cpp
+	UnixStreamEndpoint.cpp
 ;

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

@@ -0,0 +1,560 @@
+/*
+ * Copyright 2023, Trung Nguyen, [email protected].
+ * Distributed under the terms of the MIT License.
+ */
+
+
+#include "UnixDatagramEndpoint.h"
+
+#include <new>
+
+#include "unix.h"
+#include "UnixAddressManager.h"
+#include "UnixFifo.h"
+
+
+#define UNIX_DATAGRAM_ENDPOINT_DEBUG_LEVEL	0
+#define UNIX_DEBUG_LEVEL					UNIX_DATAGRAM_ENDPOINT_DEBUG_LEVEL
+#include "UnixDebug.h"
+
+
+typedef AutoLocker<UnixDatagramEndpoint> UnixDatagramEndpointLocker;
+
+
+UnixDatagramEndpoint::UnixDatagramEndpoint(net_socket* socket)
+	:
+	UnixEndpoint(socket),
+	fTargetEndpoint(NULL),
+	fReceiveFifo(NULL),
+	fShutdownWrite(false)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::UnixDatagramEndpoint()\n",
+		find_thread(NULL), this);
+}
+
+
+UnixDatagramEndpoint::~UnixDatagramEndpoint()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::~UnixDatagramEndpoint()\n",
+		find_thread(NULL), this);
+}
+
+
+status_t
+UnixDatagramEndpoint::Init()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Init()\n",
+		find_thread(NULL), this);
+
+	RETURN_ERROR(B_OK);
+}
+
+
+void
+UnixDatagramEndpoint::Uninit()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Uninit()\n",
+		find_thread(NULL), this);
+
+	ReleaseReference();
+}
+
+
+status_t
+UnixDatagramEndpoint::Open()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Open()\n",
+		find_thread(NULL), this);
+
+	status_t error = ProtocolSocket::Open();
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixDatagramEndpoint::Close()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Close()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	if (IsBound())
+		RETURN_ERROR(UnixEndpoint::_Unbind());
+
+	_UnsetReceiveFifo();
+
+	RETURN_ERROR(_Disconnect());
+}
+
+
+status_t
+UnixDatagramEndpoint::Free()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Free()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	_UnsetReceiveFifo();
+
+	RETURN_ERROR(_Disconnect());
+}
+
+
+status_t
+UnixDatagramEndpoint::Bind(const struct sockaddr* _address)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Bind(\"%s\")\n",
+		find_thread(NULL), this,
+		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
+
+	if (_address->sa_family != AF_UNIX)
+		RETURN_ERROR(EAFNOSUPPORT);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	if (IsBound())
+		RETURN_ERROR(B_BAD_VALUE);
+
+	const sockaddr_un* address = (const sockaddr_un*)_address;
+
+	RETURN_ERROR(_Bind(address));
+}
+
+
+status_t
+UnixDatagramEndpoint::Unbind()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Unbind()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	if (IsBound())
+		RETURN_ERROR(UnixEndpoint::_Unbind());
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixDatagramEndpoint::Listen(int backlog)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Listen(%d)\n", find_thread(NULL),
+		this, backlog);
+
+	RETURN_ERROR(EOPNOTSUPP);
+}
+
+
+status_t
+UnixDatagramEndpoint::Connect(const struct sockaddr* _address)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Connect(\"%s\")\n",
+		find_thread(NULL), this,
+		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	BReference<UnixDatagramEndpoint> targetEndpointReference;
+	status_t status = _InitializeEndpoint(_address, targetEndpointReference);
+	if (status != B_OK)
+		RETURN_ERROR(status);
+
+	endpointLocker.Unlock();
+
+	UnixDatagramEndpoint* targetEndpoint = targetEndpointReference.Get();
+	UnixDatagramEndpointLocker targetLocker(targetEndpoint);
+
+	if (targetEndpoint->fTargetEndpoint != NULL && targetEndpoint->fTargetEndpoint != this)
+		RETURN_ERROR(EPERM);
+
+	targetLocker.Unlock();
+	endpointLocker.Lock();
+
+	status = _Disconnect();
+	if (status != B_OK)
+		RETURN_ERROR(status);
+
+	fTargetEndpoint = targetEndpoint;
+	fTargetEndpoint->AcquireReference();
+
+	// Required by the socket layer.
+	PeerAddress().SetTo(&fTargetEndpoint->socket->address);
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixDatagramEndpoint::Accept(net_socket** _acceptedSocket)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Accept()\n",
+		find_thread(NULL), this);
+
+	RETURN_ERROR(EOPNOTSUPP);
+}
+
+
+ssize_t
+UnixDatagramEndpoint::Send(const iovec* vecs, size_t vecCount,
+	ancillary_data_container* ancillaryData, const struct sockaddr* address,
+	socklen_t addressLength)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Send()\n",
+		find_thread(NULL), this);
+
+	bigtime_t timeout = absolute_timeout(socket->send.timeout);
+	if (gStackModule->is_restarted_syscall())
+		timeout = gStackModule->restore_syscall_restart_timeout();
+	else
+		gStackModule->store_syscall_restart_timeout(timeout);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	if (fShutdownWrite)
+		RETURN_ERROR(EPIPE);
+
+	status_t status;
+
+	BReference<UnixDatagramEndpoint> targetEndpointReference;
+	if (address == NULL) {
+		if (fTargetEndpoint == NULL)
+			RETURN_ERROR(ENOTCONN);
+
+		targetEndpointReference.SetTo(fTargetEndpoint);
+	} else {
+		status = _InitializeEndpoint(address, targetEndpointReference);
+		if (status != B_OK)
+			RETURN_ERROR(status);
+	}
+
+	// Get the address before unlocking the sending endpoint.
+	struct sockaddr_storage sourceAddress;
+	memcpy(&sourceAddress, &socket->address, sizeof(struct sockaddr_storage));
+	endpointLocker.Unlock();
+
+	UnixDatagramEndpoint* targetEndpoint = targetEndpointReference.Get();
+	UnixDatagramEndpointLocker targetLocker(targetEndpoint);
+
+	if (targetEndpoint->fTargetEndpoint != NULL && targetEndpoint->fTargetEndpoint != this)
+		RETURN_ERROR(EPERM);
+
+	if (targetEndpoint->fShutdownRead)
+		RETURN_ERROR(EPIPE);
+
+	if (targetEndpoint->fReceiveFifo == NULL) {
+		targetEndpoint->fReceiveFifo
+			= new (std::nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT, UnixFifoType::Datagram);
+		if (targetEndpoint->fReceiveFifo == NULL)
+			RETURN_ERROR(B_NO_MEMORY);
+
+		status = targetEndpoint->fReceiveFifo->Init();
+		if (status != B_OK) {
+			targetEndpoint->_UnsetReceiveFifo();
+			RETURN_ERROR(status);
+		}
+	}
+
+	UnixFifo* targetFifo = targetEndpoint->fReceiveFifo;
+	BReference<UnixFifo> targetFifoReference(targetFifo);
+	UnixFifoLocker fifoLocker(targetFifo);
+
+	targetLocker.Unlock();
+
+	ssize_t result = targetFifo->Write(vecs, vecCount, ancillaryData, &sourceAddress,
+		timeout);
+
+	// Notify select()ing readers, if we successfully wrote anything.
+	size_t readable = targetFifo->Readable();
+	bool notifyRead = (readable > 0 && result >= 0);
+
+	// Notify select()ing writers, if we failed to write anything and there's
+	// still room to write.
+	size_t writable = targetFifo->Writable();
+	bool notifyWrite = (writable > 0 && result < 0);
+
+	fifoLocker.Unlock();
+	targetLocker.Lock();
+
+	if (notifyRead)
+		gSocketModule->notify(targetEndpoint->socket, B_SELECT_READ, readable);
+
+	targetLocker.Unlock();
+
+	if (notifyWrite) {
+		endpointLocker.Lock();
+		gSocketModule->notify(socket, B_SELECT_WRITE, writable);
+	}
+
+	switch (result) {
+		case EPIPE:
+			if (gStackModule->is_syscall())
+				send_signal(find_thread(NULL), SIGPIPE);
+			break;
+		case B_TIMED_OUT:
+			if (timeout == 0)
+				result = B_WOULD_BLOCK;
+			break;
+	}
+
+	RETURN_ERROR(result);
+}
+
+
+ssize_t
+UnixDatagramEndpoint::Receive(const iovec* vecs, size_t vecCount,
+	ancillary_data_container** _ancillaryData, struct sockaddr* _address,
+	socklen_t* _addressLength)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Receive()\n",
+		find_thread(NULL), this);
+
+	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
+	if (gStackModule->is_restarted_syscall())
+		timeout = gStackModule->restore_syscall_restart_timeout();
+	else
+		gStackModule->store_syscall_restart_timeout(timeout);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	// It is not clearly specified in POSIX how to treat pending
+	// datagrams when a socket has been shut down for reading.
+	// On Linux, pending messages are still read.
+	if (fShutdownRead)
+		RETURN_ERROR(0);
+
+	status_t status;
+
+	if (fReceiveFifo == NULL) {
+		fReceiveFifo = new (std::nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT,
+			UnixFifoType::Datagram);
+		if (fReceiveFifo == NULL)
+			RETURN_ERROR(B_NO_MEMORY);
+
+		status = fReceiveFifo->Init();
+		if (status != B_OK) {
+			_UnsetReceiveFifo();
+			RETURN_ERROR(status);
+		}
+	}
+
+	UnixFifo* fifo = fReceiveFifo;
+	BReference<UnixFifo> fifoReference(fifo);
+	UnixFifoLocker fifoLocker(fifo);
+
+	endpointLocker.Unlock();
+
+	struct sockaddr_storage addressStorage;
+
+	ssize_t result = fifo->Read(vecs, vecCount, _ancillaryData, &addressStorage, timeout);
+
+	// Notify select()ing writers, if we successfully read anything.
+	size_t writable = fifo->Writable();
+	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 = (result < 0 && readable > 0
+		&& !fifo->IsReadShutdown());
+
+	// re-lock our endpoint (unlock FIFO to respect locking order)
+	fifoLocker.Unlock();
+	endpointLocker.Lock();
+
+	// send notifications
+	if (notifyRead)
+		gSocketModule->notify(socket, B_SELECT_READ, readable);
+
+	if (notifyWrite) {
+		BReference<UnixDatagramEndpoint> originEndpointReference;
+		status = _InitializeEndpoint((struct sockaddr*)&addressStorage,
+			originEndpointReference);
+		if (status == B_OK) {
+			UnixDatagramEndpoint* originEndpoint = originEndpointReference.Get();
+			endpointLocker.Unlock();
+			UnixDatagramEndpointLocker originLocker(originEndpoint);
+			gSocketModule->notify(originEndpoint->socket, B_SELECT_WRITE, writable);
+			originLocker.Unlock();
+			endpointLocker.Lock();
+		}
+	}
+
+	if (result < 0) {
+		switch (result) {
+			case B_TIMED_OUT:
+				if (timeout == 0)
+					result = B_WOULD_BLOCK;
+				break;
+		}
+	} else {
+		if (_address != NULL) {
+			if (_addressLength == NULL)
+				RETURN_ERROR(B_BAD_ADDRESS);
+			struct sockaddr_un* address = (struct sockaddr_un*)&addressStorage;
+			socklen_t memoryLength = min_c(*_addressLength, address->sun_len);
+			memcpy(_address, address, memoryLength);
+			*_addressLength = address->sun_len;
+		}
+	}
+
+	RETURN_ERROR(result);
+}
+
+
+ssize_t
+UnixDatagramEndpoint::Sendable()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Sendable()\n",
+		find_thread(NULL), this);
+
+	RETURN_ERROR(EOPNOTSUPP);
+}
+
+
+ssize_t
+UnixDatagramEndpoint::Receivable()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Receivable()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker locker(this);
+
+	if (fReceiveFifo == NULL)
+		RETURN_ERROR(0);
+
+	UnixFifoLocker fifoLocker(fReceiveFifo);
+	ssize_t readable = fReceiveFifo->Readable();
+
+	RETURN_ERROR(readable);
+}
+
+
+status_t
+UnixDatagramEndpoint::SetReceiveBufferSize(size_t size)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::SetReceiveBufferSize()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker locker(this);
+
+	if (fReceiveFifo == NULL)
+		RETURN_ERROR(0);
+
+	UnixFifoLocker fifoLocker(fReceiveFifo);
+	RETURN_ERROR(fReceiveFifo->SetBufferCapacity(size));
+}
+
+
+status_t
+UnixDatagramEndpoint::GetPeerCredentials(ucred* credentials)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::GetPeerCredentials()\n",
+		find_thread(NULL), this);
+
+	RETURN_ERROR(EOPNOTSUPP);
+}
+
+
+status_t
+UnixDatagramEndpoint::Shutdown(int direction)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixDatagramEndpoint::Shutdown()\n",
+		find_thread(NULL), this);
+
+	UnixDatagramEndpointLocker endpointLocker(this);
+
+	if (direction != SHUT_RD && direction != SHUT_WR && direction != SHUT_RDWR)
+		RETURN_ERROR(B_BAD_VALUE);
+
+	if (direction != SHUT_RD)
+		fShutdownWrite = true;
+
+	if (direction != SHUT_WR)
+		fShutdownRead = true;
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixDatagramEndpoint::_InitializeEndpoint(const struct sockaddr* _address,
+	BReference<UnixDatagramEndpoint>& outEndpoint)
+{
+	if (_address->sa_family != AF_UNIX)
+		RETURN_ERROR(EAFNOSUPPORT);
+
+	UnixAddress unixAddress;
+
+	const struct sockaddr_un* address = (const struct sockaddr_un*)_address;
+
+	if (address->sun_path[0] == '\0') {
+		// internal address space (or empty address)
+		int32 internalID;
+		if (UnixAddress::IsEmptyAddress(*address))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		internalID = UnixAddress::InternalID(*address);
+		if (internalID < 0)
+			RETURN_ERROR(internalID);
+
+		unixAddress.SetTo(internalID);
+	} else {
+		// FS address space
+		size_t pathLen = strnlen(address->sun_path, sizeof(address->sun_path));
+		if (pathLen == 0 || pathLen == sizeof(address->sun_path))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		struct stat st;
+		status_t error = vfs_read_stat(-1, address->sun_path, true, &st,
+			!gStackModule->is_syscall());
+		if (error != B_OK)
+			RETURN_ERROR(error);
+
+		if (!S_ISSOCK(st.st_mode))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		unixAddress.SetTo(st.st_dev, st.st_ino, NULL);
+	}
+
+	UnixAddressManagerLocker addressLocker(gAddressManager);
+	UnixEndpoint* targetUnixEndpoint = gAddressManager.Lookup(unixAddress);
+	if (targetUnixEndpoint == NULL)
+		RETURN_ERROR(ECONNREFUSED);
+	UnixDatagramEndpoint* targetEndpoint
+		= dynamic_cast<UnixDatagramEndpoint*>(targetUnixEndpoint);
+	if (targetEndpoint == NULL)
+		RETURN_ERROR(EPROTOTYPE);
+
+	outEndpoint.SetTo(targetEndpoint);
+	addressLocker.Unlock();
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixDatagramEndpoint::_Disconnect()
+{
+	if (fTargetEndpoint != NULL)
+		fTargetEndpoint->ReleaseReference();
+
+	fTargetEndpoint = NULL;
+	RETURN_ERROR(B_OK);
+}
+
+
+void
+UnixDatagramEndpoint::_UnsetReceiveFifo()
+{
+	if (fReceiveFifo != NULL) {
+		fReceiveFifo->ReleaseReference();
+		fReceiveFifo = NULL;
+	}
+}

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2023, Trung Nguyen, [email protected].
+ * Distributed under the terms of the MIT License.
+ */
+#ifndef UNIX_DATAGRAM_ENDPOINT_H
+#define UNIX_DATAGRAM_ENDPOINT_H
+
+
+#include <Referenceable.h>
+
+#include "UnixEndpoint.h"
+
+
+class UnixFifo;
+
+
+class UnixDatagramEndpoint : public UnixEndpoint, public BReferenceable {
+public:
+								UnixDatagramEndpoint(net_socket* socket);
+	virtual						~UnixDatagramEndpoint() override;
+
+			status_t			Init() override;
+			void				Uninit() override;
+
+			status_t			Open() override;
+			status_t			Close() override;
+			status_t			Free() override;
+
+			status_t			Bind(const struct sockaddr* _address) override;
+			status_t			Unbind() override;
+			status_t			Listen(int backlog) override;
+			status_t			Connect(const struct sockaddr* address) override;
+			status_t			Accept(net_socket** _acceptedSocket) override;
+
+			ssize_t				Send(const iovec* vecs, size_t vecCount,
+									ancillary_data_container* ancillaryData,
+									const struct sockaddr* address,
+									socklen_t addressLength) override;
+			ssize_t				Receive(const iovec* vecs, size_t vecCount,
+									ancillary_data_container** _ancillaryData,
+									struct sockaddr* _address,
+									socklen_t* _addressLength) override;
+
+			ssize_t				Sendable() override;
+			ssize_t				Receivable() override;
+
+			status_t			SetReceiveBufferSize(size_t size) override;
+			status_t			GetPeerCredentials(ucred* credentials) override;
+
+			status_t			Shutdown(int direction) override;
+
+	bool IsBound() const
+	{
+		return fAddress.IsValid();
+	}
+
+private:
+	static	status_t			_InitializeEndpoint(const struct sockaddr* _address,
+									BReference<UnixDatagramEndpoint> &outEndpoint);
+
+			status_t			_Disconnect();
+			void				_UnsetReceiveFifo();
+
+private:
+		UnixDatagramEndpoint*	fTargetEndpoint;
+		UnixFifo*				fReceiveFifo;
+		bool					fShutdownWrite:1;
+		bool					fShutdownRead:1;
+};
+
+
+#endif	// UNIX_DATAGRAM_ENDPOINT_H

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

@@ -1,41 +1,46 @@
 /*
- * Copyright 2008, Ingo Weinhold, [email protected].
+ * Copyright 2023, Trung Nguyen, [email protected].
  * Distributed under the terms of the MIT License.
  */
 
 
+#include <stdio.h>
+
+#include <new>
+
 #include "UnixEndpoint.h"
 
-#include <stdio.h>
-#include <sys/stat.h>
-
-#include <AutoDeleter.h>
-
-#include <vfs.h>
-
 #include "UnixAddressManager.h"
-#include "UnixFifo.h"
+#include "UnixDatagramEndpoint.h"
+#include "UnixStreamEndpoint.h"
 
 
-#define UNIX_ENDPOINT_DEBUG_LEVEL	0
+#define UNIX_ENDPOINT_DEBUG_LEVEL	1
-#define UNIX_DEBUG_LEVEL			UNIX_ENDPOINT_DEBUG_LEVEL
+#define UNIX_DEBUG_LEVEL    		UNIX_ENDPOINT_DEBUG_LEVEL
 #include "UnixDebug.h"
 
 
-// Note on locking order (outermost -> innermost):
+status_t
-// UnixEndpoint: connecting -> listening -> child
+UnixEndpoint::Create(net_socket* socket, UnixEndpoint** _endpoint)
-// -> UnixFifo (never lock more than one at a time)
-// -> UnixAddressManager
-
-
-static inline bigtime_t
-absolute_timeout(bigtime_t timeout)
 {
-	if (timeout == 0 || timeout == B_INFINITE_TIMEOUT)
+	TRACE("[%" B_PRId32 "] UnixEndpoint::Create(%p, %p)\n", find_thread(NULL),
-		return timeout;
+		socket, _endpoint);
 
-// TODO: Make overflow safe!
+	if (socket == NULL || _endpoint == NULL)
-	return timeout + system_time();
+		return B_BAD_ADDRESS;
+
+	switch (socket->type) {
+		case SOCK_STREAM:
+			*_endpoint = new(std::nothrow) UnixStreamEndpoint(socket);
+			break;
+		case SOCK_DGRAM:
+			*_endpoint = new(std::nothrow) UnixDatagramEndpoint(socket);
+			break;
+		default:
+			return EPROTOTYPE;
+	}
+
+	return *_endpoint == NULL ? B_NO_MEMORY : B_OK;
 }
 
 
@@ -43,13 +48,7 @@ UnixEndpoint::UnixEndpoint(net_socket* socket)
 	:
 	ProtocolSocket(socket),
 	fAddress(),
-	fAddressHashLink(),
+	fAddressHashLink()
-	fPeerEndpoint(NULL),
-	fReceiveFifo(NULL),
-	fState(UNIX_ENDPOINT_CLOSED),
-	fAcceptSemaphore(-1),
-	fIsChild(false),
-	fWasConnected(false)
 {
 	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::UnixEndpoint()\n",
 		find_thread(NULL), this);
@@ -60,7 +59,7 @@ UnixEndpoint::UnixEndpoint(net_socket* socket)
 
 UnixEndpoint::~UnixEndpoint()
 {
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::~UnixEndpoint()\n",
+	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::UnixEndpoint()\n",
 		find_thread(NULL), this);
 
 	mutex_destroy(&fLock);
@@ -68,110 +67,11 @@ UnixEndpoint::~UnixEndpoint()
 
 
 status_t
-UnixEndpoint::Init()
+UnixEndpoint::_Bind(const struct sockaddr_un* address)
 {
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Init()\n", find_thread(NULL),
+	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::_Bind(\"%s\")\n",
-		this);
-
-	RETURN_ERROR(B_OK);
-}
-
-
-void
-UnixEndpoint::Uninit()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Uninit()\n", find_thread(NULL),
-		this);
-
-	// check whether we're closed
-	UnixEndpointLocker locker(this);
-	bool closed = (fState == UNIX_ENDPOINT_CLOSED);
-	locker.Unlock();
-
-	if (!closed) {
-		// That probably means, we're a child endpoint of a listener and
-		// have been fully connected, but not yet accepted. Our Close()
-		// hook isn't called in this case. Do it manually.
-		Close();
-	}
-
-	ReleaseReference();
-}
-
-
-status_t
-UnixEndpoint::Open()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Open()\n", find_thread(NULL),
-		this);
-
-	status_t error = ProtocolSocket::Open();
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	fState = UNIX_ENDPOINT_NOT_CONNECTED;
-
-	RETURN_ERROR(B_OK);
-}
-
-
-status_t
-UnixEndpoint::Close()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Close()\n", find_thread(NULL),
-		this);
-
-	UnixEndpointLocker locker(this);
-
-	if (fState == UNIX_ENDPOINT_CONNECTED) {
-		UnixEndpointLocker peerLocker;
-		if (_LockConnectedEndpoints(locker, peerLocker) == B_OK) {
-			// We're still connected. Disconnect both endpoints!
-			fPeerEndpoint->_Disconnect();
-			_Disconnect();
-		}
-	}
-
-	if (fState == UNIX_ENDPOINT_LISTENING)
-		_StopListening();
-
-	_Unbind();
-
-	fState = UNIX_ENDPOINT_CLOSED;
-	RETURN_ERROR(B_OK);
-}
-
-
-status_t
-UnixEndpoint::Free()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Free()\n", find_thread(NULL),
-		this);
-
-	UnixEndpointLocker locker(this);
-
-	_UnsetReceiveFifo();
-
-	RETURN_ERROR(B_OK);
-}
-
-
-status_t
-UnixEndpoint::Bind(const struct sockaddr *_address)
-{
-	if (_address->sa_family != AF_UNIX)
-		RETURN_ERROR(EAFNOSUPPORT);
-
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Bind(\"%s\")\n",
 		find_thread(NULL), this,
-		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
+		ConstSocketAddress(&gAddressModule, (struct sockaddr*)address).AsString().Data());
-
-	const sockaddr_un* address = (const sockaddr_un*)_address;
-
-	UnixEndpointLocker endpointLocker(this);
-
-	if (fState != UNIX_ENDPOINT_NOT_CONNECTED || IsBound())
-		RETURN_ERROR(B_BAD_VALUE);
 
 	if (address->sun_path[0] == '\0') {
 		UnixAddressManagerLocker addressLocker(gAddressManager);
@@ -227,607 +127,6 @@ UnixEndpoint::Bind(const struct sockaddr *_address)
 }
 
 
-status_t
-UnixEndpoint::Unbind()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Unbind()\n", find_thread(NULL),
-		this);
-
-	UnixEndpointLocker endpointLocker(this);
-
-	RETURN_ERROR(_Unbind());
-}
-
-
-status_t
-UnixEndpoint::Listen(int backlog)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Listen(%d)\n", find_thread(NULL),
-		this, backlog);
-
-	UnixEndpointLocker endpointLocker(this);
-
-	if (!IsBound())
-		RETURN_ERROR(EDESTADDRREQ);
-	if (fState != UNIX_ENDPOINT_NOT_CONNECTED
-		&& fState != UNIX_ENDPOINT_LISTENING)
-		RETURN_ERROR(EINVAL);
-
-	gSocketModule->set_max_backlog(socket, backlog);
-
-	if (fState == UNIX_ENDPOINT_NOT_CONNECTED) {
-		fAcceptSemaphore = create_sem(0, "unix accept");
-		if (fAcceptSemaphore < 0)
-			RETURN_ERROR(ENOBUFS);
-	
-		_UnsetReceiveFifo();
-	
-		fCredentials.pid = getpid();
-		fCredentials.uid = geteuid();
-		fCredentials.gid = getegid();
-	
-		fState = UNIX_ENDPOINT_LISTENING;
-	}
-
-	RETURN_ERROR(B_OK);
-}
-
-
-status_t
-UnixEndpoint::Connect(const struct sockaddr *_address)
-{
-	if (_address->sa_family != AF_UNIX)
-		RETURN_ERROR(EAFNOSUPPORT);
-
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Connect(\"%s\")\n",
-		find_thread(NULL), this,
-		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
-
-	const sockaddr_un* address = (const sockaddr_un*)_address;
-
-	UnixEndpointLocker endpointLocker(this);
-
-	if (fState == UNIX_ENDPOINT_CONNECTED)
-		RETURN_ERROR(EISCONN);
-
-	if (fState != UNIX_ENDPOINT_NOT_CONNECTED)
-		RETURN_ERROR(B_BAD_VALUE);
-// TODO: If listening, we could set the backlog to 0 and connect.
-
-	// check the address first
-	UnixAddress unixAddress;
-
-	if (address->sun_path[0] == '\0') {
-		// internal address space (or empty address)
-		int32 internalID;
-		if (UnixAddress::IsEmptyAddress(*address))
-			RETURN_ERROR(B_BAD_VALUE);
-
-		internalID = UnixAddress::InternalID(*address);
-		if (internalID < 0)
-			RETURN_ERROR(internalID);
-
-		unixAddress.SetTo(internalID);
-	} else {
-		// FS address space
-		size_t pathLen = strnlen(address->sun_path, sizeof(address->sun_path));
-		if (pathLen == 0 || pathLen == sizeof(address->sun_path))
-			RETURN_ERROR(B_BAD_VALUE);
-
-		struct stat st;
-		status_t error = vfs_read_stat(-1, address->sun_path, true, &st,
-			!gStackModule->is_syscall());
-		if (error != B_OK)
-			RETURN_ERROR(error);
-
-		if (!S_ISSOCK(st.st_mode))
-			RETURN_ERROR(B_BAD_VALUE);
-
-		unixAddress.SetTo(st.st_dev, st.st_ino, NULL);
-	}
-
-	// get the peer endpoint
-	UnixAddressManagerLocker addressLocker(gAddressManager);
-	UnixEndpoint* listeningEndpoint = gAddressManager.Lookup(unixAddress);
-	if (listeningEndpoint == NULL)
-		RETURN_ERROR(ECONNREFUSED);
-	BReference<UnixEndpoint> peerReference(listeningEndpoint);
-	addressLocker.Unlock();
-
-	UnixEndpointLocker peerLocker(listeningEndpoint);
-
-	if (!listeningEndpoint->IsBound()
-		|| listeningEndpoint->fState != UNIX_ENDPOINT_LISTENING
-		|| listeningEndpoint->fAddress != unixAddress) {
-		RETURN_ERROR(ECONNREFUSED);
-	}
-
-	// Allocate FIFOs for us and the socket we're going to spawn. We do that
-	// now, so that the mess we need to cleanup, if allocating them fails, is
-	// harmless.
-	UnixFifo* fifo = new(nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT);
-	UnixFifo* peerFifo = new(nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT);
-	ObjectDeleter<UnixFifo> fifoDeleter(fifo);
-	ObjectDeleter<UnixFifo> peerFifoDeleter(peerFifo);
-
-	status_t error;
-	if ((error = fifo->Init()) != B_OK || (error = peerFifo->Init()) != B_OK)
-		return error;
-
-	// spawn new endpoint for accept()
-	net_socket* newSocket;
-	error = gSocketModule->spawn_pending_socket(listeningEndpoint->socket,
-		&newSocket);
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	// init connected peer endpoint
-	UnixEndpoint* connectedEndpoint = (UnixEndpoint*)newSocket->first_protocol;
-
-	UnixEndpointLocker connectedLocker(connectedEndpoint);
-
-	connectedEndpoint->_Spawn(this, listeningEndpoint, peerFifo);
-
-	// update our attributes
-	_UnsetReceiveFifo();
-
-	fPeerEndpoint = connectedEndpoint;
-	PeerAddress().SetTo(&connectedEndpoint->socket->address);
-	fPeerEndpoint->AcquireReference();
-	fReceiveFifo = fifo;
-
-	fCredentials.pid = getpid();
-	fCredentials.uid = geteuid();
-	fCredentials.gid = getegid();
-
-	fifoDeleter.Detach();
-	peerFifoDeleter.Detach();
-
-	fState = UNIX_ENDPOINT_CONNECTED;
-	fWasConnected = true;
-
-	gSocketModule->set_connected(newSocket);
-
-	release_sem(listeningEndpoint->fAcceptSemaphore);
-
-	connectedLocker.Unlock();
-	peerLocker.Unlock();
-	endpointLocker.Unlock();
-
-	RETURN_ERROR(B_OK);
-}
-
-
-status_t
-UnixEndpoint::Accept(net_socket **_acceptedSocket)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Accept()\n", find_thread(NULL),
-		this);
-
-	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
-	if (gStackModule->is_restarted_syscall())
-		timeout = gStackModule->restore_syscall_restart_timeout();
-	else
-		gStackModule->store_syscall_restart_timeout(timeout);
-
-	UnixEndpointLocker locker(this);
-
-	status_t error;
-	do {
-		locker.Unlock();
-
-		error = acquire_sem_etc(fAcceptSemaphore, 1,
-			B_ABSOLUTE_TIMEOUT | B_CAN_INTERRUPT, timeout);
-		if (error < B_OK)
-			break;
-
-		locker.Lock();
-		error = gSocketModule->dequeue_connected(socket, _acceptedSocket);
-	} while (error != B_OK);
-
-	if (error == B_TIMED_OUT && timeout == 0) {
-		// translate non-blocking timeouts to the correct error code
-		error = B_WOULD_BLOCK;
-	}
-
-	RETURN_ERROR(error);
-}
-
-
-ssize_t
-UnixEndpoint::Send(const iovec *vecs, size_t vecCount,
-	ancillary_data_container *ancillaryData)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Send(%p, %ld, %p)\n",
-		find_thread(NULL), this, vecs, vecCount, ancillaryData);
-
-	bigtime_t timeout = absolute_timeout(socket->send.timeout);
-	if (gStackModule->is_restarted_syscall())
-		timeout = gStackModule->restore_syscall_restart_timeout();
-	else
-		gStackModule->store_syscall_restart_timeout(timeout);
-
-	UnixEndpointLocker locker(this);
-
-	BReference<UnixEndpoint> peerReference;
-	UnixEndpointLocker peerLocker;
-
-	status_t error = _LockConnectedEndpoints(locker, peerLocker);
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	UnixEndpoint* peerEndpoint = fPeerEndpoint;
-	peerReference.SetTo(peerEndpoint);
-
-	// lock the peer's FIFO
-	UnixFifo* peerFifo = peerEndpoint->fReceiveFifo;
-	BReference<UnixFifo> _(peerFifo);
-	UnixFifoLocker fifoLocker(peerFifo);
-
-	// unlock endpoints
-	locker.Unlock();
-	peerLocker.Unlock();
-
-	ssize_t result = peerFifo->Write(vecs, vecCount, ancillaryData, timeout);
-
-	// Notify select()ing readers, if we successfully wrote anything.
-	size_t readable = peerFifo->Readable();
-	bool notifyRead = (error == B_OK && readable > 0
-		&& !peerFifo->IsReadShutdown());
-
-	// Notify select()ing writers, if we failed to write anything and there's
-	// still room to write.
-	size_t writable = peerFifo->Writable();
-	bool notifyWrite = (error != B_OK && writable > 0
-		&& !peerFifo->IsWriteShutdown());
-
-	// re-lock our endpoint (unlock FIFO to respect locking order)
-	fifoLocker.Unlock();
-	locker.Lock();
-
-	bool peerLocked = (fPeerEndpoint == peerEndpoint
-		&& _LockConnectedEndpoints(locker, peerLocker) == B_OK);
-
-	// send notifications
-	if (peerLocked && notifyRead)
-		gSocketModule->notify(peerEndpoint->socket, B_SELECT_READ, readable);
-	if (notifyWrite)
-		gSocketModule->notify(socket, B_SELECT_WRITE, writable);
-
-	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.
-				result = EPIPE;
-			} else {
-				// The FD has been closed.
-				result = EBADF;
-			}
-			break;
-		case EPIPE:
-			// The peer closed connection or shutdown its read side. Reward
-			// the caller with a SIGPIPE.
-			if (gStackModule->is_syscall())
-				send_signal(find_thread(NULL), SIGPIPE);
-			break;
-		case B_TIMED_OUT:
-			// Translate non-blocking timeouts to the correct error code.
-			if (timeout == 0)
-				result = B_WOULD_BLOCK;
-			break;
-	}
-
-	RETURN_ERROR(result);
-}
-
-
-ssize_t
-UnixEndpoint::Receive(const iovec *vecs, size_t vecCount,
-	ancillary_data_container **_ancillaryData, struct sockaddr *_address,
-	socklen_t *_addressLength)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Receive(%p, %ld)\n",
-		find_thread(NULL), this, vecs, vecCount);
-
-	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
-	if (gStackModule->is_restarted_syscall())
-		timeout = gStackModule->restore_syscall_restart_timeout();
-	else
-		gStackModule->store_syscall_restart_timeout(timeout);
-
-	UnixEndpointLocker locker(this);
-
-	// We can read as long as we have a FIFO. I.e. we are still connected, or
-	// disconnected and not yet reconnected/listening/closed.
-	if (fReceiveFifo == NULL)
-		RETURN_ERROR(ENOTCONN);
-
-	UnixEndpoint* peerEndpoint = fPeerEndpoint;
-	BReference<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;
-	BReference<UnixFifo> _(fifo);
-	UnixFifoLocker fifoLocker(fifo);
-
-	// unlock endpoint
-	locker.Unlock();
-
-	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 = (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 = (result < 0 && readable > 0
-		&& !fifo->IsReadShutdown());
-
-	// re-lock our endpoint (unlock FIFO to respect locking order)
-	fifoLocker.Unlock();
-	locker.Lock();
-
-	UnixEndpointLocker peerLocker;
-	bool peerLocked = (peerEndpoint != NULL && fPeerEndpoint == peerEndpoint
-		&& _LockConnectedEndpoints(locker, peerLocker) == B_OK);
-
-	// send notifications
-	if (notifyRead)
-		gSocketModule->notify(socket, B_SELECT_READ, readable);
-	if (peerLocked && notifyWrite)
-		gSocketModule->notify(peerEndpoint->socket, B_SELECT_WRITE, writable);
-
-	switch (result) {
-		case UNIX_FIFO_SHUTDOWN:
-			// Either our socket was closed or read shutdown.
-			if (fState == UNIX_ENDPOINT_CLOSED) {
-				// The FD has been closed.
-				result = EBADF;
-			} else {
-				// if (fReceiveFifo == fifo) {
-				// 		Orderly shutdown or the peer closed the connection.
-				// } else {
-				//		Weird case: Peer closed connection and we are already
-				// 		reconnected (or listening).
-				// }
-				result = 0;
-			}
-			break;
-		case B_TIMED_OUT:
-			// translate non-blocking timeouts to the correct error code
-			if (timeout == 0)
-				result = B_WOULD_BLOCK;
-			break;
-	}
-
-	RETURN_ERROR(result);
-}
-
-
-ssize_t
-UnixEndpoint::Sendable()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Sendable()\n", find_thread(NULL),
-		this);
-
-	UnixEndpointLocker locker(this);
-	UnixEndpointLocker peerLocker;
-
-	status_t error = _LockConnectedEndpoints(locker, peerLocker);
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	// lock the peer's FIFO
-	UnixFifo* peerFifo = fPeerEndpoint->fReceiveFifo;
-	UnixFifoLocker fifoLocker(peerFifo);
-
-	RETURN_ERROR(peerFifo->Writable());
-}
-
-
-ssize_t
-UnixEndpoint::Receivable()
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Receivable()\n", find_thread(NULL),
-		this);
-
-	UnixEndpointLocker locker(this);
-
-	if (fState == UNIX_ENDPOINT_LISTENING)
-		return gSocketModule->count_connected(socket);
-
-	if (fState != UNIX_ENDPOINT_CONNECTED)
-		RETURN_ERROR(ENOTCONN);
-
-	UnixFifoLocker fifoLocker(fReceiveFifo);
-	ssize_t readable = fReceiveFifo->Readable();
-	if (readable == 0 && (fReceiveFifo->IsWriteShutdown()
-			|| fReceiveFifo->IsReadShutdown())) {
-		RETURN_ERROR(ENOTCONN);
-	}
-	RETURN_ERROR(readable);
-}
-
-
-status_t
-UnixEndpoint::SetReceiveBufferSize(size_t size)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::SetReceiveBufferSize(%lu)\n",
-		find_thread(NULL), this, size);
-
-	UnixEndpointLocker locker(this);
-
-	if (fReceiveFifo == NULL)
-		return B_BAD_VALUE;
-
-	UnixFifoLocker fifoLocker(fReceiveFifo);
-	return fReceiveFifo->SetBufferCapacity(size);
-}
-
-
-status_t
-UnixEndpoint::GetPeerCredentials(ucred* credentials)
-{
-	UnixEndpointLocker locker(this);
-	UnixEndpointLocker peerLocker;
-
-	status_t error = _LockConnectedEndpoints(locker, peerLocker);
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	*credentials = fPeerEndpoint->fCredentials;
-
-	return B_OK;
-}
-
-
-status_t
-UnixEndpoint::Shutdown(int direction)
-{
-	TRACE("[%" B_PRId32 "] %p->UnixEndpoint::Shutdown(%d)\n",
-		find_thread(NULL), this, direction);
-
-	uint32 shutdown;
-	uint32 peerShutdown;
-
-	// translate the direction into shutdown flags for our and the peer fifo
-	switch (direction) {
-		case SHUT_RD:
-			shutdown = UNIX_FIFO_SHUTDOWN_READ;
-			peerShutdown = 0;
-			break;
-		case SHUT_WR:
-			shutdown = 0;
-			peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
-			break;
-		case SHUT_RDWR:
-			shutdown = UNIX_FIFO_SHUTDOWN_READ;
-			peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
-			break;
-		default:
-			RETURN_ERROR(B_BAD_VALUE);
-	}
-
-	// lock endpoints
-	UnixEndpointLocker locker(this);
-	UnixEndpointLocker peerLocker;
-
-	status_t error = _LockConnectedEndpoints(locker, peerLocker);
-	if (error != B_OK)
-		RETURN_ERROR(error);
-
-	// shutdown our FIFO
-	fReceiveFifo->Lock();
-	fReceiveFifo->Shutdown(shutdown);
-	fReceiveFifo->Unlock();
-
-	// shutdown peer FIFO
-	fPeerEndpoint->fReceiveFifo->Lock();
-	fPeerEndpoint->fReceiveFifo->Shutdown(peerShutdown);
-	fPeerEndpoint->fReceiveFifo->Unlock();
-
-	// send select notifications
-	if (direction == SHUT_RD || direction == SHUT_RDWR) {
-		gSocketModule->notify(socket, B_SELECT_READ, EPIPE);
-		gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_WRITE, EPIPE);
-	}
-	if (direction == SHUT_WR || direction == SHUT_RDWR) {
-		gSocketModule->notify(socket, B_SELECT_WRITE, EPIPE);
-		gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_READ, EPIPE);
-	}
-
-	RETURN_ERROR(B_OK);
-}
-
-
-void
-UnixEndpoint::_Spawn(UnixEndpoint* connectingEndpoint,
-	UnixEndpoint* listeningEndpoint, UnixFifo* fifo)
-{
-	ProtocolSocket::Open();
-
-	fIsChild = true;
-	fPeerEndpoint = connectingEndpoint;
-	fPeerEndpoint->AcquireReference();
-
-	fReceiveFifo = fifo;
-
-	PeerAddress().SetTo(&connectingEndpoint->socket->address);
-
-	fCredentials = listeningEndpoint->fCredentials;
-
-	fState = UNIX_ENDPOINT_CONNECTED;
-}
-
-
-void
-UnixEndpoint::_Disconnect()
-{
-	// Both endpoints must be locked.
-
-	// Write shutdown the receive FIFO.
-	fReceiveFifo->Lock();
-	fReceiveFifo->Shutdown(UNIX_FIFO_SHUTDOWN_WRITE);
-	fReceiveFifo->Unlock();
-
-	// select() notification.
-	gSocketModule->notify(socket, B_SELECT_READ, ECONNRESET);
-	gSocketModule->notify(socket, B_SELECT_WRITE, ECONNRESET);
-
-	// Unset the peer endpoint.
-	fPeerEndpoint->ReleaseReference();
-	fPeerEndpoint = NULL;
-
-	// We're officially disconnected.
-// TODO: Deal with non accept()ed connections correctly!
-	fIsChild = false;
-	fState = UNIX_ENDPOINT_NOT_CONNECTED;
-}
-
-
-status_t
-UnixEndpoint::_LockConnectedEndpoints(UnixEndpointLocker& locker,
-	UnixEndpointLocker& peerLocker)
-{
-	if (fState != UNIX_ENDPOINT_CONNECTED)
-		RETURN_ERROR(fWasConnected ? EPIPE : ENOTCONN);
-
-	// We need to lock the peer, too. Get a reference -- we might need to
-	// unlock ourselves to get the locking order right.
-	BReference<UnixEndpoint> peerReference(fPeerEndpoint);
-	UnixEndpoint* peerEndpoint = fPeerEndpoint;
-
-	if (fIsChild) {
-		// We're the child, but locking order is the other way around.
-		locker.Unlock();
-		peerLocker.SetTo(peerEndpoint, false);
-
-		locker.Lock();
-
-		// recheck our state, also whether the peer is still the same
-		if (fState != UNIX_ENDPOINT_CONNECTED || peerEndpoint != fPeerEndpoint)
-			RETURN_ERROR(ENOTCONN);
-	} else
-		peerLocker.SetTo(peerEndpoint, false);
-
-	RETURN_ERROR(B_OK);
-}
-
-
 status_t
 UnixEndpoint::_Bind(struct vnode* vnode)
 {
@@ -852,38 +151,11 @@ UnixEndpoint::_Bind(int32 internalID)
 status_t
 UnixEndpoint::_Unbind()
 {
-	if (fState == UNIX_ENDPOINT_CONNECTED || fState == UNIX_ENDPOINT_LISTENING)
+	UnixAddressManagerLocker addressLocker(gAddressManager);
-		RETURN_ERROR(B_BAD_VALUE);
+	gAddressManager.Remove(this);
-
+	if (struct vnode* vnode = fAddress.Vnode())
-	if (IsBound()) {
+		vfs_put_vnode(vnode);
-		UnixAddressManagerLocker addressLocker(gAddressManager);
-		gAddressManager.Remove(this);
-		if (struct vnode* vnode = fAddress.Vnode())
-			vfs_put_vnode(vnode);
-
-		fAddress.Unset();
-	}
 
+	fAddress.Unset();
 	RETURN_ERROR(B_OK);
 }
-
-
-void
-UnixEndpoint::_UnsetReceiveFifo()
-{
-	if (fReceiveFifo) {
-		fReceiveFifo->ReleaseReference();
-		fReceiveFifo = NULL;
-	}
-}
-
-
-void
-UnixEndpoint::_StopListening()
-{
-	if (fState == UNIX_ENDPOINT_LISTENING) {
-		delete_sem(fAcceptSemaphore);
-		fAcceptSemaphore = -1;
-		fState = UNIX_ENDPOINT_NOT_CONNECTED;
-	}
-}

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

@@ -1,54 +1,24 @@
 /*
- * Copyright 2008, Ingo Weinhold, [email protected].
+ * Copyright 2023, Trung Nguyen, [email protected].
  * Distributed under the terms of the MIT License.
  */
 #ifndef UNIX_ENDPOINT_H
 #define UNIX_ENDPOINT_H
 
-#include <sys/stat.h>
-
-#include <Referenceable.h>
-
-#include <lock.h>
-#include <util/DoublyLinkedList.h>
-#include <util/OpenHashTable.h>
-#include <vfs.h>
 
 #include <net_protocol.h>
 #include <net_socket.h>
 #include <ProtocolUtilities.h>
 
-#include "unix.h"
+#include <lock.h>
+#include <vfs.h>
+
 #include "UnixAddress.h"
 
 
-class UnixEndpoint;
+class UnixEndpoint : public net_protocol, public ProtocolSocket {
-class UnixFifo;
-
-
-enum unix_endpoint_state {
-	UNIX_ENDPOINT_NOT_CONNECTED,
-	UNIX_ENDPOINT_LISTENING,
-	UNIX_ENDPOINT_CONNECTED,
-	UNIX_ENDPOINT_CLOSED
-};
-
-
-typedef AutoLocker<UnixEndpoint> UnixEndpointLocker;
-
-
-class UnixEndpoint : public net_protocol, public ProtocolSocket,
-	public BReferenceable {
 public:
-	UnixEndpoint(net_socket* socket);
+	virtual						~UnixEndpoint();
-	virtual ~UnixEndpoint();
-
-	status_t Init();
-	void Uninit();
-
-	status_t Open();
-	status_t Close();
-	status_t Free();
 
 	bool Lock()
 	{
@@ -60,31 +30,6 @@ public:
 		mutex_unlock(&fLock);
 	}
 
-	status_t Bind(const struct sockaddr *_address);
-	status_t Unbind();
-	status_t Listen(int backlog);
-	status_t Connect(const struct sockaddr *address);
-	status_t Accept(net_socket **_acceptedSocket);
-
-	ssize_t Send(const iovec *vecs, size_t vecCount,
-		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();
-
-	status_t SetReceiveBufferSize(size_t size);
-	status_t GetPeerCredentials(ucred* credentials);
-
-	status_t Shutdown(int direction);
-
-	bool IsBound() const
-	{
-		return !fIsChild && fAddress.IsValid();
-	}
-
 	const UnixAddress& Address() const
 	{
 		return fAddress;
@@ -95,31 +40,66 @@ public:
 		return fAddressHashLink;
 	}
 
-private:
+	virtual	status_t			Init() = 0;
-	void _Spawn(UnixEndpoint* connectingEndpoint,
+	virtual	void				Uninit() = 0;
-		UnixEndpoint* listeningEndpoint, UnixFifo* fifo);
-	void _Disconnect();
-	status_t _LockConnectedEndpoints(UnixEndpointLocker& locker,
-		UnixEndpointLocker& peerLocker);
 
-	status_t _Bind(struct vnode* vnode);
+	virtual	status_t			Open() = 0;
-	status_t _Bind(int32 internalID);
+	virtual	status_t			Close() = 0;
-	status_t _Unbind();
+	virtual	status_t			Free() = 0;
 
-	void _UnsetReceiveFifo();
+	virtual	status_t			Bind(const struct sockaddr* _address) = 0;
-	void _StopListening();
+	virtual	status_t			Unbind() = 0;
+	virtual	status_t			Listen(int backlog) = 0;
+	virtual	status_t			Connect(const struct sockaddr* address) = 0;
+	virtual	status_t			Accept(net_socket** _acceptedSocket) = 0;
+
+	virtual	ssize_t				Send(const iovec* vecs, size_t vecCount,
+									ancillary_data_container* ancillaryData,
+									const struct sockaddr* address,
+									socklen_t addressLength) = 0;
+	virtual	ssize_t				Receive(const iovec* vecs, size_t vecCount,
+									ancillary_data_container** _ancillaryData,
+									struct sockaddr* _address, socklen_t* _addressLength) = 0;
+
+	virtual	ssize_t				Sendable() = 0;
+	virtual	ssize_t				Receivable() = 0;
+
+	virtual	status_t			SetReceiveBufferSize(size_t size) = 0;
+	virtual	status_t			GetPeerCredentials(ucred* credentials) = 0;
+
+	virtual	status_t			Shutdown(int direction) = 0;
+
+	static	status_t			Create(net_socket* socket, UnixEndpoint** _endpoint);
+
+protected:
+								UnixEndpoint(net_socket* socket);
+
+	// These functions perform no locking or checking on the endpoint.
+			status_t			_Bind(const struct sockaddr_un* address);
+			status_t			_Unbind();
 
 private:
-	mutex							fLock;
+			status_t			_Bind(struct vnode* vnode);
-	UnixAddress						fAddress;
+			status_t			_Bind(int32 internalID);
-	UnixEndpoint*					fAddressHashLink;
+
-	UnixEndpoint*					fPeerEndpoint;
+protected:
-	UnixFifo*						fReceiveFifo;
+			UnixAddress			fAddress;
-	unix_endpoint_state				fState;
+
-	sem_id							fAcceptSemaphore;
+private:
-	ucred							fCredentials;
+			mutex				fLock;
-	bool							fIsChild;
+			UnixEndpoint*		fAddressHashLink;
-	bool							fWasConnected;
 };
 
+
+static inline bigtime_t
+absolute_timeout(bigtime_t timeout)
+{
+	if (timeout == 0 || timeout == B_INFINITE_TIMEOUT)
+		return timeout;
+
+// TODO: Make overflow safe!
+	return timeout + system_time();
+}
+
+
 #endif	// UNIX_ENDPOINT_H

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

@@ -24,7 +24,8 @@
 
 
 UnixRequest::UnixRequest(const iovec* vecs, size_t count,
-		ancillary_data_container* ancillaryData)
+		ancillary_data_container* ancillaryData,
+		struct sockaddr_storage* address)
 	:
 	fVecs(vecs),
 	fVecCount(count),
@@ -32,7 +33,8 @@ UnixRequest::UnixRequest(const iovec* vecs, size_t count,
 	fTotalSize(0),
 	fBytesTransferred(0),
 	fVecIndex(0),
-	fVecOffset(0)
+	fVecOffset(0),
+	fAddress(address)
 {
 	for (size_t i = 0; i < fVecCount; i++)
 		fTotalSize += fVecs[i].iov_len;
@@ -95,10 +97,11 @@ UnixRequest::AddAncillaryData(ancillary_data_container* data)
 // #pragma mark - UnixBufferQueue
 
 
-UnixBufferQueue::UnixBufferQueue(size_t capacity)
+UnixBufferQueue::UnixBufferQueue(size_t capacity, UnixFifoType type)
 	:
 	fBuffer(NULL),
-	fCapacity(capacity)
+	fCapacity(capacity),
+	fType(type)
 {
 }
 
@@ -147,6 +150,19 @@ UnixBufferQueue::Read(UnixRequest& request)
 	void* data;
 	size_t size;
 
+	DatagramEntry* datagramEntry = NULL;
+	if (fType == UnixFifoType::Datagram) {
+		datagramEntry = fDatagrams.Head();
+		if (datagramEntry == NULL)
+			return B_ERROR;
+
+		if (datagramEntry->size > readable)
+			TRACE("UnixBufferQueue::Read(): expected to read a datagram of size %lu, "
+				"but only %lu bytes are readable\n", datagramEntry->size, readable);
+		else
+			readable = datagramEntry->size;
+	}
+
 	while (readable > 0 && request.GetCurrentChunk(data, size)) {
 		if (size > readable)
 			size = readable;
@@ -184,6 +200,30 @@ UnixBufferQueue::Read(UnixRequest& request)
 		readable -= bytesRead;
 	}
 
+	if (fType == UnixFifoType::Datagram) {
+		fDatagrams.RemoveHead();
+
+		memcpy(request.Address(), &datagramEntry->address, sizeof(datagramEntry->address));
+		delete datagramEntry;
+
+		if (readable > 0) {
+			ring_buffer_flush(fBuffer, readable);
+			if (AncillaryDataEntry* entry = fAncillaryData.Head()) {
+				size_t offsetDelta = readable;
+				while (entry != NULL && offsetDelta > entry->offset) {
+					fAncillaryData.RemoveHead();
+					offsetDelta -= entry->offset;
+					delete entry;
+
+					entry = fAncillaryData.Head();
+				}
+
+				if (entry != NULL)
+					entry->offset -= offsetDelta;
+			}
+		}
+	}
+
 	return B_OK;
 }
 
@@ -197,6 +237,26 @@ UnixBufferQueue::Write(UnixRequest& request)
 	void* data;
 	size_t size;
 
+	DatagramEntry* datagramEntry = NULL;
+	ObjectDeleter<DatagramEntry> datagramEntryDeleter;
+	if (fType == UnixFifoType::Datagram) {
+		datagramEntry = new(std::nothrow) DatagramEntry;
+		if (datagramEntry == NULL)
+			return B_NO_MEMORY;
+
+		datagramEntryDeleter.SetTo(datagramEntry);
+		memcpy(&datagramEntry->address, request.Address(),
+			sizeof(datagramEntry->address));
+		datagramEntry->size = request.TotalSize();
+
+		// This should have been handled in UnixFifo
+		if (writable < datagramEntry->size) {
+			TRACE("UnixBufferQueue::Write(): not enough space for"
+				"datagram of size %lu (%lu bytes left)\n", datagramEntry->size, writable);
+			return B_ERROR;
+		}
+	}
+
 	// If the request has ancillary data create an entry first.
 	AncillaryDataEntry* ancillaryEntry = NULL;
 	ObjectDeleter<AncillaryDataEntry> ancillaryEntryDeleter;
@@ -244,6 +304,11 @@ UnixBufferQueue::Write(UnixRequest& request)
 		writable -= bytesWritten;
 	}
 
+	if (fType == UnixFifoType::Datagram) {
+		fDatagrams.Add(datagramEntry);
+		datagramEntryDeleter.Detach();
+	}
+
 	return B_OK;
 }
 
@@ -259,9 +324,9 @@ return B_ERROR;
 // #pragma mark -
 
 
-UnixFifo::UnixFifo(size_t capacity)
+UnixFifo::UnixFifo(size_t capacity, UnixFifoType type)
 	:
-	fBuffer(capacity),
+	fBuffer(capacity, type),
 	fReaders(),
 	fWriters(),
 	fReadRequested(0),
@@ -306,7 +371,8 @@ UnixFifo::Shutdown(uint32 shutdown)
 
 ssize_t
 UnixFifo::Read(const iovec* vecs, size_t vecCount,
-	ancillary_data_container** _ancillaryData, bigtime_t timeout)
+	ancillary_data_container** _ancillaryData,
+	struct sockaddr_storage* address, bigtime_t timeout)
 {
 	TRACE("[%" B_PRId32 "] %p->UnixFifo::Read(%p, %ld, %" B_PRIdBIGTIME ")\n",
 		find_thread(NULL), this, vecs, vecCount, timeout);
@@ -314,7 +380,7 @@ UnixFifo::Read(const iovec* vecs, size_t vecCount,
 	if (IsReadShutdown() && fBuffer.Readable() == 0)
 		RETURN_ERROR(UNIX_FIFO_SHUTDOWN);
 
-	UnixRequest request(vecs, vecCount, NULL);
+	UnixRequest request(vecs, vecCount, NULL, address);
 	fReaders.Add(&request);
 	fReadRequested += request.TotalSize();
 
@@ -351,7 +417,8 @@ UnixFifo::Read(const iovec* vecs, size_t vecCount,
 
 ssize_t
 UnixFifo::Write(const iovec* vecs, size_t vecCount,
-	ancillary_data_container* ancillaryData, bigtime_t timeout)
+	ancillary_data_container* ancillaryData,
+	const struct sockaddr_storage* address, bigtime_t timeout)
 {
 	TRACE("[%" B_PRId32 "] %p->UnixFifo::Write(%p, %ld, %p, %" B_PRIdBIGTIME
 		")\n", find_thread(NULL), this, vecs, vecCount, ancillaryData,
@@ -363,7 +430,8 @@ UnixFifo::Write(const iovec* vecs, size_t vecCount,
 	if (IsReadShutdown())
 		RETURN_ERROR(EPIPE);
 
-	UnixRequest request(vecs, vecCount, ancillaryData);
+	UnixRequest request(vecs, vecCount, ancillaryData,
+		(struct sockaddr_storage*)address);
 	fWriters.Add(&request);
 	fWriteRequested += request.TotalSize();
 
@@ -530,8 +598,8 @@ UnixFifo::_Write(UnixRequest& request, bigtime_t timeout)
 
 	while (error == B_OK && request.BytesRemaining() > 0) {
 		// wait for any space to become available
-		while (error == B_OK && fBuffer.Writable() == 0 && !IsWriteShutdown()
+		while (error == B_OK && fBuffer.Writable() < _MinimumWritableSize(request)
-				&& !IsReadShutdown()) {
+				&& !IsWriteShutdown() && !IsReadShutdown()) {
 			ConditionVariableEntry entry;
 			fWriteCondition.Add(&entry);
 
@@ -567,7 +635,7 @@ UnixFifo::_WriteNonBlocking(UnixRequest& request)
 {
 	// 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)
+	if (fWriters.Head() != &request || fBuffer.Writable() < _MinimumWritableSize(request))
 		RETURN_ERROR(B_WOULD_BLOCK);
 
 	if (request.TotalSize() == 0)
@@ -577,3 +645,15 @@ UnixFifo::_WriteNonBlocking(UnixRequest& request)
 	RETURN_ERROR(fBuffer.Write(request));
 }
 
+
+size_t
+UnixFifo::_MinimumWritableSize(const UnixRequest& request) const
+{
+	switch (fType) {
+		case UnixFifoType::Datagram:
+			return request.TotalSize();
+		case UnixFifoType::Stream:
+		default:
+			return 1;
+	}
+}

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

@@ -25,12 +25,19 @@
 #define UNIX_FIFO_MAXIMAL_CAPACITY	(128 * 1024)
 
 
+enum class UnixFifoType {
+	Stream,
+	Datagram
+};
+
+
 struct ring_buffer;
 
 class UnixRequest : public DoublyLinkedListLinkImpl<UnixRequest> {
 public:
 	UnixRequest(const iovec* vecs, size_t count,
-			ancillary_data_container* ancillaryData);
+			ancillary_data_container* ancillaryData,
+			struct sockaddr_storage* address);
 
 	off_t TotalSize() const			{ return fTotalSize; }
 	off_t BytesTransferred() const	{ return fBytesTransferred; }
@@ -43,20 +50,23 @@ public:
 	void SetAncillaryData(ancillary_data_container* data);
 	void AddAncillaryData(ancillary_data_container* data);
 
+	struct sockaddr_storage* Address() const	{ return fAddress; }
+
 private:
-	const iovec*				fVecs;
+	const iovec*					fVecs;
-	size_t						fVecCount;
+	size_t							fVecCount;
-	ancillary_data_container*	fAncillaryData;
+	ancillary_data_container*		fAncillaryData;
-	off_t						fTotalSize;
+	off_t							fTotalSize;
-	off_t						fBytesTransferred;
+	off_t							fBytesTransferred;
-	size_t						fVecIndex;
+	size_t							fVecIndex;
-	size_t						fVecOffset;
+	size_t							fVecOffset;
+	struct sockaddr_storage*		fAddress;
 };
 
 
 class UnixBufferQueue {
 public:
-	UnixBufferQueue(size_t capacity);
+	UnixBufferQueue(size_t capacity, UnixFifoType type);
 	~UnixBufferQueue();
 
 	status_t Init();
@@ -78,15 +88,24 @@ private:
 
 	typedef DoublyLinkedList<AncillaryDataEntry> AncillaryDataList;
 
+	struct DatagramEntry : DoublyLinkedListLinkImpl<DatagramEntry> {
+		struct sockaddr_storage	address;
+		size_t	size;
+	};
+
+	typedef DoublyLinkedList<DatagramEntry> DatagramList;
+
 	ring_buffer*		fBuffer;
 	size_t				fCapacity;
 	AncillaryDataList	fAncillaryData;
+	DatagramList		fDatagrams;
+	UnixFifoType		fType;
 };
 
 
 class UnixFifo : public BReferenceable {
 public:
-	UnixFifo(size_t capacity);
+	UnixFifo(size_t capacity, UnixFifoType type);
 	~UnixFifo();
 
 	status_t Init();
@@ -114,9 +133,11 @@ public:
 	}
 
 	ssize_t Read(const iovec* vecs, size_t vecCount,
-		ancillary_data_container** _ancillaryData, bigtime_t timeout);
+		ancillary_data_container** _ancillaryData,
+		struct sockaddr_storage* address, bigtime_t timeout);
 	ssize_t Write(const iovec* vecs, size_t vecCount,
-		ancillary_data_container* ancillaryData, bigtime_t timeout);
+		ancillary_data_container* ancillaryData,
+		const struct sockaddr_storage* address, bigtime_t timeout);
 
 	size_t Readable() const;
 	size_t Writable() const;
@@ -130,6 +151,7 @@ private:
 	status_t _Read(UnixRequest& request, bigtime_t timeout);
 	status_t _Write(UnixRequest& request, bigtime_t timeout);
 	status_t _WriteNonBlocking(UnixRequest& request);
+	size_t _MinimumWritableSize(const UnixRequest& request) const;
 
 private:
 	mutex				fLock;
@@ -141,6 +163,7 @@ private:
 	ConditionVariable	fReadCondition;
 	ConditionVariable	fWriteCondition;
 	uint32				fShutdown;
+	UnixFifoType		fType;
 };
 
 

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

@@ -0,0 +1,801 @@
+/*
+ * Copyright 2008, Ingo Weinhold, [email protected].
+ * Distributed under the terms of the MIT License.
+ */
+
+
+#include "UnixStreamEndpoint.h"
+
+#include <stdio.h>
+#include <sys/stat.h>
+
+#include <AutoDeleter.h>
+
+#include <vfs.h>
+
+#include "UnixAddressManager.h"
+#include "UnixFifo.h"
+
+
+#define UNIX_STREAM_ENDPOINT_DEBUG_LEVEL	0
+#define UNIX_DEBUG_LEVEL					UNIX_STREAM_ENDPOINT_DEBUG_LEVEL
+#include "UnixDebug.h"
+
+
+// Note on locking order (outermost -> innermost):
+// UnixStreamEndpoint: connecting -> listening -> child
+// -> UnixFifo (never lock more than one at a time)
+// -> UnixAddressManager
+
+
+UnixStreamEndpoint::UnixStreamEndpoint(net_socket* socket)
+	:
+	UnixEndpoint(socket),
+	fPeerEndpoint(NULL),
+	fReceiveFifo(NULL),
+	fState(unix_stream_endpoint_state::Closed),
+	fAcceptSemaphore(-1),
+	fIsChild(false),
+	fWasConnected(false)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::UnixStreamEndpoint()\n",
+		find_thread(NULL), this);
+}
+
+
+UnixStreamEndpoint::~UnixStreamEndpoint()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::~UnixStreamEndpoint()\n",
+		find_thread(NULL), this);
+}
+
+
+status_t
+UnixStreamEndpoint::Init()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Init()\n", find_thread(NULL),
+		this);
+
+	RETURN_ERROR(B_OK);
+}
+
+
+void
+UnixStreamEndpoint::Uninit()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Uninit()\n", find_thread(NULL),
+		this);
+
+	// check whether we're closed
+	UnixStreamEndpointLocker locker(this);
+	bool closed = (fState == unix_stream_endpoint_state::Closed);
+	locker.Unlock();
+
+	if (!closed) {
+		// That probably means, we're a child endpoint of a listener and
+		// have been fully connected, but not yet accepted. Our Close()
+		// hook isn't called in this case. Do it manually.
+		Close();
+	}
+
+	ReleaseReference();
+}
+
+
+status_t
+UnixStreamEndpoint::Open()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Open()\n", find_thread(NULL),
+		this);
+
+	status_t error = ProtocolSocket::Open();
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	fState = unix_stream_endpoint_state::NotConnected;
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::Close()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Close()\n", find_thread(NULL),
+		this);
+
+	UnixStreamEndpointLocker locker(this);
+
+	if (fState == unix_stream_endpoint_state::Connected) {
+		UnixStreamEndpointLocker peerLocker;
+		if (_LockConnectedEndpoints(locker, peerLocker) == B_OK) {
+			// We're still connected. Disconnect both endpoints!
+			fPeerEndpoint->_Disconnect();
+			_Disconnect();
+		}
+	}
+
+	if (fState == unix_stream_endpoint_state::Listening)
+		_StopListening();
+
+	_Unbind();
+
+	fState = unix_stream_endpoint_state::Closed;
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::Free()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Free()\n", find_thread(NULL),
+		this);
+
+	UnixStreamEndpointLocker locker(this);
+
+	_UnsetReceiveFifo();
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::Bind(const struct sockaddr* _address)
+{
+	if (_address->sa_family != AF_UNIX)
+		RETURN_ERROR(EAFNOSUPPORT);
+
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Bind(\"%s\")\n",
+		find_thread(NULL), this,
+		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
+
+	const sockaddr_un* address = (const sockaddr_un*)_address;
+
+	UnixStreamEndpointLocker endpointLocker(this);
+
+	if (fState != unix_stream_endpoint_state::NotConnected || IsBound())
+		RETURN_ERROR(B_BAD_VALUE);
+
+	RETURN_ERROR(_Bind(address));
+}
+
+
+status_t
+UnixStreamEndpoint::Unbind()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Unbind()\n", find_thread(NULL),
+		this);
+
+	UnixStreamEndpointLocker endpointLocker(this);
+
+	RETURN_ERROR(_Unbind());
+}
+
+
+status_t
+UnixStreamEndpoint::Listen(int backlog)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Listen(%d)\n", find_thread(NULL),
+		this, backlog);
+
+	UnixStreamEndpointLocker endpointLocker(this);
+
+	if (!IsBound())
+		RETURN_ERROR(EDESTADDRREQ);
+	if (fState != unix_stream_endpoint_state::NotConnected
+		&& fState != unix_stream_endpoint_state::Listening)
+		RETURN_ERROR(EINVAL);
+
+	gSocketModule->set_max_backlog(socket, backlog);
+
+	if (fState == unix_stream_endpoint_state::NotConnected) {
+		fAcceptSemaphore = create_sem(0, "unix accept");
+		if (fAcceptSemaphore < 0)
+			RETURN_ERROR(ENOBUFS);
+
+		_UnsetReceiveFifo();
+
+		fCredentials.pid = getpid();
+		fCredentials.uid = geteuid();
+		fCredentials.gid = getegid();
+
+		fState = unix_stream_endpoint_state::Listening;
+	}
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::Connect(const struct sockaddr* _address)
+{
+	if (_address->sa_family != AF_UNIX)
+		RETURN_ERROR(EAFNOSUPPORT);
+
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Connect(\"%s\")\n",
+		find_thread(NULL), this,
+		ConstSocketAddress(&gAddressModule, _address).AsString().Data());
+
+	const sockaddr_un* address = (const sockaddr_un*)_address;
+
+	UnixStreamEndpointLocker endpointLocker(this);
+
+	if (fState == unix_stream_endpoint_state::Connected)
+		RETURN_ERROR(EISCONN);
+
+	if (fState != unix_stream_endpoint_state::NotConnected)
+		RETURN_ERROR(B_BAD_VALUE);
+// TODO: If listening, we could set the backlog to 0 and connect.
+
+	// check the address first
+	UnixAddress unixAddress;
+
+	if (address->sun_path[0] == '\0') {
+		// internal address space (or empty address)
+		int32 internalID;
+		if (UnixAddress::IsEmptyAddress(*address))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		internalID = UnixAddress::InternalID(*address);
+		if (internalID < 0)
+			RETURN_ERROR(internalID);
+
+		unixAddress.SetTo(internalID);
+	} else {
+		// FS address space
+		size_t pathLen = strnlen(address->sun_path, sizeof(address->sun_path));
+		if (pathLen == 0 || pathLen == sizeof(address->sun_path))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		struct stat st;
+		status_t error = vfs_read_stat(-1, address->sun_path, true, &st,
+			!gStackModule->is_syscall());
+		if (error != B_OK)
+			RETURN_ERROR(error);
+
+		if (!S_ISSOCK(st.st_mode))
+			RETURN_ERROR(B_BAD_VALUE);
+
+		unixAddress.SetTo(st.st_dev, st.st_ino, NULL);
+	}
+
+	// get the peer endpoint
+	UnixAddressManagerLocker addressLocker(gAddressManager);
+	UnixEndpoint* listeningUnixEndpoint = gAddressManager.Lookup(unixAddress);
+	if (listeningUnixEndpoint == NULL)
+		RETURN_ERROR(ECONNREFUSED);
+	UnixStreamEndpoint* listeningEndpoint
+		= dynamic_cast<UnixStreamEndpoint*>(listeningUnixEndpoint);
+	if (listeningEndpoint == NULL)
+		RETURN_ERROR(EPROTOTYPE);
+	BReference<UnixStreamEndpoint> peerReference(listeningEndpoint);
+	addressLocker.Unlock();
+
+	UnixStreamEndpointLocker peerLocker(listeningEndpoint);
+
+	if (!listeningEndpoint->IsBound()
+		|| listeningEndpoint->fState != unix_stream_endpoint_state::Listening
+		|| listeningEndpoint->fAddress != unixAddress) {
+		RETURN_ERROR(ECONNREFUSED);
+	}
+
+	// Allocate FIFOs for us and the socket we're going to spawn. We do that
+	// now, so that the mess we need to cleanup, if allocating them fails, is
+	// harmless.
+	UnixFifo* fifo = new(nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT, UnixFifoType::Stream);
+	UnixFifo* peerFifo = new(nothrow) UnixFifo(UNIX_MAX_TRANSFER_UNIT, UnixFifoType::Stream);
+	ObjectDeleter<UnixFifo> fifoDeleter(fifo);
+	ObjectDeleter<UnixFifo> peerFifoDeleter(peerFifo);
+
+	status_t error;
+	if ((error = fifo->Init()) != B_OK || (error = peerFifo->Init()) != B_OK)
+		return error;
+
+	// spawn new endpoint for accept()
+	net_socket* newSocket;
+	error = gSocketModule->spawn_pending_socket(listeningEndpoint->socket,
+		&newSocket);
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	// init connected peer endpoint
+	UnixStreamEndpoint* connectedEndpoint = (UnixStreamEndpoint*)newSocket->first_protocol;
+
+	UnixStreamEndpointLocker connectedLocker(connectedEndpoint);
+
+	connectedEndpoint->_Spawn(this, listeningEndpoint, peerFifo);
+
+	// update our attributes
+	_UnsetReceiveFifo();
+
+	fPeerEndpoint = connectedEndpoint;
+	PeerAddress().SetTo(&connectedEndpoint->socket->address);
+	fPeerEndpoint->AcquireReference();
+	fReceiveFifo = fifo;
+
+	fCredentials.pid = getpid();
+	fCredentials.uid = geteuid();
+	fCredentials.gid = getegid();
+
+	fifoDeleter.Detach();
+	peerFifoDeleter.Detach();
+
+	fState = unix_stream_endpoint_state::Connected;
+	fWasConnected = true;
+
+	gSocketModule->set_connected(newSocket);
+
+	release_sem(listeningEndpoint->fAcceptSemaphore);
+
+	connectedLocker.Unlock();
+	peerLocker.Unlock();
+	endpointLocker.Unlock();
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::Accept(net_socket** _acceptedSocket)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Accept()\n", find_thread(NULL),
+		this);
+
+	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
+	if (gStackModule->is_restarted_syscall())
+		timeout = gStackModule->restore_syscall_restart_timeout();
+	else
+		gStackModule->store_syscall_restart_timeout(timeout);
+
+	UnixStreamEndpointLocker locker(this);
+
+	status_t error;
+	do {
+		locker.Unlock();
+
+		error = acquire_sem_etc(fAcceptSemaphore, 1,
+			B_ABSOLUTE_TIMEOUT | B_CAN_INTERRUPT, timeout);
+		if (error < B_OK)
+			break;
+
+		locker.Lock();
+		error = gSocketModule->dequeue_connected(socket, _acceptedSocket);
+	} while (error != B_OK);
+
+	if (error == B_TIMED_OUT && timeout == 0) {
+		// translate non-blocking timeouts to the correct error code
+		error = B_WOULD_BLOCK;
+	}
+
+	RETURN_ERROR(error);
+}
+
+
+ssize_t
+UnixStreamEndpoint::Send(const iovec* vecs, size_t vecCount,
+	ancillary_data_container* ancillaryData,
+	const struct sockaddr* address, socklen_t addressLength)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Send(%p, %ld, %p)\n",
+		find_thread(NULL), this, vecs, vecCount, ancillaryData);
+
+	bigtime_t timeout = absolute_timeout(socket->send.timeout);
+	if (gStackModule->is_restarted_syscall())
+		timeout = gStackModule->restore_syscall_restart_timeout();
+	else
+		gStackModule->store_syscall_restart_timeout(timeout);
+
+	UnixStreamEndpointLocker locker(this);
+
+	BReference<UnixStreamEndpoint> peerReference;
+	UnixStreamEndpointLocker peerLocker;
+
+	status_t error = _LockConnectedEndpoints(locker, peerLocker);
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;
+	peerReference.SetTo(peerEndpoint);
+
+	// lock the peer's FIFO
+	UnixFifo* peerFifo = peerEndpoint->fReceiveFifo;
+	BReference<UnixFifo> _(peerFifo);
+	UnixFifoLocker fifoLocker(peerFifo);
+
+	// unlock endpoints
+	locker.Unlock();
+	peerLocker.Unlock();
+
+	ssize_t result = peerFifo->Write(vecs, vecCount, ancillaryData, NULL, timeout);
+
+	// Notify select()ing readers, if we successfully wrote anything.
+	size_t readable = peerFifo->Readable();
+	bool notifyRead = (error == B_OK && readable > 0
+		&& !peerFifo->IsReadShutdown());
+
+	// Notify select()ing writers, if we failed to write anything and there's
+	// still room to write.
+	size_t writable = peerFifo->Writable();
+	bool notifyWrite = (error != B_OK && writable > 0
+		&& !peerFifo->IsWriteShutdown());
+
+	// re-lock our endpoint (unlock FIFO to respect locking order)
+	fifoLocker.Unlock();
+	locker.Lock();
+
+	bool peerLocked = (fPeerEndpoint == peerEndpoint
+		&& _LockConnectedEndpoints(locker, peerLocker) == B_OK);
+
+	// send notifications
+	if (peerLocked && notifyRead)
+		gSocketModule->notify(peerEndpoint->socket, B_SELECT_READ, readable);
+	if (notifyWrite)
+		gSocketModule->notify(socket, B_SELECT_WRITE, writable);
+
+	switch (result) {
+		case UNIX_FIFO_SHUTDOWN:
+			if (fPeerEndpoint == peerEndpoint
+					&& fState == unix_stream_endpoint_state::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.
+				result = EPIPE;
+			} else {
+				// The FD has been closed.
+				result = EBADF;
+			}
+			break;
+		case EPIPE:
+			// The peer closed connection or shutdown its read side. Reward
+			// the caller with a SIGPIPE.
+			if (gStackModule->is_syscall())
+				send_signal(find_thread(NULL), SIGPIPE);
+			break;
+		case B_TIMED_OUT:
+			// Translate non-blocking timeouts to the correct error code.
+			if (timeout == 0)
+				result = B_WOULD_BLOCK;
+			break;
+	}
+
+	RETURN_ERROR(result);
+}
+
+
+ssize_t
+UnixStreamEndpoint::Receive(const iovec* vecs, size_t vecCount,
+	ancillary_data_container** _ancillaryData, struct sockaddr* _address,
+	socklen_t* _addressLength)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Receive(%p, %ld)\n",
+		find_thread(NULL), this, vecs, vecCount);
+
+	bigtime_t timeout = absolute_timeout(socket->receive.timeout);
+	if (gStackModule->is_restarted_syscall())
+		timeout = gStackModule->restore_syscall_restart_timeout();
+	else
+		gStackModule->store_syscall_restart_timeout(timeout);
+
+	UnixStreamEndpointLocker locker(this);
+
+	// We can read as long as we have a FIFO. I.e. we are still connected, or
+	// disconnected and not yet reconnected/listening/closed.
+	if (fReceiveFifo == NULL)
+		RETURN_ERROR(ENOTCONN);
+
+	UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;
+	BReference<UnixStreamEndpoint> 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;
+	BReference<UnixFifo> _(fifo);
+	UnixFifoLocker fifoLocker(fifo);
+
+	// unlock endpoint
+	locker.Unlock();
+
+	ssize_t result = fifo->Read(vecs, vecCount, _ancillaryData, NULL, timeout);
+
+	// Notify select()ing writers, if we successfully read anything.
+	size_t writable = fifo->Writable();
+	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 = (result < 0 && readable > 0
+		&& !fifo->IsReadShutdown());
+
+	// re-lock our endpoint (unlock FIFO to respect locking order)
+	fifoLocker.Unlock();
+	locker.Lock();
+
+	UnixStreamEndpointLocker peerLocker;
+	bool peerLocked = (peerEndpoint != NULL && fPeerEndpoint == peerEndpoint
+		&& _LockConnectedEndpoints(locker, peerLocker) == B_OK);
+
+	// send notifications
+	if (notifyRead)
+		gSocketModule->notify(socket, B_SELECT_READ, readable);
+	if (peerLocked && notifyWrite)
+		gSocketModule->notify(peerEndpoint->socket, B_SELECT_WRITE, writable);
+
+	switch (result) {
+		case UNIX_FIFO_SHUTDOWN:
+			// Either our socket was closed or read shutdown.
+			if (fState == unix_stream_endpoint_state::Closed) {
+				// The FD has been closed.
+				result = EBADF;
+			} else {
+				// if (fReceiveFifo == fifo) {
+				// 		Orderly shutdown or the peer closed the connection.
+				// } else {
+				//		Weird case: Peer closed connection and we are already
+				// 		reconnected (or listening).
+				// }
+				result = 0;
+			}
+			break;
+		case B_TIMED_OUT:
+			// translate non-blocking timeouts to the correct error code
+			if (timeout == 0)
+				result = B_WOULD_BLOCK;
+			break;
+	}
+
+	RETURN_ERROR(result);
+}
+
+
+ssize_t
+UnixStreamEndpoint::Sendable()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Sendable()\n", find_thread(NULL),
+		this);
+
+	UnixStreamEndpointLocker locker(this);
+	UnixStreamEndpointLocker peerLocker;
+
+	status_t error = _LockConnectedEndpoints(locker, peerLocker);
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	// lock the peer's FIFO
+	UnixFifo* peerFifo = fPeerEndpoint->fReceiveFifo;
+	UnixFifoLocker fifoLocker(peerFifo);
+
+	RETURN_ERROR(peerFifo->Writable());
+}
+
+
+ssize_t
+UnixStreamEndpoint::Receivable()
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Receivable()\n", find_thread(NULL),
+		this);
+
+	UnixStreamEndpointLocker locker(this);
+
+	if (fState == unix_stream_endpoint_state::Listening)
+		return gSocketModule->count_connected(socket);
+
+	if (fState != unix_stream_endpoint_state::Connected)
+		RETURN_ERROR(ENOTCONN);
+
+	UnixFifoLocker fifoLocker(fReceiveFifo);
+	ssize_t readable = fReceiveFifo->Readable();
+	if (readable == 0 && (fReceiveFifo->IsWriteShutdown()
+			|| fReceiveFifo->IsReadShutdown())) {
+		RETURN_ERROR(ENOTCONN);
+	}
+	RETURN_ERROR(readable);
+}
+
+
+status_t
+UnixStreamEndpoint::SetReceiveBufferSize(size_t size)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::SetReceiveBufferSize(%lu)\n",
+		find_thread(NULL), this, size);
+
+	UnixStreamEndpointLocker locker(this);
+
+	if (fReceiveFifo == NULL)
+		return B_BAD_VALUE;
+
+	UnixFifoLocker fifoLocker(fReceiveFifo);
+	return fReceiveFifo->SetBufferCapacity(size);
+}
+
+
+status_t
+UnixStreamEndpoint::GetPeerCredentials(ucred* credentials)
+{
+	UnixStreamEndpointLocker locker(this);
+	UnixStreamEndpointLocker peerLocker;
+
+	status_t error = _LockConnectedEndpoints(locker, peerLocker);
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	*credentials = fPeerEndpoint->fCredentials;
+
+	return B_OK;
+}
+
+
+status_t
+UnixStreamEndpoint::Shutdown(int direction)
+{
+	TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Shutdown(%d)\n",
+		find_thread(NULL), this, direction);
+
+	uint32 shutdown;
+	uint32 peerShutdown;
+
+	// translate the direction into shutdown flags for our and the peer fifo
+	switch (direction) {
+		case SHUT_RD:
+			shutdown = UNIX_FIFO_SHUTDOWN_READ;
+			peerShutdown = 0;
+			break;
+		case SHUT_WR:
+			shutdown = 0;
+			peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
+			break;
+		case SHUT_RDWR:
+			shutdown = UNIX_FIFO_SHUTDOWN_READ;
+			peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
+			break;
+		default:
+			RETURN_ERROR(B_BAD_VALUE);
+	}
+
+	// lock endpoints
+	UnixStreamEndpointLocker locker(this);
+	UnixStreamEndpointLocker peerLocker;
+
+	status_t error = _LockConnectedEndpoints(locker, peerLocker);
+	if (error != B_OK)
+		RETURN_ERROR(error);
+
+	// shutdown our FIFO
+	fReceiveFifo->Lock();
+	fReceiveFifo->Shutdown(shutdown);
+	fReceiveFifo->Unlock();
+
+	// shutdown peer FIFO
+	fPeerEndpoint->fReceiveFifo->Lock();
+	fPeerEndpoint->fReceiveFifo->Shutdown(peerShutdown);
+	fPeerEndpoint->fReceiveFifo->Unlock();
+
+	// send select notifications
+	if (direction == SHUT_RD || direction == SHUT_RDWR) {
+		gSocketModule->notify(socket, B_SELECT_READ, EPIPE);
+		gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_WRITE, EPIPE);
+	}
+	if (direction == SHUT_WR || direction == SHUT_RDWR) {
+		gSocketModule->notify(socket, B_SELECT_WRITE, EPIPE);
+		gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_READ, EPIPE);
+	}
+
+	RETURN_ERROR(B_OK);
+}
+
+
+void
+UnixStreamEndpoint::_Spawn(UnixStreamEndpoint* connectingEndpoint,
+	UnixStreamEndpoint* listeningEndpoint, UnixFifo* fifo)
+{
+	ProtocolSocket::Open();
+
+	fIsChild = true;
+	fPeerEndpoint = connectingEndpoint;
+	fPeerEndpoint->AcquireReference();
+
+	fReceiveFifo = fifo;
+
+	PeerAddress().SetTo(&connectingEndpoint->socket->address);
+
+	fCredentials = listeningEndpoint->fCredentials;
+
+	fState = unix_stream_endpoint_state::Connected;
+}
+
+
+void
+UnixStreamEndpoint::_Disconnect()
+{
+	// Both endpoints must be locked.
+
+	// Write shutdown the receive FIFO.
+	fReceiveFifo->Lock();
+	fReceiveFifo->Shutdown(UNIX_FIFO_SHUTDOWN_WRITE);
+	fReceiveFifo->Unlock();
+
+	// select() notification.
+	gSocketModule->notify(socket, B_SELECT_READ, ECONNRESET);
+	gSocketModule->notify(socket, B_SELECT_WRITE, ECONNRESET);
+
+	// Unset the peer endpoint.
+	fPeerEndpoint->ReleaseReference();
+	fPeerEndpoint = NULL;
+
+	// We're officially disconnected.
+// TODO: Deal with non accept()ed connections correctly!
+	fIsChild = false;
+	fState = unix_stream_endpoint_state::NotConnected;
+}
+
+
+status_t
+UnixStreamEndpoint::_LockConnectedEndpoints(UnixStreamEndpointLocker& locker,
+	UnixStreamEndpointLocker& peerLocker)
+{
+	if (fState != unix_stream_endpoint_state::Connected)
+		RETURN_ERROR(fWasConnected ? EPIPE : ENOTCONN);
+
+	// We need to lock the peer, too. Get a reference -- we might need to
+	// unlock ourselves to get the locking order right.
+	BReference<UnixStreamEndpoint> peerReference(fPeerEndpoint);
+	UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;
+
+	if (fIsChild) {
+		// We're the child, but locking order is the other way around.
+		locker.Unlock();
+		peerLocker.SetTo(peerEndpoint, false);
+
+		locker.Lock();
+
+		// recheck our state, also whether the peer is still the same
+		if (fState != unix_stream_endpoint_state::Connected || peerEndpoint != fPeerEndpoint)
+			RETURN_ERROR(ENOTCONN);
+	} else
+		peerLocker.SetTo(peerEndpoint, false);
+
+	RETURN_ERROR(B_OK);
+}
+
+
+status_t
+UnixStreamEndpoint::_Unbind()
+{
+	if (fState == unix_stream_endpoint_state::Connected
+		|| fState == unix_stream_endpoint_state::Listening)
+		RETURN_ERROR(B_BAD_VALUE);
+
+	if (IsBound())
+		RETURN_ERROR(UnixEndpoint::_Unbind());
+
+	RETURN_ERROR(B_OK);
+}
+
+
+void
+UnixStreamEndpoint::_UnsetReceiveFifo()
+{
+	if (fReceiveFifo) {
+		fReceiveFifo->ReleaseReference();
+		fReceiveFifo = NULL;
+	}
+}
+
+
+void
+UnixStreamEndpoint::_StopListening()
+{
+	if (fState == unix_stream_endpoint_state::Listening) {
+		delete_sem(fAcceptSemaphore);
+		fAcceptSemaphore = -1;
+		fState = unix_stream_endpoint_state::NotConnected;
+	}
+}

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2008, Ingo Weinhold, [email protected].
+ * Distributed under the terms of the MIT License.
+ */
+#ifndef UNIX_STREAM_ENDPOINT_H
+#define UNIX_STREAM_ENDPOINT_H
+
+#include <sys/stat.h>
+
+#include <Referenceable.h>
+
+#include <util/DoublyLinkedList.h>
+#include <util/OpenHashTable.h>
+
+#include "unix.h"
+#include "UnixEndpoint.h"
+
+class UnixStreamEndpoint;
+class UnixFifo;
+
+
+enum class unix_stream_endpoint_state {
+	NotConnected,
+	Listening,
+	Connected,
+	Closed
+};
+
+
+typedef AutoLocker<UnixStreamEndpoint> UnixStreamEndpointLocker;
+
+
+class UnixStreamEndpoint : public UnixEndpoint, public BReferenceable {
+public:
+								UnixStreamEndpoint(net_socket* socket);
+	virtual						~UnixStreamEndpoint() override;
+
+			status_t			Init() override;
+			void				Uninit() override;
+
+			status_t			Open() override;
+			status_t			Close() override;
+			status_t			Free() override;
+
+			status_t			Bind(const struct sockaddr* _address) override;
+			status_t			Unbind() override;
+			status_t			Listen(int backlog) override;
+			status_t			Connect(const struct sockaddr* address) override;
+			status_t			Accept(net_socket** _acceptedSocket) override;
+
+			ssize_t				Send(const iovec* vecs, size_t vecCount,
+									ancillary_data_container* ancillaryData,
+									const struct sockaddr* address,
+									socklen_t addressLength) override;
+			ssize_t				Receive(const iovec* vecs, size_t vecCount,
+									ancillary_data_container** _ancillaryData,
+									struct sockaddr* _address,
+									socklen_t* _addressLength) override;
+
+			ssize_t				Sendable() override;
+			ssize_t				Receivable() override;
+
+			status_t			SetReceiveBufferSize(size_t size) override;
+			status_t			GetPeerCredentials(ucred* credentials) override;
+
+			status_t			Shutdown(int direction) override;
+
+	bool IsBound() const
+	{
+		return !fIsChild && fAddress.IsValid();
+	}
+
+private:
+			void				_Spawn(UnixStreamEndpoint* connectingEndpoint,
+									UnixStreamEndpoint* listeningEndpoint, UnixFifo* fifo);
+			void				_Disconnect();
+			status_t			_LockConnectedEndpoints(UnixStreamEndpointLocker& locker,
+									UnixStreamEndpointLocker& peerLocker);
+
+			status_t			_Unbind();
+
+			void				_UnsetReceiveFifo();
+			void				_StopListening();
+
+private:
+	UnixStreamEndpoint*			fPeerEndpoint;
+	UnixFifo*					fReceiveFifo;
+	unix_stream_endpoint_state	fState;
+	sem_id						fAcceptSemaphore;
+	ucred						fCredentials;
+	bool						fIsChild;
+	bool						fWasConnected;
+};
+
+#endif	// UNIX_STREAM_ENDPOINT_H

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

@@ -21,6 +21,7 @@
 #include <net_socket.h>
 #include <net_stack.h>
 
+#include "unix.h"
 #include "UnixAddressManager.h"
 #include "UnixEndpoint.h"
 
@@ -67,11 +68,12 @@ unix_init_protocol(net_socket *socket)
 	TRACE("[%" B_PRId32 "] unix_init_protocol(%p)\n", find_thread(NULL),
 		socket);
 
-	UnixEndpoint* endpoint = new(std::nothrow) UnixEndpoint(socket);
+	UnixEndpoint* endpoint;
-	if (endpoint == NULL)
+	status_t error = UnixEndpoint::Create(socket, &endpoint);
+	if (error != B_OK)
 		return NULL;
 
-	status_t error = endpoint->Init();
+	error = endpoint->Init();
 	if (error != B_OK) {
 		delete endpoint;
 		return NULL;
@@ -408,7 +410,8 @@ 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);
+	return ((UnixEndpoint*)_protocol)->Send(vecs, vecCount, ancillaryData,
+		address, addressLength);
 }
 
 
@@ -435,6 +438,11 @@ init_unix()
 
 	error = gStackModule->register_domain_protocols(AF_UNIX, SOCK_STREAM, 0,
 		"network/protocols/unix/v1", NULL);
+	if (error == B_OK) {
+		error = gStackModule->register_domain_protocols(AF_UNIX, SOCK_DGRAM, 0,
+			"network/protocols/unix/v1", NULL);
+	}
+
 	if (error != B_OK) {
 		gAddressManager.~UnixAddressManager();
 		return error;

src/tests/system/network/Jamfile

@@ -17,6 +17,8 @@ SimpleTest getpeername : getpeername.cpp : $(TARGET_NETWORK_LIBS) ;
 
 SimpleTest if_nameindex : if_nameindex.c : $(TARGET_NETWORK_LIBS) ;
 
+SimpleTest unix_dgram_test : unix_dgram_test.cpp : $(TARGET_NETWORK_LIBS) ;
+
 SimpleTest tcp_connection_test : tcp_connection_test.cpp
 	: $(TARGET_NETWORK_LIBS) ;
 

src/tests/system/network/unix_dgram_test.cpp

@@ -0,0 +1,620 @@
+/*
+ * Copyright 2023, Trung Nguyen, trungnt282910@gmail.com.
+ * Distributed under the terms of the MIT License.
+ */
+
+
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <fcntl.h>
+#include <sys/mman.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+
+#define REPORT_ERROR(msg, ...) \
+	fprintf(stderr, "%s:%d: " msg "\n", __FILE__, __LINE__, ##__VA_ARGS__)
+
+
+int
+connect_test()
+{
+	unlink("test.sock");
+	unlink("test1.sock");
+	unlink("test2.sock");
+
+	int status;
+
+	int sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr;
+	addr.sun_family = AF_UNIX;
+	strcpy(addr.sun_path, "test.sock");
+	status = bind(sock, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int sock1 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock1 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr1;
+	addr1.sun_family = AF_UNIX;
+	strcpy(addr1.sun_path, "test1.sock");
+	status = bind(sock1, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Set non-blocking on both sockets
+	int flags1 = fcntl(sock, F_GETFL, 0);
+	if (flags1 == -1) {
+		REPORT_ERROR("fcntl() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	status = fcntl(sock, F_SETFL, flags1 | O_NONBLOCK);
+	if (status == -1) {
+		REPORT_ERROR("fcntl() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	status = fcntl(sock1, F_SETFL, flags1 | O_NONBLOCK);
+	if (status == -1) {
+		REPORT_ERROR("fcntl() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = connect(sock, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Connect in the opposite way
+	status = connect(sock1, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Reconnect a connected DGRAM socket
+	status = connect(sock, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int sock2 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock2 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr2;
+	addr2.sun_family = AF_UNIX;
+	strcpy(addr2.sun_path, "test2.sock");
+	status = bind(sock2, (struct sockaddr*)&addr2, sizeof(addr2));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Connect to a socket that are already connected
+	status = connect(sock2, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status != -1) {
+		REPORT_ERROR("connect() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != EPERM) {
+		REPORT_ERROR("connect() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = close(sock2);
+	if (status == -1) {
+		REPORT_ERROR("close() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Connect to a closed socket
+	status = connect(sock, (struct sockaddr*)&addr2, sizeof(addr2));
+	if (status != -1) {
+		REPORT_ERROR("connect() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != ECONNREFUSED) {
+		REPORT_ERROR("connect() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	close(sock);
+	close(sock1);
+
+	unlink("test.sock");
+	unlink("test1.sock");
+	unlink("test2.sock");
+
+	return 0;
+}
+
+
+int
+send_test()
+{
+	unlink("test.sock");
+	unlink("test1.sock");
+	unlink("test2.sock");
+
+	int status;
+
+	int sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr;
+	memset(&addr, 0, sizeof(addr));
+	addr.sun_family = AF_UNIX;
+	strcpy(addr.sun_path, "test.sock");
+	status = bind(sock, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = send(sock, "test", 4, 0);
+	if (status != -1) {
+		REPORT_ERROR("send() succeeded unexpectedly\n");
+		return 1;
+	}
+	// if (errno != ENOTCONN) {
+	// 	REPORT_ERROR("send() failed with unexpected error: %s\n", strerror(errno));
+	// 	return 1;
+	// }
+
+	int sock1 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock1 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr1;
+	addr1.sun_family = AF_UNIX;
+	strcpy(addr1.sun_path, "test1.sock");
+	status = bind(sock1, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Set non-blocking on both sockets
+	status = fcntl(sock, F_SETFL, O_NONBLOCK);
+	if (status == -1) {
+		REPORT_ERROR("fcntl() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	status = fcntl(sock1, F_SETFL, O_NONBLOCK);
+	if (status == -1) {
+		REPORT_ERROR("fcntl() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = sendto(sock, "test1", 5, 0, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("sendto() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = connect(sock, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	status = connect(sock1, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = send(sock, "test2", 5, 0);
+	if (status == -1) {
+		REPORT_ERROR("send() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int sock2 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock2 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr2;
+	addr2.sun_family = AF_UNIX;
+	strcpy(addr2.sun_path, "test2.sock");
+	status = bind(sock2, (struct sockaddr*)&addr2, sizeof(addr2));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = sendto(sock2, "test3", 5, 0, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status != -1) {
+		REPORT_ERROR("sendto() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != EPERM) {
+		REPORT_ERROR("sendto() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	char buf[16];
+	memset(buf, 0, sizeof(buf));
+	status = recv(sock1, buf, sizeof(buf), 0);
+	if (status == -1) {
+		REPORT_ERROR("recv() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	if (strcmp(buf, "test1") != 0) {
+		REPORT_ERROR("recv() received unexpected data: %s\n", buf);
+		return 1;
+	}
+
+	memset(buf, 0, sizeof(buf));
+	struct sockaddr_un addr3;
+	memset(&addr3, 0, sizeof(addr3));
+	socklen_t addrlen = sizeof(addr3);
+	status = recvfrom(sock1, buf, sizeof(buf), 0, (struct sockaddr*)&addr3, &addrlen);
+	if (status == -1) {
+		REPORT_ERROR("recv() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	if (strcmp(buf, "test2") != 0) {
+		REPORT_ERROR("recv() received unexpected data: %s\n", buf);
+		return 1;
+	}
+	if (strcmp(addr.sun_path, addr3.sun_path) != 0) {
+		REPORT_ERROR("recv() received unexpected address: %s\n", addr3.sun_path);
+		return 1;
+	}
+
+	status = send(sock, "test4", 4, 0);
+	if (status == -1) {
+		REPORT_ERROR("send() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = send(sock, "test5", 5, 0);
+	if (status == -1) {
+		REPORT_ERROR("send() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	memset(buf, 0, sizeof(buf));
+	status = recv(sock1, buf, 4, 0);
+	if (status == -1) {
+		REPORT_ERROR("recv() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	if (strcmp(buf, "test") != 0) {
+		REPORT_ERROR("recv() received unexpected data: %s\n", buf);
+		return 1;
+	}
+
+	// The last byte of the previous datagram should be discarded.
+	memset(buf, 0, sizeof(buf));
+	status = recv(sock1, buf, sizeof(buf), 0);
+	if (status == -1) {
+		REPORT_ERROR("recv() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	if (strcmp(buf, "test5") != 0) {
+		REPORT_ERROR("recv() received unexpected data: %s\n", buf);
+		return 1;
+	}
+
+	close(sock1);
+	status = send(sock, "test6", 5, 0);
+	if (status != -1) {
+		REPORT_ERROR("send() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != ECONNREFUSED) {
+		REPORT_ERROR("send() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	close(sock);
+	close(sock2);
+
+	unlink("test.sock");
+	unlink("test1.sock");
+	unlink("test2.sock");
+
+	return 0;
+}
+
+
+int
+shutdown_test()
+{
+	unlink("test.sock");
+	unlink("test1.sock");
+
+	int status;
+
+	int sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr;
+	addr.sun_family = AF_UNIX;
+	strcpy(addr.sun_path, "test.sock");
+	status = bind(sock, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int sock1 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock1 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr1;
+	addr1.sun_family = AF_UNIX;
+	strcpy(addr1.sun_path, "test1.sock");
+	status = bind(sock1, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = shutdown(sock, SHUT_WR);
+	if (status == -1) {
+		REPORT_ERROR("shutdown() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = sendto(sock, "test", 4, 0, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status != -1) {
+		REPORT_ERROR("send() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != EPIPE) {
+		REPORT_ERROR("send() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = sendto(sock1, "test", 4, 0, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("send() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = shutdown(sock, SHUT_RD);
+	if (status == -1) {
+		REPORT_ERROR("shutdown() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = sendto(sock1, "test", 4, 0, (struct sockaddr*)&addr, sizeof(addr));
+	if (status != -1) {
+		REPORT_ERROR("send() succeeded unexpectedly\n");
+		return 1;
+	}
+	if (errno != EPIPE) {
+		REPORT_ERROR("send() failed with unexpected error: %s\n", strerror(errno));
+		return 1;
+	}
+
+	char buf[16];
+	memset(buf, 0, sizeof(buf));
+	status = recv(sock, buf, sizeof(buf), 0);
+	if (status == -1) {
+		REPORT_ERROR("recv() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	if (status != 0) {
+		REPORT_ERROR("recv() received unexpected data\n");
+		return 1;
+	}
+
+	close(sock);
+	close(sock1);
+
+	unlink("test.sock");
+	unlink("test1.sock");
+
+	return 0;
+}
+
+
+int
+send_fd_test()
+{
+	unlink("test.sock");
+	unlink("test1.sock");
+
+	int status;
+
+	int sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr;
+	addr.sun_family = AF_UNIX;
+	strcpy(addr.sun_path, "test.sock");
+	status = bind(sock, (struct sockaddr*)&addr, sizeof(addr));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int sock1 = socket(AF_UNIX, SOCK_DGRAM, 0);
+	if (sock1 == -1) {
+		REPORT_ERROR("socket() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct sockaddr_un addr1;
+	addr1.sun_family = AF_UNIX;
+	strcpy(addr1.sun_path, "test1.sock");
+	status = bind(sock1, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("bind() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	status = connect(sock, (struct sockaddr*)&addr1, sizeof(addr1));
+	if (status == -1) {
+		REPORT_ERROR("connect() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	int fd = shm_open("test_shm", O_CREAT | O_RDWR, 0666);
+	if (fd == -1) {
+		REPORT_ERROR("shm_open() failed: %s\n", strerror(errno));
+		return 1;
+	}
+	shm_unlink("test_shm");
+
+	// Send FD
+	char iobuf[] = "test";
+	struct iovec iov {
+		.iov_base = iobuf,
+		.iov_len = sizeof(iobuf),
+	};
+
+	struct msghdr msg;
+	memset(&msg, 0, sizeof(msg));
+
+	struct cmsghdr *cmsg;
+	char buf[CMSG_SPACE(sizeof(fd))];
+	memset(buf, 0, sizeof(buf));
+	msg.msg_control = buf;
+	msg.msg_controllen = sizeof(buf);
+	msg.msg_iov = &iov;
+	msg.msg_iovlen = 1;
+
+	cmsg = CMSG_FIRSTHDR(&msg);
+	cmsg->cmsg_level = SOL_SOCKET;
+	cmsg->cmsg_type = SCM_RIGHTS;
+	cmsg->cmsg_len = CMSG_LEN(sizeof(fd));
+	memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
+	msg.msg_controllen = cmsg->cmsg_len;
+
+	status = sendmsg(sock, &msg, 0);
+	if (status == -1) {
+		REPORT_ERROR("sendmsg() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	// Receive FD
+	memset(buf, 0, sizeof(buf));
+	msg.msg_control = buf;
+	msg.msg_controllen = sizeof(buf);
+	msg.msg_iov = &iov;
+	msg.msg_iovlen = 1;
+
+	status = recvmsg(sock1, &msg, 0);
+	if (status == -1) {
+		REPORT_ERROR("recvmsg() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	cmsg = CMSG_FIRSTHDR(&msg);
+	if (cmsg == NULL) {
+		REPORT_ERROR("recvmsg() failed: no control message\n");
+		return 1;
+	}
+	if (cmsg->cmsg_level != SOL_SOCKET) {
+		REPORT_ERROR("recvmsg() failed: unexpected level %d\n", cmsg->cmsg_level);
+		return 1;
+	}
+	if (cmsg->cmsg_type != SCM_RIGHTS) {
+		REPORT_ERROR("recvmsg() failed: unexpected type %d\n", cmsg->cmsg_type);
+		return 1;
+	}
+	if (cmsg->cmsg_len != CMSG_LEN(sizeof(fd))) {
+		REPORT_ERROR("recvmsg() failed: unexpected length %ld\n", cmsg->cmsg_len);
+		return 1;
+	}
+
+	int fd1;
+	memcpy(&fd1, CMSG_DATA(cmsg), sizeof(fd1));
+	if (fd1 == -1) {
+		REPORT_ERROR("recvmsg() failed: unexpected fd %d\n", fd1);
+		return 1;
+	}
+
+	// Check that the FD refers to the same file
+	struct stat statbuf;
+	status = fstat(fd, &statbuf);
+	if (status == -1) {
+		REPORT_ERROR("fstat() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	struct stat statbuf1;
+	status = fstat(fd1, &statbuf1);
+	if (status == -1) {
+		REPORT_ERROR("fstat() failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	if (statbuf.st_dev != statbuf1.st_dev) {
+		REPORT_ERROR("recvmsg() failed: unexpected device %ld\n", (long)statbuf1.st_dev);
+		return 1;
+	}
+	if (statbuf.st_ino != statbuf1.st_ino) {
+		REPORT_ERROR("recvmsg() failed: unexpected inode %ld\n", (long)statbuf1.st_ino);
+		return 1;
+	}
+
+	close(sock);
+	close(sock1);
+	close(fd);
+	close(fd1);
+
+	unlink("test.sock");
+	unlink("test1.sock");
+
+	return 0;
+}
+
+
+int
+main()
+{
+	if (connect_test() != 0)
+		return 1;
+
+	if (send_test() != 0)
+		return 1;
+
+	if (shutdown_test() != 0)
+		return 1;
+
+	if (send_fd_test() != 0)
+		return 1;
+
+	return 0;
+}