From b7b57869e8a1241bcc52c8850510105385317c2b Mon Sep 17 00:00:00 2001 From: Trung Nguyen Date: Sun, 18 Jun 2023 21:00:57 +1000 Subject: [PATCH] unix: Implement datagram sockets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Tested-by: Commit checker robot --- .../kernel/network/protocols/unix/Jamfile | 3 + .../protocols/unix/UnixDatagramEndpoint.cpp | 560 ++++++++++++ .../protocols/unix/UnixDatagramEndpoint.h | 72 ++ .../network/protocols/unix/UnixEndpoint.cpp | 804 +----------------- .../network/protocols/unix/UnixEndpoint.h | 144 ++-- .../network/protocols/unix/UnixFifo.cpp | 106 ++- .../kernel/network/protocols/unix/UnixFifo.h | 47 +- .../protocols/unix/UnixStreamEndpoint.cpp | 801 +++++++++++++++++ .../protocols/unix/UnixStreamEndpoint.h | 95 +++ .../kernel/network/protocols/unix/unix.cpp | 16 +- src/tests/system/network/Jamfile | 2 + src/tests/system/network/unix_dgram_test.cpp | 620 ++++++++++++++ 12 files changed, 2393 insertions(+), 877 deletions(-) create mode 100644 src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.cpp create mode 100644 src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.h create mode 100644 src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.cpp create mode 100644 src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.h create mode 100644 src/tests/system/network/unix_dgram_test.cpp diff --git a/src/add-ons/kernel/network/protocols/unix/Jamfile b/src/add-ons/kernel/network/protocols/unix/Jamfile index 57ba6bf778..753a4f815a 100644 --- a/src/add-ons/kernel/network/protocols/unix/Jamfile +++ b/src/add-ons/kernel/network/protocols/unix/Jamfile @@ -9,4 +9,7 @@ KernelAddon unix : UnixAddress.cpp UnixEndpoint.cpp UnixFifo.cpp + + UnixDatagramEndpoint.cpp + UnixStreamEndpoint.cpp ; diff --git a/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.cpp b/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.cpp new file mode 100644 index 0000000000..30a3d08c96 --- /dev/null +++ b/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.cpp @@ -0,0 +1,560 @@ +/* + * Copyright 2023, Trung Nguyen, trungnt282910@gmail.com. + * Distributed under the terms of the MIT License. + */ + + +#include "UnixDatagramEndpoint.h" + +#include + +#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 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 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 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 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 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 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& 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(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; + } +} diff --git a/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.h b/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.h new file mode 100644 index 0000000000..1b142c864f --- /dev/null +++ b/src/add-ons/kernel/network/protocols/unix/UnixDatagramEndpoint.h @@ -0,0 +1,72 @@ +/* + * Copyright 2023, Trung Nguyen, trungnt282910@gmail.com. + * Distributed under the terms of the MIT License. + */ +#ifndef UNIX_DATAGRAM_ENDPOINT_H +#define UNIX_DATAGRAM_ENDPOINT_H + + +#include + +#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 &outEndpoint); + + status_t _Disconnect(); + void _UnsetReceiveFifo(); + +private: + UnixDatagramEndpoint* fTargetEndpoint; + UnixFifo* fReceiveFifo; + bool fShutdownWrite:1; + bool fShutdownRead:1; +}; + + +#endif // UNIX_DATAGRAM_ENDPOINT_H diff --git a/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.cpp b/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.cpp index 745a0197f6..0e207dd03d 100644 --- a/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.cpp +++ b/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.cpp @@ -1,41 +1,46 @@ /* - * Copyright 2008, Ingo Weinhold, ingo_weinhold@gmx.de. + * Copyright 2023, Trung Nguyen, trungnt282910@gmail.com. * Distributed under the terms of the MIT License. */ +#include + +#include + #include "UnixEndpoint.h" -#include -#include - -#include - -#include - #include "UnixAddressManager.h" -#include "UnixFifo.h" +#include "UnixDatagramEndpoint.h" +#include "UnixStreamEndpoint.h" -#define UNIX_ENDPOINT_DEBUG_LEVEL 0 -#define UNIX_DEBUG_LEVEL UNIX_ENDPOINT_DEBUG_LEVEL +#define UNIX_ENDPOINT_DEBUG_LEVEL 1 +#define UNIX_DEBUG_LEVEL UNIX_ENDPOINT_DEBUG_LEVEL #include "UnixDebug.h" -// Note on locking order (outermost -> innermost): -// UnixEndpoint: connecting -> listening -> child -// -> UnixFifo (never lock more than one at a time) -// -> UnixAddressManager - - -static inline bigtime_t -absolute_timeout(bigtime_t timeout) +status_t +UnixEndpoint::Create(net_socket* socket, UnixEndpoint** _endpoint) { - if (timeout == 0 || timeout == B_INFINITE_TIMEOUT) - return timeout; + TRACE("[%" B_PRId32 "] UnixEndpoint::Create(%p, %p)\n", find_thread(NULL), + socket, _endpoint); -// TODO: Make overflow safe! - return timeout + system_time(); + if (socket == NULL || _endpoint == NULL) + 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(), - fPeerEndpoint(NULL), - fReceiveFifo(NULL), - fState(UNIX_ENDPOINT_CLOSED), - fAcceptSemaphore(-1), - fIsChild(false), - fWasConnected(false) + fAddressHashLink() { 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), - 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", + TRACE("[%" B_PRId32 "] %p->UnixEndpoint::_Bind(\"%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_NOT_CONNECTED || IsBound()) - RETURN_ERROR(B_BAD_VALUE); + ConstSocketAddress(&gAddressModule, (struct sockaddr*)address).AsString().Data()); 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 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 fifoDeleter(fifo); - ObjectDeleter 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 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 _(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 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 _(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 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) - RETURN_ERROR(B_BAD_VALUE); - - if (IsBound()) { - UnixAddressManagerLocker addressLocker(gAddressManager); - gAddressManager.Remove(this); - if (struct vnode* vnode = fAddress.Vnode()) - vfs_put_vnode(vnode); - - fAddress.Unset(); - } + UnixAddressManagerLocker addressLocker(gAddressManager); + gAddressManager.Remove(this); + if (struct vnode* vnode = fAddress.Vnode()) + vfs_put_vnode(vnode); + 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; - } -} diff --git a/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.h b/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.h index 717ac7a97f..be816fc71e 100644 --- a/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.h +++ b/src/add-ons/kernel/network/protocols/unix/UnixEndpoint.h @@ -1,54 +1,24 @@ /* - * Copyright 2008, Ingo Weinhold, ingo_weinhold@gmx.de. + * Copyright 2023, Trung Nguyen, trungnt282910@gmail.com. * Distributed under the terms of the MIT License. */ #ifndef UNIX_ENDPOINT_H #define UNIX_ENDPOINT_H -#include - -#include - -#include -#include -#include -#include #include #include #include -#include "unix.h" +#include +#include + #include "UnixAddress.h" -class UnixEndpoint; -class UnixFifo; - - -enum unix_endpoint_state { - UNIX_ENDPOINT_NOT_CONNECTED, - UNIX_ENDPOINT_LISTENING, - UNIX_ENDPOINT_CONNECTED, - UNIX_ENDPOINT_CLOSED -}; - - -typedef AutoLocker UnixEndpointLocker; - - -class UnixEndpoint : public net_protocol, public ProtocolSocket, - public BReferenceable { +class UnixEndpoint : public net_protocol, public ProtocolSocket { public: - UnixEndpoint(net_socket* socket); - virtual ~UnixEndpoint(); - - status_t Init(); - void Uninit(); - - status_t Open(); - status_t Close(); - status_t Free(); + virtual ~UnixEndpoint(); 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: - void _Spawn(UnixEndpoint* connectingEndpoint, - UnixEndpoint* listeningEndpoint, UnixFifo* fifo); - void _Disconnect(); - status_t _LockConnectedEndpoints(UnixEndpointLocker& locker, - UnixEndpointLocker& peerLocker); + virtual status_t Init() = 0; + virtual void Uninit() = 0; - status_t _Bind(struct vnode* vnode); - status_t _Bind(int32 internalID); - status_t _Unbind(); + virtual status_t Open() = 0; + virtual status_t Close() = 0; + virtual status_t Free() = 0; - void _UnsetReceiveFifo(); - void _StopListening(); + virtual status_t Bind(const struct sockaddr* _address) = 0; + 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; - UnixAddress fAddress; - UnixEndpoint* fAddressHashLink; - UnixEndpoint* fPeerEndpoint; - UnixFifo* fReceiveFifo; - unix_endpoint_state fState; - sem_id fAcceptSemaphore; - ucred fCredentials; - bool fIsChild; - bool fWasConnected; + status_t _Bind(struct vnode* vnode); + status_t _Bind(int32 internalID); + +protected: + UnixAddress fAddress; + +private: + mutex fLock; + UnixEndpoint* fAddressHashLink; }; + +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 diff --git a/src/add-ons/kernel/network/protocols/unix/UnixFifo.cpp b/src/add-ons/kernel/network/protocols/unix/UnixFifo.cpp index d07db9e091..6ceb9af648 100644 --- a/src/add-ons/kernel/network/protocols/unix/UnixFifo.cpp +++ b/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 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 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() - && !IsReadShutdown()) { + while (error == B_OK && fBuffer.Writable() < _MinimumWritableSize(request) + && !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; + } +} diff --git a/src/add-ons/kernel/network/protocols/unix/UnixFifo.h b/src/add-ons/kernel/network/protocols/unix/UnixFifo.h index 80935ef87c..89345e9f22 100644 --- a/src/add-ons/kernel/network/protocols/unix/UnixFifo.h +++ b/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 { 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; - size_t fVecCount; - ancillary_data_container* fAncillaryData; - off_t fTotalSize; - off_t fBytesTransferred; - size_t fVecIndex; - size_t fVecOffset; + const iovec* fVecs; + size_t fVecCount; + ancillary_data_container* fAncillaryData; + off_t fTotalSize; + off_t fBytesTransferred; + size_t fVecIndex; + 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 AncillaryDataList; + struct DatagramEntry : DoublyLinkedListLinkImpl { + struct sockaddr_storage address; + size_t size; + }; + + typedef DoublyLinkedList 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; }; diff --git a/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.cpp b/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.cpp new file mode 100644 index 0000000000..6a2302bbfe --- /dev/null +++ b/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.cpp @@ -0,0 +1,801 @@ +/* + * Copyright 2008, Ingo Weinhold, ingo_weinhold@gmx.de. + * Distributed under the terms of the MIT License. + */ + + +#include "UnixStreamEndpoint.h" + +#include +#include + +#include + +#include + +#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(listeningUnixEndpoint); + if (listeningEndpoint == NULL) + RETURN_ERROR(EPROTOTYPE); + BReference 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 fifoDeleter(fifo); + ObjectDeleter 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 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 _(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 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 _(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 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; + } +} diff --git a/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.h b/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.h new file mode 100644 index 0000000000..03a4421f7e --- /dev/null +++ b/src/add-ons/kernel/network/protocols/unix/UnixStreamEndpoint.h @@ -0,0 +1,95 @@ +/* + * Copyright 2008, Ingo Weinhold, ingo_weinhold@gmx.de. + * Distributed under the terms of the MIT License. + */ +#ifndef UNIX_STREAM_ENDPOINT_H +#define UNIX_STREAM_ENDPOINT_H + +#include + +#include + +#include +#include + +#include "unix.h" +#include "UnixEndpoint.h" + +class UnixStreamEndpoint; +class UnixFifo; + + +enum class unix_stream_endpoint_state { + NotConnected, + Listening, + Connected, + Closed +}; + + +typedef AutoLocker 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 diff --git a/src/add-ons/kernel/network/protocols/unix/unix.cpp b/src/add-ons/kernel/network/protocols/unix/unix.cpp index 2198fc76f2..83fd98383b 100644 --- a/src/add-ons/kernel/network/protocols/unix/unix.cpp +++ b/src/add-ons/kernel/network/protocols/unix/unix.cpp @@ -21,6 +21,7 @@ #include #include +#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); - if (endpoint == NULL) + UnixEndpoint* endpoint; + 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; diff --git a/src/tests/system/network/Jamfile b/src/tests/system/network/Jamfile index 2399a3f3f5..04693d6545 100644 --- a/src/tests/system/network/Jamfile +++ b/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) ; diff --git a/src/tests/system/network/unix_dgram_test.cpp b/src/tests/system/network/unix_dgram_test.cpp new file mode 100644 index 0000000000..e5b0c8606c --- /dev/null +++ b/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 +#include +#include + +#include +#include +#include +#include +#include +#include + + +#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; +}