The Tor Browser / file revision

 /* -*- 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