pjs/ipc/glue/AsyncChannel.cpp

439 строки
11 KiB
C++

/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
* vim: sw=4 ts=4 et :
*/
/* ***** BEGIN LICENSE BLOCK *****
* Version: MPL 1.1/GPL 2.0/LGPL 2.1
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is Mozilla Plugin App.
*
* The Initial Developer of the Original Code is
* Chris Jones <jones.chris.g@gmail.com>
* Portions created by the Initial Developer are Copyright (C) 2009
* the Initial Developer. All Rights Reserved.
*
* Contributor(s):
*
* Alternatively, the contents of this file may be used under the terms of
* either the GNU General Public License Version 2 or later (the "GPL"), or
* the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
* in which case the provisions of the GPL or the LGPL are applicable instead
* of those above. If you wish to allow use of your version of this file only
* under the terms of either the GPL or the LGPL, and not to allow others to
* use your version of this file under the terms of the MPL, indicate your
* decision by deleting the provisions above and replace them with the notice
* and other provisions required by the GPL or the LGPL. If you do not delete
* the provisions above, a recipient may use your version of this file under
* the terms of any one of the MPL, the GPL or the LGPL.
*
* ***** END LICENSE BLOCK ***** */
#include "mozilla/ipc/AsyncChannel.h"
#include "mozilla/ipc/GeckoThread.h"
#include "mozilla/ipc/ProtocolUtils.h"
#include "nsDebug.h"
#include "nsTraceRefcnt.h"
#include "nsXULAppAPI.h"
using mozilla::MutexAutoLock;
template<>
struct RunnableMethodTraits<mozilla::ipc::AsyncChannel>
{
static void RetainCallee(mozilla::ipc::AsyncChannel* obj) { }
static void ReleaseCallee(mozilla::ipc::AsyncChannel* obj) { }
};
namespace mozilla {
namespace ipc {
AsyncChannel::AsyncChannel(AsyncListener* aListener)
: mTransport(0),
mListener(aListener),
mChannelState(ChannelClosed),
mMutex("mozilla.ipc.AsyncChannel.mMutex"),
mCvar(mMutex, "mozilla.ipc.AsyncChannel.mCvar"),
mIOLoop(),
mWorkerLoop(),
mChannelErrorTask(NULL)
{
MOZ_COUNT_CTOR(AsyncChannel);
}
AsyncChannel::~AsyncChannel()
{
MOZ_COUNT_DTOR(AsyncChannel);
Clear();
}
bool
AsyncChannel::Open(Transport* aTransport, MessageLoop* aIOLoop)
{
NS_PRECONDITION(!mTransport, "Open() called > once");
NS_PRECONDITION(aTransport, "need transport layer");
// FIXME need to check for valid channel
mTransport = aTransport;
mTransport->set_listener(this);
// FIXME figure out whether we're in parent or child, grab IO loop
// appropriately
bool needOpen = true;
if(!aIOLoop) {
// parent
needOpen = false;
aIOLoop = BrowserProcessSubThread
::GetMessageLoop(BrowserProcessSubThread::IO);
// FIXME assuming that the parent waits for the OnConnected event.
// FIXME see GeckoChildProcessHost.cpp. bad assumption!
mChannelState = ChannelConnected;
}
mChild = needOpen;
mIOLoop = aIOLoop;
mWorkerLoop = MessageLoop::current();
NS_ASSERTION(mIOLoop, "need an IO loop");
NS_ASSERTION(mWorkerLoop, "need a worker loop");
if (needOpen) { // child process
MutexAutoLock lock(mMutex);
mIOLoop->PostTask(FROM_HERE,
NewRunnableMethod(this,
&AsyncChannel::OnChannelOpened));
// FIXME/cjones: handle errors
while (mChannelState != ChannelConnected) {
mCvar.Wait();
}
}
return true;
}
void
AsyncChannel::Close()
{
{
MutexAutoLock lock(mMutex);
if (ChannelError == mChannelState)
return;
if (ChannelConnected != mChannelState)
// XXX be strict about this until there's a compelling reason
// to relax
NS_RUNTIMEABORT("Close() called on closed channel!");
AssertWorkerThread();
// notify the other side that we're about to close our socket
SendGoodbye();
mChannelState = ChannelClosing;
// and post the task will do the actual close
mIOLoop->PostTask(
FROM_HERE, NewRunnableMethod(this, &AsyncChannel::OnCloseChannel));
while (ChannelClosing == mChannelState)
mCvar.Wait();
// TODO sort out Close() on this side racing with Close() on the
// other side
mChannelState = ChannelClosed;
}
return NotifyChannelClosed();
}
bool
AsyncChannel::Send(Message* msg)
{
AssertWorkerThread();
mMutex.AssertNotCurrentThreadOwns();
NS_ABORT_IF_FALSE(MSG_ROUTING_NONE != msg->routing_id(), "need a route");
{
MutexAutoLock lock(mMutex);
if (!Connected()) {
ReportConnectionError("AsyncChannel");
return false;
}
mIOLoop->PostTask(FROM_HERE,
NewRunnableMethod(this, &AsyncChannel::OnSend, msg));
}
return true;
}
void
AsyncChannel::OnDispatchMessage(const Message& msg)
{
AssertWorkerThread();
NS_ASSERTION(!msg.is_reply(), "can't process replies here");
NS_ASSERTION(!(msg.is_sync() || msg.is_rpc()), "async dispatch only");
if (MaybeInterceptGoodbye(msg))
// there's a NotifyMaybeChannelError event waiting for us, or
// will be soon
return;
// it's OK to dispatch messages if the channel is closed/error'd,
// since we don't have a reply to send back
(void)MaybeHandleError(mListener->OnMessageReceived(msg), "AsyncChannel");
}
// This is an async message
class GoodbyeMessage : public IPC::Message
{
public:
enum { ID = GOODBYE_MESSAGE_TYPE };
GoodbyeMessage() :
IPC::Message(MSG_ROUTING_NONE, ID, PRIORITY_NORMAL)
{
}
// XXX not much point in implementing this; maybe could help with
// debugging?
static bool Read(const Message* msg)
{
return true;
}
void Log(const std::string& aPrefix,
FILE* aOutf) const
{
fputs("(special `Goodbye' message)", aOutf);
}
};
void
AsyncChannel::SendGoodbye()
{
AssertWorkerThread();
mIOLoop->PostTask(
FROM_HERE,
NewRunnableMethod(this, &AsyncChannel::OnSend, new GoodbyeMessage()));
}
bool
AsyncChannel::MaybeInterceptGoodbye(const Message& msg)
{
// IPDL code isn't allowed to send MSG_ROUTING_NONE messages, so
// there's no chance of confusion here
if (MSG_ROUTING_NONE != msg.routing_id())
return false;
if (msg.is_sync() || msg.is_rpc() || GOODBYE_MESSAGE_TYPE != msg.type())
NS_RUNTIMEABORT("received unknown MSG_ROUTING_NONE message when expecting `Goodbye'");
MutexAutoLock lock(mMutex);
// TODO sort out Close() on this side racing with Close() on the
// other side
mChannelState = ChannelClosing;
printf("NOTE: %s process received `Goodbye', closing down\n",
mChild ? "child" : "parent");
return true;
}
void
AsyncChannel::NotifyChannelClosed()
{
if (ChannelClosed != mChannelState)
NS_RUNTIMEABORT("channel should have been closed!");
// OK, the IO thread just closed the channel normally. Let the
// listener know about it.
mListener->OnChannelClose();
Clear();
}
void
AsyncChannel::NotifyMaybeChannelError()
{
// TODO sort out Close() on this side racing with Close() on the
// other side
if (ChannelClosing == mChannelState) {
// the channel closed, but we received a "Goodbye" message
// warning us about it. no worries
mChannelState = ChannelClosed;
return NotifyChannelClosed();
}
// Oops, error! Let the listener know about it.
mChannelState = ChannelError;
mListener->OnChannelError();
Clear();
}
void
AsyncChannel::Clear()
{
mListener = 0;
mIOLoop = 0;
mWorkerLoop = 0;
if (mTransport) {
mTransport->set_listener(0);
// we only hold a weak ref to the transport, which is "owned"
// by GeckoChildProcess/GeckoThread
mTransport = 0;
}
if (mChannelErrorTask) {
mChannelErrorTask->Cancel();
mChannelErrorTask = NULL;
}
}
bool
AsyncChannel::MaybeHandleError(Result code, const char* channelName)
{
if (MsgProcessed == code)
return true;
const char* errorMsg;
switch (code) {
case MsgNotKnown:
errorMsg = "Unknown message: not processed";
break;
case MsgNotAllowed:
errorMsg = "Message not allowed: cannot be sent/recvd in this state";
break;
case MsgPayloadError:
errorMsg = "Payload error: message could not be deserialized";
break;
case MsgRouteError:
errorMsg = "Route error: message sent to unknown actor ID";
break;
case MsgValueError:
errorMsg = "Value error: message was deserialized, but contained an illegal value";
break;
default:
NS_RUNTIMEABORT("unknown Result code");
return false;
}
PrintErrorMessage(channelName, errorMsg);
return false;
}
void
AsyncChannel::ReportConnectionError(const char* channelName)
{
const char* errorMsg;
switch (mChannelState) {
case ChannelClosed:
errorMsg = "Closed channel: cannot send/recv";
break;
case ChannelOpening:
errorMsg = "Opening channel: not yet ready for send/recv";
break;
case ChannelError:
errorMsg = "Channel error: cannot send/recv";
break;
default:
NOTREACHED();
}
PrintErrorMessage(channelName, errorMsg);
}
//
// The methods below run in the context of the IO thread
//
void
AsyncChannel::OnMessageReceived(const Message& msg)
{
AssertIOThread();
NS_ASSERTION(mChannelState != ChannelError, "Shouldn't get here!");
// wake up the worker, there's work to do
mWorkerLoop->PostTask(
FROM_HERE,
NewRunnableMethod(this, &AsyncChannel::OnDispatchMessage, msg));
}
void
AsyncChannel::OnChannelOpened()
{
AssertIOThread();
mChannelState = ChannelOpening;
/*assert*/mTransport->Connect();
}
void
AsyncChannel::OnChannelConnected(int32 peer_pid)
{
AssertIOThread();
MutexAutoLock lock(mMutex);
mChannelState = ChannelConnected;
mCvar.Notify();
}
void
AsyncChannel::OnChannelError()
{
AssertIOThread();
MutexAutoLock lock(mMutex);
// NB: this can race with the `Goodbye' event being processed by
// the worker thread
if (ChannelClosing != mChannelState)
mChannelState = ChannelError;
NS_ASSERTION(!mChannelErrorTask, "OnChannelError called twice?");
mChannelErrorTask =
NewRunnableMethod(this, &AsyncChannel::NotifyMaybeChannelError);
mWorkerLoop->PostTask(FROM_HERE, mChannelErrorTask);
}
void
AsyncChannel::OnSend(Message* aMsg)
{
AssertIOThread();
mTransport->Send(aMsg);
// mTransport assumes ownership of aMsg
}
void
AsyncChannel::OnCloseChannel()
{
AssertIOThread();
mTransport->Close();
MutexAutoLock lock(mMutex);
mChannelState = ChannelClosed;
mCvar.Notify();
}
} // namespace ipc
} // namespace mozilla