gecko/ipc/unixsocket/UnixSocket.cpp

922 lines
22 KiB
C++

/* -*- Mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; tab-width: 40 -*- */
/* vim: set ts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "UnixSocket.h"
#include "nsTArray.h"
#include "nsXULAppAPI.h"
#include <fcntl.h>
#ifdef MOZ_TASK_TRACER
#include "GeckoTaskTracer.h"
using namespace mozilla::tasktracer;
#endif
static const size_t MAX_READ_SIZE = 1 << 16;
namespace mozilla {
namespace ipc {
class UnixSocketImpl : public UnixSocketWatcher
{
public:
UnixSocketImpl(MessageLoop* mIOLoop,
UnixSocketConsumer* aConsumer, UnixSocketConnector* aConnector,
const nsACString& aAddress)
: UnixSocketWatcher(mIOLoop)
, mConsumer(aConsumer)
, mConnector(aConnector)
, mShuttingDownOnIOThread(false)
, mAddress(aAddress)
, mDelayedConnectTask(nullptr)
{
}
~UnixSocketImpl()
{
MOZ_ASSERT(NS_IsMainThread());
MOZ_ASSERT(IsShutdownOnMainThread());
}
void QueueWriteData(UnixSocketRawData* aData)
{
mOutgoingQ.AppendElement(aData);
AddWatchers(WRITE_WATCHER, false);
}
bool IsShutdownOnMainThread()
{
MOZ_ASSERT(NS_IsMainThread());
return mConsumer == nullptr;
}
void ShutdownOnMainThread()
{
MOZ_ASSERT(NS_IsMainThread());
MOZ_ASSERT(!IsShutdownOnMainThread());
mConsumer = nullptr;
}
bool IsShutdownOnIOThread()
{
return mShuttingDownOnIOThread;
}
void ShutdownOnIOThread()
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(!mShuttingDownOnIOThread);
Close(); // will also remove fd from I/O loop
mShuttingDownOnIOThread = true;
}
void SetDelayedConnectTask(CancelableTask* aTask)
{
MOZ_ASSERT(NS_IsMainThread());
mDelayedConnectTask = aTask;
}
void ClearDelayedConnectTask()
{
MOZ_ASSERT(NS_IsMainThread());
mDelayedConnectTask = nullptr;
}
void CancelDelayedConnectTask()
{
MOZ_ASSERT(NS_IsMainThread());
if (!mDelayedConnectTask) {
return;
}
mDelayedConnectTask->Cancel();
ClearDelayedConnectTask();
}
/**
* Connect to a socket
*/
void Connect();
/**
* Run bind/listen to prepare for further runs of accept()
*/
void Listen();
void GetSocketAddr(nsAString& aAddrStr)
{
if (!mConnector) {
NS_WARNING("No connector to get socket address from!");
aAddrStr.Truncate();
return;
}
mConnector->GetSocketAddr(mAddr, aAddrStr);
}
/**
* Consumer pointer. Non-thread safe RefPtr, so should only be manipulated
* directly from main thread. All non-main-thread accesses should happen with
* mImpl as container.
*/
RefPtr<UnixSocketConsumer> mConsumer;
void OnAccepted(int aFd, const sockaddr_any* aAddr,
socklen_t aAddrLen) MOZ_OVERRIDE;
void OnConnected() MOZ_OVERRIDE;
void OnError(const char* aFunction, int aErrno) MOZ_OVERRIDE;
void OnListening() MOZ_OVERRIDE;
void OnSocketCanReceiveWithoutBlocking() MOZ_OVERRIDE;
void OnSocketCanSendWithoutBlocking() MOZ_OVERRIDE;
private:
// Set up flags on whatever our current file descriptor is.
static bool SetSocketFlags(int aFd);
void FireSocketError();
/**
* Raw data queue. Must be pushed/popped from IO thread only.
*/
typedef nsTArray<UnixSocketRawData* > UnixSocketRawDataQueue;
UnixSocketRawDataQueue mOutgoingQ;
/**
* Connector object used to create the connection we are currently using.
*/
nsAutoPtr<UnixSocketConnector> mConnector;
/**
* If true, do not requeue whatever task we're running
*/
bool mShuttingDownOnIOThread;
/**
* Address we are connecting to, assuming we are creating a client connection.
*/
nsCString mAddress;
/**
* Size of the socket address struct
*/
socklen_t mAddrSize;
/**
* Address struct of the socket currently in use
*/
sockaddr_any mAddr;
/**
* Task member for delayed connect task. Should only be access on main thread.
*/
CancelableTask* mDelayedConnectTask;
};
template<class T>
class DeleteInstanceRunnable : public nsRunnable
{
public:
DeleteInstanceRunnable(T* aInstance)
: mInstance(aInstance)
{ }
NS_IMETHOD Run()
{
delete mInstance;
return NS_OK;
}
private:
T* mInstance;
};
class UnixSocketImplRunnable : public nsRunnable
{
public:
UnixSocketImpl* GetImpl() const
{
return mImpl;
}
protected:
UnixSocketImplRunnable(UnixSocketImpl* aImpl)
: mImpl(aImpl)
{
MOZ_ASSERT(aImpl);
}
virtual ~UnixSocketImplRunnable()
{ }
private:
UnixSocketImpl* mImpl;
};
class OnSocketEventRunnable : public UnixSocketImplRunnable
{
public:
enum SocketEvent {
CONNECT_SUCCESS,
CONNECT_ERROR,
DISCONNECT
};
OnSocketEventRunnable(UnixSocketImpl* aImpl, SocketEvent e)
: UnixSocketImplRunnable(aImpl)
, mEvent(e)
{
MOZ_ASSERT(!NS_IsMainThread());
}
NS_IMETHOD Run() MOZ_OVERRIDE
{
MOZ_ASSERT(NS_IsMainThread());
UnixSocketImpl* impl = GetImpl();
if (impl->IsShutdownOnMainThread()) {
NS_WARNING("CloseSocket has already been called!");
// Since we've already explicitly closed and the close happened before
// this, this isn't really an error. Since we've warned, return OK.
return NS_OK;
}
if (mEvent == CONNECT_SUCCESS) {
impl->mConsumer->NotifySuccess();
} else if (mEvent == CONNECT_ERROR) {
impl->mConsumer->NotifyError();
} else if (mEvent == DISCONNECT) {
impl->mConsumer->NotifyDisconnect();
}
return NS_OK;
}
private:
SocketEvent mEvent;
};
class SocketReceiveRunnable : public UnixSocketImplRunnable
{
public:
SocketReceiveRunnable(UnixSocketImpl* aImpl, UnixSocketRawData* aData)
: UnixSocketImplRunnable(aImpl)
, mRawData(aData)
{
MOZ_ASSERT(aData);
}
NS_IMETHOD Run() MOZ_OVERRIDE
{
MOZ_ASSERT(NS_IsMainThread());
UnixSocketImpl* impl = GetImpl();
if (impl->IsShutdownOnMainThread()) {
NS_WARNING("mConsumer is null, aborting receive!");
// Since we've already explicitly closed and the close happened before
// this, this isn't really an error. Since we've warned, return OK.
return NS_OK;
}
MOZ_ASSERT(impl->mConsumer);
impl->mConsumer->ReceiveSocketData(mRawData);
return NS_OK;
}
private:
nsAutoPtr<UnixSocketRawData> mRawData;
};
class RequestClosingSocketRunnable : public UnixSocketImplRunnable
{
public:
RequestClosingSocketRunnable(UnixSocketImpl* aImpl)
: UnixSocketImplRunnable(aImpl)
{ }
NS_IMETHOD Run() MOZ_OVERRIDE
{
MOZ_ASSERT(NS_IsMainThread());
UnixSocketImpl* impl = GetImpl();
if (impl->IsShutdownOnMainThread()) {
NS_WARNING("CloseSocket has already been called!");
// Since we've already explicitly closed and the close happened before
// this, this isn't really an error. Since we've warned, return OK.
return NS_OK;
}
// Start from here, same handling flow as calling CloseSocket() from
// upper layer
impl->mConsumer->CloseSocket();
return NS_OK;
}
};
class UnixSocketImplTask : public CancelableTask
{
public:
UnixSocketImpl* GetImpl() const
{
return mImpl;
}
void Cancel() MOZ_OVERRIDE
{
mImpl = nullptr;
}
bool IsCanceled() const
{
return !mImpl;
}
protected:
UnixSocketImplTask(UnixSocketImpl* aImpl)
: mImpl(aImpl)
{
MOZ_ASSERT(mImpl);
}
private:
UnixSocketImpl* mImpl;
};
class SocketSendTask : public UnixSocketImplTask
{
public:
SocketSendTask(UnixSocketImpl* aImpl,
UnixSocketConsumer* aConsumer,
UnixSocketRawData* aData)
: UnixSocketImplTask(aImpl)
, mConsumer(aConsumer)
, mData(aData)
{
MOZ_ASSERT(aConsumer);
MOZ_ASSERT(aData);
}
void Run() MOZ_OVERRIDE
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(!IsCanceled());
UnixSocketImpl* impl = GetImpl();
MOZ_ASSERT(!impl->IsShutdownOnIOThread());
impl->QueueWriteData(mData);
}
private:
nsRefPtr<UnixSocketConsumer> mConsumer;
UnixSocketRawData* mData;
};
class SocketListenTask : public UnixSocketImplTask
{
public:
SocketListenTask(UnixSocketImpl* aImpl)
: UnixSocketImplTask(aImpl)
{ }
void Run() MOZ_OVERRIDE
{
MOZ_ASSERT(!NS_IsMainThread());
if (!IsCanceled()) {
GetImpl()->Listen();
}
}
};
class SocketConnectTask : public UnixSocketImplTask
{
public:
SocketConnectTask(UnixSocketImpl* aImpl)
: UnixSocketImplTask(aImpl)
{ }
void Run() MOZ_OVERRIDE
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(!IsCanceled());
GetImpl()->Connect();
}
};
class SocketDelayedConnectTask : public UnixSocketImplTask
{
public:
SocketDelayedConnectTask(UnixSocketImpl* aImpl)
: UnixSocketImplTask(aImpl)
{ }
void Run() MOZ_OVERRIDE
{
MOZ_ASSERT(NS_IsMainThread());
if (IsCanceled()) {
return;
}
UnixSocketImpl* impl = GetImpl();
if (impl->IsShutdownOnMainThread()) {
return;
}
impl->ClearDelayedConnectTask();
XRE_GetIOMessageLoop()->PostTask(FROM_HERE, new SocketConnectTask(impl));
}
};
class ShutdownSocketTask : public UnixSocketImplTask
{
public:
ShutdownSocketTask(UnixSocketImpl* aImpl)
: UnixSocketImplTask(aImpl)
{ }
void Run() MOZ_OVERRIDE
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(!IsCanceled());
UnixSocketImpl* impl = GetImpl();
// At this point, there should be no new events on the IO thread after this
// one with the possible exception of a SocketListenTask that
// ShutdownOnIOThread will cancel for us. We are now fully shut down, so we
// can send a message to the main thread that will delete impl safely knowing
// that no more tasks reference it.
impl->ShutdownOnIOThread();
nsRefPtr<nsIRunnable> r(new DeleteInstanceRunnable<UnixSocketImpl>(impl));
nsresult rv = NS_DispatchToMainThread(r);
NS_ENSURE_SUCCESS_VOID(rv);
}
};
void
UnixSocketImpl::FireSocketError()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
// Clean up watchers, statuses, fds
Close();
// Tell the main thread we've errored
nsRefPtr<OnSocketEventRunnable> r =
new OnSocketEventRunnable(this, OnSocketEventRunnable::CONNECT_ERROR);
NS_DispatchToMainThread(r);
}
void
UnixSocketImpl::Listen()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(mConnector);
// This will set things we don't particularly care about, but it will hand
// back the correct structure size which is what we do care about.
if (!mConnector->CreateAddr(true, mAddrSize, mAddr, nullptr)) {
NS_WARNING("Cannot create socket address!");
FireSocketError();
return;
}
if (!IsOpen()) {
int fd = mConnector->Create();
if (fd < 0) {
NS_WARNING("Cannot create socket fd!");
FireSocketError();
return;
}
if (!SetSocketFlags(fd)) {
NS_WARNING("Cannot set socket flags!");
FireSocketError();
return;
}
SetFd(fd);
// calls OnListening on success, or OnError otherwise
nsresult rv = UnixSocketWatcher::Listen(
reinterpret_cast<struct sockaddr*>(&mAddr), mAddrSize);
NS_WARN_IF(NS_FAILED(rv));
}
}
void
UnixSocketImpl::Connect()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(mConnector);
if (!IsOpen()) {
int fd = mConnector->Create();
if (fd < 0) {
NS_WARNING("Cannot create socket fd!");
FireSocketError();
return;
}
if (!SetSocketFlags(fd)) {
NS_WARNING("Cannot set socket flags!");
FireSocketError();
return;
}
SetFd(fd);
}
if (!mConnector->CreateAddr(false, mAddrSize, mAddr, mAddress.get())) {
NS_WARNING("Cannot create socket address!");
FireSocketError();
return;
}
// calls OnConnected() on success, or OnError() otherwise
nsresult rv = UnixSocketWatcher::Connect(
reinterpret_cast<struct sockaddr*>(&mAddr), mAddrSize);
NS_WARN_IF(NS_FAILED(rv));
}
bool
UnixSocketImpl::SetSocketFlags(int aFd)
{
// Set socket addr to be reused even if kernel is still waiting to close
int n = 1;
if (setsockopt(aFd, SOL_SOCKET, SO_REUSEADDR, &n, sizeof(n)) < 0) {
return false;
}
// Set close-on-exec bit.
int flags = TEMP_FAILURE_RETRY(fcntl(aFd, F_GETFD));
if (-1 == flags) {
return false;
}
flags |= FD_CLOEXEC;
if (-1 == TEMP_FAILURE_RETRY(fcntl(aFd, F_SETFD, flags))) {
return false;
}
// Set non-blocking status flag.
flags = TEMP_FAILURE_RETRY(fcntl(aFd, F_GETFL));
if (-1 == flags) {
return false;
}
flags |= O_NONBLOCK;
if (-1 == TEMP_FAILURE_RETRY(fcntl(aFd, F_SETFL, flags))) {
return false;
}
return true;
}
void
UnixSocketImpl::OnAccepted(int aFd,
const sockaddr_any* aAddr,
socklen_t aAddrLen)
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_LISTENING);
MOZ_ASSERT(aAddr);
MOZ_ASSERT(aAddrLen <= sizeof(mAddr));
memcpy (&mAddr, aAddr, aAddrLen);
mAddrSize = aAddrLen;
if (!mConnector->SetUp(aFd)) {
NS_WARNING("Could not set up socket!");
return;
}
RemoveWatchers(READ_WATCHER|WRITE_WATCHER);
Close();
if (!SetSocketFlags(aFd)) {
return;
}
SetSocket(aFd, SOCKET_IS_CONNECTED);
nsRefPtr<OnSocketEventRunnable> r =
new OnSocketEventRunnable(this, OnSocketEventRunnable::CONNECT_SUCCESS);
NS_DispatchToMainThread(r);
AddWatchers(READ_WATCHER, true);
if (!mOutgoingQ.IsEmpty()) {
AddWatchers(WRITE_WATCHER, false);
}
}
void
UnixSocketImpl::OnConnected()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_CONNECTED);
if (!SetSocketFlags(GetFd())) {
NS_WARNING("Cannot set socket flags!");
FireSocketError();
return;
}
if (!mConnector->SetUp(GetFd())) {
NS_WARNING("Could not set up socket!");
FireSocketError();
return;
}
nsRefPtr<OnSocketEventRunnable> r =
new OnSocketEventRunnable(this, OnSocketEventRunnable::CONNECT_SUCCESS);
NS_DispatchToMainThread(r);
AddWatchers(READ_WATCHER, true);
if (!mOutgoingQ.IsEmpty()) {
AddWatchers(WRITE_WATCHER, false);
}
}
void
UnixSocketImpl::OnListening()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_LISTENING);
if (!mConnector->SetUpListenSocket(GetFd())) {
NS_WARNING("Could not set up listen socket!");
FireSocketError();
return;
}
AddWatchers(READ_WATCHER, true);
}
void
UnixSocketImpl::OnError(const char* aFunction, int aErrno)
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
UnixFdWatcher::OnError(aFunction, aErrno);
FireSocketError();
}
void
UnixSocketImpl::OnSocketCanReceiveWithoutBlocking()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_CONNECTED); // see bug 990984
// Read all of the incoming data.
while (true) {
nsAutoPtr<UnixSocketRawData> incoming(new UnixSocketRawData(MAX_READ_SIZE));
ssize_t ret = read(GetFd(), incoming->mData, incoming->mSize);
if (ret <= 0) {
if (ret == -1) {
if (errno == EINTR) {
continue; // retry system call when interrupted
}
if (errno == EAGAIN || errno == EWOULDBLOCK) {
return; // no data available: return and re-poll
}
#ifdef DEBUG
NS_WARNING("Cannot read from network");
#endif
// else fall through to error handling on other errno's
}
// We're done with our descriptors. Ensure that spurious events don't
// cause us to end up back here.
RemoveWatchers(READ_WATCHER|WRITE_WATCHER);
nsRefPtr<RequestClosingSocketRunnable> r =
new RequestClosingSocketRunnable(this);
NS_DispatchToMainThread(r);
return;
}
#ifdef MOZ_TASK_TRACER
// Make unix socket creation events to be the source events of TaskTracer,
// and originate the rest correlation tasks from here.
AutoSourceEvent taskTracerEvent(SourceEventType::UNIXSOCKET);
#endif
incoming->mSize = ret;
nsRefPtr<SocketReceiveRunnable> r =
new SocketReceiveRunnable(this, incoming.forget());
NS_DispatchToMainThread(r);
// If ret is less than MAX_READ_SIZE, there's no
// more data in the socket for us to read now.
if (ret < ssize_t(MAX_READ_SIZE)) {
return;
}
}
}
void
UnixSocketImpl::OnSocketCanSendWithoutBlocking()
{
MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_CONNECTED); // see bug 990984
// Try to write the bytes of mCurrentRilRawData. If all were written, continue.
//
// Otherwise, save the byte position of the next byte to write
// within mCurrentWriteOffset, and request another write when the
// system won't block.
//
while (true) {
UnixSocketRawData* data;
if (mOutgoingQ.IsEmpty()) {
return;
}
data = mOutgoingQ.ElementAt(0);
const uint8_t *toWrite;
toWrite = data->mData;
while (data->mCurrentWriteOffset < data->mSize) {
ssize_t write_amount = data->mSize - data->mCurrentWriteOffset;
ssize_t written;
written = write (GetFd(), toWrite + data->mCurrentWriteOffset,
write_amount);
if (written > 0) {
data->mCurrentWriteOffset += written;
}
if (written != write_amount) {
break;
}
}
if (data->mCurrentWriteOffset != data->mSize) {
AddWatchers(WRITE_WATCHER, false);
return;
}
mOutgoingQ.RemoveElementAt(0);
delete data;
}
}
UnixSocketConsumer::UnixSocketConsumer() : mImpl(nullptr)
, mConnectionStatus(SOCKET_DISCONNECTED)
, mConnectTimestamp(0)
, mConnectDelayMs(0)
{
}
UnixSocketConsumer::~UnixSocketConsumer()
{
MOZ_ASSERT(mConnectionStatus == SOCKET_DISCONNECTED);
MOZ_ASSERT(!mImpl);
}
bool
UnixSocketConsumer::SendSocketData(UnixSocketRawData* aData)
{
MOZ_ASSERT(NS_IsMainThread());
if (!mImpl) {
return false;
}
MOZ_ASSERT(!mImpl->IsShutdownOnMainThread());
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new SocketSendTask(mImpl, this, aData));
return true;
}
bool
UnixSocketConsumer::SendSocketData(const nsACString& aStr)
{
MOZ_ASSERT(NS_IsMainThread());
if (!mImpl) {
return false;
}
if (aStr.Length() > MAX_READ_SIZE) {
return false;
}
MOZ_ASSERT(!mImpl->IsShutdownOnMainThread());
UnixSocketRawData* d = new UnixSocketRawData(aStr.BeginReading(),
aStr.Length());
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new SocketSendTask(mImpl, this, d));
return true;
}
void
UnixSocketConsumer::CloseSocket()
{
MOZ_ASSERT(NS_IsMainThread());
if (!mImpl) {
return;
}
mImpl->CancelDelayedConnectTask();
// From this point on, we consider mImpl as being deleted.
// We sever the relationship here so any future calls to listen or connect
// will create a new implementation.
mImpl->ShutdownOnMainThread();
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new ShutdownSocketTask(mImpl));
mImpl = nullptr;
NotifyDisconnect();
}
void
UnixSocketConsumer::GetSocketAddr(nsAString& aAddrStr)
{
aAddrStr.Truncate();
if (!mImpl || mConnectionStatus != SOCKET_CONNECTED) {
NS_WARNING("No socket currently open!");
return;
}
mImpl->GetSocketAddr(aAddrStr);
}
void
UnixSocketConsumer::NotifySuccess()
{
MOZ_ASSERT(NS_IsMainThread());
mConnectionStatus = SOCKET_CONNECTED;
mConnectTimestamp = PR_IntervalNow();
OnConnectSuccess();
}
void
UnixSocketConsumer::NotifyError()
{
MOZ_ASSERT(NS_IsMainThread());
mConnectionStatus = SOCKET_DISCONNECTED;
mConnectDelayMs = CalculateConnectDelayMs();
OnConnectError();
}
void
UnixSocketConsumer::NotifyDisconnect()
{
MOZ_ASSERT(NS_IsMainThread());
mConnectionStatus = SOCKET_DISCONNECTED;
mConnectDelayMs = CalculateConnectDelayMs();
OnDisconnect();
}
bool
UnixSocketConsumer::ConnectSocket(UnixSocketConnector* aConnector,
const char* aAddress,
int aDelayMs)
{
MOZ_ASSERT(aConnector);
MOZ_ASSERT(NS_IsMainThread());
nsAutoPtr<UnixSocketConnector> connector(aConnector);
if (mImpl) {
NS_WARNING("Socket already connecting/connected!");
return false;
}
nsCString addr(aAddress);
MessageLoop* ioLoop = XRE_GetIOMessageLoop();
mImpl = new UnixSocketImpl(ioLoop, this, connector.forget(), addr);
mConnectionStatus = SOCKET_CONNECTING;
if (aDelayMs > 0) {
SocketDelayedConnectTask* connectTask = new SocketDelayedConnectTask(mImpl);
mImpl->SetDelayedConnectTask(connectTask);
MessageLoop::current()->PostDelayedTask(FROM_HERE, connectTask, aDelayMs);
} else {
ioLoop->PostTask(FROM_HERE, new SocketConnectTask(mImpl));
}
return true;
}
bool
UnixSocketConsumer::ListenSocket(UnixSocketConnector* aConnector)
{
MOZ_ASSERT(aConnector);
MOZ_ASSERT(NS_IsMainThread());
nsAutoPtr<UnixSocketConnector> connector(aConnector);
if (mImpl) {
NS_WARNING("Socket already connecting/connected!");
return false;
}
mImpl = new UnixSocketImpl(XRE_GetIOMessageLoop(), this, connector.forget(),
EmptyCString());
mConnectionStatus = SOCKET_LISTENING;
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new SocketListenTask(mImpl));
return true;
}
uint32_t
UnixSocketConsumer::CalculateConnectDelayMs() const
{
MOZ_ASSERT(NS_IsMainThread());
uint32_t connectDelayMs = mConnectDelayMs;
if ((PR_IntervalNow()-mConnectTimestamp) > connectDelayMs) {
// reset delay if connection has been opened for a while, or...
connectDelayMs = 0;
} else if (!connectDelayMs) {
// ...start with a delay of ~1 sec, or...
connectDelayMs = 1<<10;
} else if (connectDelayMs < (1<<16)) {
// ...otherwise increase delay by a factor of 2
connectDelayMs <<= 1;
}
return connectDelayMs;
}
} // namespace ipc
} // namespace mozilla