corysanin/tubestation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backing out e8e7e0cf43d8 (Bug 790739) due to bustage

Browse Source
This commit is contained in:
Kyle Machulis
2012-09-30 23:17:46 -07:00
parent 1be51f9284
commit cd306b0e65
2 changed files with 55 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files

303
ipc/unixsocket/UnixSocket.cpp
View File

@@ -24,31 +24,16 @@
#include "nsTArray.h"
#include "nsXULAppAPI.h"
#undef LOG
#if defined(MOZ_WIDGET_GONK)
#include <android/log.h>
#define LOG(args...) __android_log_print(ANDROID_LOG_INFO, "GonkDBus", args);
#else
#define BTDEBUG true
#define LOG(args...) if (BTDEBUG) printf(args);
#endif
static const int SOCKET_RETRY_TIME_MS = 1000;
namespace mozilla {
namespace ipc {
class UnixSocketImpl : public MessageLoopForIO::Watcher
{
public:
UnixSocketImpl(UnixSocketConsumer* aConsumer, UnixSocketConnector* aConnector,
const nsACString& aAddress)
UnixSocketImpl(UnixSocketConsumer* aConsumer, int aFd)
: mConsumer(aConsumer)
, mIOLoop(nullptr)
, mFd(-1)
, mConnector(aConnector)
, mCurrentTaskIsCanceled(false)
, mAddress(aAddress)
, mFd(aFd)
{
}
@@ -69,42 +54,6 @@ public:
return mFd > 0;
}
void CancelTask()
{
if (!mTask) {
return;
}
mTask->Cancel();
mTask = nullptr;
mCurrentTaskIsCanceled = true;
}
void UnsetTask()
{
mTask = nullptr;
}
void EnqueueTask(int aDelayMs, CancelableTask* aTask)
{
MessageLoopForIO* ioLoop = MessageLoopForIO::current();
if (!ioLoop) {
NS_WARNING("No IOLoop to attach to, cancelling self!");
return;
}
if (mTask) {
return;
}
if (mCurrentTaskIsCanceled) {
return;
}
mTask = aTask;
if (aDelayMs) {
ioLoop->PostDelayedTask(FROM_HERE, mTask, aDelayMs);
} else {
ioLoop->PostTask(FROM_HERE, mTask);
}
}
void SetUpIO()
{
MOZ_ASSERT(!mIOLoop);
@@ -121,33 +70,6 @@ public:
mConsumer.forget();
}
/**
* Connect to a socket
*/
void Connect();
/**
* Run bind/listen to prepare for further runs of accept()
*/
void Listen();
/**
* Accept an incoming connection
*/
void Accept();
/**
* Stop whatever connect/accept task is running
*/
void Stop();
/**
* Set up nonblocking flags on whatever our current file descriptor is.
*
* @return true if successful, false otherwise
*/
bool SetNonblockFlags();
/**
* Consumer pointer. Non-thread safe RefPtr, so should only be manipulated
* directly from main thread. All non-main-thread accesses should happen with
@@ -183,6 +105,12 @@ private:
typedef nsTArray<UnixSocketRawData* > UnixSocketRawDataQueue;
UnixSocketRawDataQueue mOutgoingQ;
/**
* File descriptor to read from/write to. Connection happens on user provided
* thread. Read/write/close happens on IO thread.
*/
ScopedClose mFd;
/**
* Incoming packet. Only to be accessed on IO Thread.
*/
@@ -197,33 +125,6 @@ private:
* Write watcher for libevent. Only to be accessed on IO Thread.
*/
MessageLoopForIO::FileDescriptorWatcher mWriteWatcher;
/**
* File descriptor to read from/write to. Connection happens on user provided
* thread. Read/write/close happens on IO thread.
*/
ScopedClose mFd;
/**
* 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 mCurrentTaskIsCanceled;
/**
* Pointer to the task we're currently running. DO NOT DELETE MANUALLY. This
* will be taken care of by the IO loop. Just set to nullptr.
*/
CancelableTask* mTask;
/**
* Address we are connecting to, assuming we are creating a client connection.
*/
nsCString mAddress;
};
static void
@@ -300,160 +201,27 @@ private:
UnixSocketImpl* mImpl;
};
class SocketAcceptTask : public CancelableTask {
virtual void Run();
bool mCanceled;
UnixSocketImpl* mImpl;
public:
virtual void Cancel() { mCanceled = true; }
SocketAcceptTask(UnixSocketImpl* aImpl) : mCanceled(false), mImpl(aImpl) { }
};
void SocketAcceptTask::Run() {
mImpl->UnsetTask();
if (mCanceled) {
return;
}
mImpl->Accept();
}
class SocketConnectTask : public CancelableTask {
virtual void Run();
bool mCanceled;
UnixSocketImpl* mImpl;
public:
SocketConnectTask(UnixSocketImpl* aImpl) : mCanceled(false), mImpl(aImpl) { }
virtual void Cancel() { mCanceled = true; }
};
void SocketConnectTask::Run() {
mImpl->UnsetTask();
if (mCanceled) {
return;
}
mImpl->Connect();
}
void
UnixSocketImpl::Accept()
{
socklen_t addr_sz;
struct sockaddr addr;
// 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.
mConnector->CreateAddr(true, addr_sz, &addr, nullptr);
if(mFd.get() < 0)
{
mFd = mConnector->Create();
if (mFd.get() < 0) {
return;
}
if (!SetNonblockFlags()) {
return;
}
if (bind(mFd.get(), &addr, addr_sz)) {
#ifdef DEBUG
LOG("...bind(%d) gave errno %d", mFd.get(), errno);
#endif
return;
}
if (listen(mFd.get(), 1)) {
#ifdef DEBUG
LOG("...listen(%d) gave errno %d", mFd.get(), errno);
#endif
return;
}
}
int client_fd;
client_fd = accept(mFd.get(), &addr, &addr_sz);
if (client_fd < 0) {
#if DEBUG
LOG("Socket accept errno=%d\n", errno);
#endif
EnqueueTask(SOCKET_RETRY_TIME_MS, new SocketAcceptTask(this));
return;
}
if(client_fd < 0)
{
EnqueueTask(SOCKET_RETRY_TIME_MS, new SocketAcceptTask(this));
return;
}
if (!mConnector->Setup(client_fd)) {
NS_WARNING("Could not set up socket!");
return;
}
mFd.reset(client_fd);
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new StartImplReadingTask(this));
}
void
UnixSocketImpl::Connect()
{
if(mFd.get() < 0)
{
mFd = mConnector->Create();
if (mFd.get() < 0) {
return;
}
}
int ret;
socklen_t addr_sz;
struct sockaddr addr;
mConnector->CreateAddr(false, addr_sz, &addr, mAddress.get());
ret = connect(mFd.get(), &addr, addr_sz);
if (ret) {
#if DEBUG
LOG("Socket connect errno=%d\n", errno);
#endif
mFd.reset(-1);
return;
}
if (!mConnector->Setup(mFd)) {
NS_WARNING("Could not set up socket!");
return;
}
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new StartImplReadingTask(this));
}
bool
UnixSocketImpl::SetNonblockFlags()
UnixSocketConnector::Connect(int aFd, const char* aAddress)
{
// Set socket addr to be reused even if kernel is still waiting to close
int n = 1;
setsockopt(mFd, SOL_SOCKET, SO_REUSEADDR, &n, sizeof(n));
if (!ConnectInternal(aFd, aAddress))
{
return false;
}
// Set close-on-exec bit.
int flags = fcntl(mFd, F_GETFD);
int flags = fcntl(aFd, F_GETFD);
if (-1 == flags) {
return false;
}
flags |= FD_CLOEXEC;
if (-1 == fcntl(mFd, F_SETFD, flags)) {
if (-1 == fcntl(aFd, F_SETFD, flags)) {
return false;
}
// Select non-blocking IO.
if (-1 == fcntl(mFd, F_SETFL, O_NONBLOCK)) {
if (-1 == fcntl(aFd, F_SETFL, O_NONBLOCK)) {
return false;
}
@@ -605,48 +373,23 @@ UnixSocketImpl::OnFileCanWriteWithoutBlocking(int aFd)
}
bool
UnixSocketConsumer::ConnectSocket(UnixSocketConnector* aConnector,
UnixSocketConsumer::ConnectSocket(UnixSocketConnector& aConnector,
const char* aAddress)
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(aConnector);
if (mImpl) {
NS_WARNING("Socket already connecting/connected!");
MOZ_ASSERT(!mImpl);
ScopedClose fd(aConnector.Create());
if (fd < 0) {
return false;
}
nsCString addr;
addr.Assign(aAddress);
mImpl = new UnixSocketImpl(this, aConnector, addr);
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new SocketConnectTask(mImpl));
return true;
}
bool
UnixSocketConsumer::ListenSocket(UnixSocketConnector* aConnector)
{
MOZ_ASSERT(!NS_IsMainThread());
MOZ_ASSERT(aConnector);
if (mImpl) {
NS_WARNING("Socket already connecting/connected!");
if (!aConnector.Connect(fd, aAddress)) {
return false;
}
nsCString addr;
mImpl = new UnixSocketImpl(this, aConnector, addr);
mImpl = new UnixSocketImpl(this, fd.forget());
XRE_GetIOMessageLoop()->PostTask(FROM_HERE,
new SocketAcceptTask(mImpl));
new StartImplReadingTask(mImpl));
return true;
}
void
UnixSocketConsumer::CancelSocketTask()
{
if(!mImpl) {
NS_WARNING("No socket implementation to cancel task on!");
return;
}
mImpl->CancelTask();
}
} // namespace ipc
} // namespace mozilla