Binder学习之startThreadPool,joinThreadPool

我们先不去分析ServiceManager的服务端，也先不忙分析Drvier部分，我们先把ServiceManager客户端的工作分析完。

接上一篇：http://blog.csdn.net/cs_lht/article/details/8171373

我们先来看看startThreadPool

ProcessState::self()->startThreadPool();

只要前面的理解了，这个还是比较的简单，我们一步一步的走：

void ProcessState::startThreadPool()
{
AutoMutex _l(mLock);
if (!mThreadPoolStarted) {
mThreadPoolStarted = true;
spawnPooledThread(true);
}
}
void ProcessState::spawnPooledThread(bool isMain)
{
if (mThreadPoolStarted) {
int32_t s = android_atomic_add(1, &mThreadPoolSeq);
char buf[32];
sprintf(buf, "Binder Thread #%d", s);
LOGV("Spawning new pooled thread, name=%s\n", buf);
sp<Thread> t = new PoolThread(isMain);
t->run(buf);
}
}

以上就是在ProcessState中创建一个线程子类对象并加入线程池，但是并没有create线程，然后调用run方法。

frameworks/base/libs/utils/Threads.cpp

status_t Thread::run(const char* name, int32_t priority, size_t stack)
{
Mutex::Autolock _l(mLock);

if (mRunning) {
// thread already started
return INVALID_OPERATION;
}

// reset status and exitPending to their default value, so we can
// try again after an error happened (either below, or in readyToRun())
mStatus = NO_ERROR;
mExitPending = false;
mThread = thread_id_t(-1);

// hold a strong reference on ourself
mHoldSelf = this;

mRunning = true;

bool res;
if (mCanCallJava) {//这儿才真正的创建线程
res = createThreadEtc(_threadLoop,//这个是线程函数
this, name, priority, stack, &mThread);
} else {
res = androidCreateRawThreadEtc(_threadLoop,
this, name, priority, stack, &mThread);
}

if (res == false) {
mStatus = UNKNOWN_ERROR;   // something happened!
mRunning = false;
mThread = thread_id_t(-1);
mHoldSelf.clear();  // "this" may have gone away after this.

return UNKNOWN_ERROR;
}

// Do not refer to mStatus here: The thread is already running (may, in fact
// already have exited with a valid mStatus result). The NO_ERROR indication
// here merely indicates successfully starting the thread and does not
// imply successful termination/execution.
return NO_ERROR;
}

而_threadLoop调用到了它自已的threadLoop，而这个函数在ThreadLoop这个子类中是重载了的。

virtual bool threadLoop()
{
IPCThreadState::self()->joinThreadPool(mIsMain);
return false;
}

void IPCThreadState::joinThreadPool(bool isMain)
{
LOG_THREADPOOL("**** THREAD %p (PID %d) IS JOINING THE THREAD POOL\n", (void*)pthread_self(), getpid());

mOut.writeInt32(isMain ? BC_ENTER_LOOPER : BC_REGISTER_LOOPER);//这儿就是告诉Binder，我们要注册一个Looper

// This thread may have been spawned by a thread that was in the background
// scheduling group, so first we will make sure it is in the default/foreground
// one to avoid performing an initial transaction in the background.
androidSetThreadSchedulingGroup(mMyThreadId, ANDROID_TGROUP_DEFAULT);

status_t result;
do {
int32_t cmd;

// When we've cleared the incoming command queue, process any pending derefs
if (mIn.dataPosition() >= mIn.dataSize()) {
size_t numPending = mPendingWeakDerefs.size();
if (numPending > 0) {
for (size_t i = 0; i < numPending; i++) {
RefBase::weakref_type* refs = mPendingWeakDerefs[i];
refs->decWeak(mProcess.get());
}
mPendingWeakDerefs.clear();
}
//处理已经死亡的BBinder对象
numPending = mPendingStrongDerefs.size();
if (numPending > 0) {
for (size_t i = 0; i < numPending; i++) {
BBinder* obj = mPendingStrongDerefs[i];
obj->decStrong(mProcess.get());
}
mPendingStrongDerefs.clear();
}
}

// now get the next command to be processed, waiting if necessary
result = talkWithDriver();
if (result >= NO_ERROR) {
size_t IN = mIn.dataAvail();
if (IN < sizeof(int32_t)) continue;
cmd = mIn.readInt32();
IF_LOG_COMMANDS() {
alog << "Processing top-level Command: "
<< getReturnString(cmd) << endl;
}

result = executeCommand(cmd);
}

// After executing the command, ensure that the thread is returned to the
// default cgroup before rejoining the pool.  The driver takes care of
// restoring the priority, but doesn't do anything with cgroups so we
// need to take care of that here in userspace.  Note that we do make
// sure to go in the foreground after executing a transaction, but
// there are other callbacks into user code that could have changed
// our group so we want to make absolutely sure it is put back.
androidSetThreadSchedulingGroup(mMyThreadId, ANDROID_TGROUP_DEFAULT);

// Let this thread exit the thread pool if it is no longer
// needed and it is not the main process thread.
if(result == TIMED_OUT && !isMain) {
break;
}
} while (result != -ECONNREFUSED && result != -EBADF);

LOG_THREADPOOL("**** THREAD %p (PID %d) IS LEAVING THE THREAD POOL err=%p\n",
(void*)pthread_self(), getpid(), (void*)result);

mOut.writeInt32(BC_EXIT_LOOPER);
talkWithDriver(false);
}
status_t IPCThreadState::executeCommand(int32_t cmd)
{
BBinder* obj;
RefBase::weakref_type* refs;
status_t result = NO_ERROR;

switch (cmd) {
case BR_ERROR:
result = mIn.readInt32();
break;

case BR_OK:
break;

case BR_ACQUIRE:
refs = (RefBase::weakref_type*)mIn.readInt32();
obj = (BBinder*)mIn.readInt32();
LOG_ASSERT(refs->refBase() == obj,
"BR_ACQUIRE: object %p does not match cookie %p (expected %p)",
refs, obj, refs->refBase());
obj->incStrong(mProcess.get());
IF_LOG_REMOTEREFS() {
LOG_REMOTEREFS("BR_ACQUIRE from driver on %p", obj);
obj->printRefs();
}
mOut.writeInt32(BC_ACQUIRE_DONE);
mOut.writeInt32((int32_t)refs);
mOut.writeInt32((int32_t)obj);
break;

case BR_RELEASE:
refs = (RefBase::weakref_type*)mIn.readInt32();
obj = (BBinder*)mIn.readInt32();
LOG_ASSERT(refs->refBase() == obj,
"BR_RELEASE: object %p does not match cookie %p (expected %p)",
refs, obj, refs->refBase());
IF_LOG_REMOTEREFS() {
LOG_REMOTEREFS("BR_RELEASE from driver on %p", obj);
obj->printRefs();
}
mPendingStrongDerefs.push(obj);
break;

case BR_INCREFS:
refs = (RefBase::weakref_type*)mIn.readInt32();
obj = (BBinder*)mIn.readInt32();
refs->incWeak(mProcess.get());
mOut.writeInt32(BC_INCREFS_DONE);
mOut.writeInt32((int32_t)refs);
mOut.writeInt32((int32_t)obj);
break;

case BR_DECREFS:
refs = (RefBase::weakref_type*)mIn.readInt32();
obj = (BBinder*)mIn.readInt32();
// NOTE: This assertion is not valid, because the object may no
// longer exist (thus the (BBinder*)cast above resulting in a different
// memory address).
//LOG_ASSERT(refs->refBase() == obj,
//           "BR_DECREFS: object %p does not match cookie %p (expected %p)",
//           refs, obj, refs->refBase());
mPendingWeakDerefs.push(refs);
break;

case BR_ATTEMPT_ACQUIRE:
refs = (RefBase::weakref_type*)mIn.readInt32();
obj = (BBinder*)mIn.readInt32();

{
const bool success = refs->attemptIncStrong(mProcess.get());
LOG_ASSERT(success && refs->refBase() == obj,
"BR_ATTEMPT_ACQUIRE: object %p does not match cookie %p (expected %p)",
refs, obj, refs->refBase());

mOut.writeInt32(BC_ACQUIRE_RESULT);
mOut.writeInt32((int32_t)success);
}
break;

case BR_TRANSACTION:
{
binder_transaction_data tr;
result = mIn.read(&tr, sizeof(tr));
LOG_ASSERT(result == NO_ERROR,
"Not enough command data for brTRANSACTION");
if (result != NO_ERROR) break;

Parcel buffer;
buffer.ipcSetDataReference(
reinterpret_cast<const uint8_t*>(tr.data.ptr.buffer),
tr.data_size,
reinterpret_cast<const size_t*>(tr.data.ptr.offsets),
tr.offsets_size/sizeof(size_t), freeBuffer, this);

const pid_t origPid = mCallingPid;
const uid_t origUid = mCallingUid;

mCallingPid = tr.sender_pid;
mCallingUid = tr.sender_euid;

int curPrio = getpriority(PRIO_PROCESS, mMyThreadId);
if (gDisableBackgroundScheduling) {
if (curPrio > ANDROID_PRIORITY_NORMAL) {
// We have inherited a reduced priority from the caller, but do not
// want to run in that state in this process.  The driver set our
// priority already (though not our scheduling class), so bounce
// it back to the default before invoking the transaction.
setpriority(PRIO_PROCESS, mMyThreadId, ANDROID_PRIORITY_NORMAL);
}
} else {
if (curPrio >= ANDROID_PRIORITY_BACKGROUND) {
// We want to use the inherited priority from the caller.
// Ensure this thread is in the background scheduling class,
// since the driver won't modify scheduling classes for us.
// The scheduling group is reset to default by the caller
// once this method returns after the transaction is complete.
androidSetThreadSchedulingGroup(mMyThreadId,
ANDROID_TGROUP_BG_NONINTERACT);
}
}

//LOGI(">>>> TRANSACT from pid %d uid %d\n", mCallingPid, mCallingUid);

Parcel reply;
IF_LOG_TRANSACTIONS() {
TextOutput::Bundle _b(alog);
alog << "BR_TRANSACTION thr " << (void*)pthread_self()
<< " / obj " << tr.target.ptr << " / code "
<< TypeCode(tr.code) << ": " << indent << buffer
<< dedent << endl
<< "Data addr = "
<< reinterpret_cast<const uint8_t*>(tr.data.ptr.buffer)
<< ", offsets addr="
<< reinterpret_cast<const size_t*>(tr.data.ptr.offsets) << endl;
}
if (tr.target.ptr) {
sp<BBinder> b((BBinder*)tr.cookie);
const status_t error = b->transact(tr.code, buffer, &reply, 0); //我们说BnXXXService继承自BBinder,所以实际上就是调用了BnXXXService的onTransact函数
if (error < NO_ERROR) reply.setError(error);

} else {
const status_t error = the_context_object->transact(tr.code, buffer, &reply, 0);
if (error < NO_ERROR) reply.setError(error);
}

//LOGI("<<<< TRANSACT from pid %d restore pid %d uid %d\n",
//     mCallingPid, origPid, origUid);

if ((tr.flags & TF_ONE_WAY) == 0) {
LOG_ONEWAY("Sending reply to %d!", mCallingPid);
sendReply(reply, 0);
} else {
LOG_ONEWAY("NOT sending reply to %d!", mCallingPid);
}

mCallingPid = origPid;
mCallingUid = origUid;

IF_LOG_TRANSACTIONS() {
TextOutput::Bundle _b(alog);
alog << "BC_REPLY thr " << (void*)pthread_self() << " / obj "
<< tr.target.ptr << ": " << indent << reply << dedent << endl;
}

}
break;

case BR_DEAD_BINDER:
{
BpBinder *proxy = (BpBinder*)mIn.readInt32();
proxy->sendObituary();
mOut.writeInt32(BC_DEAD_BINDER_DONE);
mOut.writeInt32((int32_t)proxy);
} break;

case BR_CLEAR_DEATH_NOTIFICATION_DONE:
{
BpBinder *proxy = (BpBinder*)mIn.readInt32();
proxy->getWeakRefs()->decWeak(proxy);
} break;

case BR_FINISHED:
result = TIMED_OUT;
break;

case BR_NOOP:
break;

case BR_SPAWN_LOOPER:
mProcess->spawnPooledThread(false);
break;

default:
printf("*** BAD COMMAND %d received from Binder driver\n", cmd);
result = UNKNOWN_ERROR;
break;
}

if (result != NO_ERROR) {
mLastError = result;
}

return result;
}

以上部分是创建一个主线程与Binder通信。
IPCThreadState::self()->joinThreadPool()是把进程的主线程着为一个Binder通信的主线程，其实这儿有两个线程干着同样的工作，有人问可不可以去掉一个，答案是肯定的，少一个也没有什么影响，至于为什么这么设计就不知道了。